[core-http] Throttling retry policy fix in core-http

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

sdk/core/core-http/CHANGELOG.md

@@ -13,6 +13,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/package.json

@@ -179,6 +179,7 @@
     "karma-sourcemap-loader": "^0.3.8",
     "mocha": "^7.1.1",
     "mocha-junit-reporter": "^1.18.0",
+    "nock": "^12.0.3",
     "npm-run-all": "^4.1.5",
     "nyc": "^14.0.0",
     "prettier": "^1.16.4",

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

@@ -181,7 +181,7 @@ export class DefaultHttpClient extends FetchHttpClient {
     }
 
 // @public
-export function delay<T>(t: number, value?: T): Promise<T | void>;
+export function delay<T>(delayInMs: number, abortSignal?: AbortSignalLike, abortErrorMsg?: string, value?: T): 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";
@@ -185,14 +186,13 @@ function getRegistrationStatus(
   reqOptions.url = url;
   reqOptions.method = "GET";
 
-  return policy._nextPolicy.sendRequest(reqOptions).then((res) => {
+  return policy._nextPolicy.sendRequest(reqOptions).then(async (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));
+      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,11 @@ export class ThrottlingRetryPolicy extends BaseRequestPolicy {
         retryAfterHeader
       );
       if (delayInMs) {
-        return delay(delayInMs).then((_: any) => this._nextPolicy.sendRequest(httpRequest));
+        await delay(delayInMs, httpRequest.abortSignal, 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,57 @@
+// 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 abortSignal - The abortSignal associated with containing operation.
+ * @param abortErrorMsg - The abort error message associated with containing operation.
+ * @param value - The value to be resolved with after a timeout of t milliseconds.
+ * @returns - Resolved promise
+ */
+export function delay<T>(
+  delayInMs: number,
+  abortSignal?: AbortSignalLike,
+  abortErrorMsg?: string,
+  value?: T
+): 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(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(value);
+    }, delayInMs);
+
+    if (abortSignal) {
+      abortSignal.addEventListener("abort", onAborted);
+    }
+  });
+}

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

@@ -0,0 +1,11 @@
+// Copyright (c) Microsoft Corporation.
+// Licensed under the MIT license.
+
+/**
+ * Helper TypeGuard that checks if something is defined or not.
+ * @param thing - Anything
+ * @internal
+ */
+export function isDefined<T>(thing: T | undefined | null): thing is T {
+  return typeof thing !== "undefined" && thing !== null;
+}

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

@@ -113,16 +113,6 @@ export function executePromisesSequentially(
   return result;
 }
 
-/**
- * A wrapper for setTimeout that resolves a promise after t milliseconds.
- * @param t - The number of milliseconds to be delayed.
- * @param value - The value to be resolved with after a timeout of t milliseconds.
- * @returns Resolved promise
- */
-export function delay<T>(t: number, value?: T): Promise<T | void> {
-  return new Promise((resolve) => setTimeout(() => resolve(value), t));
-}
-
 /**
  * Service callback that is returned for REST requests initiated by the service client.
  */

sdk/core/core-http/test/policies/throttlingRetryPolicy.node.ts

@@ -0,0 +1,52 @@
+// Copyright (c) Microsoft Corporation.
+// Licensed under the MIT license.
+
+import nock from "nock";
+import { Constants, ServiceClient } from "../../src/coreHttp";
+import { AbortController } from "@azure/abort-controller";
+import { assert } from "chai";
+
+describe("Throttling retry policy", () => {
+  let client: ServiceClient;
+
+  beforeEach(function() {
+    if (!nock.isActive()) {
+      nock.activate();
+    }
+    nock("https://fakeservice.io:443")
+      .persist()
+      .put(/.*/g)
+      .reply(
+        Constants.HttpConstants.StatusCodes.TooManyRequests,
+        {
+          type: "https://fakeservice.io/errors/too-many-requests",
+          title: "Resource utilization has surpassed the assigned quota",
+          policy: "Total Requests",
+          status: Constants.HttpConstants.StatusCodes.TooManyRequests
+        },
+        ["Retry-After", "10000"]
+      );
+    client = new ServiceClient();
+  });
+
+  afterEach(async function() {
+    nock.restore();
+    nock.cleanAll();
+    nock.enableNetConnect();
+  });
+
+  it("Should not retry forever (honors the abort signal passed)", async () => {
+    let errorWasThrown = false;
+    try {
+      await client.sendRequest({
+        url: "https://fakeservice.io/ABCD",
+        abortSignal: AbortController.timeout(100),
+        method: "PUT"
+      });
+    } catch (error) {
+      errorWasThrown = true;
+      assert.equal((error as any).name, "AbortError", "Unexpected error thrown");
+    }
+    assert.equal(errorWasThrown, true, "Error was not thrown");
+  });
+});