[ServiceBus] Pass timeout to server when accepting next session

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

@@ -77,6 +77,8 @@ export function onMessageSettled(
   }
 }
 
+// Placed in Service Bus for now and can be promoted to core-amqp if also useful for Event Hubs in the future.
+const timeoutName = `${Constants.vendorString}:timeout`;
 /**
  * Creates the options that need to be specified while creating an AMQP receiver link.
  *
@@ -87,8 +89,13 @@ export function createReceiverOptions(
   receiveMode: ReceiveMode,
   source: Source,
   clientId: string,
-  handlers: ReceiverHandlers
+  handlers: ReceiverHandlers,
+  timeoutInMs?: number
 ): ReceiverOptions {
+  const properties =
+    timeoutInMs !== undefined
+      ? { [Constants.receiverIdentifierName]: clientId, [timeoutName]: timeoutInMs }
+      : { [Constants.receiverIdentifierName]: clientId };
   const rcvrOptions: ReceiverOptions = {
     name,
     // "autoaccept" being true in the "receiveAndDelete" mode sets the "settled" flag to true on the deliveries
@@ -101,7 +108,7 @@ export function createReceiverOptions(
     source,
     target: clientId,
     credit_window: 0,
-    properties: { [Constants.receiverIdentifierName]: clientId },
+    properties,
     ...handlers,
   };
 

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

@@ -1,7 +1,7 @@
 // Copyright (c) Microsoft Corporation.
 // Licensed under the MIT license.
 
-import { ConnectionConfig } from "@azure/core-amqp";
+import { ConnectionConfig, Constants } from "@azure/core-amqp";
 import { TokenCredential, NamedKeyCredential, SASCredential } from "@azure/core-auth";
 import {
   ServiceBusClientOptions,
@@ -438,6 +438,19 @@ export class ServiceBusClient {
       options3
     );
 
+    const operationTimeout =
+      this._clientOptions?.retryOptions?.timeoutInMs ?? Constants.defaultOperationTimeoutInMs;
+    // The number of milliseconds to use as the basis for calculating a random jitter amount
+    // opening receiver links. This is intended to ensure that multiple
+    // session operations don't timeout at the same exact moment.
+    const jitterBaseInMs = Math.min(operationTimeout * 0.01, 100);
+    // Subtract a random amount up to 100ms from the operation timeout as the jitter when attempting to open next available session link.
+    // This prevents excessive resource usage when using high amounts of concurrency and accepting the next available session.
+    // Take the min of 1% of the total timeout and the base jitter amount so that we don't end up subtracting more than 1% of the total timeout.
+    const timeoutWithJitterInMs = operationTimeout - jitterBaseInMs * Math.random();
+    // We set the operation timeout on the properties not only to include the jitter, but also because the server will otherwise
+    // restrict the maximum timeout to 1 minute and 5 seconds, regardless of the client timeout. We only do this for accepting next available
+    // session as this is the only long-polling scenario.
     const messageSession = await MessageSession.create(
       ensureValidIdentifier(entityPath, options?.identifier),
       this._connectionContext,
@@ -447,7 +460,7 @@ export class ServiceBusClient {
         maxAutoLockRenewalDurationInMs: options?.maxAutoLockRenewalDurationInMs,
         receiveMode,
         abortSignal: options?.abortSignal,
-        retryOptions: this._clientOptions.retryOptions,
+        retryOptions: { ...this._clientOptions.retryOptions, timeoutInMs: timeoutWithJitterInMs },
         skipParsingBodyAsJson: options?.skipParsingBodyAsJson ?? false,
       }
     );

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

@@ -258,10 +258,12 @@ export class MessageSession extends LinkEntity<Receiver> {
   /**
    * Creates a new AMQP receiver under a new AMQP session.
    */
-  private async _init(abortSignal?: AbortSignalLike): Promise<void> {
+  private async _init(
+    opts: { abortSignal?: AbortSignalLike; timeoutInMs?: number } = {}
+  ): Promise<void> {
     try {
-      const options = this._createMessageSessionOptions(this.identifier);
-      await this.initLink(options, abortSignal);
+      const sessionOptions = this._createMessageSessionOptions(this.identifier, opts.timeoutInMs);
+      await this.initLink(sessionOptions, opts.abortSignal);
 
       if (this.link == null) {
         throw new Error("INTERNAL ERROR: failed to create receiver but without an error.");
@@ -302,7 +304,11 @@ export class MessageSession extends LinkEntity<Receiver> {
         this.sessionId,
         this.sessionLockedUntilUtc.toISOString()
       );
-      logger.verbose("%s Receiver created with receiver options: %O", this.logPrefix, options);
+      logger.verbose(
+        "%s Receiver created with receiver options: %O",
+        this.logPrefix,
+        sessionOptions
+      );
       if (!this._context.messageSessions[this.name]) {
         this._context.messageSessions[this.name] = this;
       }
@@ -327,7 +333,7 @@ export class MessageSession extends LinkEntity<Receiver> {
   /**
    * Creates the options that need to be specified while creating an AMQP receiver link.
    */
-  private _createMessageSessionOptions(clientId: string): ReceiverOptions {
+  private _createMessageSessionOptions(clientId: string, timeoutInMs?: number): ReceiverOptions {
     const rcvrOptions: ReceiverOptions = createReceiverOptions(
       this.name,
       this.receiveMode,
@@ -348,7 +354,8 @@ export class MessageSession extends LinkEntity<Receiver> {
         onError: this._onAmqpError,
         onSessionError: this._onSessionError,
         onSettled: this._onSettled,
-      }
+      },
+      timeoutInMs
     );
 
     return rcvrOptions;
@@ -943,7 +950,10 @@ export class MessageSession extends LinkEntity<Receiver> {
   ): Promise<MessageSession> {
     throwErrorIfConnectionClosed(context);
     const messageSession = new MessageSession(identifier, context, entityPath, sessionId, options);
-    await messageSession._init(options?.abortSignal);
+    await messageSession._init({
+      abortSignal: options?.abortSignal,
+      timeoutInMs: options.retryOptions?.timeoutInMs,
+    });
     return messageSession;
   }