Timeouts: add "last-in" for the last payload's length to exception messages (#2276)

In a multiplexed setup we often see a timeout behind a large payload in the process of parsing (or not). This is meant to help better diagnose that case, by recording what the last payload size was that came through this connection.

Needs some @mgravell eyes on the record point here and if we want to adjust (or if it's wrong due to buffers in a way I'm not understanding). I realize this won't work on a payload so large enough it blows all buffers completely, but I don't see how to solve that either the way we properly stream parse. I think it'll still improve most cases we see this in.

Author: NickCraver

docs/ReleaseNotes.md

@@ -8,7 +8,7 @@ Current package versions:
 
 ## Unreleased
 
-No pending changes for the next release yet.
+- Adds: `last-in` and `cur-in` (bytes) to timeout exceptions to help identify timeouts that were just-behind another large payload off the wire ([#2276 by NickCraver](https://github.com/StackExchange/StackExchange.Redis/pull/2276))
 
 ## 2.6.70
 

src/StackExchange.Redis/ExceptionFactory.cs

@@ -320,6 +320,8 @@ private static void AddCommonDetail(
                 if (bs.Connection.BytesAvailableOnSocket >= 0) Add(data, sb, "Inbound-Bytes", "in", bs.Connection.BytesAvailableOnSocket.ToString());
                 if (bs.Connection.BytesInReadPipe >= 0) Add(data, sb, "Inbound-Pipe-Bytes", "in-pipe", bs.Connection.BytesInReadPipe.ToString());
                 if (bs.Connection.BytesInWritePipe >= 0) Add(data, sb, "Outbound-Pipe-Bytes", "out-pipe", bs.Connection.BytesInWritePipe.ToString());
+                Add(data, sb, "Last-Result-Bytes", "last-in", bs.Connection.BytesLastResult.ToString());
+                Add(data, sb, "Inbound-Buffer-Bytes", "cur-in", bs.Connection.BytesInBuffer.ToString());
 
                 if (multiplexer.StormLogThreshold >= 0 && bs.Connection.MessagesSentAwaitingResponse >= multiplexer.StormLogThreshold && Interlocked.CompareExchange(ref multiplexer.haveStormLog, 1, 0) == 0)
                 {

src/StackExchange.Redis/PhysicalConnection.cs

@@ -52,6 +52,9 @@ private static readonly Message
 
         private int lastWriteTickCount, lastReadTickCount, lastBeatTickCount;
 
+        private long bytesLastResult;
+        private long bytesInBuffer;
+
         internal void GetBytes(out long sent, out long received)
         {
             if (_ioPipe is IMeasuredDuplexPipe sc)
@@ -1283,6 +1286,14 @@ internal readonly struct ConnectionStatus
             /// Bytes in the writer pipe, waiting to be written to the socket.
             /// </summary>
             public long BytesInWritePipe { get; init; }
+            /// <summary>
+            /// Byte size of the last result we processed.
+            /// </summary>
+            public long BytesLastResult { get; init; }
+            /// <summary>
+            /// Byte size on the buffer that isn't processed yet.
+            /// </summary>
+            public long BytesInBuffer { get; init; }
 
             /// <summary>
             /// The inbound pipe reader status.
@@ -1334,6 +1345,8 @@ public ConnectionStatus GetStatus()
                     BytesInWritePipe = counters.BytesWaitingToBeSent,
                     ReadStatus = _readStatus,
                     WriteStatus = _writeStatus,
+                    BytesLastResult = bytesLastResult,
+                    BytesInBuffer = bytesInBuffer,
                 };
             }
 
@@ -1356,6 +1369,8 @@ public ConnectionStatus GetStatus()
                 BytesInWritePipe = -1,
                 ReadStatus = _readStatus,
                 WriteStatus = _writeStatus,
+                BytesLastResult = bytesLastResult,
+                BytesInBuffer = bytesInBuffer,
             };
         }
 
@@ -1702,6 +1717,7 @@ private async Task ReadFromPipe()
         private int ProcessBuffer(ref ReadOnlySequence<byte> buffer)
         {
             int messageCount = 0;
+            bytesInBuffer = buffer.Length;
 
             while (!buffer.IsEmpty)
             {
@@ -1718,6 +1734,10 @@ private int ProcessBuffer(ref ReadOnlySequence<byte> buffer)
                         Trace(result.ToString());
                         _readStatus = ReadStatus.MatchResult;
                         MatchResult(result);
+
+                        // Track the last result size *after* processing for the *next* error message
+                        bytesInBuffer = buffer.Length;
+                        bytesLastResult = result.Payload.Length;
                     }
                     else
                     {

tests/StackExchange.Redis.Tests/AsyncTests.cs

@@ -48,7 +48,7 @@ public async Task AsyncTimeoutIsNoticed()
         using var conn = Create(syncTimeout: 1000);
         var opt = ConfigurationOptions.Parse(conn.Configuration);
         if (!Debugger.IsAttached)
-        { // we max the timeouts if a degugger is detected
+        { // we max the timeouts if a debugger is detected
             Assert.Equal(1000, opt.AsyncTimeout);
         }
 
@@ -65,14 +65,20 @@ public async Task AsyncTimeoutIsNoticed()
         var ex = await Assert.ThrowsAsync<RedisTimeoutException>(async () =>
         {
             await db.StringGetAsync(key).ForAwait(); // but *subsequent* operations are paused
-                ms.Stop();
+            ms.Stop();
             Writer.WriteLine($"Unexpectedly succeeded after {ms.ElapsedMilliseconds}ms");
         }).ForAwait();
         ms.Stop();
         Writer.WriteLine($"Timed out after {ms.ElapsedMilliseconds}ms");
 
+        Writer.WriteLine("Exception message: " + ex.Message);
         Assert.Contains("Timeout awaiting response", ex.Message);
-        Writer.WriteLine(ex.Message);
+        // Ensure we are including the last payload size
+        Assert.Contains("last-in:", ex.Message);
+        Assert.DoesNotContain("last-in: 0", ex.Message);
+        Assert.NotNull(ex.Data["Redis-Last-Result-Bytes"]);
+
+        Assert.Contains("cur-in:", ex.Message);
 
         string status = conn.GetStatus();
         Writer.WriteLine(status);

tests/StackExchange.Redis.Tests/ExceptionFactoryTests.cs

@@ -112,11 +112,11 @@ public void TimeoutException()
             var ex = Assert.IsType<RedisTimeoutException>(rawEx);
             Writer.WriteLine("Exception: " + ex.Message);
 
-            // Example format: "Test Timeout, command=PING, inst: 0, qu: 0, qs: 0, aw: False, in: 0, in-pipe: 0, out-pipe: 0, serverEndpoint: 127.0.0.1:6379, mgr: 10 of 10 available, clientName: TimeoutException, IOCP: (Busy=0,Free=1000,Min=8,Max=1000), WORKER: (Busy=2,Free=2045,Min=8,Max=2047), v: 2.1.0 (Please take a look at this article for some common client-side issues that can cause timeouts: https://stackexchange.github.io/StackExchange.Redis/Timeouts)";
+            // Example format: "Test Timeout, command=PING, inst: 0, qu: 0, qs: 0, aw: False, in: 0, in-pipe: 0, out-pipe: 0, last-in: 0, cur-in: 0, serverEndpoint: 127.0.0.1:6379, mgr: 10 of 10 available, clientName: TimeoutException, IOCP: (Busy=0,Free=1000,Min=8,Max=1000), WORKER: (Busy=2,Free=2045,Min=8,Max=2047), v: 2.1.0 (Please take a look at this article for some common client-side issues that can cause timeouts: https://stackexchange.github.io/StackExchange.Redis/Timeouts)";
             Assert.StartsWith("Test Timeout, command=PING", ex.Message);
             Assert.Contains("clientName: " + nameof(TimeoutException), ex.Message);
             // Ensure our pipe numbers are in place
-            Assert.Contains("inst: 0, qu: 0, qs: 0, aw: False, bw: Inactive, in: 0, in-pipe: 0, out-pipe: 0", ex.Message);
+            Assert.Contains("inst: 0, qu: 0, qs: 0, aw: False, bw: Inactive, in: 0, in-pipe: 0, out-pipe: 0, last-in: 0, cur-in: 0", ex.Message);
             Assert.Contains("mc: 1/1/0", ex.Message);
             Assert.Contains("serverEndpoint: " + server.EndPoint, ex.Message);
             Assert.Contains("IOCP: ", ex.Message);
@@ -183,19 +183,19 @@ public void NoConnectionException(bool abortOnConnect, int connCount, int comple
                 var ex = Assert.IsType<RedisConnectionException>(rawEx);
                 Writer.WriteLine("Exception: " + ex.Message);
 
-                // Example format: "Exception: No connection is active/available to service this operation: PING, inst: 0, qu: 0, qs: 0, aw: False, in: 0, in-pipe: 0, out-pipe: 0, serverEndpoint: 127.0.0.1:6379, mc: 1/1/0, mgr: 10 of 10 available, clientName: NoConnectionException, IOCP: (Busy=0,Free=1000,Min=8,Max=1000), WORKER: (Busy=2,Free=2045,Min=8,Max=2047), Local-CPU: 100%, v: 2.1.0.5";
+                // Example format: "Exception: No connection is active/available to service this operation: PING, inst: 0, qu: 0, qs: 0, aw: False, in: 0, in-pipe: 0, out-pipe: 0, last-in: 0, cur-in: 0, serverEndpoint: 127.0.0.1:6379, mc: 1/1/0, mgr: 10 of 10 available, clientName: NoConnectionException, IOCP: (Busy=0,Free=1000,Min=8,Max=1000), WORKER: (Busy=2,Free=2045,Min=8,Max=2047), Local-CPU: 100%, v: 2.1.0.5";
                 Assert.StartsWith(messageStart, ex.Message);
 
                 // Ensure our pipe numbers are in place if they should be
                 if (hasDetail)
                 {
-                    Assert.Contains("inst: 0, qu: 0, qs: 0, aw: False, bw: Inactive, in: 0, in-pipe: 0, out-pipe: 0", ex.Message);
+                    Assert.Contains("inst: 0, qu: 0, qs: 0, aw: False, bw: Inactive, in: 0, in-pipe: 0, out-pipe: 0, last-in: 0, cur-in: 0", ex.Message);
                     Assert.Contains($"mc: {connCount}/{completeCount}/0", ex.Message);
                     Assert.Contains("serverEndpoint: " + server.EndPoint.ToString()?.Replace("Unspecified/", ""), ex.Message);
                 }
                 else
                 {
-                    Assert.DoesNotContain("inst: 0, qu: 0, qs: 0, aw: False, bw: Inactive, in: 0, in-pipe: 0, out-pipe: 0", ex.Message);
+                    Assert.DoesNotContain("inst: 0, qu: 0, qs: 0, aw: False, bw: Inactive, in: 0, in-pipe: 0, out-pipe: 0, last-in: 0, cur-in: 0", ex.Message);
                     Assert.DoesNotContain($"mc: {connCount}/{completeCount}/0", ex.Message);
                     Assert.DoesNotContain("serverEndpoint: " + server.EndPoint.ToString()?.Replace("Unspecified/", ""), ex.Message);
                 }