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