[App Config] Handle throttling - do not hang - should honor abort signal

sdk/appconfiguration/app-configuration/package.json

@@ -86,6 +86,7 @@
     ]
   },
   "dependencies": {
+    "@azure/abort-controller": "^1.0.0",
     "@azure/core-asynciterator-polyfill": "^1.0.0",
     "@azure/core-http": "^1.2.0",
     "@azure/core-paging": "^1.1.1",

sdk/appconfiguration/app-configuration/src/appConfigurationClient.ts

@@ -204,7 +204,6 @@ export class AppConfigurationClient {
         entity: keyValue,
         ...newOptions
       });
-
       return transformKeyValueResponse(originalResponse);
     });
   }

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

@@ -1,6 +1,7 @@
 // Copyright (c) Microsoft Corporation.
 // Licensed under the MIT license.
 
+import { AbortError, AbortSignalLike } from "@azure/abort-controller";
 import {
   BaseRequestPolicy,
   RequestPolicy,
@@ -9,7 +10,6 @@ import {
   WebResource,
   HttpOperationResponse,
   Constants,
-  delay,
   RestError
 } from "@azure/core-http";
 
@@ -24,6 +24,89 @@ export function throttlingRetryPolicy(): RequestPolicyFactory {
   };
 }
 
+const StandardAbortMessage = "The operation was aborted.";
+
+/**
+ * An executor for a function that returns a Promise that obeys both a timeout and an
+ * optional AbortSignal.
+ * @param actionFn - The callback that we want to resolve.
+ * @param timeoutMs - The number of milliseconds to allow before throwing an OperationTimeoutError.
+ * @param timeoutMessage - The message to place in the .description field for the thrown exception for Timeout.
+ * @param abortSignal - The abortSignal associated with containing operation.
+ *
+ * @internal
+ */
+export async function waitForTimeoutOrAbortOrResolve<T>(args: {
+  actionFn: () => Promise<T>;
+  timeoutMs: number;
+  timeoutMessage: string;
+  abortSignal: AbortSignalLike | undefined;
+}): Promise<T> {
+  if (args.abortSignal && args.abortSignal.aborted) {
+    throw new AbortError(StandardAbortMessage);
+  }
+
+  let timer: any | undefined = undefined;
+  let clearAbortSignal: (() => void) | undefined = undefined;
+
+  const clearAbortSignalAndTimer = (): void => {
+    clearTimeout(timer);
+
+    if (clearAbortSignal) {
+      clearAbortSignal();
+    }
+  };
+
+  const abortOrTimeoutPromise = new Promise<T>((_resolve, reject) => {
+    clearAbortSignal = checkAndRegisterWithAbortSignal(reject, args.abortSignal);
+
+    timer = setTimeout(() => {
+      reject(new Error(args.timeoutMessage));
+    }, args.timeoutMs);
+  });
+
+  try {
+    return await Promise.race([abortOrTimeoutPromise, args.actionFn()]);
+  } finally {
+    clearAbortSignalAndTimer();
+  }
+}
+
+/**
+ * Registers listener to the abort event on the abortSignal to call your abortFn and
+ * returns a function that will clear the same listener.
+ *
+ * If abort signal is already aborted, then throws an AbortError and returns a function that does nothing
+ *
+ * @returns A function that removes any of our attached event listeners on the abort signal or an empty function if
+ * the abortSignal was not defined.
+ *
+ * @internal
+ */
+export function checkAndRegisterWithAbortSignal(
+  onAbortFn: (abortError: AbortError) => void,
+  abortSignal?: AbortSignalLike
+): () => void {
+  if (abortSignal == null) {
+    return () => {
+      /** Nothing to do here, no abort signal */
+    };
+  }
+
+  if (abortSignal.aborted) {
+    throw new AbortError(StandardAbortMessage);
+  }
+
+  const onAbort = (): void => {
+    abortSignal.removeEventListener("abort", onAbort);
+    onAbortFn(new AbortError(StandardAbortMessage));
+  };
+
+  abortSignal.addEventListener("abort", onAbort);
+
+  return () => abortSignal.removeEventListener("abort", onAbort);
+}
+
 /**
  * This policy is a close copy of the ThrottlingRetryPolicy class from
  * core-http with modifications to work with how AppConfig is currently
@@ -37,15 +120,22 @@ export class ThrottlingRetryPolicy extends BaseRequestPolicy {
   }
 
   public async sendRequest(httpRequest: WebResource): Promise<HttpOperationResponse> {
-    return this._nextPolicy.sendRequest(httpRequest.clone()).catch((err) => {
+    return this._nextPolicy.sendRequest(httpRequest.clone()).catch(async (err) => {
       if (isRestErrorWithHeaders(err)) {
         const delayInMs = getDelayInMs(err.response.headers);
 
         if (delayInMs == null) {
           throw err;
         }
 
-        return delay(delayInMs).then((_: any) => this.sendRequest(httpRequest.clone()));
+        return await waitForTimeoutOrAbortOrResolve({
+          timeoutMs: delayInMs,
+          abortSignal: httpRequest.abortSignal,
+          actionFn: async () => {
+            return await this.sendRequest(httpRequest.clone());
+          },
+          timeoutMessage: `ServiceBusy: Unable to fulfill the request in ${delayInMs}ms when retried.`
+        });
       } else {
         throw err;
       }

sdk/appconfiguration/app-configuration/test/public/throttling.spec.ts

@@ -0,0 +1,63 @@
+// Copyright (c) Microsoft Corporation.
+// Licensed under the MIT license.
+
+import { createAppConfigurationClientForTests, startRecorder } from "./utils/testHelpers";
+import { AppConfigurationClient } from "../../src";
+import { Recorder } from "@azure/test-utils-recorder";
+import { Context } from "mocha";
+import { AbortController } from "@azure/abort-controller";
+
+describe("AppConfigurationClient", () => {
+  let client: AppConfigurationClient;
+  let recorder: Recorder;
+
+  beforeEach(function(this: Context) {
+    recorder = startRecorder(this);
+    client = createAppConfigurationClientForTests() || this.skip();
+  });
+
+  afterEach(async function(this: Context) {
+    await recorder.stop();
+  });
+
+  describe.only("simple usages", () => {
+    it("Add and query a setting without a label", async () => {
+      const key = recorder.getUniqueName("noLabelTests");
+      const numberOfSettings = 200;
+      const times = 1000;
+      for (let time = 0; time < times; time++) {
+        const promises = [];
+        try {
+          for (let index = 0; index < numberOfSettings; index++) {
+            promises.push(
+              client.addConfigurationSetting(
+                {
+                  key: key + " " + +index,
+                  value: "added"
+                },
+                {
+                  abortSignal: AbortController.timeout(10000)
+                }
+              )
+            );
+          }
+          await Promise.all(promises);
+        } catch (error) {
+          console.log(error);
+        }
+      }
+
+      await cleanupSampleValues([key], client);
+    });
+  });
+});
+
+async function cleanupSampleValues(keys: string[], client: AppConfigurationClient) {
+  const settingsIterator = client.listConfigurationSettings({
+    keyFilter: keys.join(",")
+  });
+
+  for await (const setting of settingsIterator) {
+    await client.deleteConfigurationSetting({ key: setting.key, label: setting.label });
+  }
+}