MultiplexingStream transmit errors when shutting down channels

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,13 @@ 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,
+            /// when using protocol version 2 or later.
+            /// This is sent before a <see cref="ContentWritingCompleted"/> frame closes that channel.
+            /// </summary>
+            ContentWritingError,
         }
     }
 }

src/Nerdbank.Streams/MultiplexingStream.Formatters.cs

@@ -5,6 +5,7 @@ namespace Nerdbank.Streams
 {
     using System;
     using System.Buffers;
+    using System.Diagnostics;
     using System.IO;
     using System.IO.Pipelines;
     using System.Threading;
@@ -495,6 +496,53 @@ 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="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(WriteError error)
+            {
+                // Create the payload
+                Sequence<byte> errorSequence = new();
+                MessagePackWriter writer = new(errorSequence);
+
+                // Write the error message and the protocol version to the payload
+                writer.WriteArrayHeader(1);
+                writer.Write(error.Message);
+
+                // 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>
+            /// <returns>A <see cref="WriteError"/> object.</returns>
+            internal WriteError DeserializeWriteError(ReadOnlySequence<byte> serializedError)
+            {
+                MessagePackReader reader = new(serializedError);
+                int numElements = reader.ReadArrayHeader();
+
+                string? errorMessage = null;
+                for (int i = 0; i < numElements; i++)
+                {
+                    switch (i)
+                    {
+                        case 0:
+                            errorMessage = reader.ReadString();
+                            break;
+                        default:
+                            reader.Skip();
+                            break;
+                    }
+                }
+
+                return new WriteError(errorMessage ?? string.Empty);
+            }
+
             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.Message = message;
+            }
+
+            /// <summary>
+            /// Gets the error message associated with this error.
+            /// </summary>
+            internal string? Message { get; }
+        }
+    }
+}

src/Nerdbank.Streams/MultiplexingStream.cs

@@ -209,6 +209,16 @@ private enum TraceEventId
             /// Raised when the protocol handshake is starting, to annouce the major version being used.
             /// </summary>
             HandshakeStarted,
+
+            /// <summary>
+            /// Raised when receiving or sending a <see cref="ControlCode.ContentWritingError"/>.
+            /// </summary>
+            WriteError,
+
+            /// <summary>
+            /// An error occurred that is likely not fatal.
+            /// </summary>
+            NonFatalInternalError,
         }
 
         /// <summary>
@@ -220,7 +230,7 @@ private enum TraceEventId
         /// Gets the logger used by this instance.
         /// </summary>
         /// <value>Never null.</value>
-        public TraceSource TraceSource { get; }
+        public TraceSource TraceSource { get; private set; }
 
         /// <summary>
         /// Gets the default window size used for new channels that do not specify a value for <see cref="ChannelOptions.ChannelReceivingWindowSize"/>.
@@ -244,6 +254,11 @@ private enum TraceEventId
         /// </summary>
         private Func<QualifiedChannelId, string, TraceSource?>? DefaultChannelTraceSourceFactory { get; }
 
+        /// <summary>
+        ///  Gets a value indicating whether this stream can send or receive frames of type<see cref="ControlCode.ContentWritingError"/>.
+        /// </summary>
+        private bool ContentWritingErrorSupported => this.protocolMajorVersion > 1;
+
         /// <summary>
         /// Initializes a new instance of the <see cref="MultiplexingStream"/> class
         /// with <see cref="Options.ProtocolMajorVersion"/> set to 3.
@@ -837,6 +852,9 @@ private async Task ReadStreamAsync()
                         case ControlCode.ChannelTerminated:
                             await this.OnChannelTerminatedAsync(header.RequiredChannelId).ConfigureAwait(false);
                             break;
+                        case ControlCode.ContentWritingError:
+                            this.OnContentWritingError(header.RequiredChannelId, frame.Value.Payload);
+                            break;
                         default:
                             break;
                     }
@@ -919,6 +937,56 @@ private async Task OnChannelTerminatedAsync(QualifiedChannelId channelId)
             }
         }
 
+        /// <summary>
+        /// Occurs when the channel receives a frame with code <see cref="ControlCode.ContentWritingError"/> from the remote.
+        /// </summary>
+        /// <param name="channelId">The channel id of the sender of the frame.</param>
+        /// <param name="payload">The payload that the sender sent in the frame.</param>
+        private void OnContentWritingError(QualifiedChannelId channelId, ReadOnlySequence<byte> payload)
+        {
+            // Get the channel that send this frame
+            Channel channel;
+            lock (this.syncObject)
+            {
+                channel = this.openChannels[channelId];
+            }
+
+            // Determines if the channel is in a state to receive messages
+            bool channelInValidState = channelId.Source != ChannelSource.Local || channel.IsAccepted;
+
+            // If the channel is in a valid state and we have a valid protocol version, then process the message
+            if (channelInValidState && this.ContentWritingErrorSupported)
+            {
+                // Deserialize the payload and verify that it was in an expected state
+                V2Formatter errorDeserializingFormattter = (V2Formatter)this.formatter;
+                WriteError error = errorDeserializingFormattter.DeserializeWriteError(payload);
+
+                // Get the error message and complete the channel using it
+                string errorMessage = error.Message ?? "<unspecified>";
+                MultiplexingProtocolException channelClosingException = new MultiplexingProtocolException($"Remote party indicated writing error: {errorMessage}");
+                channel.OnContentWritingCompleted(channelClosingException);
+            }
+            else if (channelInValidState && !this.ContentWritingErrorSupported)
+            {
+                // The channel is in a valid state but we have a protocol version that doesn't support processing errrors
+                // so don't do anything.
+                if (this.TraceSource?.Switch.ShouldTrace(TraceEventType.Warning) ?? false)
+                {
+                    this.TraceSource.TraceEvent(
+                        TraceEventType.Warning,
+                        (int)TraceEventId.WriteError,
+                        "Rejecting writing error from channel {0} as MultiplexingStream has protocol version of {1}",
+                        channelId,
+                        this.protocolMajorVersion);
+                }
+            }
+            else
+            {
+                // The channel is in an invalid state so throw an error indicating so
+                throw new MultiplexingProtocolException($"Remote party indicated they encountered errors writing to channel {channelId} before accepting it.");
+            }
+        }
+
         private void OnContentWritingCompleted(QualifiedChannelId channelId)
         {
             Channel channel;
@@ -1064,12 +1132,7 @@ private bool TryAcceptChannel(Channel channel, ChannelOptions options)
             Requires.NotNull(channel, nameof(channel));
             Requires.NotNull(options, nameof(options));
 
-            if (channel.TryAcceptOffer(options))
-            {
-                return true;
-            }
-
-            return false;
+            return channel.TryAcceptOffer(options);
         }
 
         private void AcceptChannelOrThrow(Channel channel, ChannelOptions options)
@@ -1079,7 +1142,12 @@ private void AcceptChannelOrThrow(Channel channel, ChannelOptions options)
 
             if (!this.TryAcceptChannel(channel, options))
             {
-                if (channel.IsAccepted)
+                // If we disposed of the channel due to a user provided error then ignore the error.
+                if (channel.IsDisposed && (channel.Completion.IsFaulted || channel.Acceptance.IsFaulted))
+                {
+                    return;
+                }
+                else if (channel.IsAccepted)
                 {
                     throw new InvalidOperationException("Channel is already accepted.");
                 }
@@ -1113,6 +1181,50 @@ private void OnChannelDisposed(Channel channel)
             }
         }
 
+        /// <summary>
+        /// Informs the remote party of a local error that prevents sending all the required data to this channel
+        /// by transmitting a <see cref="ControlCode.ContentWritingError"/> frame.
+        /// </summary>
+        /// <param name="channel">The channel whose writing was halted.</param>
+        /// <param name="exception">The exception that caused the writing to be haulted.</param>
+        private void OnChannelWritingError(Channel channel, Exception exception)
+        {
+            if (this.TraceSource.Switch.ShouldTrace(TraceEventType.Error))
+            {
+                this.TraceSource.TraceEvent(TraceEventType.Error, (int)TraceEventId.WriteError, "Local channel {0} encountered write error {1}", channel.QualifiedId, exception.Message);
+            }
+
+            // Verify that we can send a message over this channel.
+            // The race condition here is handled within SendFrameAsync which will drop the frame
+            // if the conditions we're checking for here change after we check them, so our check here
+            // is just an optimization to avoid work when we can predict its failure.
+            bool channelInValidState = true;
+            lock (this.syncObject)
+            {
+                channelInValidState = !this.channelsPendingTermination.Contains(channel.QualifiedId)
+                    && this.openChannels.ContainsKey(channel.QualifiedId);
+            }
+
+            // If we can send messages over this channel and we have the correct protocol version then send the error
+            if (channelInValidState && this.ContentWritingErrorSupported)
+            {
+                // Create the payload to send to the remote side
+                V2Formatter errorSerializationFormatter = (V2Formatter)this.formatter;
+                WriteError error = new(exception.Message);
+                ReadOnlySequence<byte> serializedError = errorSerializationFormatter.SerializeWriteError(error);
+
+                // Create the frame header indicating that we encountered a content writing error
+                FrameHeader header = new FrameHeader
+                {
+                    Code = ControlCode.ContentWritingError,
+                    ChannelId = channel.QualifiedId,
+                };
+
+                // Send the frame alongside the payload to the remote side
+                this.SendFrame(header, serializedError, CancellationToken.None);
+            }
+        }
+
         /// <summary>
         /// Indicates that the local end will not be writing any more data to this channel,
         /// leading to the transmission of a <see cref="ControlCode.ContentWritingCompleted"/> frame being sent for this channel.
@@ -1191,7 +1303,7 @@ private async Task SendFrameAsync(FrameHeader header, ReadOnlySequence<byte> pay
                             // In such cases, we should just suppress transmission of the frame because the other side does not care.
                             // ContentWritingCompleted can be sent to SendFrame after a ChannelTerminated message such that neither have been transmitted yet
                             // and thus wasn't in the termination collection until later, so forgive that too.
-                            if (header.Code is ControlCode.ContentProcessed or ControlCode.ContentWritingCompleted)
+                            if (header.Code is ControlCode.ContentProcessed or ControlCode.ContentWritingCompleted or ControlCode.ContentWritingError)
                             {
                                 this.TraceSource.TraceEvent(TraceEventType.Information, (int)TraceEventId.FrameSendSkipped, "Skipping {0} frame for channel {1} because we're about to terminate it.", header.Code, header.ChannelId);
                                 return;

src/nerdbank-streams/src/Channel.ts

@@ -53,6 +53,15 @@ export abstract class Channel implements IDisposableObservable {
         return this._isDisposed;
     }
 
+    /**
+     * Closes this channel after transmitting an error to the remote party.
+     * @param error The error to transmit to the remote party.
+     */
+    public fault(error: Error): Promise<void> {
+        // The interesting stuff is in the derived class.
+        return Promise.resolve();
+    }
+
     /**
      * Closes this channel.
      */
@@ -71,6 +80,7 @@ export class ChannelClass extends Channel {
     private readonly _completion = new Deferred<void>();
     public localWindowSize?: number;
     private remoteWindowSize?: number;
+    private remoteError?: Error;
 
     /**
      * The number of bytes transmitted from here but not yet acknowledged as processed from there,
@@ -214,7 +224,13 @@ export class ChannelClass extends Channel {
         return this._acceptance.resolve();
     }
 
-    public onContent(buffer: Buffer | null) {
+    public onContent(buffer: Buffer | null, error?: Error) {
+        // If we have already received an error from the remote party, then don't process any future messages.
+        if (this.remoteError) {
+            return;
+        }
+
+        this.remoteError = error;
         this._duplex.push(buffer);
 
         // We should find a way to detect when we *actually* share the received buffer with the Channel's user
@@ -244,18 +260,43 @@ export class ChannelClass extends Channel {
         }
     }
 
+    public async fault(error: Error) {
+        // If the channel is already disposed then don't do anything
+        if (this.isDisposed) {
+            return;
+        }
+
+        // Send the error message to the remote side
+        await this._multiplexingStream.onChannelWritingError(this, error);
+
+        // Set the remote exception to the passed in error so that the channel is
+        // completed with this error
+        this.remoteError = error;
+
+        // Dispose of the channel
+        await this.dispose();
+    }
+
     public async dispose() {
         if (!this.isDisposed) {
             super.dispose();
 
             this._acceptance.reject(new CancellationToken.CancellationError("disposed"));
 
             // For the pipes, we Complete *our* ends, and leave the user's ends alone.
-            // The completion will propagate when it's ready to.
+            // The completion will propagate when it's ready to. No need to destroy the duplex
+            // as the frame containing the error message has already been sent.
             this._duplex.end();
             this._duplex.push(null);
 
-            this._completion.resolve();
+            // If we are sending an error to the remote side or received an error from the remote,
+            // relay that information to the clients.
+            if (this.remoteError) {
+                this._completion.reject(this.remoteError);
+            } else {
+                this._completion.resolve();
+            }
+
             await this._multiplexingStream.onChannelDisposed(this);
         }
     }