From b3321f5792c217b6f3ea08eadae24f80df02bff7 Mon Sep 17 00:00:00 2001 From: Marc Gravell Date: Thu, 22 Oct 2026 11:18:01 +0100 Subject: [PATCH 1/2] Complete a message refused by a poisoned connection; divert a batch's tail on reconnect (#3282) Since #3205 a connection whose write faulted is poisoned and refuses further writes, so that no reply can be matched to the wrong caller. That refusal returned WriteFailure without completing the message. Single commands were unaffected, as ExecuteAsyncImpl/ExecuteSyncImpl translate a non-success result into an exception for the caller; but RedisBatch.Execute writes a list via PhysicalBridge.TryEnqueue, which does not act on the result, so once one message of a batch faulted the connection, the rest of that batch was neither queued nor completed and its tasks were awaited forever (3.2.0 to 3.4.5). - WriteMessageToServerInsideWriteLock: the IsWriteFaulted refusal now fails and completes the message, as every other refusal to write already does (FailDueToNoConnection, TimedOutBeforeWrite, QueueOrFailMessage, the write-lock catch, the transaction sub-command loop). - TryEnqueue(list): once NeedsReconnect is set part-way through the batch, the remaining messages take the same path a single command would at the top of TryWriteSync - backlog or fail, per policy - instead of being written to a connection known to be dead. BatchWriteFailureTests injects a write failure mid-batch under both backlog policies and asserts that every task completes and that any successful result is real. The reporter's repro (400 batches in flight, CLIENT KILL in a loop) hung within 13 s on main and ran 6.6 M batches in 120 s with this change. Fixes #3282 --- src/StackExchange.Redis/PhysicalBridge.cs | 15 +++- .../BatchWriteFailureTests.cs | 75 +++++++++++++++++++ 2 files changed, 89 insertions(+), 1 deletion(-) create mode 100644 tests/StackExchange.Redis.Tests/BatchWriteFailureTests.cs diff --git a/src/StackExchange.Redis/PhysicalBridge.cs b/src/StackExchange.Redis/PhysicalBridge.cs index 7cf39edeb..14e6270e8 100644 --- a/src/StackExchange.Redis/PhysicalBridge.cs +++ b/src/StackExchange.Redis/PhysicalBridge.cs @@ -810,6 +810,15 @@ internal bool TryEnqueue(List messages, bool isReplica) } foreach (var message in messages) { + if (NeedsReconnect) + { + // the connection went away part-way through the batch (typically an earlier message's write + // faulted it); give the remainder the treatment a single command gets at the top of + // TryWriteSync - backlog or fail, per policy - rather than writing to a dead connection (#3282) + QueueOrFailMessage(message); + continue; + } + // deliberately not taking a single lock here; we don't care if // other threads manage to interleave - in fact, it would be desirable // (to avoid a batch monopolising the connection) @@ -1653,7 +1662,11 @@ private WriteResult WriteMessageToServerInsideWriteLock(PhysicalConnection conne if (connection.IsWriteFaulted) { - // never add to the response queue of a connection whose queue we no longer trust + // never add to the response queue of a connection whose queue we no longer trust - but, like every + // other refusal to write, complete the message *here*: a caller writing a list (a batch) does not + // act on the result, and a message that is neither queued nor completed is awaited forever (#3282) + message.Fail(ConnectionFailureType.SocketFailure, null, "an earlier write on this connection failed; this command was not written", Multiplexer); + message.Complete(connection); return WriteResult.WriteFailure; } diff --git a/tests/StackExchange.Redis.Tests/BatchWriteFailureTests.cs b/tests/StackExchange.Redis.Tests/BatchWriteFailureTests.cs new file mode 100644 index 000000000..e61256a38 --- /dev/null +++ b/tests/StackExchange.Redis.Tests/BatchWriteFailureTests.cs @@ -0,0 +1,75 @@ +using System; +using System.Linq; +using System.Threading.Tasks; +using Xunit; + +namespace StackExchange.Redis.Tests; + +/// +/// A write failure part-way through an poisons the connection; the batch's remaining +/// messages must still complete - faulted, or backlogged and delivered after the reconnect - never be left +/// awaiting a reply that cannot come (#3282). +/// +public class BatchWriteFailureTests(ITestOutputHelper log) +{ + private sealed class ThrowingMessage(Exception toThrow) : Message(-1, CommandFlags.None, RedisCommand.PING) + { + public override int ArgCount => 0; + + protected override void WriteImpl(in MessageWriter writer) => throw toThrow; + } + + [Theory] + [InlineData(false)] + [InlineData(true)] + public async Task EveryBatchCommandCompletesAfterMidBatchWriteFailure(bool queueWhileDisconnected) + { + using var server = new InProcessTestServer(log); + var config = server.GetClientConfig(withPubSub: false, defaultOnly: true); + config.BacklogPolicy = queueWhileDisconnected ? BacklogPolicy.Default : BacklogPolicy.FailFast; + config.AbortOnConnectFail = false; + await using var conn = await ConnectionMultiplexer.ConnectAsync(config); + var db = conn.GetDatabase(); + RedisKey key = nameof(EveryBatchCommandCompletesAfterMidBatchWriteFailure); + await db.KeyDeleteAsync(key); + + var batch = (RedisBatch)db.CreateBatch(); + var before = batch.StringIncrementAsync(key); + var boom = batch.ExecuteAsync(new ThrowingMessage(new InvalidOperationException("simulated WriteImpl failure")), ResultProcessor.DemandOK, false); + var after1 = batch.StringIncrementAsync(key); + var after2 = batch.StringIncrementAsync(key); + batch.Execute(); + + Task[] all = [before, boom, after1, after2]; + var everything = Task.WhenAll(all); + if (await Task.WhenAny(everything, Task.Delay(TimeSpan.FromSeconds(10))) != everything) + { + Assert.Fail("hung: " + string.Join(", ", all.Select((t, i) => $"cmd{i}={t.Status}"))); + } + // individual faults are inspected below + + for (int i = 0; i < all.Length; i++) + { + log.WriteLine($"cmd{i}: {all[i].Status} {all[i].Exception?.GetBaseException().GetType().Name} {all[i].Exception?.GetBaseException().Message}"); + } + Assert.All(all, t => Assert.True(t.IsCompleted)); + Assert.True(boom.IsFaulted); + // "before" was written and then torn down with the connection: answered first, or failed with it. + // The tail is backlogged and delivered on the new connection, or refused (cancelled before the write, + // or faulted at the poisoned connection) - never left pending. Whatever succeeded must be real: the + // counter reflects every INCR the server ran, which is at least every one reported as successful + Task[] increments = [before, after1, after2]; + if (queueWhileDisconnected) + { + Assert.True(after1.IsCompletedSuccessfully, after1.Status.ToString()); + Assert.True(after2.IsCompletedSuccessfully, after2.Status.ToString()); + } + else + { + Assert.All([after1, after2], t => Assert.True( + t.IsCompletedSuccessfully || t.IsCanceled || t.Exception?.GetBaseException() is RedisConnectionException, t.Status.ToString())); + } + long succeeded = increments.Count(t => t.IsCompletedSuccessfully); + Assert.InRange((long)await db.StringGetAsync(key), succeeded, increments.Length); + } +} From 6242f1d94b4a2597b507f30b52e0b7119fdd2b54 Mon Sep 17 00:00:00 2001 From: Marc Gravell Date: Thu, 22 Oct 2026 12:50:52 +0100 Subject: [PATCH 2/2] Batch: route the tail to the backlog when the held connection is poisoned, not just while the reconnect flag is set The reconnect flag is cleared as soon as TryConnect starts a replacement connection, on another thread; a batch still holding the previous PhysicalConnection could then write its remaining messages to it and have them refused with WriteFailure. Seen on the Windows CI runner (2 of 3 attempts); reproduces locally 12/25 pinned to two cores. Also make the test's fail-fast variant wait for the reconnect before its verification read, which is the other way the two-core run failed. --- src/StackExchange.Redis/PhysicalBridge.cs | 6 ++++-- .../BatchWriteFailureTests.cs | 13 ++++++++++++- 2 files changed, 16 insertions(+), 3 deletions(-) diff --git a/src/StackExchange.Redis/PhysicalBridge.cs b/src/StackExchange.Redis/PhysicalBridge.cs index 14e6270e8..48cee4aaa 100644 --- a/src/StackExchange.Redis/PhysicalBridge.cs +++ b/src/StackExchange.Redis/PhysicalBridge.cs @@ -810,11 +810,13 @@ internal bool TryEnqueue(List messages, bool isReplica) } foreach (var message in messages) { - if (NeedsReconnect) + if (NeedsReconnect || physical.IsWriteFaulted) { // the connection went away part-way through the batch (typically an earlier message's write // faulted it); give the remainder the treatment a single command gets at the top of - // TryWriteSync - backlog or fail, per policy - rather than writing to a dead connection (#3282) + // TryWriteSync - backlog or fail, per policy - rather than writing to a dead connection (#3282). + // Both flags matter: the reconnect flag is cleared as soon as a replacement connection starts + // (TryConnect, on another thread), but the connection we are still holding stays poisoned QueueOrFailMessage(message); continue; } diff --git a/tests/StackExchange.Redis.Tests/BatchWriteFailureTests.cs b/tests/StackExchange.Redis.Tests/BatchWriteFailureTests.cs index e61256a38..5ca7a2789 100644 --- a/tests/StackExchange.Redis.Tests/BatchWriteFailureTests.cs +++ b/tests/StackExchange.Redis.Tests/BatchWriteFailureTests.cs @@ -1,4 +1,4 @@ -using System; +using System; using System.Linq; using System.Threading.Tasks; using Xunit; @@ -70,6 +70,17 @@ public async Task EveryBatchCommandCompletesAfterMidBatchWriteFailure(bool queue t.IsCompletedSuccessfully || t.IsCanceled || t.Exception?.GetBaseException() is RedisConnectionException, t.Status.ToString())); } long succeeded = increments.Count(t => t.IsCompletedSuccessfully); + + // under fail-fast the tail is refused rather than backlogged, so nothing above waited for the reconnect; + // the verification read must not fail for the same reason + var reconnected = Task.Run(async () => + { + while (!conn.IsConnected) await Task.Delay(50); + }); + if (await Task.WhenAny(reconnected, Task.Delay(TimeSpan.FromSeconds(10))) != reconnected) + { + Assert.Fail("did not reconnect"); + } Assert.InRange((long)await db.StringGetAsync(key), succeeded, increments.Length); } }