From 96a16910aa47ff5554de725baed9ac00931251f2 Mon Sep 17 00:00:00 2001 From: mgravell Date: Sun, 1 Nov 2026 23:01:21 +0000 Subject: [PATCH] The read loop stops matching replies once teardown owns the response queue; a close with commands in flight no longer completes a call with the previous call's reply RecordConnectionFailed drains _writtenAwaitingResponse one message at a time, completing each outside the lock, while the read loop can still be matching replies that were already in the receive buffer. Each reply the loop matched after the drain had taken a message went to the message *after* the one it answered: GET key N completed, successfully, with key N-1's value, and every reply behind it was off by one. Reproduced on main with the reporter's loop: 616 wrong replies in 20,000 calls. _isShutdown was already set under the queue's lock before the drain, and the loop already consulted it - but only once the queue was empty. Checking it before the dequeue closes the race: a reply arriving after teardown has begun is dropped with the connection, and the message it answered is failed by the drain as before. The test re-creates the scenario against the real server (20 rounds of 500 in-flight GETs closed with allowCommandsToComplete: false) and asserts that no call completes with another call's value; without the fix it sees ~200 in a run. Fixes #3329 --- .../PhysicalConnection.Read.cs | 19 ++--- .../CloseInFlightTests.cs | 70 +++++++++++++++++++ 2 files changed, 80 insertions(+), 9 deletions(-) create mode 100644 tests/StackExchange.Redis.Tests/CloseInFlightTests.cs diff --git a/src/StackExchange.Redis/PhysicalConnection.Read.cs b/src/StackExchange.Redis/PhysicalConnection.Read.cs index 4fe44624f..5b572c471 100644 --- a/src/StackExchange.Redis/PhysicalConnection.Read.cs +++ b/src/StackExchange.Redis/PhysicalConnection.Read.cs @@ -741,17 +741,18 @@ private void MatchNextResult(ReadOnlySpan 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] diff --git a/tests/StackExchange.Redis.Tests/CloseInFlightTests.cs b/tests/StackExchange.Redis.Tests/CloseInFlightTests.cs new file mode 100644 index 000000000..cc9472045 --- /dev/null +++ b/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); + } +}