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); + } +}