diff --git a/src/StackExchange.Redis/PhysicalBridge.cs b/src/StackExchange.Redis/PhysicalBridge.cs index 7cf39edeb..48cee4aaa 100644 --- a/src/StackExchange.Redis/PhysicalBridge.cs +++ b/src/StackExchange.Redis/PhysicalBridge.cs @@ -810,6 +810,17 @@ internal bool TryEnqueue(List messages, bool isReplica) } foreach (var message in messages) { + 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). + // 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; + } + // 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 +1664,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..5ca7a2789 --- /dev/null +++ b/tests/StackExchange.Redis.Tests/BatchWriteFailureTests.cs @@ -0,0 +1,86 @@ +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); + + // 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); + } +}