From e333078ec0f9be506a8bc8e8b9c6bb86de9ce2fa Mon Sep 17 00:00:00 2001 From: Marc Gravell Date: Tue, 18 Aug 2026 17:35:35 +0100 Subject: [PATCH] Treat a write losing its connection as a closure, not an internal fault (#3167) Shutdown() discards the output writer, and it cannot wait for the write lock: teardown is usually spotted on the read loop, which must not block. So a writer already inside the lock can have the pipe pulled out from under it at any point. That is an ordinary closure, but the write path reported it as InternalFailure wrapping InvalidOperationException("Output pipe not initialized") - which reads as a client bug, is not a recognised connection failure, and also gets announced through ConnectionMultiplexer.InternalError. IdentifyFailureType already maps ObjectDisposedException to SocketClosed; the write path just never consulted it. Changes: - PhysicalConnection keeps an _isShutdown flag, set before the output is discarded, so a writer that finds it gone can tell a closure from a genuinely uninitialised pipe: the former now throws ObjectDisposedException, the latter still InvalidOperationException. This also removes a duplicate copy of that throw in Flush(). - New PhysicalConnection.ClassifyWriteFailure centralises the decision: honour an inner RedisConnectionException's own failure type, defer to IdentifyFailureType, and only fall back to InternalFailure when there is nothing better. PhysicalBridge (both write-failure paths) and Message.WriteTo now use it, and only report OnInternalError when it really was internal. Two further faults found while reproducing this, both in the same window: - ProcessBridgeBacklog re-read the nullable `physical` field to pass to WriteMessageInsideLock (whose parameter is not nullable) after testing it in the loop guard. OnDisconnected nulls that field, so the drain could hand it null and NRE - surfacing as InternalFailure "Failed to write". Now snapshotted per message, matching the idiom used elsewhere in the file. - OperationCanceledException was not classified. RESPite already documents it as expected teardown noise alongside ObjectDisposedException, and the read loop treats a cancelled read as SocketClosed; the write path now matches, gated on our own OutputCancel having fired so a caller's cancellation is not mislabelled. Tests: a deterministic test that shuts the output down before the write and asserts the failure is not badged internal, plus an explicit (opt-in) stress test that races real teardowns against concurrent writers - the latter is what found all three faults, but it has to saturate the box to hit the window, so it is not part of the default run. --- src/StackExchange.Redis/Message.cs | 24 +- src/StackExchange.Redis/PhysicalBridge.cs | 21 +- .../PhysicalConnection.Write.cs | 32 ++- src/StackExchange.Redis/PhysicalConnection.cs | 32 ++- .../Issues/Issue3167Tests.cs | 218 ++++++++++++++++++ 5 files changed, 303 insertions(+), 24 deletions(-) create mode 100644 tests/StackExchange.Redis.Tests/Issues/Issue3167Tests.cs diff --git a/src/StackExchange.Redis/Message.cs b/src/StackExchange.Redis/Message.cs index 0eceec89e..837651106 100644 --- a/src/StackExchange.Redis/Message.cs +++ b/src/StackExchange.Redis/Message.cs @@ -954,8 +954,7 @@ internal void WriteTo(PhysicalConnection physical) } catch (Exception ex) when (ex is not RedisCommandException) // these have specific meaning; don't wrap { - physical?.OnInternalError(ex); - Fail(ConnectionFailureType.InternalFailure, ex, null, physical?.BridgeCouldBeNull?.Multiplexer); + FailWrite(physical, ex); // Re-throw so the outer write path (PhysicalBridge.HandleWriteException) can tear down the // connection. A partial write would otherwise leave bytes on the wire while the response // queue still considers the slot healthy, allowing a subsequent reply to match the wrong @@ -973,8 +972,7 @@ internal void WriteTo(PhysicalConnection physical, CommandMap commandMap, byte[] } catch (Exception ex) when (ex is not RedisCommandException) // these have specific meaning; don't wrap { - physical?.OnInternalError(ex); - Fail(ConnectionFailureType.InternalFailure, ex, null, physical?.BridgeCouldBeNull?.Multiplexer); + FailWrite(physical, ex); // Re-throw so the outer write path (PhysicalBridge.HandleWriteException) can tear down the // connection. A partial write would otherwise leave bytes on the wire while the response // queue still considers the slot healthy, allowing a subsequent reply to match the wrong @@ -983,6 +981,21 @@ internal void WriteTo(PhysicalConnection physical, CommandMap commandMap, byte[] } } + /// + /// Fail this message after a write fault, only shouting via OnInternalError when it really was an internal + /// fault; see . + /// + private void FailWrite(PhysicalConnection? physical, Exception ex) + { + var failureType = PhysicalConnection.ClassifyWriteFailure(ex, physical); + if (failureType == ConnectionFailureType.InternalFailure) + { + physical?.OnInternalError(ex); + } + + Fail(failureType, ex, null, physical?.BridgeCouldBeNull?.Multiplexer); + } + private static ReadOnlySpan ChecksumTemplate => "$4\r\nXXXX\r\n"u8; internal void WriteHighIntegrityChecksumRequest(PhysicalConnection physical) @@ -1001,8 +1014,7 @@ internal void WriteHighIntegrityChecksumRequest(PhysicalConnection physical) } catch (Exception ex) { - physical?.OnInternalError(ex); - Fail(ConnectionFailureType.InternalFailure, ex, null, physical?.BridgeCouldBeNull?.Multiplexer); + FailWrite(physical, ex); } } diff --git a/src/StackExchange.Redis/PhysicalBridge.cs b/src/StackExchange.Redis/PhysicalBridge.cs index 6a8857a39..a974d523a 100644 --- a/src/StackExchange.Redis/PhysicalBridge.cs +++ b/src/StackExchange.Redis/PhysicalBridge.cs @@ -1197,8 +1197,13 @@ private void ProcessBridgeBacklog() // Timeouts are handled above, so we're exclusively into backlog items eligible to write at this point. // If we can't write them, abort and wait for the next heartbeat or activation to try this again. bool flush = false; - while (IsConnected && physical is { HasOutputPipe: true }) + while (IsConnected) { + // Snapshot the connection for the whole of this message: OnDisconnected nulls the field + // without waiting for the write lock, so re-reading it (as the loop guard used to) can hand + // null to a write path that requires a connection. See #3167. + if (this.physical is not { HasOutputPipe: true } physical) break; + Message? message; _backlogStatus = BacklogStatus.CheckingForWork; @@ -1437,13 +1442,14 @@ private async ValueTask CompleteWriteAndReleaseLockAsync( private WriteResult HandleWriteException(PhysicalConnection? physical, Message message, Exception ex) { - var inner = new RedisConnectionException(ConnectionFailureType.InternalFailure, message.Flags, "Failed to write", ex); + var failureType = PhysicalConnection.ClassifyWriteFailure(ex, physical); + var inner = new RedisConnectionException(failureType, message.Flags, "Failed to write", ex); message.SetExceptionAndComplete(inner, physical); // Tear down the physical connection. A write that throws may have left a partial frame on the // wire, and continuing to use the same socket would let the next reply match the wrong message // in the response queue. Forcing a reconnect drains the in-flight queue with failures and // restores wire-level synchronization. - physical?.RecordConnectionFailed(ConnectionFailureType.InternalFailure, inner); + physical?.RecordConnectionFailed(failureType, inner); return WriteResult.WriteFailure; } @@ -1713,11 +1719,14 @@ private WriteResult WriteMessageToServerInsideWriteLock(PhysicalConnection conne catch (Exception ex) { Trace("Write failed: " + ex.Message); - message.Fail(ConnectionFailureType.InternalFailure, ex, null, Multiplexer); + + // Most likely an IOException, or the connection being torn down underneath us + var failureType = PhysicalConnection.ClassifyWriteFailure(ex, connection); + message.Fail(failureType, ex, null, Multiplexer); message.Complete(connection); - // We're not sure *what* happened here - probably an IOException; kill the connection - connection?.RecordConnectionFailed(ConnectionFailureType.InternalFailure, ex); + // We don't know how far the write got; kill the connection + connection?.RecordConnectionFailed(failureType, ex); return WriteResult.WriteFailure; } } diff --git a/src/StackExchange.Redis/PhysicalConnection.Write.cs b/src/StackExchange.Redis/PhysicalConnection.Write.cs index 08edefd68..a51774e13 100644 --- a/src/StackExchange.Redis/PhysicalConnection.Write.cs +++ b/src/StackExchange.Redis/PhysicalConnection.Write.cs @@ -1,6 +1,8 @@ using System; using System.Buffers; +using System.Diagnostics.CodeAnalysis; using System.IO; +using System.Runtime.CompilerServices; using System.Threading.Tasks; using RESPite.Streams; @@ -9,15 +11,29 @@ namespace StackExchange.Redis; internal partial class PhysicalConnection { private BufferedStreamWriter? _output; + + /// + /// Set by before it discards , so a writer that finds the + /// output gone can tell the two cases apart; see . + /// + private volatile bool _isShutdown; + private long TotalBytesSent => _output?.TotalBytesWritten ?? 0; - public IBufferWriter Output - { - get - { - return _output ?? Throw(); - static IBufferWriter Throw() => throw new InvalidOperationException("Output pipe not initialized"); - } - } + public IBufferWriter Output => _output ?? ThrowOutputUnavailable(); + + /// + /// discards the output writer, and it can run concurrently with a writer that is + /// already inside the write lock - teardown is usually detected on the read loop, which must not block on + /// that lock. A write that loses the pipe that way is an ordinary closure, not a bug, so report it as one: + /// is what maps to + /// . A missing output when we were *not* shut down really + /// is a bug, and stays an . See #3167. + /// + [MethodImpl(MethodImplOptions.NoInlining)] + [DoesNotReturn] + private BufferedStreamWriter ThrowOutputUnavailable() => throw (_isShutdown + ? new ObjectDisposedException(nameof(PhysicalConnection), "The connection was closed while writing") + : (Exception)new InvalidOperationException("Output pipe not initialized")); private void InitOutput(Stream? stream) { diff --git a/src/StackExchange.Redis/PhysicalConnection.cs b/src/StackExchange.Redis/PhysicalConnection.cs index 0d93eefeb..9930b1066 100644 --- a/src/StackExchange.Redis/PhysicalConnection.cs +++ b/src/StackExchange.Redis/PhysicalConnection.cs @@ -325,6 +325,7 @@ public void SetProtocol(RedisProtocol value) [System.Diagnostics.CodeAnalysis.SuppressMessage("Microsoft.Usage", "CA2202:Do not dispose objects multiple times", Justification = "Trust me yo")] internal void Shutdown(ConnectionFailureType failureType = ConnectionFailureType.ConnectionDisposed) { + _isShutdown = true; // *before* discarding the output, so an observed-null output implies this flag var output = Interlocked.Exchange(ref _output, null); // compare to the critical read var socket = Interlocked.Exchange(ref _socket, null); var transport = Interlocked.Exchange(ref _transport, null); @@ -599,6 +600,32 @@ internal enum WriteStatus /// A string that represents the current object. public override string ToString() => $"{_physicalName} ({_writeStatus})"; + /// + /// Classify a fault from the write path. Prefer what the exception already knows - an inner + /// carries its own failure type, and a discarded output pipe or a + /// dead socket is a closure - falling back to only when + /// there is nothing better to go on. Writes can lose the connection underneath them at any point, because + /// does not (and must not) wait on the write lock; that is an ordinary connection + /// failure, not an internal fault, and badging it as the latter also reports it via OnInternalError. See #3167. + /// + internal static ConnectionFailureType ClassifyWriteFailure(Exception exception, PhysicalConnection? connection) + { + var failureType = exception is RedisConnectionException rce ? rce.FailureType : ConnectionFailureType.InternalFailure; + IdentifyFailureType(exception, ref failureType); + + // A write killed by *our own* output cancellation is the same closure that the read loop reports as + // SocketClosed (see ReadAllAsync), and RESPite calls it out as expected teardown noise. A cancellation + // that came from the caller is a different thing, so check whose token actually fired. + if (failureType == ConnectionFailureType.InternalFailure + && exception is OperationCanceledException + && connection?.OutputCancel.IsCancellationRequested == true) + { + failureType = ConnectionFailureType.SocketClosed; + } + + return failureType; + } + internal static void IdentifyFailureType(Exception? exception, ref ConnectionFailureType failureType) { if (exception != null && failureType == ConnectionFailureType.InternalFailure) @@ -864,14 +891,11 @@ internal void RecordQuit() internal void Flush() { - var tmp = _output; - if (tmp is null) Throw(); + var tmp = _output ?? ThrowOutputUnavailable(); _writeStatus = WriteStatus.Flushing; tmp.Flush(); _writeStatus = WriteStatus.Flushed; UpdateLastWriteTime(); - [DoesNotReturn] - static void Throw() => throw new InvalidOperationException("Output pipe not initialized"); } internal readonly struct ConnectionStatus diff --git a/tests/StackExchange.Redis.Tests/Issues/Issue3167Tests.cs b/tests/StackExchange.Redis.Tests/Issues/Issue3167Tests.cs new file mode 100644 index 000000000..c9bb0c569 --- /dev/null +++ b/tests/StackExchange.Redis.Tests/Issues/Issue3167Tests.cs @@ -0,0 +1,218 @@ +using System; +using System.Collections.Concurrent; +using System.Collections.Generic; +using System.Linq; +using System.Threading; +using System.Threading.Tasks; +using Xunit; + +namespace StackExchange.Redis.Tests.Issues; + +/// +/// A write can lose its connection at any point: Shutdown discards the output writer, and it cannot wait +/// for the write lock, because teardown is usually spotted on the read loop - which must not block. That is an +/// ordinary closure, so it should be reported as a connection failure, not as InternalFailure carrying +/// InvalidOperationException("Output pipe not initialized"), which reads as a client bug and is also +/// announced through . See #3167. +/// +[Collection(NonParallelCollection.Name)] // these kill the connection out from under a shared server +public class Issue3167Tests(ITestOutputHelper output) : TestBase(output) +{ + protected override string GetConfiguration() => TestConfig.Current.PrimaryServerAndPort; + + /// + /// The deterministic half: teardown has already discarded the output writer by the time the write starts, + /// which is exactly what the race produces, minus the racing. + /// + [Fact] + public async Task WriteLosingItsConnectionIsReportedAsAClosure() + { + try + { + await using var conn = Create(shared: false); + await conn.GetDatabase().PingAsync(); + + var bridge = conn.GetServerSnapshot()[0].GetBridge(ConnectionType.Interactive); + Assert.NotNull(bridge); + var physical = bridge.TryConnect(null); + Assert.NotNull(physical); + + var internalErrors = new List(); + conn.UnderlyingMultiplexer.InternalError += (_, args) => internalErrors.Add(args.Exception); + + // teardown wins the race + physical.Shutdown(); + + var message = Message.Create(0, CommandFlags.None, RedisCommand.GET, (RedisKey)Me()); + var resultBox = SimpleResultBox.Create(); + message.SetSource(ResultProcessor.String, resultBox); + + var result = await bridge.WriteMessageTakingWriteLockAsync(physical, message, bypassBacklog: true); + Assert.Equal(WriteResult.WriteFailure, result); + + resultBox.GetResult(out var ex); + Assert.NotNull(ex); + Log(ex.ToString()); + + // Note the message can also be completed by the teardown that our own Shutdown kicked off, carrying + // the underlying socket error - that is fine, and is why this doesn't demand one exact exception. What + // must not happen is the write path reporting a closure as an internal fault: before #3167 this was + // "InternalFailure on [0]:GET ...", wrapping InvalidOperationException("Output pipe not initialized"). + Assert.DoesNotContain("Output pipe not initialized", ex.ToString(), StringComparison.Ordinal); + for (Exception? walk = ex; walk is not null; walk = walk.InnerException) + { + if (walk is RedisConnectionException rce) + { + Assert.NotEqual(ConnectionFailureType.InternalFailure, rce.FailureType); + } + } + + // ...and a routine disconnect should not be announced as an internal library fault + Assert.Empty(internalErrors); + } + finally + { + ClearAmbientFailures(); + } + } + + /// + /// The racing half: writes in flight while the connection is repeatedly torn down underneath them. This is what + /// found the issue, and it covers the whole write path rather than one throw site - including the backlog drain, + /// which the deterministic test above does not reach. + /// + /// + /// Explicit: it needs to saturate the box to hit the window reliably, which starves anything running alongside + /// it. Run it directly when touching the write path, via + /// dotnet run -c Release -f net10.0 -- -explicit only -method "*WritesRacingTeardown*". Note this may + /// show as Inconclusive, depending on the runner. + /// + [Fact(Explicit = true)] + [Trait(TestCategories.Category, TestCategories.SimulatedConnectionFailure)] + public async Task WritesRacingTeardownAreNeverInternalFailures() + { + var options = new ConfigurationOptions + { + BacklogPolicy = BacklogPolicy.Default, + AbortOnConnectFail = false, + ConnectTimeout = 1000, + ConnectRetry = 2, + SyncTimeout = 5000, + AsyncTimeout = 5000, + KeepAlive = 10000, + AllowAdmin = true, + AllowSimulateConnectionFailure = true, + }; + options.EndPoints.Add(TestConfig.Current.PrimaryServerAndPort); + + try + { + await using var conn = await ConnectionMultiplexer.ConnectAsync(options, Writer); + var db = conn.GetDatabase(); + await db.PingAsync(); + + var server = conn.GetServerSnapshot()[0]; + Assert.SkipUnless(server.CanSimulateConnectionFailure, "Skipping because server cannot simulate connection failure"); + + var outputPipeFaults = new ConcurrentQueue(); + var internalFailures = new ConcurrentQueue(); + var internalErrors = new ConcurrentQueue(); + long totalOps = 0, totalFaults = 0; + using var cts = new CancellationTokenSource(TimeSpan.FromSeconds(5)); + + void Record(Exception ex) + { + Interlocked.Increment(ref totalFaults); + + // the bug's fingerprint: the write path complaining as if we had never connected + if (ex.ToString().Contains("Output pipe not initialized", StringComparison.Ordinal)) + { + outputPipeFaults.Enqueue(ex); + } + + for (Exception? walk = ex; walk is not null; walk = walk.InnerException) + { + if (walk is RedisConnectionException { FailureType: ConnectionFailureType.InternalFailure }) + { + internalFailures.Enqueue(ex); + } + } + } + + conn.InternalError += (_, args) => + { + internalErrors.Enqueue(args.Exception); + Record(args.Exception); + }; + + var key = Me(); + var token = cts.Token; + + // lots of concurrent writers, so the write lock is contended and a backlog forms + var writers = Enumerable.Range(0, 16).Select(_ => Task.Run(async () => + { + while (!token.IsCancellationRequested) + { + try + { + await db.StringGetAsync(key).ForAwait(); + Interlocked.Increment(ref totalOps); + } + catch (Exception ex) + { + Record(ex); + } + } + })).ToArray(); + + // ...while the connection is repeatedly torn down under them. SimulateConnectionFailure runs + // RecordConnectionFailed -> Shutdown() synchronously on *this* thread, which is what discards the + // output writer, so it lands while the writers are inside the write lock. + int kills = 0; + while (!token.IsCancellationRequested) + { + server.SimulateConnectionFailure(SimulatedFailureType.AllInteractive); + kills++; + try + { + await Task.Delay(20, token).ForAwait(); + } + catch (OperationCanceledException) + { + break; + } + } + + await Task.WhenAll(writers).ForAwait(); + + var ops = Volatile.Read(ref totalOps); + Log($"ops: {ops}, faults: {Volatile.Read(ref totalFaults)}, kills: {kills}"); + Log($"internal failures: {internalFailures.Count}, output-pipe faults: {outputPipeFaults.Count}, internal errors: {internalErrors.Count}"); + + void Dump(string label, ConcurrentQueue queue) + { + foreach (var ex in queue.Take(3)) + { + Log($"---- {label} ----"); + Log(ex.ToString()); + } + } + + Dump("output-pipe fault", outputPipeFaults); + Dump("internal failure", internalFailures); + + // this only means anything if we actually raced teardown; if these trip, the writers or the kill + // loop stopped doing their job, rather than the bug being fixed + Assert.True(kills > 10, $"expected the kill loop to run, got {kills}"); + Assert.True(ops > 1000, $"expected the writers to run, got {ops} ops"); + Assert.True(Volatile.Read(ref totalFaults) > 0, "expected the teardowns to have visible fallout"); + + Assert.Empty(outputPipeFaults); + Assert.Empty(internalFailures); + } + finally + { + ClearAmbientFailures(); + } + } +}