[Service Bus Client] Detect Closed Shared Connection (Azure#20541)

The focus of these changes is to guard against a shared connection being
closed when the `ServiceBusClient` that spawned a sender/receiver/processor
is.

Previously, the sender and receiver did not validate the connection status
and would trigger exception behavior in the AMQP library, where the type of
exception could vary depending on the library's internal state. This scenario
will now trigger a consistent `ObjectDisposedException` with an error message
stating that the connection was closed.

The processor would trigger the AMQP exception and fail to recognize it as
fatal; signaling the customer error handler and continuing to retry forever,
though it cannot recover.  The exception surfaced did not provide customers
enough context to differentiate between the fatal scenario and one that was
recoverable.

The processor will now detect the connection being closed and interpret that
 as fatal, signaling the customer exception handler with an `ObjectDisposedException`
 with explicit message indicating an unrecoverable scenario and will then stop processing.

Author: jsquire

sdk/servicebus/Azure.Messaging.ServiceBus/src/Administration/ServiceBusAdministrationClient.cs

@@ -3,7 +3,6 @@
 
 using System;
 using System.Globalization;
-using System.IO;
 using System.Threading;
 using System.Threading.Tasks;
 using Azure.Core;

sdk/servicebus/Azure.Messaging.ServiceBus/src/Amqp/AmqpConnectionScope.cs

@@ -37,6 +37,9 @@ internal class AmqpConnectionScope : TransportConnectionScope
         /// <summary>The URI scheme to apply when using web sockets for service communication.</summary>
         private const string WebSocketsUriScheme = "wss";
 
+        /// <summary>Indicates whether or not this instance has been disposed.</summary>
+        private volatile bool _disposed;
+
         /// <summary>
         ///   The version of AMQP to use within the scope.
         /// </summary>
@@ -83,7 +86,12 @@ internal class AmqpConnectionScope : TransportConnectionScope
         /// </summary>
         ///
         /// <value><c>true</c> if disposed; otherwise, <c>false</c>.</value>
-        public override bool IsDisposed { get; protected set; }
+        ///
+        public override bool IsDisposed
+        {
+            get => _disposed;
+            protected set => _disposed = value;
+        }
 
         /// <summary>
         ///   The cancellation token to use with operations initiated by the scope.
@@ -254,6 +262,7 @@ public virtual async Task<RequestResponseAmqpLink> OpenManagementLinkAsync(
             ServiceBusEventSource.Log.CreateManagementLinkStart(identifier);
             try
             {
+                Argument.AssertNotDisposed(_disposed, nameof(AmqpConnectionScope));
                 cancellationToken.ThrowIfCancellationRequested<TaskCanceledException>();
 
                 var stopWatch = ValueStopwatch.StartNew();
@@ -303,6 +312,7 @@ public virtual async Task<ReceivingAmqpLink> OpenReceiverLinkAsync(
             bool isSessionReceiver,
             CancellationToken cancellationToken)
         {
+            Argument.AssertNotDisposed(_disposed, nameof(AmqpConnectionScope));
             cancellationToken.ThrowIfCancellationRequested<TaskCanceledException>();
 
             var stopWatch = ValueStopwatch.StartNew();
@@ -347,6 +357,7 @@ public virtual async Task<SendingAmqpLink> OpenSenderLinkAsync(
             TimeSpan timeout,
             CancellationToken cancellationToken)
         {
+            Argument.AssertNotDisposed(_disposed, nameof(AmqpConnectionScope));
             cancellationToken.ThrowIfCancellationRequested<TaskCanceledException>();
             var stopWatch = ValueStopwatch.StartNew();
 

sdk/servicebus/Azure.Messaging.ServiceBus/src/Diagnostics/ServiceBusEventSource.cs

@@ -190,6 +190,8 @@ internal ServiceBusEventSource() { }
         internal const int RequestAuthorizationCompleteEvent = 106;
         internal const int RequestAuthorizationExceptionEvent = 107;
 
+        internal const int ProcessorClientClosedExceptionEvent = 108;
+
         #endregion
         // add new event numbers here incrementing from previous
 
@@ -778,6 +780,15 @@ public void ProcessorMessageHandlerException(string identifier, long sequenceNum
             }
         }
 
+        [Event(ProcessorClientClosedExceptionEvent, Level = EventLevel.Error, Message = "{0}: The Service Bus client associated with the processor was closed by the host application.  The processor cannot continue and is shutting down.")]
+        public void ProcessorClientClosedException(string identifier)
+        {
+            if (IsEnabled())
+            {
+                WriteEvent(ProcessorClientClosedExceptionEvent, identifier);
+            }
+        }
+
         #endregion region
 
         #region Rule management

sdk/servicebus/Azure.Messaging.ServiceBus/src/Processor/ReceiverManager.cs

@@ -77,7 +77,7 @@ public virtual async Task ReceiveAndProcessMessagesAsync(CancellationToken cance
             try
             {
                 // loop within the context of this thread
-                while (!cancellationToken.IsCancellationRequested)
+                while (!cancellationToken.IsCancellationRequested && !Processor.Connection.IsClosed)
                 {
                     errorSource = ServiceBusErrorSource.Receive;
                     ServiceBusReceivedMessage message = await Receiver.ReceiveMessageAsync(

sdk/servicebus/Azure.Messaging.ServiceBus/src/Processor/ServiceBusProcessor.cs

@@ -434,7 +434,7 @@ internal event Func<ProcessSessionEventArgs, Task> SessionClosingAsync
 
         /// <summary>
         /// Invokes the process message event handler after a message has been received.
-        /// This method can be overriden to raise an event manually for testing purposes.
+        /// This method can be overridden to raise an event manually for testing purposes.
         /// </summary>
         /// <param name="args">The event args containing information related to the message.</param>
         protected internal virtual async Task OnProcessMessageAsync(ProcessMessageEventArgs args)
@@ -443,8 +443,8 @@ protected internal virtual async Task OnProcessMessageAsync(ProcessMessageEventA
         }
 
         /// <summary>
-        /// Invokes the error event handler when an error has occured during processing.
-        /// This method can be overriden to raise an event manually for testing purposes.
+        /// Invokes the error event handler when an error has occurred during processing.
+        /// This method can be overridden to raise an event manually for testing purposes.
         /// </summary>
         /// <param name="args">The event args containing information related to the error.</param>
         protected internal async virtual Task OnProcessErrorAsync(ProcessErrorEventArgs args)
@@ -486,6 +486,7 @@ public virtual async Task StartProcessingAsync(
             CancellationToken cancellationToken = default)
         {
             Argument.AssertNotDisposed(IsClosed, nameof(ServiceBusProcessor));
+            Connection.ThrowIfClosed();
             cancellationToken.ThrowIfCancellationRequested<TaskCanceledException>();
             bool releaseGuard = false;
             try
@@ -672,7 +673,7 @@ private async Task RunReceiveTaskAsync(
             List<Task> tasks = new List<Task>(_maxConcurrentCalls + _receiverManagers.Count);
             try
             {
-                while (!cancellationToken.IsCancellationRequested)
+                while (!cancellationToken.IsCancellationRequested && !Connection.IsClosed)
                 {
                     foreach (ReceiverManager receiverManager in _receiverManagers)
                     {
@@ -692,6 +693,42 @@ private async Task RunReceiveTaskAsync(
                         tasks.RemoveAll(t => t.IsCompleted);
                     }
                 }
+
+                // If the main loop is aborting due to the connection being canceled, then
+                // force the processor to stop.
+                if (!cancellationToken.IsCancellationRequested && Connection.IsClosed)
+                {
+                    Logger.ProcessorClientClosedException(Identifier);
+
+                    // Because this is a highly unusual situation
+                    // and goes against the goal of resiliency for the processor, surface
+                    // a fatal exception with an explicit message detailing that processing
+                    // will be stopped.
+                    try
+                    {
+                        await OnProcessErrorAsync(
+                            new ProcessErrorEventArgs(
+                                new ObjectDisposedException(nameof(ServiceBusConnection), Resources.DisposedConnectionMessageProcessorMustStop),
+                                ServiceBusErrorSource.Receive,
+                                FullyQualifiedNamespace,
+                                EntityPath,
+                                cancellationToken))
+                        .ConfigureAwait(false);
+                    }
+                    catch (Exception ex)
+                    {
+                        // Don't bubble up exceptions raised from customer exception handler.
+                        Logger.ProcessorErrorHandlerThrewException(ex.ToString());
+                    }
+
+                    // This call will deadlock if awaited, as StopProcessingAsync awaits
+                    // completion of this task.  The processor is already known to be in a bad
+                    // state and exceptions in StopProcessingAsync are likely and will be logged
+                    // in that call.  There is little value in surfacing those expected exceptions
+                    // to the customer error handler as well; allow StopProcessingAsync to run
+                    // in a fire-and-forget manner.
+                    _ = StopProcessingAsync();
+                }
             }
             finally
             {

sdk/servicebus/Azure.Messaging.ServiceBus/src/Receiver/ServiceBusReceiver.cs

@@ -252,6 +252,8 @@ public virtual async Task<IReadOnlyList<ServiceBusReceivedMessage>> ReceiveMessa
         {
             Argument.AssertAtLeast(maxMessages, 1, nameof(maxMessages));
             Argument.AssertNotDisposed(IsClosed, nameof(ServiceBusReceiver));
+            _connection.ThrowIfClosed();
+
             if (maxWaitTime.HasValue)
             {
                 Argument.AssertPositive(maxWaitTime.Value, nameof(maxWaitTime));
@@ -438,8 +440,9 @@ private async Task<IReadOnlyList<ServiceBusReceivedMessage>> PeekMessagesInterna
             int maxMessages,
             CancellationToken cancellationToken)
         {
-            Argument.AssertNotDisposed(IsClosed, nameof(ServiceBusReceiver));
             Argument.AssertAtLeast(maxMessages, 1, nameof(maxMessages));
+            Argument.AssertNotDisposed(IsClosed, nameof(ServiceBusReceiver));
+            _connection.ThrowIfClosed();
             cancellationToken.ThrowIfCancellationRequested<TaskCanceledException>();
 
             Logger.PeekMessageStart(Identifier, sequenceNumber, maxMessages);
@@ -527,6 +530,7 @@ internal virtual async Task CompleteMessageAsync(
             CancellationToken cancellationToken = default)
         {
             Argument.AssertNotDisposed(IsClosed, nameof(ServiceBusReceiver));
+            _connection.ThrowIfClosed();
             ThrowIfNotPeekLockMode();
             ThrowIfLockTokenIsEmpty(lockToken);
             cancellationToken.ThrowIfCancellationRequested<TaskCanceledException>();
@@ -614,6 +618,7 @@ internal virtual async Task AbandonMessageAsync(
             CancellationToken cancellationToken = default)
         {
             Argument.AssertNotDisposed(IsClosed, nameof(ServiceBusReceiver));
+            _connection.ThrowIfClosed();
             ThrowIfNotPeekLockMode();
             ThrowIfLockTokenIsEmpty(lockToken);
             cancellationToken.ThrowIfCancellationRequested<TaskCanceledException>();
@@ -798,6 +803,7 @@ private async Task DeadLetterInternalAsync(
             CancellationToken cancellationToken = default)
         {
             Argument.AssertNotDisposed(IsClosed, nameof(ServiceBusReceiver));
+            _connection.ThrowIfClosed();
             ThrowIfNotPeekLockMode();
             ThrowIfLockTokenIsEmpty(lockToken);
             cancellationToken.ThrowIfCancellationRequested<TaskCanceledException>();
@@ -891,6 +897,7 @@ internal virtual async Task DeferMessageAsync(
             CancellationToken cancellationToken = default)
         {
             Argument.AssertNotDisposed(IsClosed, nameof(ServiceBusReceiver));
+            _connection.ThrowIfClosed();
             ThrowIfNotPeekLockMode();
             ThrowIfLockTokenIsEmpty(lockToken);
             cancellationToken.ThrowIfCancellationRequested<TaskCanceledException>();
@@ -978,8 +985,9 @@ public virtual async Task<IReadOnlyList<ServiceBusReceivedMessage>> ReceiveDefer
             IEnumerable<long> sequenceNumbers,
             CancellationToken cancellationToken = default)
         {
-            Argument.AssertNotDisposed(IsClosed, nameof(ServiceBusReceiver));
             Argument.AssertNotNull(sequenceNumbers, nameof(sequenceNumbers));
+            Argument.AssertNotDisposed(IsClosed, nameof(ServiceBusReceiver));
+            _connection.ThrowIfClosed();
             cancellationToken.ThrowIfCancellationRequested<TaskCanceledException>();
 
             // the sequence numbers MUST be in array form for them to be encoded correctly
@@ -1067,6 +1075,7 @@ internal virtual async Task<DateTimeOffset> RenewMessageLockAsync(
             CancellationToken cancellationToken = default)
         {
             Argument.AssertNotDisposed(IsClosed, nameof(ServiceBusReceiver));
+            _connection.ThrowIfClosed();
             ThrowIfNotPeekLockMode();
             ThrowIfSessionReceiver();
             ThrowIfLockTokenIsEmpty(lockToken);

sdk/servicebus/Azure.Messaging.ServiceBus/src/Receiver/ServiceBusSessionReceiver.cs

@@ -20,6 +20,13 @@ namespace Azure.Messaging.ServiceBus
     /// </summary>
     public class ServiceBusSessionReceiver : ServiceBusReceiver
     {
+        /// <summary>
+        /// The active connection to the Azure Service Bus service, enabling client communications for metadata
+        /// about the associated Service Bus entity and access to transport-aware receivers.
+        /// </summary>
+        ///
+        private readonly ServiceBusConnection _connection;
+
         /// <summary>
         /// The Session Id associated with the receiver.
         /// </summary>
@@ -89,6 +96,7 @@ internal ServiceBusSessionReceiver(
             string sessionId = default) :
             base(connection, entityPath, true, plugins, options?.ToReceiverOptions(), sessionId, cancellationToken)
         {
+            _connection = connection;
         }
 
         /// <summary>
@@ -111,6 +119,7 @@ protected ServiceBusSessionReceiver() : base() { }
         public virtual async Task<BinaryData> GetSessionStateAsync(CancellationToken cancellationToken = default)
         {
             Argument.AssertNotDisposed(IsClosed, nameof(ServiceBusSessionReceiver));
+            _connection.ThrowIfClosed();
             cancellationToken.ThrowIfCancellationRequested<TaskCanceledException>();
             Logger.GetSessionStateStart(Identifier, SessionId);
             using DiagnosticScope scope = ScopeFactory.CreateScope(
@@ -154,6 +163,7 @@ public virtual async Task SetSessionStateAsync(
             CancellationToken cancellationToken = default)
         {
             Argument.AssertNotDisposed(IsClosed, nameof(ServiceBusSessionReceiver));
+            _connection.ThrowIfClosed();
             cancellationToken.ThrowIfCancellationRequested<TaskCanceledException>();
             Logger.SetSessionStateStart(Identifier, SessionId);
             using DiagnosticScope scope = ScopeFactory.CreateScope(

sdk/servicebus/Azure.Messaging.ServiceBus/src/Resources.Designer.cs

@@ -309,4 +309,7 @@
   <data name="InvalidAmqpMessageValueBody" xml:space="preserve">
     <value>{0} is not a supported value body type.</value>
   </data>
-</root>
+  <data name="DisposedConnectionMessageProcessorMustStop" xml:space="preserve">
+    <value>The message processor is unable to continue and will stop processing; the host application has closed the connection to the Service Bus service.</value>
+  </data>
+</root>

sdk/servicebus/Azure.Messaging.ServiceBus/src/Resources.resx

@@ -4,7 +4,6 @@
 using System;
 using System.Collections.Generic;
 using System.ComponentModel;
-using System.Diagnostics;
 using System.Diagnostics.CodeAnalysis;
 using System.Globalization;
 using System.Linq;
@@ -189,6 +188,8 @@ public virtual async Task SendMessagesAsync(
         {
             Argument.AssertNotNull(messages, nameof(messages));
             Argument.AssertNotDisposed(IsClosed, nameof(ServiceBusSender));
+            _connection.ThrowIfClosed();
+
             IReadOnlyList<ServiceBusMessage> messageList = messages switch
             {
                 IReadOnlyList<ServiceBusMessage> alreadyList => alreadyList,
@@ -299,6 +300,8 @@ public virtual async ValueTask<ServiceBusMessageBatch> CreateMessageBatchAsync(
             CancellationToken cancellationToken = default)
         {
             Argument.AssertNotDisposed(IsClosed, nameof(ServiceBusSender));
+            _connection.ThrowIfClosed();
+
             options = options?.Clone() ?? new CreateMessageBatchOptions();
             cancellationToken.ThrowIfCancellationRequested<TaskCanceledException>();
             Logger.CreateMessageBatchStart(Identifier);
@@ -334,6 +337,8 @@ public virtual async Task SendMessagesAsync(
         {
             Argument.AssertNotNull(messageBatch, nameof(messageBatch));
             Argument.AssertNotDisposed(IsClosed, nameof(ServiceBusSender));
+            _connection.ThrowIfClosed();
+
             if (messageBatch.Count == 0)
             {
                 return;
@@ -419,6 +424,7 @@ public virtual async Task<IReadOnlyList<long>> ScheduleMessagesAsync(
         {
             Argument.AssertNotNull(messages, nameof(messages));
             Argument.AssertNotDisposed(IsClosed, nameof(ServiceBusSender));
+            _connection.ThrowIfClosed();
             cancellationToken.ThrowIfCancellationRequested<TaskCanceledException>();
 
             IReadOnlyList<ServiceBusMessage> messageList = messages switch
@@ -486,8 +492,9 @@ public virtual async Task CancelScheduledMessagesAsync(
             IEnumerable<long> sequenceNumbers,
             CancellationToken cancellationToken = default)
         {
-            Argument.AssertNotDisposed(IsClosed, nameof(ServiceBusSender));
             Argument.AssertNotNull(sequenceNumbers, nameof(sequenceNumbers));
+            Argument.AssertNotDisposed(IsClosed, nameof(ServiceBusSender));
+            _connection.ThrowIfClosed();
             cancellationToken.ThrowIfCancellationRequested<TaskCanceledException>();
 
             // the sequence numbers MUST be in array form for them to be encoded correctly