[service-bus] Pass in ProcessErrorArgs to subscribe({ processError }) for extra diagnostics

sdk/servicebus/service-bus/README.md

@@ -192,9 +192,13 @@ When you are done, call `receiver.close()` to stop receiving any more messages.
 ```javascript
 const myMessageHandler = async (message) => {
   // your code here
+  console.log(`message.body: ${message.body}`);
 };
-const myErrorHandler = async (error) => {
-  console.log(error);
+const myErrorHandler = async (error, context) => {
+  console.log(
+    `Error occurred with ${context.entityPath} within ${context.fullyQualifiedNamespace}: `,
+    error
+  );
 };
 receiver.subscribe({
   processMessage: myMessageHandler,

sdk/servicebus/service-bus/review/service-bus.api.md

@@ -193,7 +193,7 @@ export interface GetMessageIteratorOptions extends OperationOptionsBase {
 
 // @public
 export interface MessageHandlers<ReceivedMessageT> {
-    processError(err: Error): Promise<void>;
+    processError(err: Error, context: ProcessErrorContext): Promise<void>;
     processMessage(message: ReceivedMessageT): Promise<void>;
 }
 
@@ -223,6 +223,44 @@ export interface PeekMessagesOptions extends OperationOptionsBase {
     fromSequenceNumber?: Long;
 }
 
+// @public
+export interface ProcessErrorContext {
+    entityPath: string;
+    errorSource: "complete"
+    /**
+     * Covers errors that occur when the user (or autoComplete) abandons a message.
+     */
+     | "abandon"
+    /**
+     * Errors thrown from the user's `processMessage` callback passed to `subscribe`
+     */
+     | "userCallback"
+    /**
+     * Errors thrown when receiving messages.
+     */
+     | "receive"
+    /**
+     * Errors thrown when the user calls renewLock or when the internal lock renewer calls renewLock.
+     * Automatic lock renewal can be enabled via the `CreateReceiverOptions.maxAutoLockRenewalDurationInMs`
+     * property passed when calling `ServiceBusClient.createReceiver()`
+     */
+     | "renewLock"
+    /**
+     * Broadly covers a series of phases - the actual initialization that occurs when an operation
+     * is first started as well as errors that occur when reconnecting.
+     */
+     | "initialize"
+    /**
+     * Errors thrown when accepting a session.
+     */
+     | "acceptSession"
+    /**
+     * Errors thrown when closing a session.
+     */
+     | "closeSession";
+    fullyQualifiedNamespace: string;
+}
+
 // @public
 export interface QueueProperties {
     authorizationRules?: AuthorizationRule[];

sdk/servicebus/service-bus/src/core/autoLockRenewer.ts

@@ -6,7 +6,7 @@ import { receiverLogger as logger } from "../log";
 import { ServiceBusMessageImpl } from "../serviceBusMessage";
 import { calculateRenewAfterDuration } from "../util/utils";
 import { LinkEntity } from "./linkEntity";
-import { OnError } from "./messageReceiver";
+import { OnErrorNoContext } from "./messageReceiver";
 
 /**
  * @internal
@@ -118,7 +118,7 @@ export class LockRenewer {
    *
    * @param bMessage The message whose lock renewal we will start.
    */
-  start(linkEntity: MinimalLink, bMessage: RenewableMessageProperties, onError: OnError) {
+  start(linkEntity: MinimalLink, bMessage: RenewableMessageProperties, onError: OnErrorNoContext) {
     try {
       const logPrefix = linkEntity.logPrefix;
 

sdk/servicebus/service-bus/src/core/messageReceiver.ts

@@ -14,7 +14,7 @@ import { LinkEntity, ReceiverType } from "./linkEntity";
 import { ConnectionContext } from "../connectionContext";
 import { DispositionType, ServiceBusMessageImpl } from "../serviceBusMessage";
 import { getUniqueName } from "../util/utils";
-import { ReceiveMode, SubscribeOptions } from "../models";
+import { ProcessErrorContext, ReceiveMode, SubscribeOptions } from "../models";
 import { DispositionStatusOptions } from "./managementClient";
 import { AbortSignalLike } from "@azure/core-http";
 import { onMessageSettled, DeferredPromiseAndTimer } from "./shared";
@@ -80,7 +80,21 @@ export interface OnMessage {
 export interface OnError {
   /**
    * Handler for any error that occurs while receiving or processing messages.
+   *
+   * NOTE: if this signature changes make sure you reflect those same changes in the
+   * `OnErrorNoContext` definition below.
    */
+  (error: MessagingError | Error, context: ProcessErrorContext): void;
+}
+
+/**
+ * An onError method but without the context property. Used when wrapping OnError
+ * with an implicit ProcessErrorContext. Used by LockRenewer.
+ *
+ * @internal
+ * @ignore
+ */
+export interface OnErrorNoContext {
   (error: MessagingError | Error): void;
 }
 

sdk/servicebus/service-bus/src/core/streamingReceiver.ts

@@ -199,7 +199,11 @@ export class StreamingReceiver extends MessageReceiver {
                 "retryable, we let the user know about it by calling the user's error handler.",
               this.logPrefix
             );
-            this._onError!(sbError);
+            this._onError!(sbError, {
+              errorSource: "receive",
+              entityPath: this.entityPath,
+              fullyQualifiedNamespace: this._context.config.host
+            });
           } else {
             logger.verbose(
               "%s The received error is not retryable. However, the receiver was " +
@@ -234,7 +238,11 @@ export class StreamingReceiver extends MessageReceiver {
               "retryable, we let the user know about it by calling the user's error handler.",
             this.logPrefix
           );
-          this._onError!(sbError);
+          this._onError!(sbError, {
+            errorSource: "receive",
+            entityPath: this.entityPath,
+            fullyQualifiedNamespace: this._context.config.host
+          });
         }
       }
     };
@@ -262,7 +270,11 @@ export class StreamingReceiver extends MessageReceiver {
 
       this._lockRenewer?.start(this, bMessage, (err) => {
         if (this._onError) {
-          this._onError(err);
+          this._onError(err, {
+            errorSource: "renewLock",
+            entityPath: this.entityPath,
+            fullyQualifiedNamespace: this._context.config.host
+          });
         }
       });
 
@@ -279,7 +291,11 @@ export class StreamingReceiver extends MessageReceiver {
             bMessage.messageId,
             this.name
           );
-          this._onError!(err);
+          this._onError!(err, {
+            errorSource: "userCallback",
+            entityPath: this.entityPath,
+            fullyQualifiedNamespace: this._context.config.host
+          });
         }
 
         // Do not want renewLock to happen unnecessarily, while abandoning the message. Hence,
@@ -314,7 +330,11 @@ export class StreamingReceiver extends MessageReceiver {
               bMessage.messageId,
               this.name
             );
-            this._onError!(translatedError);
+            this._onError!(translatedError, {
+              errorSource: "abandon",
+              entityPath: this.entityPath,
+              fullyQualifiedNamespace: this._context.config.host
+            });
           }
         }
         return;
@@ -346,7 +366,11 @@ export class StreamingReceiver extends MessageReceiver {
             bMessage.messageId,
             this.name
           );
-          this._onError!(translatedError);
+          this._onError!(translatedError, {
+            errorSource: "complete",
+            entityPath: this.entityPath,
+            fullyQualifiedNamespace: this._context.config.host
+          });
         }
       }
     };
@@ -513,7 +537,11 @@ export class StreamingReceiver extends MessageReceiver {
       if (typeof this._onError === "function") {
         logger.verbose(`${this.logPrefix} Unable to automatically reconnect`);
         try {
-          this._onError(err);
+          this._onError(err, {
+            errorSource: "initialize",
+            entityPath: this.entityPath,
+            fullyQualifiedNamespace: this._context.config.host
+          });
         } catch (err) {
           logger.logError(
             err,

sdk/servicebus/service-bus/src/index.ts

@@ -10,7 +10,7 @@ export {
   RetryOptions,
   TokenCredential,
   TokenType,
-  WebSocketOptions
+  WebSocketOptions,
 } from "@azure/core-amqp";
 export { OperationOptions } from "@azure/core-http";
 export { Delivery, WebSocketImpl } from "rhea-promise";
@@ -22,11 +22,12 @@ export {
   AcceptSessionOptions,
   GetMessageIteratorOptions,
   MessageHandlers,
+  ProcessErrorContext,
   PeekMessagesOptions,
   ReceiveMessagesOptions,
   ReceiveMode,
   SubQueue,
-  SubscribeOptions
+  SubscribeOptions,
 } from "./models";
 export { OperationOptionsBase, TryAddOptions } from "./modelsToBeSharedWithEventHubs";
 export { ServiceBusReceiver } from "./receivers/receiver";
@@ -36,18 +37,18 @@ export { NamespaceProperties } from "./serializers/namespaceResourceSerializer";
 export {
   CreateQueueOptions,
   QueueProperties,
-  QueueRuntimeProperties
+  QueueRuntimeProperties,
 } from "./serializers/queueResourceSerializer";
 export { RuleProperties, SqlRuleAction, SqlRuleFilter } from "./serializers/ruleResourceSerializer";
 export {
   CreateSubscriptionOptions,
   SubscriptionProperties,
-  SubscriptionRuntimeProperties
+  SubscriptionRuntimeProperties,
 } from "./serializers/subscriptionResourceSerializer";
 export {
   CreateTopicOptions,
   TopicProperties,
-  TopicRuntimeProperties
+  TopicRuntimeProperties,
 } from "./serializers/topicResourceSerializer";
 export {
   EntitiesResponse,
@@ -60,7 +61,7 @@ export {
   SubscriptionResponse,
   SubscriptionRuntimePropertiesResponse,
   TopicResponse,
-  TopicRuntimePropertiesResponse
+  TopicRuntimePropertiesResponse,
 } from "./serviceBusAtomManagementClient";
 export { ServiceBusClient } from "./serviceBusClient";
 export {
@@ -70,7 +71,7 @@ export {
   DeadLetterOptions,
   ServiceBusReceivedMessage,
   ServiceBusReceivedMessageWithLock,
-  ServiceBusMessage
+  ServiceBusMessage,
 } from "./serviceBusMessage";
 export { ServiceBusMessageBatch } from "./serviceBusMessageBatch";
 export { AuthorizationRule, EntityStatus, EntityAvailabilityStatus } from "./util/utils";

sdk/servicebus/service-bus/src/models.ts

@@ -4,6 +4,59 @@
 import { OperationOptionsBase } from "./modelsToBeSharedWithEventHubs";
 import Long from "long";
 
+/**
+ * Context for the error given in processError.
+ */
+export interface ProcessErrorContext {
+  /**
+   * The operation where the error originated.
+   */
+  errorSource: /**
+   * Covers errors that occur when the user (or autoComplete) completes a message.
+   */
+  | "complete"
+    /**
+     * Covers errors that occur when the user (or autoComplete) abandons a message.
+     */
+    | "abandon"
+    /**
+     * Errors thrown from the user's `processMessage` callback passed to `subscribe`
+     */
+    | "userCallback"
+    /**
+     * Errors thrown when receiving messages.
+     */
+    | "receive"
+    /**
+     * Errors thrown when the user calls renewLock or when the internal lock renewer calls renewLock.
+     * Automatic lock renewal can be enabled via the `CreateReceiverOptions.maxAutoLockRenewalDurationInMs`
+     * property passed when calling `ServiceBusClient.createReceiver()`
+     */
+    | "renewLock"
+    /**
+     * Broadly covers a series of phases - the actual initialization that occurs when an operation
+     * is first started as well as errors that occur when reconnecting.
+     */
+    | "initialize"
+    /**
+     * Errors thrown when accepting a session.
+     */
+    | "acceptSession"
+    /**
+     * Errors thrown when closing a session.
+     */
+    | "closeSession";
+
+  /**
+   * The entity path for the current receiver.
+   */
+  entityPath: string;
+  /**
+   * The fully qualified namespace for the Service Bus.
+   */
+  fullyQualifiedNamespace: string;
+}
+
 /**
  * The general message handler interface (used for streamMessages).
  */
@@ -18,7 +71,7 @@ export interface MessageHandlers<ReceivedMessageT> {
    * Handler that processes errors that occur during receiving.
    * @param err An error from Service Bus.
    */
-  processError(err: Error): Promise<void>;
+  processError(err: Error, context: ProcessErrorContext): Promise<void>;
 }
 
 /**

sdk/servicebus/service-bus/src/receivers/receiver.ts

@@ -10,7 +10,7 @@ import {
   InternalMessageHandlers
 } from "../models";
 import { OperationOptionsBase, trace } from "../modelsToBeSharedWithEventHubs";
-import { ServiceBusReceivedMessage } from "..";
+import { ServiceBusReceivedMessage } from "../serviceBusMessage";
 import { ConnectionContext } from "../connectionContext";
 import {
   getAlreadyReceivingErrorMsg,
@@ -262,7 +262,11 @@ export class ServiceBusReceiverImpl<
         try {
           await onInitialize();
         } catch (err) {
-          onError(err);
+          onError(err, {
+            errorSource: "userCallback",
+            entityPath: this.entityPath,
+            fullyQualifiedNamespace: this._context.config.host
+          });
         }
 
         if (!this.isClosed) {
@@ -275,7 +279,13 @@ export class ServiceBusReceiverImpl<
         return;
       })
       .catch((err) => {
-        onError(err);
+        // TODO: being a bit broad here but the only errors that should filter out this
+        // far are going to be bootstrapping the subscription.
+        onError(err, {
+          errorSource: "initialize",
+          entityPath: this.entityPath,
+          fullyQualifiedNamespace: this._context.config.host
+        });
       });
   }
 

sdk/servicebus/service-bus/src/receivers/sessionReceiver.ts

@@ -472,7 +472,11 @@ export class ServiceBusSessionReceiverImpl<
     try {
       this._messageSession.subscribe(onMessage, onError, options);
     } catch (err) {
-      onError(err);
+      onError(err, {
+        errorSource: "initialize",
+        entityPath: this.entityPath,
+        fullyQualifiedNamespace: this._context.config.host
+      });
     }
   }