[core-http] Throttling retry policy fix in core-http (Azure#15832)

Fixes Azure#15796
## Problem

The throttlingRetryPolicy in core-http has the potential to retry for an extended period if the service continues returning "retry after" headers on subsequent calls.

Here's the snippet of code that handles the "retry after" retries:

```typescript
  public async sendRequest(httpRequest: WebResource): Promise<HttpOperationResponse> {
    return this._nextPolicy.sendRequest(httpRequest.clone()).catch((err) => {
        // other code elided....
        return delay(delayInMs).then((_: any) => this.sendRequest(httpRequest.clone()));
```

## Solution
Update delay such that it respects abort signal.
Similar to what I had to do for app-config at Azure#15721

Author: HarshaNalluru

sdk/appconfiguration/app-configuration/src/policies/throttlingRetryPolicy.ts

@@ -1,7 +1,7 @@
 // Copyright (c) Microsoft Corporation.
 // Licensed under the MIT license.
 
-import { AbortError, AbortSignalLike } from "@azure/abort-controller";
+import { AbortError } from "@azure/abort-controller";
 import {
   BaseRequestPolicy,
   RequestPolicy,
@@ -12,7 +12,7 @@ import {
   Constants,
   RestError
 } from "@azure/core-http";
-import { isDefined } from "../internal/typeguards";
+import { delay } from "@azure/core-http";
 
 /**
  * @internal
@@ -27,55 +27,6 @@ export function throttlingRetryPolicy(): RequestPolicyFactory {
 
 const StandardAbortMessage = "The operation was aborted.";
 
-/**
- * A wrapper for setTimeout that resolves a promise after t milliseconds.
- * @param delayInMs - The number of milliseconds to be delayed.
- * @param abortSignal - The abortSignal associated with containing operation.
- * @param abortErrorMsg - The abort error message associated with containing operation.
- * @returns - Resolved promise
- */
-export function delay(
-  delayInMs: number,
-  abortSignal?: AbortSignalLike,
-  abortErrorMsg?: string
-): Promise<void> {
-  return new Promise((resolve, reject) => {
-    let timer: ReturnType<typeof setTimeout> | undefined = undefined;
-    let onAborted: (() => void) | undefined = undefined;
-
-    const rejectOnAbort = (): void => {
-      return reject(new AbortError(abortErrorMsg ? abortErrorMsg : StandardAbortMessage));
-    };
-
-    const removeListeners = (): void => {
-      if (abortSignal && onAborted) {
-        abortSignal.removeEventListener("abort", onAborted);
-      }
-    };
-
-    onAborted = (): void => {
-      if (isDefined(timer)) {
-        clearTimeout(timer);
-      }
-      removeListeners();
-      return rejectOnAbort();
-    };
-
-    if (abortSignal && abortSignal.aborted) {
-      return rejectOnAbort();
-    }
-
-    timer = setTimeout(() => {
-      removeListeners();
-      resolve();
-    }, delayInMs);
-
-    if (abortSignal) {
-      abortSignal.addEventListener("abort", onAborted);
-    }
-  });
-}
-
 /**
  * This policy is a close copy of the ThrottlingRetryPolicy class from
  * core-http with modifications to work with how AppConfig is currently
@@ -97,7 +48,10 @@ export class ThrottlingRetryPolicy extends BaseRequestPolicy {
           throw err;
         }
 
-        await delay(delayInMs, httpRequest.abortSignal, StandardAbortMessage);
+        await delay(delayInMs, undefined, {
+          abortSignal: httpRequest.abortSignal,
+          abortErrorMsg: StandardAbortMessage
+        });
         if (httpRequest.abortSignal?.aborted) {
           throw new AbortError(StandardAbortMessage);
         }

sdk/core/core-http/CHANGELOG.md

@@ -15,6 +15,7 @@
 ### Fixed
 
 - Fixed an issue where `proxySettings` does not work when there is username but no password [Issue 15720](https://github.com/Azure/azure-sdk-for-js/issues/15720)
+- Throttling retry policy respects abort signal [#15796](https://github.com/Azure/azure-sdk-for-js/issues/15796)
 
 ## 1.2.6 (2021-06-14)
 

sdk/core/core-http/review/core-http.api.md

@@ -181,7 +181,10 @@ export class DefaultHttpClient extends FetchHttpClient {
     }
 
 // @public
-export function delay<T>(t: number, value?: T): Promise<T | void>;
+export function delay<T>(delayInMs: number, value?: T, options?: {
+    abortSignal?: AbortSignalLike;
+    abortErrorMsg?: string;
+}): Promise<T | void>;
 
 // @public
 export interface DeserializationContentTypes {

sdk/core/core-http/src/coreHttp.ts

@@ -99,7 +99,6 @@ export {
 export {
   stripRequest,
   stripResponse,
-  delay,
   executePromisesSequentially,
   generateUuid,
   encodeUri,
@@ -113,7 +112,7 @@ export {
 } from "./util/utils";
 export { URLBuilder, URLQuery } from "./url";
 export { AbortSignalLike } from "@azure/abort-controller";
-
+export { delay } from "./util/delay";
 // legacy exports. Use core-tracing instead (and remove on next major version update of core-http).
 export { createSpanFunction, SpanConfig } from "./createSpanLegacy";
 

sdk/core/core-http/src/policies/bearerTokenAuthenticationPolicy.ts

@@ -11,7 +11,7 @@ import {
 import { Constants } from "../util/constants";
 import { HttpOperationResponse } from "../httpOperationResponse";
 import { WebResourceLike } from "../webResource";
-import { delay } from "../util/utils";
+import { delay } from "../util/delay";
 
 // #region Access Token Cycler
 
@@ -71,7 +71,7 @@ async function beginRefresh(
 ): Promise<AccessToken> {
   // This wrapper handles exceptions gracefully as long as we haven't exceeded
   // the timeout.
-  async function tryGetAccessToken() {
+  async function tryGetAccessToken(): Promise<AccessToken | null> {
     if (Date.now() < timeoutInMs) {
       try {
         return await getAccessToken();

sdk/core/core-http/src/policies/exponentialRetryPolicy.ts

@@ -2,7 +2,6 @@
 // Licensed under the MIT license.
 
 import { HttpOperationResponse } from "../httpOperationResponse";
-import * as utils from "../util/utils";
 import { WebResourceLike } from "../webResource";
 import {
   BaseRequestPolicy,
@@ -22,6 +21,7 @@ import {
 } from "../util/exponentialBackoffStrategy";
 import { RestError } from "../restError";
 import { logger } from "../log";
+import { delay } from "../util/delay";
 
 export function exponentialRetryPolicy(
   retryCount?: number,
@@ -164,7 +164,7 @@ async function retry(
   if (!isAborted && shouldRetry(policy.retryCount, shouldPolicyRetry, retryData, response)) {
     logger.info(`Retrying request in ${retryData.retryInterval}`);
     try {
-      await utils.delay(retryData.retryInterval);
+      await delay(retryData.retryInterval);
       const res = await policy._nextPolicy.sendRequest(request.clone());
       return retry(policy, request, res, retryData);
     } catch (err) {

sdk/core/core-http/src/policies/rpRegistrationPolicy.ts

@@ -1,6 +1,7 @@
 // Copyright (c) Microsoft Corporation.
 // Licensed under the MIT license.
 
+import { delay } from "../util/delay";
 import { HttpOperationResponse } from "../httpOperationResponse";
 import * as utils from "../util/utils";
 import { WebResourceLike } from "../webResource";
@@ -145,9 +146,8 @@ function extractSubscriptionUrl(url: string): string {
  * @param provider - The provider name to be registered.
  * @param originalRequest - The original request sent by the user that returned a 409 response
  * with a message that the provider is not registered.
- * @param callback - The callback that handles the RP registration
  */
-function registerRP(
+async function registerRP(
   policy: RPRegistrationPolicy,
   urlPrefix: string,
   provider: string,
@@ -159,12 +159,11 @@ function registerRP(
   reqOptions.method = "POST";
   reqOptions.url = postUrl;
 
-  return policy._nextPolicy.sendRequest(reqOptions).then((response) => {
-    if (response.status !== 200) {
-      throw new Error(`Autoregistration of ${provider} failed. Please try registering manually.`);
-    }
-    return getRegistrationStatus(policy, getUrl, originalRequest);
-  });
+  const response = await policy._nextPolicy.sendRequest(reqOptions);
+  if (response.status !== 200) {
+    throw new Error(`Autoregistration of ${provider} failed. Please try registering manually.`);
+  }
+  return getRegistrationStatus(policy, getUrl, originalRequest);
 }
 
 /**
@@ -176,7 +175,7 @@ function registerRP(
  * with a message that the provider is not registered.
  * @returns True if RP Registration is successful.
  */
-function getRegistrationStatus(
+async function getRegistrationStatus(
   policy: RPRegistrationPolicy,
   url: string,
   originalRequest: WebResourceLike
@@ -185,14 +184,12 @@ function getRegistrationStatus(
   reqOptions.url = url;
   reqOptions.method = "GET";
 
-  return policy._nextPolicy.sendRequest(reqOptions).then((res) => {
-    const obj = res.parsedBody as any;
-    if (res.parsedBody && obj.registrationState && obj.registrationState === "Registered") {
-      return true;
-    } else {
-      return utils
-        .delay(policy._retryTimeout * 1000)
-        .then(() => getRegistrationStatus(policy, url, originalRequest));
-    }
-  });
+  const res = await policy._nextPolicy.sendRequest(reqOptions);
+  const obj = res.parsedBody;
+  if (res.parsedBody && obj.registrationState && obj.registrationState === "Registered") {
+    return true;
+  } else {
+    await delay(policy._retryTimeout * 1000);
+    return getRegistrationStatus(policy, url, originalRequest);
+  }
 }

sdk/core/core-http/src/policies/systemErrorRetryPolicy.ts

@@ -2,7 +2,6 @@
 // Licensed under the MIT license.
 
 import { HttpOperationResponse } from "../httpOperationResponse";
-import * as utils from "../util/utils";
 import { WebResourceLike } from "../webResource";
 import {
   BaseRequestPolicy,
@@ -21,6 +20,7 @@ import {
   DEFAULT_CLIENT_MIN_RETRY_INTERVAL,
   isNumber
 } from "../util/exponentialBackoffStrategy";
+import { delay } from "../util/delay";
 
 export function systemErrorRetryPolicy(
   retryCount?: number,
@@ -107,7 +107,7 @@ async function retry(
   if (shouldRetry(policy.retryCount, shouldPolicyRetry, retryData, operationResponse, err)) {
     // If previous operation ended with an error and the policy allows a retry, do that
     try {
-      await utils.delay(retryData.retryInterval);
+      await delay(retryData.retryInterval);
       return policy._nextPolicy.sendRequest(request.clone());
     } catch (nestedErr) {
       return retry(policy, request, operationResponse, nestedErr, retryData);

sdk/core/core-http/src/policies/throttlingRetryPolicy.ts

@@ -10,7 +10,8 @@ import {
 import { WebResourceLike } from "../webResource";
 import { HttpOperationResponse } from "../httpOperationResponse";
 import { Constants } from "../util/constants";
-import { delay } from "../util/utils";
+import { delay } from "../util/delay";
+import { AbortError } from "@azure/abort-controller";
 
 type ResponseHandler = (
   httpRequest: WebResourceLike,
@@ -26,6 +27,8 @@ export function throttlingRetryPolicy(): RequestPolicyFactory {
   };
 }
 
+const StandardAbortMessage = "The operation was aborted.";
+
 /**
  * To learn more, please refer to
  * https://docs.microsoft.com/en-us/azure/azure-resource-manager/resource-manager-request-limits,
@@ -67,7 +70,14 @@ export class ThrottlingRetryPolicy extends BaseRequestPolicy {
         retryAfterHeader
       );
       if (delayInMs) {
-        return delay(delayInMs).then((_: any) => this._nextPolicy.sendRequest(httpRequest));
+        await delay(delayInMs, undefined, {
+          abortSignal: httpRequest.abortSignal,
+          abortErrorMsg: StandardAbortMessage
+        });
+        if (httpRequest.abortSignal?.aborted) {
+          throw new AbortError(StandardAbortMessage);
+        }
+        return this._nextPolicy.sendRequest(httpRequest);
       }
     }
 

sdk/core/core-http/src/util/delay.ts

@@ -0,0 +1,62 @@
+// Copyright (c) Microsoft Corporation.
+// Licensed under the MIT license.
+
+import { isDefined } from "./typeguards";
+import { AbortError, AbortSignalLike } from "@azure/abort-controller";
+const StandardAbortMessage = "The operation was aborted.";
+
+/**
+ * A wrapper for setTimeout that resolves a promise after delayInMs milliseconds.
+ * @param delayInMs - The number of milliseconds to be delayed.
+ * @param value - The value to be resolved with after a timeout of t milliseconds.
+ * @param options - The options for delay - currently abort options
+ *   @param abortSignal - The abortSignal associated with containing operation.
+ *   @param abortErrorMsg - The abort error message associated with containing operation.
+ * @returns - Resolved promise
+ */
+export function delay<T>(
+  delayInMs: number,
+  value?: T,
+  options?: {
+    abortSignal?: AbortSignalLike;
+    abortErrorMsg?: string;
+  }
+): Promise<T | void> {
+  return new Promise((resolve, reject) => {
+    let timer: ReturnType<typeof setTimeout> | undefined = undefined;
+    let onAborted: (() => void) | undefined = undefined;
+
+    const rejectOnAbort = (): void => {
+      return reject(
+        new AbortError(options?.abortErrorMsg ? options?.abortErrorMsg : StandardAbortMessage)
+      );
+    };
+
+    const removeListeners = (): void => {
+      if (options?.abortSignal && onAborted) {
+        options.abortSignal.removeEventListener("abort", onAborted);
+      }
+    };
+
+    onAborted = (): void => {
+      if (isDefined(timer)) {
+        clearTimeout(timer);
+      }
+      removeListeners();
+      return rejectOnAbort();
+    };
+
+    if (options?.abortSignal && options.abortSignal.aborted) {
+      return rejectOnAbort();
+    }
+
+    timer = setTimeout(() => {
+      removeListeners();
+      resolve(value);
+    }, delayInMs);
+
+    if (options?.abortSignal) {
+      options.abortSignal.addEventListener("abort", onAborted);
+    }
+  });
+}