MultiplexingStream transmit errors when shutting down channels

src/Nerdbank.Streams/MultiplexingStream.Channel.cs

@@ -110,7 +110,8 @@ public class Channel : IDisposableObservable, IDuplexPipe
             private Exception? faultingException;
 
             /// <summary>
-            /// The <see cref="PipeReader"/> to use to get data to be transmitted over the <see cref="Streams.MultiplexingStream"/>.
+            /// The <see cref="PipeReader"/> to use to get data to be transmitted over the <see cref="Streams.MultiplexingStream"/>. Any errors passed to this
+            /// <see cref="PipeReader"/> are transmitted to the remote side.
             /// </summary>
             private PipeReader? mxStreamIOReader;
 
@@ -142,6 +143,11 @@ public class Channel : IDisposableObservable, IDuplexPipe
             /// </summary>
             private bool? existingPipeGiven;
 
+            /// <summary>
+            /// A value indicating whether this <see cref="Channel"/> received an error from a remote party in <see cref="OnContentWritingCompleted(MultiplexingProtocolException?)"/>.
+            /// </summary>
+            private bool receivedRemoteException;
+
             /// <summary>
             /// Initializes a new instance of the <see cref="Channel"/> class.
             /// </summary>
@@ -361,7 +367,7 @@ public void Dispose()
                                         mxStreamIOWriter = self.mxStreamIOWriter;
                                     }
 
-                                    mxStreamIOWriter?.Complete();
+                                    mxStreamIOWriter?.Complete(self.GetRemoteException());
                                     self.mxStreamIOWriterCompleted.Set();
                                 }
                                 finally
@@ -401,7 +407,17 @@ public void Dispose()
                     this.remoteWindowHasCapacity.Set();
 
                     this.disposalTokenSource.Cancel();
-                    this.completionSource.TrySetResult(null);
+
+                    // If we are disposing due to receiving or sending an exception, relay that to our client.
+                    if (this.faultingException != null)
+                    {
+                        this.completionSource.TrySetException(this.faultingException);
+                    }
+                    else
+                    {
+                        this.completionSource.TrySetResult(null);
+                    }
+
                     this.MultiplexingStream.OnChannelDisposed(this);
                 }
             }
@@ -418,7 +434,7 @@ internal async Task OnChannelTerminatedAsync()
                     // We Complete the writer because only the writing (logical) thread should complete it
                     // to avoid race conditions, and Channel.Dispose can be called from any thread.
                     using PipeWriterRental writerRental = await this.GetReceivedMessagePipeWriterAsync().ConfigureAwait(false);
-                    await writerRental.Writer.CompleteAsync().ConfigureAwait(false);
+                    await writerRental.Writer.CompleteAsync(this.GetRemoteException()).ConfigureAwait(false);
                 }
                 catch (ObjectDisposedException)
                 {
@@ -497,32 +513,49 @@ internal async ValueTask OnContentAsync(ReadOnlySequence<byte> payload, Cancella
             /// <summary>
             /// Called by the <see cref="MultiplexingStream"/> when when it will not be writing any more data to the channel.
             /// </summary>
-            internal void OnContentWritingCompleted()
+            /// <param name="error">The error in writing that originated on the remote side, if applicable.</param>
+            internal void OnContentWritingCompleted(MultiplexingProtocolException? error = null)
             {
+                // If we have already received an error from the remote side then no need to complete the channel again.
+                if (this.receivedRemoteException)
+                {
+                    return;
+                }
+
+                // Set the state of the channel based on whether we are completing due to an error.
+                lock (this.SyncObject)
+                {
+                    this.faultingException ??= error;
+                    this.receivedRemoteException = error != null;
+                }
+
                 this.DisposeSelfOnFailure(Task.Run(async delegate
                 {
                     if (!this.IsDisposed)
                     {
                         try
                         {
+                            // If the channel is not disposed, then first try to close the writer used by the channel owner
                             using PipeWriterRental writerRental = await this.GetReceivedMessagePipeWriterAsync().ConfigureAwait(false);
-                            await writerRental.Writer.CompleteAsync().ConfigureAwait(false);
+                            await writerRental.Writer.CompleteAsync(this.GetRemoteException()).ConfigureAwait(false);
                         }
                         catch (ObjectDisposedException)
                         {
+                            // If not, try to close the underlying writer.
                             if (this.mxStreamIOWriter != null)
                             {
                                 using AsyncSemaphore.Releaser releaser = await this.mxStreamIOWriterSemaphore.EnterAsync().ConfigureAwait(false);
-                                await this.mxStreamIOWriter.CompleteAsync().ConfigureAwait(false);
+                                await this.mxStreamIOWriter.CompleteAsync(this.GetRemoteException()).ConfigureAwait(false);
                             }
                         }
                     }
                     else
                     {
+                        // If the channel has not been disposed then just close the underlying writer
                         if (this.mxStreamIOWriter != null)
                         {
                             using AsyncSemaphore.Releaser releaser = await this.mxStreamIOWriterSemaphore.EnterAsync().ConfigureAwait(false);
-                            await this.mxStreamIOWriter.CompleteAsync().ConfigureAwait(false);
+                            await this.mxStreamIOWriter.CompleteAsync(this.GetRemoteException()).ConfigureAwait(false);
                         }
                     }
 
@@ -644,6 +677,15 @@ private async ValueTask<PipeWriterRental> GetReceivedMessagePipeWriterAsync(Canc
                 }
             }
 
+            /// <summary>
+            /// Gets the <see cref="Exception"/> exception that we received from the remote side when completing this channel.
+            /// </summary>
+            /// <returns>The exception sent from the remote if there is one, null otherwise.</returns>
+            private Exception? GetRemoteException()
+            {
+                return this.receivedRemoteException ? this.faultingException : null;
+            }
+
             /// <summary>
             /// Apply channel options to this channel, including setting up or linking to an user-supplied pipe writer/reader pair.
             /// </summary>
@@ -832,13 +874,22 @@ private async Task ProcessOutboundTransmissionsAsync()
                 }
                 catch (Exception ex)
                 {
+                    // If the operation had been cancelled then we are expecting to receive this error so don't transmit it.
                     if (ex is OperationCanceledException && this.DisposalToken.IsCancellationRequested)
                     {
                         await mxStreamIOReader!.CompleteAsync().ConfigureAwait(false);
                     }
                     else
                     {
+                        // If not record it as the error to dispose this channel with
+                        lock (this.SyncObject)
+                        {
+                            this.faultingException = ex;
+                        }
+
+                        // Since we're not expecting to receive this error, transmit the error to the remote side.
                         await mxStreamIOReader!.CompleteAsync(ex).ConfigureAwait(false);
+                        this.MultiplexingStream.OnChannelWritingError(this, ex);
                     }
 
                     throw;
@@ -916,11 +967,7 @@ private async Task AutoCloseOnPipesClosureAsync()
 
             private void Fault(Exception exception)
             {
-                if (this.TraceSource?.Switch.ShouldTrace(TraceEventType.Critical) ?? false)
-                {
-                    this.TraceSource!.TraceEvent(TraceEventType.Critical, (int)TraceEventId.FatalError, "Channel Closing self due to exception: {0}", exception);
-                }
-
+                // Record the faulting exception unless it is not the original exception.
                 lock (this.SyncObject)
                 {
                     this.faultingException ??= exception;

src/Nerdbank.Streams/MultiplexingStream.ChannelOptions.cs

@@ -50,6 +50,10 @@ public ChannelOptions()
             /// The <see cref="PipeWriter"/> specified in <see cref="IDuplexPipe.Output"/> *must* be created with <see cref="PipeOptions.PauseWriterThreshold"/> that *exceeds*
             /// the value of <see cref="ChannelReceivingWindowSize"/> and <see cref="Options.DefaultChannelReceivingWindowSize"/>.
             /// </para>
+            /// <para>
+            /// A faulted <see cref="IDuplexPipe.Input" /> (one where <see cref="PipeReader.CompleteAsync(Exception)" /> is called with an exception)
+            /// will have its exception relayed to the remote party before closing the channel.
+            /// </para>
             /// </remarks>
             /// <exception cref="ArgumentException">Thrown if set to an <see cref="IDuplexPipe"/> that returns <c>null</c> for either of its properties.</exception>
             public IDuplexPipe? ExistingPipe

src/Nerdbank.Streams/MultiplexingStream.ControlCode.cs

@@ -44,6 +44,12 @@ internal enum ControlCode : byte
             /// allowing them to send more data.
             /// </summary>
             ContentProcessed,
+
+            /// <summary>
+            /// Sent when one party experiences an exception related to a particular channel and carries details regarding the error.
+            /// This is sent right before a <see cref="ContentWritingCompleted"/> frame closes that channel.
+            /// </summary>
+            ContentWritingError,
         }
     }
 }

src/Nerdbank.Streams/MultiplexingStream.Formatters.cs

@@ -495,6 +495,57 @@ internal override Channel.AcceptanceParameters DeserializeAcceptanceParameters(R
                 return new Channel.AcceptanceParameters(remoteWindowSize);
             }
 
+            /// <summary>
+            /// Returns the serialized representation of a <see cref="WriteError"/> object using <see cref="MessagePack"/>.
+            /// </summary>
+            /// <param name="protocolVersion">The protocol version to include in the serialized error buffer.</param>
+            /// <param name="error">An instance of <see cref="WriteError"/> that we want to seralize.</param>
+            /// <returns>A <see cref="Sequence{T}"/> which is the serialized version of the error.</returns>
+            internal ReadOnlySequence<byte> SerializeWriteError(int protocolVersion, WriteError error)
+            {
+                // Create the payload
+                using Sequence<byte> errorSequence = new(ArrayPool<byte>.Shared);
+                MessagePackWriter writer = new(errorSequence);
+
+                // Write the error message and the protocol version to the payload
+                writer.WriteArrayHeader(2);
+                writer.WriteInt32(protocolVersion);
+                writer.Write(error.ErrorMessage);
+
+                // Return the payload to the caller
+                writer.Flush();
+                return errorSequence.AsReadOnlySequence;
+            }
+
+            /// <summary>
+            /// Extracts an <see cref="WriteError"/> object from the payload using <see cref="MessagePack"/>.
+            /// </summary>
+            /// <param name="serializedError">The payload we are trying to extract the error object from.</param>
+            /// <param name="expectedVersion">The protocol version we expect to be associated with the error object.</param>
+            /// <returns>A <see cref="WriteError"/> object if the payload is correctly formatted and has the expected protocol version,
+            ///          null otherwise. </returns>
+            internal WriteError? DeserializeWriteError(ReadOnlySequence<byte> serializedError, int expectedVersion)
+            {
+                MessagePackReader reader = new(serializedError);
+
+                // The payload should only have the error message and the protocol version.
+                if (reader.ReadArrayHeader() != 2)
+                {
+                    return null;
+                }
+
+                // Verify that the protocol version of the payload matches our expected value
+                int senderVersion = reader.ReadInt32();
+                if (senderVersion != expectedVersion)
+                {
+                    return null;
+                }
+
+                // Extract the error message and use that to create the write error object
+                string errorMessage = reader.ReadString();
+                return new WriteError(errorMessage);
+            }
+
             protected virtual (FrameHeader Header, ReadOnlySequence<byte> Payload) DeserializeFrame(ReadOnlySequence<byte> frameSequence)
             {
                 var reader = new MessagePackReader(frameSequence);

src/Nerdbank.Streams/MultiplexingStream.WriteError.cs

@@ -0,0 +1,32 @@
+// Copyright (c) Andrew Arnott. All rights reserved.
+// Licensed under the MIT license. See LICENSE file in the project root for full license information.
+
+namespace Nerdbank.Streams
+{
+    /// <content>
+    /// Contains the <see cref="WriteError"/> nested type.
+    /// </content>
+    public partial class MultiplexingStream
+    {
+        /// <summary>
+        /// A class containing information about a write error and which is sent to the
+        /// remote alongside <see cref="MultiplexingStream.ControlCode.ContentWritingError"/>.
+        /// </summary>
+        internal class WriteError
+        {
+            /// <summary>
+            /// Initializes a new instance of the <see cref="WriteError"/> class.
+            /// </summary>
+            /// <param name="message">The error message we want to send to the receiver.</param>
+            internal WriteError(string message)
+            {
+                this.ErrorMessage = message;
+            }
+
+            /// <summary>
+            /// Gets the error message associated with this error.
+            /// </summary>
+            internal string ErrorMessage { get; }
+        }
+    }
+}