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

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 } from "../receivers/shared";
 
 /**
  * Describes the batching receiver where the user can receive a specified number of messages for
@@ -490,15 +491,17 @@ export class BatchingReceiverLite {
       reject(err);
     }, args.abortSignal);
 
-    logger.verbose(
-      `${loggingPrefix} Adding credit for receiving ${args.maxMessageCount} messages.`
-    );
+    const creditsToAdd =
+      this._receiveMode === "peekLock"
+        ? Math.min(args.maxMessageCount, numberOfEmptyIncomingSlots(receiver) - 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/streamingReceiver.ts

@@ -26,8 +26,8 @@ import { receiverLogger as logger } from "../log";
 import { AmqpError, EventContext, OnAmqpEvent } from "rhea-promise";
 import { ServiceBusMessageImpl } from "../serviceBusMessage";
 import { AbortSignalLike } from "@azure/abort-controller";
-import { translateServiceBusError } from "../serviceBusError";
-import { abandonMessage, completeMessage } from "../receivers/shared";
+import { ServiceBusError, translateServiceBusError } from "../serviceBusError";
+import { abandonMessage, completeMessage, numberOfEmptyIncomingSlots } from "../receivers/shared";
 import { ReceiverHandlers } from "./shared";
 
 /**
@@ -294,7 +294,24 @@ export class StreamingReceiver extends MessageReceiver {
         }
         return;
       } finally {
-        this._receiverHelper.addCredit(1);
+        if (this.receiveMode === "receiveAndDelete") {
+          this._receiverHelper.addCredit(1);
+        } else if (numberOfEmptyIncomingSlots(this.link) - 1 > 1) {
+          this._receiverHelper.addCredit(1);
+        } else {
+          // Additionally.. have a checkWithTimeout that keeps checking if the above if-condition satisfies
+          // If it ever satisfies - add the credit
+          this._onError?.({
+            error: new ServiceBusError(
+              `Circular buffer that contains the incoming deliveries is full, please settle the messages using settlement methods such as .completeMessage() on the receiver.
+                Or set the "autoComplete" flag to true to let the library complete the messages on your behalf.`,
+              "GeneralError"
+            ),
+            errorSource: "receive",
+            entityPath: this.entityPath,
+            fullyQualifiedNamespace: this._context.config.host
+          });
+        }
       }
 
       // If we've made it this far, then user's message handler completed fine. Let us try
@@ -434,15 +451,19 @@ 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._receiverHelper.addCredit(
+      this.receiveMode === "peekLock"
+        ? Math.min(this.maxConcurrentCalls, numberOfEmptyIncomingSlots(this.link) - 1)
+        : this.maxConcurrentCalls
+    );
   }
 
   /**
@@ -503,10 +524,13 @@ export class StreamingReceiver extends MessageReceiver {
         connectionId: this._context.connectionId,
         onError: (args) => this._onError && this._onError(args)
       });
-
-      this._receiverHelper.addCredit(this.maxConcurrentCalls);
+      const creditsToAdd =
+        this.receiveMode === "peekLock"
+          ? Math.min(this.maxConcurrentCalls, numberOfEmptyIncomingSlots(this.link) - 1)
+          : this.maxConcurrentCalls;
+      this._receiverHelper.addCredit(creditsToAdd);
       logger.verbose(
-        `${this.logPrefix} onDetached: link has been reestablished, added ${this.maxConcurrentCalls} credits.`
+        `${this.logPrefix} onDetached: link has been reestablished, added ${creditsToAdd} credits.`
       );
     } finally {
       this._isDetaching = false;

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

@@ -226,3 +226,10 @@ function settleMessage(
     throw translateServiceBusError(err);
   });
 }
+
+export function numberOfEmptyIncomingSlots(receiver: any): number {
+  return (
+    (receiver as any).session._session.incoming.deliveries.capacity -
+    (receiver as any).session._session.incoming.deliveries.size
+  );
+}

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

@@ -33,8 +33,8 @@ import {
   SubscribeOptions
 } from "../models";
 import { OperationOptionsBase } from "../modelsToBeSharedWithEventHubs";
-import { translateServiceBusError } from "../serviceBusError";
-import { abandonMessage, completeMessage } from "../receivers/shared";
+import { ServiceBusError, translateServiceBusError } from "../serviceBusError";
+import { abandonMessage, completeMessage, numberOfEmptyIncomingSlots } from "../receivers/shared";
 
 /**
  * Describes the options that need to be provided while creating a message session receiver link.
@@ -672,7 +672,24 @@ export class MessageSession extends LinkEntity<Receiver> {
           }
           return;
         } finally {
-          this.receiverHelper.addCredit(1);
+          if (this.receiveMode === "receiveAndDelete") {
+            this._receiverHelper.addCredit(1);
+          } else if (numberOfEmptyIncomingSlots(this.link) - 1 > 1) {
+            this._receiverHelper.addCredit(1);
+          } else {
+            // Additionally.. have a checkWithTimeout that keeps checking if the above if-condition satisfies
+            // If it ever satisfies - add the credit
+            this._notifyError?.({
+              error: new ServiceBusError(
+                `Circular buffer that contains the incoming deliveries is full, please settle the messages using settlement methods such as .completeMessage() on the receiver.
+                Or set the "autoComplete" flag to true to let the library complete the messages on your behalf.`,
+                "GeneralError"
+              ),
+              errorSource: "receive",
+              entityPath: this.entityPath,
+              fullyQualifiedNamespace: this._context.config.host
+            });
+          }
         }
 
         // If we've made it this far, then user's message handler completed fine. Let us try
@@ -709,7 +726,11 @@ export class MessageSession extends LinkEntity<Receiver> {
       // setting the "message" event listener.
       this.link.on(ReceiverEvents.message, onSessionMessage);
       // adding credit
-      this.receiverHelper.addCredit(this.maxConcurrentCalls);
+      const creditsToAdd = Math.min(
+        this.maxConcurrentCalls,
+        numberOfEmptyIncomingSlots(this.link) - 1
+      );
+      this._receiverHelper.addCredit(creditsToAdd);
     } else {
       this._isReceivingMessagesForSubscriber = false;
       const msg =