[ServiceBus] wait and throw error if sender link is not sendable (Azure#26927)

We already have the logic in `MessageSender._trySend()`. This PR adds
the same to `ManagementClient._makeManagementRequest()` so that when the
link is not sendable, we wait for one second then check again. A
`SendBusyError` is thrown if we still cannot send messages.

### Packages impacted by this PR
`@azure/service-bus`

### Issues associated with this PR

Azure#26855

Author: jeremymeng

sdk/servicebus/service-bus/CHANGELOG.md

@@ -10,6 +10,8 @@
 
 ### Other Changes
 
+- Check whether we can send messages before making management requests. [PR #26927](https://github.com/Azure/azure-sdk-for-js/pull/26927)
+
 ## 7.9.0 (2023-04-11)
 
 ### Bugs Fixed

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

@@ -37,7 +37,7 @@ import {
 } from "../serviceBusMessage";
 import { LinkEntity, RequestResponseLinkOptions } from "./linkEntity";
 import { managementClientLogger, receiverLogger, senderLogger, ServiceBusLogger } from "../log";
-import { toBuffer } from "../util/utils";
+import { toBuffer, waitForSendable } from "../util/utils";
 import {
   InvalidMaxMessageCountError,
   throwErrorIfConnectionClosed,
@@ -257,12 +257,12 @@ export class ManagementClient extends LinkEntity<RequestResponseLink> {
       this._initLock,
       async () => {
         managementClientLogger.verbose(
-          `{this._logPrefix} lock acquired for initializing replyTo address and link`
+          `${this.logPrefix} lock acquired for initializing replyTo address and link`
         );
         if (!this.isOpen()) {
           this.replyTo = generate_uuid();
           managementClientLogger.verbose(
-            `{this._logPrefix} new replyTo address: ${this.replyTo} generated`
+            `${this.logPrefix} new replyTo address: ${this.replyTo} generated`
           );
         }
         const { abortSignal } = options ?? {};
@@ -403,6 +403,16 @@ export class ManagementClient extends LinkEntity<RequestResponseLink> {
     }
 
     try {
+      const { timeoutInMs } = sendRequestOptions;
+      await waitForSendable(
+        internalLogger,
+        this.logPrefix,
+        this.name,
+        timeoutInMs ?? Constants.defaultOperationTimeoutInMs,
+        this.link?.sender,
+        this.link?.session?.outgoing?.available()
+      );
+
       return await this.link!.sendRequest(request, sendRequestOptions);
     } catch (err: any) {
       const translatedError = translateServiceBusError(err);

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

@@ -17,14 +17,13 @@ import {
   RetryConfig,
   RetryOperationType,
   RetryOptions,
-  delay,
   retry,
   AmqpAnnotatedMessage,
 } from "@azure/core-amqp";
 import { ServiceBusMessage, toRheaMessage } from "../serviceBusMessage";
 import { ConnectionContext } from "../connectionContext";
 import { LinkEntity } from "./linkEntity";
-import { getUniqueName, waitForTimeoutOrAbortOrResolve } from "../util/utils";
+import { getUniqueName, waitForSendable, waitForTimeoutOrAbortOrResolve } from "../util/utils";
 import { throwErrorIfConnectionClosed } from "../util/errors";
 import { ServiceBusMessageBatch, ServiceBusMessageBatchImpl } from "../serviceBusMessageBatch";
 import { CreateMessageBatchOptions } from "../models";
@@ -205,41 +204,16 @@ export class MessageSender extends LinkEntity<AwaitableSender> {
         this.link?.session?.outgoing?.available()
       );
 
-      let waitTimeForSendable = 1000;
-      if (!this.link?.sendable() && timeoutInMs - timeTakenByInit > waitTimeForSendable) {
-        logger.verbose(
-          "%s Sender '%s', waiting for 1 second for sender to become sendable",
-          this.logPrefix,
-          this.name
-        );
-
-        await delay(waitTimeForSendable);
-
-        logger.verbose(
-          "%s Sender '%s' after waiting for a second, credit: %d available: %d",
-          this.logPrefix,
-          this.name,
-          this.link?.credit,
-          this.link?.session?.outgoing?.available()
-        );
-      } else {
-        waitTimeForSendable = 0;
-      }
-
-      if (!this.link?.sendable()) {
-        // let us retry to send the message after some time.
-        const msg =
-          `[${this.logPrefix}] Sender "${this.name}", ` +
-          `cannot send the message right now. Please try later.`;
-        logger.warning(msg);
-        const amqpError: AmqpError = {
-          condition: ErrorNameConditionMapper.SenderBusyError,
-          description: msg,
-        };
-        throw translateServiceBusError(amqpError);
-      }
+      const waitingTime = await waitForSendable(
+        logger,
+        this.logPrefix,
+        this.name,
+        timeoutInMs - timeTakenByInit,
+        this.link,
+        this.link?.session?.outgoing?.available()
+      );
 
-      if (timeoutInMs <= timeTakenByInit + waitTimeForSendable) {
+      if (timeoutInMs <= timeTakenByInit + waitingTime) {
         const desc: string =
           `${this.logPrefix} Sender "${this.name}" ` +
           `with address "${this.address}", was not able to send the message right now, due ` +
@@ -255,7 +229,7 @@ export class MessageSender extends LinkEntity<AwaitableSender> {
       try {
         const delivery = await this.link!.send(encodedMessage, {
           format: sendBatch ? 0x80013700 : 0,
-          timeoutInSeconds: (timeoutInMs - timeTakenByInit - waitTimeForSendable) / 1000,
+          timeoutInSeconds: (timeoutInMs - timeTakenByInit - waitingTime) / 1000,
           abortSignal,
         });
         logger.verbose(

sdk/servicebus/service-bus/src/util/utils.ts

@@ -2,16 +2,17 @@
 // Licensed under the MIT license.
 
 import Long from "long";
-import { logger, receiverLogger, messageLogger } from "../log";
-import { OperationTimeoutError, generate_uuid } from "rhea-promise";
+import { logger, receiverLogger, messageLogger, ServiceBusLogger } from "../log";
+import { AmqpError, OperationTimeoutError, generate_uuid } from "rhea-promise";
 import isBuffer from "is-buffer";
 import { Buffer } from "buffer";
 import * as Constants from "../util/constants";
 import { AbortError, AbortSignalLike } from "@azure/abort-controller";
 import { PipelineResponse } from "@azure/core-rest-pipeline";
 import { isDefined } from "@azure/core-util";
 import { HttpResponse, toHttpResponse } from "./compat";
-import { StandardAbortMessage } from "@azure/core-amqp";
+import { ErrorNameConditionMapper, StandardAbortMessage, delay } from "@azure/core-amqp";
+import { translateServiceBusError } from "../serviceBusError";
 
 // This is the only dependency we have on DOM types, so rather than require
 // the DOM lib we can just shim this in.
@@ -654,3 +655,58 @@ export const getHttpResponseOnly = (pipelineResponse: PipelineResponse): HttpRes
  * Type with the service versions for the ATOM API.
  */
 export type ServiceBusAtomAPIVersion = "2021-05" | "2017-04";
+
+/**
+ * @internal
+ * Waits for one second if a sender is not sendable then check again. Throws
+ * SenderBusyError if it is still not sendable.
+ * Only waits when operation timeout is greater than one second.
+ * @returns the actual waiting time.
+ */
+export async function waitForSendable(
+  sendLogger: ServiceBusLogger,
+  logPrefix: string,
+  name: string,
+  timeout: number,
+  sender:
+    | {
+        sendable: () => boolean;
+        credit: number;
+      }
+    | undefined,
+  outgoingAvaiable: number
+): Promise<number> {
+  let waitTimeForSendable = 1000;
+  if (!sender?.sendable() && timeout > waitTimeForSendable) {
+    sendLogger.verbose(
+      "%s Sender '%s', waiting for 1 second for sender to become sendable",
+      logPrefix,
+      name
+    );
+
+    await delay(waitTimeForSendable);
+
+    sendLogger.verbose(
+      "%s Sender '%s' after waiting for a second, credit: %d available: %d",
+      logPrefix,
+      name,
+      sender?.credit,
+      outgoingAvaiable
+    );
+  } else {
+    waitTimeForSendable = 0;
+  }
+
+  if (!sender?.sendable()) {
+    // let us retry to send the message after some time.
+    const msg =
+      `[${logPrefix}] Sender "${name}", ` + `cannot send the message right now. Please try later.`;
+    sendLogger.warning(msg);
+    const amqpError: AmqpError = {
+      condition: ErrorNameConditionMapper.SenderBusyError,
+      description: msg,
+    };
+    throw translateServiceBusError(amqpError);
+  }
+  return waitTimeForSendable;
+}