Treat a write losing its connection as a closure, not an internal fault (#3167)

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[]
             }
         }
 
+        /// <summary>
+        /// Fail this message after a write fault, only shouting via OnInternalError when it really was an internal
+        /// fault; see <see cref="PhysicalConnection.ClassifyWriteFailure"/>.
+        /// </summary>
+        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<byte> 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);
             }
         }
 

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<WriteResult> 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;
             }
         }

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;
+
+    /// <summary>
+    /// Set by <see cref="Shutdown"/> before it discards <see cref="_output"/>, so a writer that finds the
+    /// output gone can tell the two cases apart; see <see cref="ThrowOutputUnavailable"/>.
+    /// </summary>
+    private volatile bool _isShutdown;
+
     private long TotalBytesSent => _output?.TotalBytesWritten ?? 0;
-    public IBufferWriter<byte> Output
-    {
-        get
-        {
-            return _output ?? Throw();
-            static IBufferWriter<byte> Throw() => throw new InvalidOperationException("Output pipe not initialized");
-        }
-    }
+    public IBufferWriter<byte> Output => _output ?? ThrowOutputUnavailable();
+
+    /// <summary>
+    /// <see cref="Shutdown"/> 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:
+    /// <see cref="ObjectDisposedException"/> is what <see cref="IdentifyFailureType"/> maps to
+    /// <see cref="ConnectionFailureType.SocketClosed"/>. A missing output when we were *not* shut down really
+    /// is a bug, and stays an <see cref="InvalidOperationException"/>. See #3167.
+    /// </summary>
+    [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)
     {

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
         /// <returns>A string that represents the current object.</returns>
         public override string ToString() => $"{_physicalName} ({_writeStatus})";
 
+        /// <summary>
+        /// Classify a fault from the write path. Prefer what the exception already knows - an inner
+        /// <see cref="RedisConnectionException"/> carries its own failure type, and a discarded output pipe or a
+        /// dead socket is a closure - falling back to <see cref="ConnectionFailureType.InternalFailure"/> only when
+        /// there is nothing better to go on. Writes can lose the connection underneath them at any point, because
+        /// <see cref="Shutdown"/> 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.
+        /// </summary>
+        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