[Service Bus] Fix credit management based on the buffer capacity for the receivers

sdk/servicebus/service-bus/CHANGELOG.md

@@ -11,7 +11,8 @@
 
 ### Bug fixes
 
-- [Bug Fix] `expiresAtUtc` is `Invalid Date` in the received message when the ttl is not defined. Has been fixed in [#13543](https://github.com/Azure/azure-sdk-for-js/pull/13543)
+- If there are many outstanding messages that are not settled, `receiveMessages` hangs forever and doesn't return. [#14060](https://github.com/Azure/azure-sdk-for-js/pull/14060) fixes the problem by throwing an error with code `UnsettledMessagesLimitExceeded` for receiveMessages. If used `subscribe`, `processError` will be notified upon exceeding the limit.
+- `expiresAtUtc` is `Invalid Date` in the received message when the ttl is not defined. Has been fixed in [#13543](https://github.com/Azure/azure-sdk-for-js/pull/13543)
 - Some of the queue properties such as "forwardTo" and "autoDeleteOnIdle" were not being set as requested through the `ServiceBusAdministrationClient.createQueue` method because of a bug w.r.t the ordering of XML properties. The issue has been fixed in [#14692](https://github.com/Azure/azure-sdk-for-js/pull/14692).
 - Settling messages now use the `retryOptions` passed to `ServiceBusClient`, making it more resilient against network failures.
   [PR#14867](https://github.com/Azure/azure-sdk-for-js/pull/14867/files)

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

@@ -370,7 +370,11 @@ export type ServiceBusErrorCode =
 /**
  * The user doesn't have access to the entity.
  */
- | "UnauthorizedAccess";
+ | "UnauthorizedAccess"
+/**
+ * The number of unsettled messages that can be held is reached, please settle the received messages.
+ */
+ | "UnsettledMessagesLimitExceeded";
 
 // @public
 export interface ServiceBusMessage {

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

@@ -21,6 +21,7 @@ import { OperationOptionsBase } from "../modelsToBeSharedWithEventHubs";
 import { createAndEndProcessingSpan } from "../diagnostics/instrumentServiceBusMessage";
 import { ReceiveMode } from "../models";
 import { ServiceBusError, translateServiceBusError } from "../serviceBusError";
+import { numberOfEmptyIncomingSlots, UnsettledMessagesLimitExceededError } from "./shared";
 
 /**
  * Describes the batching receiver where the user can receive a specified number of messages for
@@ -195,7 +196,10 @@ type EventEmitterLike<T extends Receiver | Session> = Pick<T, "once" | "removeLi
  *
  * @internal
  */
-export type MinimalReceiver = Pick<Receiver, "name" | "isOpen" | "credit" | "addCredit" | "drain"> &
+export type MinimalReceiver = Pick<
+  Receiver,
+  "name" | "isOpen" | "credit" | "addCredit" | "drain" | "session"
+> &
   EventEmitterLike<Receiver> & {
     session: EventEmitterLike<Session>;
   } & {
@@ -493,15 +497,24 @@ export class BatchingReceiverLite {
       reject(err);
     }, args.abortSignal);
 
-    logger.verbose(
-      `${loggingPrefix} Adding credit for receiving ${args.maxMessageCount} messages.`
-    );
+    const emptySlots = numberOfEmptyIncomingSlots(receiver);
+    if (emptySlots <= 1) {
+      throw new ServiceBusError(
+        UnsettledMessagesLimitExceededError,
+        "UnsettledMessagesLimitExceeded"
+      );
+    }
+    const creditsToAdd =
+      this._receiveMode === "peekLock"
+        ? Math.min(args.maxMessageCount, emptySlots <= 1 ? 0 : emptySlots - 1)
+        : args.maxMessageCount;
+    logger.verbose(`${loggingPrefix} Adding credit for receiving ${creditsToAdd} messages.`);
 
-    // By adding credit here, we let the service know that at max we can handle `maxMessageCount`
+    // By adding credit here, we let the service know that at max we can handle `creditsToAdd`
     // number of messages concurrently. We will return the user an array of messages that can
-    // be of size upto maxMessageCount. Then the user needs to accordingly dispose
+    // be of size upto `creditsToAdd`. Then the user needs to accordingly dispose
     // (complete/abandon/defer/deadletter) the messages from the array.
-    receiver.addCredit(args.maxMessageCount);
+    receiver.addCredit(creditsToAdd);
 
     logger.verbose(
       `${loggingPrefix} Setting the wait timer for ${args.maxWaitTimeInMs} milliseconds.`

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

@@ -2,9 +2,13 @@
 // Licensed under the MIT license.
 
 import { Delivery, ReceiverOptions, Source } from "rhea-promise";
-import { translateServiceBusError } from "../serviceBusError";
+import { ServiceBusError, translateServiceBusError } from "../serviceBusError";
 import { receiverLogger } from "../log";
 import { ReceiveMode } from "../models";
+import { Receiver } from "rhea-promise";
+import { OnError } from "./messageReceiver";
+import { ReceiverHelper } from "./receiverHelper";
+import { delay } from "@azure/core-amqp";
 
 /**
  * @internal
@@ -103,3 +107,99 @@ export function createReceiverOptions(
 
   return rcvrOptions;
 }
+
+export const UnsettledMessagesLimitExceededError =
+  "Failed to fetch new messages as the limit for unsettled messages is reached. Please settle received messages using settlement methods(such as `completeMessage()`) on the receiver to receive the next message.";
+
+/**
+ * Returns the number of empty slots in the Circular buffer of incoming deliveries
+ * based on the capacity and size of the buffer.
+ *
+ * @internal
+ */
+export function numberOfEmptyIncomingSlots(
+  receiver: Pick<Receiver, "session"> | undefined
+): number {
+  const incomingDeliveries = receiver?.session?.incoming?.deliveries;
+  return incomingDeliveries ? incomingDeliveries.capacity - incomingDeliveries.size : 0;
+}
+
+/**
+ * Provides helper methods to manage the credits on the link for the
+ * streaming messages scenarios.
+ * (Used by both sessions(MessageSession) and non-sessions(StreamingReceiver))
+ *
+ * @internal
+ */
+export class StreamingReceiverCreditManager {
+  constructor(
+    private _getCurrentReceiver: () => { receiver: Receiver | undefined; logPrefix: string },
+    private receiverHelper: ReceiverHelper,
+    private receiveMode: ReceiveMode,
+    private entityPath: string,
+    private fullyQualifiedNamespace: string
+  ) {}
+
+  addCreditsInit(maxConcurrentCalls: number) {
+    const emptySlots = numberOfEmptyIncomingSlots(this._getCurrentReceiver().receiver);
+    this.receiverHelper.addCredit(
+      this.receiveMode === "peekLock"
+        ? Math.min(maxConcurrentCalls, emptySlots <= 1 ? 0 : emptySlots - 1)
+        : maxConcurrentCalls
+    );
+    // TODO: Add log message
+  }
+  /**
+   * Upon receiving a new message, this method can be called to add a credit to receive one more message.
+   * If no empty slots, calls the onError callback with the `UnsettledMessagesLimitExceeded` error to
+   * let users know about the excess unsettled messages.
+   *
+   * @internal
+   */
+  onReceive(notifyError: OnError | undefined) {
+    const receiver = this._getCurrentReceiver().receiver;
+    if (this.receiveMode === "receiveAndDelete" || numberOfEmptyIncomingSlots(receiver) > 1) {
+      this.receiverHelper.addCredit(1);
+    } else if (receiver) {
+      notifyError?.({
+        error: new ServiceBusError(
+          UnsettledMessagesLimitExceededError,
+          "UnsettledMessagesLimitExceeded"
+        ),
+        errorSource: "receive",
+        entityPath: this.entityPath,
+        fullyQualifiedNamespace: this.fullyQualifiedNamespace
+      });
+    } else {
+      // Link doesn't exist
+      // SessionLockLost for sessions/onAMQPError for non-sessions will be notified in one of the listeners
+      // So, not notifying here
+      // TODO: Validate above
+    }
+  }
+
+  /**
+   * After processing the message, if no empty slots,
+   * - keeps checking if the link has more empty slots in a loop with a delay of 1 sec,
+   * - adds a credit if there are empty slots.
+   *
+   * @internal
+   */
+  async postProcessing() {
+    const receiver = this._getCurrentReceiver().receiver;
+    if (this.receiveMode === "peekLock" && numberOfEmptyIncomingSlots(receiver) <= 1) {
+      // Wait for the user to clear the deliveries before adding more credits
+      while (receiver?.isOpen() && numberOfEmptyIncomingSlots(receiver) <= 1) {
+        // TODO: check for canReceiveMessages too to exit from the loop
+        await delay(1000); // TODO: Not have hard-coded 1000ms as delay - move it to constants maybe
+        // TODO: Add jitter
+      }
+      // TODO: Instead of adding one credit, make it maxConcurrentCalls
+      // Example:
+      //   - Suppose maxConcurrentCalls=1000 and the user is not settling the messages
+      //   - After 2048 messages, 1000 while loops are running(for all the processMessage callbacks) so that they can do their part of adding a credit
+      //   - Instead, replenish with maxConcurrentCalls with a single while loop and end the rest of the processMessage callbacks
+      this.receiverHelper.addCredit(1);
+    }
+  }
+}

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

@@ -28,7 +28,7 @@ import { ServiceBusMessageImpl } from "../serviceBusMessage";
 import { AbortSignalLike } from "@azure/abort-controller";
 import { translateServiceBusError } from "../serviceBusError";
 import { abandonMessage, completeMessage } from "../receivers/receiverCommon";
-import { ReceiverHandlers } from "./shared";
+import { ReceiverHandlers, StreamingReceiverCreditManager } from "./shared";
 
 /**
  * @internal
@@ -95,6 +95,11 @@ export class StreamingReceiver extends MessageReceiver {
    */
   private _onAmqpError: OnAmqpEvent;
 
+  /**
+   * Provides helper methods to allow adding credits during initialization, on receiving a message, and after processing a message.
+   */
+  private _creditManager: StreamingReceiverCreditManager;
+
   /**
    * The message handler that will be set as the handler on the
    * underlying rhea receiver for the "message" event.
@@ -132,6 +137,17 @@ export class StreamingReceiver extends MessageReceiver {
       logPrefix: this.logPrefix
     }));
 
+    this._creditManager = new StreamingReceiverCreditManager(
+      () => ({
+        receiver: this.link,
+        logPrefix: this.logPrefix
+      }),
+      this._receiverHelper,
+      this.receiveMode,
+      this.entityPath,
+      this._context.config.host
+    );
+
     this._onAmqpClose = async (context: EventContext) => {
       const receiverError = context.receiver && context.receiver.error;
       const receiver = this.link || context.receiver!;
@@ -300,7 +316,7 @@ export class StreamingReceiver extends MessageReceiver {
         }
         return;
       } finally {
-        this._receiverHelper.addCredit(1);
+        this._creditManager.onReceive(this._onError);
       }
 
       // If we've made it this far, then user's message handler completed fine. Let us try
@@ -335,6 +351,8 @@ export class StreamingReceiver extends MessageReceiver {
           });
         }
       }
+
+      await this._creditManager.postProcessing();
     };
   }
 
@@ -440,15 +458,14 @@ export class StreamingReceiver extends MessageReceiver {
    * Starts the receiver by establishing an AMQP session and an AMQP receiver link on the session.
    *
    * @param onMessage - The message handler to receive servicebus messages.
-   * @param onError - The error handler to receive an error that occurs while receivin messages.
+   * @param onError - The error handler to receive an error that occurs while receiving messages.
    */
   subscribe(onMessage: OnMessage, onError: OnError): void {
     throwErrorIfConnectionClosed(this._context);
 
     this._onMessage = onMessage;
     this._onError = onError;
-
-    this._receiverHelper.addCredit(this.maxConcurrentCalls);
+    this._creditManager.addCreditsInit(this.maxConcurrentCalls);
   }
 
   /**
@@ -510,10 +527,10 @@ export class StreamingReceiver extends MessageReceiver {
         onError: (args) => this._onError && this._onError(args)
       });
 
-      this._receiverHelper.addCredit(this.maxConcurrentCalls);
       logger.verbose(
-        `${this.logPrefix} onDetached: link has been reestablished, added ${this.maxConcurrentCalls} credits.`
+        `${this.logPrefix} onDetached: link has been reestablished, attempting to add credits.`
       );
+      this._creditManager.addCreditsInit(this.maxConcurrentCalls);
     } finally {
       this._isDetaching = false;
     }

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

@@ -65,7 +65,11 @@ export type ServiceBusErrorCode =
   /**
    * The user doesn't have access to the entity.
    */
-  | "UnauthorizedAccess";
+  | "UnauthorizedAccess"
+  /**
+   * The number of unsettled messages that can be held is reached, please settle the received messages.
+   */
+  | "UnsettledMessagesLimitExceeded";
 
 /**
  * Translation between the MessagingErrorCodes into a ServiceBusCode

sdk/servicebus/service-bus/src/session/messageSession.ts

@@ -25,7 +25,12 @@ import { DispositionType, ServiceBusMessageImpl } from "../serviceBusMessage";
 import { throwErrorIfConnectionClosed } from "../util/errors";
 import { calculateRenewAfterDuration, convertTicksToDate } from "../util/utils";
 import { BatchingReceiverLite, MinimalReceiver } from "../core/batchingReceiver";
-import { onMessageSettled, DeferredPromiseAndTimer, createReceiverOptions } from "../core/shared";
+import {
+  onMessageSettled,
+  DeferredPromiseAndTimer,
+  createReceiverOptions,
+  StreamingReceiverCreditManager
+} from "../core/shared";
 import { AbortError, AbortSignalLike } from "@azure/abort-controller";
 import { ReceiverHelper } from "../core/receiverHelper";
 import {
@@ -607,6 +612,16 @@ export class MessageSession extends LinkEntity<Receiver> {
     this._onError = onError;
 
     if (this.link && this.link.isOpen()) {
+      const creditManager = new StreamingReceiverCreditManager(
+        () => ({
+          receiver: this.link,
+          logPrefix: this.logPrefix
+        }),
+        this._receiverHelper,
+        this.receiveMode,
+        this.entityPath,
+        this._context.config.host
+      );
       const onSessionMessage = async (context: EventContext): Promise<void> => {
         // If the receiver got closed in PeekLock mode, avoid processing the message as we
         // cannot settle the message.
@@ -684,7 +699,7 @@ export class MessageSession extends LinkEntity<Receiver> {
           }
           return;
         } finally {
-          this.receiverHelper.addCredit(1);
+          creditManager.onReceive(this._notifyError);
         }
 
         // If we've made it this far, then user's message handler completed fine. Let us try
@@ -717,11 +732,12 @@ export class MessageSession extends LinkEntity<Receiver> {
             });
           }
         }
+        await creditManager.postProcessing();
       };
       // setting the "message" event listener.
       this.link.on(ReceiverEvents.message, onSessionMessage);
       // adding credit
-      this.receiverHelper.addCredit(this.maxConcurrentCalls);
+      creditManager.addCreditsInit(this.maxConcurrentCalls);
     } else {
       this._isReceivingMessagesForSubscriber = false;
       const msg =