engineer-bot: address @mention on PR #600

Signed-off-by: peco-engineer-bot[bot] <peco-engineer-bot[bot]@users.noreply.github.com>

Author: peco-engineer-bot[bot]

lib/connection/connections/HttpRetryPolicy.ts

@@ -3,9 +3,24 @@ import { HttpTransactionDetails } from '../contracts/IConnectionProvider';
 import IClientContext from '../../contracts/IClientContext';
 import RetryError, { RetryErrorCode } from '../../errors/RetryError';
 
-function delay(milliseconds: number): Promise<void> {
-  return new Promise<void>((resolve) => {
-    setTimeout(() => resolve(), milliseconds);
+// Sleeps for `milliseconds`; aborting `signal` clears the timer and rejects with
+// `signal.reason`, so a cancelled backoff does not keep the process alive.
+function delay(milliseconds: number, signal?: AbortSignal): Promise<void> {
+  return new Promise<void>((resolve, reject) => {
+    if (signal?.aborted) {
+      reject(signal.reason);
+      return;
+    }
+    let timer: NodeJS.Timeout | undefined;
+    const onAbort = () => {
+      clearTimeout(timer);
+      reject(signal?.reason);
+    };
+    timer = setTimeout(() => {
+      signal?.removeEventListener('abort', onAbort);
+      resolve();
+    }, milliseconds);
+    signal?.addEventListener('abort', onAbort, { once: true });
   });
 }
 
@@ -52,7 +67,10 @@ export default class HttpRetryPolicy implements IRetryPolicy<HttpTransactionDeta
     return { shouldRetry: false };
   }
 
-  public async invokeWithRetry(operation: RetryableOperation<HttpTransactionDetails>): Promise<HttpTransactionDetails> {
+  public async invokeWithRetry(
+    operation: RetryableOperation<HttpTransactionDetails>,
+    signal?: AbortSignal,
+  ): Promise<HttpTransactionDetails> {
     for (;;) {
       // Capture either the resolved response or the thrown error so the
       // retry-decision logic below can flow without an early `continue` and
@@ -73,7 +91,7 @@ export default class HttpRetryPolicy implements IRetryPolicy<HttpTransactionDeta
           return outcome.details;
         }
         // eslint-disable-next-line no-await-in-loop
-        await delay(status.retryAfter);
+        await delay(status.retryAfter, signal);
       } else {
         // The operation threw before producing a response. This is typically a
         // transient network failure (stale keep-alive socket reset by a load
@@ -91,7 +109,7 @@ export default class HttpRetryPolicy implements IRetryPolicy<HttpTransactionDeta
           throw outcome.error;
         }
         // eslint-disable-next-line no-await-in-loop
-        await delay(status.retryAfter);
+        await delay(status.retryAfter, signal);
       }
     }
   }

lib/connection/contracts/IRetryPolicy.ts

@@ -12,5 +12,7 @@ export type RetryableOperation<R> = () => Promise<R>;
 export default interface IRetryPolicy<R> {
   shouldRetry(details: R): Promise<ShouldRetryResult>;
 
-  invokeWithRetry(operation: RetryableOperation<R>): Promise<R>;
+  // `signal` cancels a pending backoff sleep: aborting it rejects with `signal.reason`
+  // instead of waiting for the sleep to finish (the operation itself must honour the signal).
+  invokeWithRetry(operation: RetryableOperation<R>, signal?: AbortSignal): Promise<R>;
 }

lib/result/CloudFetchResultHandler.ts

@@ -159,9 +159,11 @@ export default class CloudFetchResultHandler implements IResultsProvider<ArrowBa
     // The retry policy checks `retriesTimeout` only between attempts, so a stalled attempt
     // (headers or body never arrive) would block forever. Enforce it as a deadline over the
     // whole download - every attempt plus the backoff between them - and abort the in-flight
-    // request when it expires (#520). Non-positive values keep the deadline off.
+    // request and any pending backoff sleep when it expires (#520). Non-positive values keep
+    // the deadline off.
     const deadlineMs = retriesTimeout > 0 ? Math.min(retriesTimeout, MAX_TIMER_DELAY_MS) : undefined;
-    let controller: AbortController | undefined;
+    // Aborted only once, by the deadline, so a single controller serves every attempt and backoff.
+    const controller = deadlineMs === undefined ? undefined : new AbortController();
     let timeoutError: RetryError | undefined;
     let lastFailure: unknown;
     let timer: NodeJS.Timeout | undefined;
@@ -181,7 +183,6 @@ export default class CloudFetchResultHandler implements IResultsProvider<ArrowBa
           throw timeoutError;
         }
         downloaded = undefined;
-        controller = deadlineMs === undefined ? undefined : new AbortController();
         const request = new Request(url, {
           ...requestConfig,
           signal: controller?.signal as unknown as RequestInit['signal'],
@@ -201,15 +202,16 @@ export default class CloudFetchResultHandler implements IResultsProvider<ArrowBa
           lastFailure = error;
           throw error;
         }
-      });
+      }, controller?.signal);
 
       if (deadlineMs === undefined) {
         result = await download;
       } else {
         const deadline = new Promise<never>((_, reject) => {
           timer = setTimeout(() => {
             timeoutError = new RetryError(RetryErrorCode.TimeoutExceeded, lastFailure);
-            controller?.abort();
+            // Aborts the in-flight request, and rejects a pending backoff sleep with `timeoutError`
+            controller?.abort(timeoutError);
             reject(timeoutError);
           }, deadlineMs);
         });

tests/unit/connection/connections/HttpRetryPolicy.test.ts

@@ -400,5 +400,78 @@ describe('HttpRetryPolicy', () => {
         expect(operation.callCount).to.equal(clientConfig.retryMaxAttempts);
       }
     });
+
+    it('should cancel a pending backoff sleep when the signal is aborted', async () => {
+      const context = new ClientContextStub({
+        retryDelayMin: 10000,
+        retryDelayMax: 60000,
+        retryMaxAttempts: 20,
+      });
+      const policy = new HttpRetryPolicy(context);
+      const operation = sinon.stub().resolves({
+        request: new Request('http://localhost'),
+        response: new Response(undefined, { status: 503 }),
+      });
+      const controller = new AbortController();
+      const reason = new Error('deadline');
+      const clearTimeoutSpy = sinon.spy(global, 'clearTimeout');
+      const setTimeoutSpy = sinon.spy(global, 'setTimeout');
+
+      try {
+        const invocation = policy.invokeWithRetry(operation, controller.signal);
+        // Let the first attempt complete and the 20s backoff sleep start
+        await new Promise((resolve) => {
+          setImmediate(resolve);
+        });
+        const backoffCall = setTimeoutSpy.getCalls().find((call) => call.args[1] === 20000);
+        expect(backoffCall).to.not.be.undefined;
+
+        controller.abort(reason);
+
+        try {
+          await invocation;
+          expect.fail('It should reject with the abort reason');
+        } catch (error) {
+          if (error instanceof AssertionError) {
+            throw error;
+          }
+          expect(error).to.equal(reason);
+        }
+        expect(clearTimeoutSpy.calledWith(backoffCall!.returnValue)).to.be.true;
+        expect(operation.callCount).to.equal(1);
+      } finally {
+        setTimeoutSpy.restore();
+        clearTimeoutSpy.restore();
+      }
+    });
+
+    it('should not start another attempt when the signal is already aborted before the backoff', async () => {
+      const context = new ClientContextStub({
+        retryDelayMin: 10000,
+        retryDelayMax: 60000,
+        retryMaxAttempts: 20,
+      });
+      const policy = new HttpRetryPolicy(context);
+      const controller = new AbortController();
+      const reason = new Error('deadline');
+      const operation = sinon.stub().callsFake(async () => {
+        controller.abort(reason);
+        return {
+          request: new Request('http://localhost'),
+          response: new Response(undefined, { status: 503 }),
+        };
+      });
+
+      try {
+        await policy.invokeWithRetry(operation, controller.signal);
+        expect.fail('It should reject with the abort reason');
+      } catch (error) {
+        if (error instanceof AssertionError) {
+          throw error;
+        }
+        expect(error).to.equal(reason);
+      }
+      expect(operation.callCount).to.equal(1);
+    });
   });
 });

tests/unit/result/CloudFetchResultHandler.test.ts

@@ -834,8 +834,26 @@ describe('CloudFetchResultHandler', () => {
       });
       // The first backoff (2 * retryDelayMin = 2s) alone outlasts the 300ms budget
       const handler = createHandler(url, { retriesTimeout: 300, retryDelayMin: 1000, retryMaxAttempts: 100 });
+      const setTimeoutSpy = sinon.spy(global, 'setTimeout');
+      const clearTimeoutSpy = sinon.spy(global, 'clearTimeout');
 
-      const { error, elapsedMs } = await fetchWithin(handler, 1500);
+      let outcome: Awaited<ReturnType<typeof fetchWithin>>;
+      let backoffTimers: Array<NodeJS.Timeout>;
+      try {
+        outcome = await fetchWithin(handler, 1500);
+        backoffTimers = setTimeoutSpy
+          .getCalls()
+          .filter((call) => call.args[1] === 2000)
+          .map((call) => call.returnValue);
+      } finally {
+        setTimeoutSpy.restore();
+        clearTimeoutSpy.restore();
+      }
+      const { error, elapsedMs } = outcome;
+
+      // The deadline cancels the pending backoff sleep instead of leaving its timer to keep Node alive
+      expect(backoffTimers).to.have.length(1);
+      expect(clearTimeoutSpy.calledWith(backoffTimers[0])).to.be.true;
 
       expect(error).to.be.instanceOf(RetryError);
       expect(error.errorCode).to.be.equal(RetryErrorCode.TimeoutExceeded);
@@ -845,9 +863,9 @@ describe('CloudFetchResultHandler', () => {
       expect(elapsedMs).to.be.lt(1000);
       expect(hits).to.be.equal(1);
 
-      // Once the backoff sleep ends, the retry loop must stop instead of sending another request
+      // The cancelled backoff ends the retry loop instead of sending another request
       await new Promise((resolve) => {
-        setTimeout(resolve, 2200);
+        setTimeout(resolve, 100);
       });
       expect(hits).to.be.equal(1);
     });