A close with commands in flight no longer completes a call with the previous call's reply (#3329)

src/StackExchange.Redis/PhysicalConnection.Read.cs

@@ -741,17 +741,18 @@ private void MatchNextResult(ReadOnlySpan<byte> frame)
                 _awaitingToken = null;
             }
 
-            if (!_writtenAwaitingResponse.TryDequeue(out msg))
+            // A failure can race bytes that were already in the socket or read buffer. Once shutdown has
+            // started, no new command can enter this queue, and RecordConnectionFailed is draining it one
+            // message at a time (completing each outside the lock); if we kept dequeuing here, every reply
+            // we matched would be handed to the message *after* the one it belongs to (#3329). Drop the
+            // bytes with the dead connection instead - those messages are being failed anyway.
+            if (_isShutdown)
             {
-                // A failure can race bytes that were already in the socket or read buffer. Once shutdown has
-                // started, no new command can enter this queue, so these are replies to commands the failure
-                // path has already completed. Drop them with the dead connection instead of reporting a fresh
-                // protocol failure.
-                if (_isShutdown)
-                {
-                    return;
-                }
+                return;
+            }
 
+            if (!_writtenAwaitingResponse.TryDequeue(out msg))
+            {
                 Throw(frame, connectionType, _protocol);
 
                 [DoesNotReturn]

tests/StackExchange.Redis.Tests/CloseInFlightTests.cs

@@ -0,0 +1,70 @@
+using System;
+using System.Linq;
+using System.Threading.Tasks;
+using Xunit;
+
+namespace StackExchange.Redis.Tests;
+
+public class CloseInFlightTests(ITestOutputHelper output) : TestBase(output)
+{
+    // #3329: CloseAsync(allowCommandsToComplete: false) with commands in flight races the read loop, still
+    // matching replies already in the buffer, against the failure path draining the sent-awaiting-reply
+    // queue one message at a time. Each reply the read loop matched after the drain had taken a message was
+    // handed to the message *after* the one it belonged to: GET key N completed, successfully, with key N-1's
+    // value. The only acceptable outcomes are the right value or a connection exception.
+    [Fact]
+    public async Task CloseWithCommandsInFlight_NeverCompletesACallWithAnotherCallsReply()
+    {
+        const int Rounds = 20, PerRound = 500;
+        SetExpectedAmbientFailureCount(-1); // every round tears a connection down with work in flight
+        await using var admin = Create();
+        var adminDb = admin.GetDatabase();
+        var prefix = Me();
+        long ok = 0, wrong = 0, failed = 0;
+
+        for (int round = 0; round < Rounds; round++)
+        {
+            var keys = Enumerable.Range(0, PerRound).Select(i => (RedisKey)$"{prefix}:{round}:{i}").ToArray();
+            await Task.WhenAll(keys.Select(k => adminDb.StringSetAsync(k, k.ToString(), TimeSpan.FromMinutes(1))));
+
+            var conn = Create(shared: false);
+            var db = conn.GetDatabase();
+            var calls = keys.Select(k => db.StringGetAsync(k)).ToArray();
+            await conn.CloseAsync(allowCommandsToComplete: false);
+
+            for (int i = 0; i < calls.Length; i++)
+            {
+                string? value;
+                try
+                {
+                    value = await calls[i];
+                }
+                catch (RedisConnectionException)
+                {
+                    failed++;
+                    continue;
+                }
+                catch (ObjectDisposedException)
+                {
+                    failed++;
+                    continue;
+                }
+
+                if (value == keys[i].ToString())
+                {
+                    ok++;
+                }
+                else if (wrong++ < 5)
+                {
+                    Log($"round {round}: GET {keys[i]} returned '{value}'");
+                }
+            }
+
+            await conn.DisposeAsync();
+            await adminDb.KeyDeleteAsync(keys);
+        }
+
+        Log($"ok={ok} wrong={wrong} failed={failed}");
+        Assert.Equal(0, wrong);
+    }
+}