fix(sdk): honor caller abort signals in retry.fetch

packages/trigger-sdk/src/v3/retry.test.ts

@@ -1,11 +1,18 @@
 import { SEMATTRS_HTTP_URL } from "@opentelemetry/semantic-conventions";
-import { describe, expect, it } from "vitest";
+import { afterEach, beforeEach, describe, expect, it, vi } from "vitest";
 import {
   createAttributesFromHeaders,
   createFetchAttributes,
   createFetchResponseAttributes,
+  retry,
 } from "./retry.js";
 
+vi.mock("./wait.js", () => ({
+  wait: {
+    until: vi.fn(async () => undefined),
+  },
+}));
+
 describe("retry fetch telemetry", () => {
   it("omits URL credentials, query parameters, and fragments", () => {
     const attributes = createFetchAttributes(
@@ -54,3 +61,77 @@ describe("retry fetch telemetry", () => {
     expect(attributes).not.toHaveProperty("http.response.header.set-cookie");
   });
 });
+
+describe("retry.fetch abort signal", () => {
+  const originalFetch = globalThis.fetch;
+  let fetchCount = 0;
+
+  beforeEach(() => {
+    fetchCount = 0;
+  });
+
+  afterEach(() => {
+    globalThis.fetch = originalFetch;
+  });
+
+  it("rejects immediately when init.signal is already aborted", async () => {
+    globalThis.fetch = (async () => {
+      fetchCount += 1;
+      return new Response("ok");
+    }) as typeof fetch;
+
+    await expect(retry.fetch("https://example.com", { signal: AbortSignal.abort() })).rejects.toMatchObject({
+      name: "AbortError",
+    });
+    expect(fetchCount).toBe(0);
+  });
+
+  it("rejects when a Request input carries an already-aborted signal", async () => {
+    globalThis.fetch = (async () => {
+      fetchCount += 1;
+      return new Response("ok");
+    }) as typeof fetch;
+
+    await expect(
+      retry.fetch(new Request("https://example.com", { signal: AbortSignal.abort() }))
+    ).rejects.toMatchObject({ name: "AbortError" });
+    expect(fetchCount).toBe(0);
+  });
+
+  it("does not retry when the caller aborts mid-request", async () => {
+    globalThis.fetch = (async (_input, init) => {
+      fetchCount += 1;
+      return await new Promise<Response>((_resolve, reject) => {
+        const signal = init?.signal;
+        if (!signal) {
+          reject(new Error("missing signal"));
+          return;
+        }
+
+        if (signal.aborted) {
+          const error = new Error("This operation was aborted.");
+          error.name = "AbortError";
+          reject(error);
+          return;
+        }
+
+        signal.addEventListener(
+          "abort",
+          () => {
+            const error = new Error("This operation was aborted.");
+            error.name = "AbortError";
+            reject(error);
+          },
+          { once: true }
+        );
+      });
+    }) as typeof fetch;
+
+    const controller = new AbortController();
+    const pending = retry.fetch("https://example.com/slow", { signal: controller.signal });
+    controller.abort();
+
+    await expect(pending).rejects.toMatchObject({ name: "AbortError" });
+    expect(fetchCount).toBe(1);
+  });
+});

packages/trigger-sdk/src/v3/retry.ts

@@ -152,6 +152,31 @@ class FetchErrorWithSpan extends Error {
 
 const MAX_ATTEMPTS = 10;
 
+function resolveCallerAbortSignal(
+  input: RequestInfo | URL,
+  init?: RequestInit
+): AbortSignal | undefined {
+  if (init?.signal) {
+    return init.signal;
+  }
+
+  if (typeof Request !== "undefined" && input instanceof Request) {
+    return input.signal;
+  }
+
+  return undefined;
+}
+
+function createCallerAbortError(signal?: AbortSignal): Error {
+  if (signal?.reason instanceof Error) {
+    return signal.reason;
+  }
+
+  const error = new Error("This operation was aborted.");
+  error.name = "AbortError";
+  return error;
+}
+
 async function retryFetch(
   input: RequestInfo | URL,
   init?: RetryFetchRequestInit | undefined
@@ -160,36 +185,59 @@ async function retryFetch(
     "retry.fetch()",
     async (span) => {
       let attempt = 1;
+      const callerSignal = resolveCallerAbortSignal(input, init);
+
+      if (callerSignal?.aborted) {
+        throw createCallerAbortError(callerSignal);
+      }
 
       while (true) {
         try {
-          const abortController = new AbortController();
+          if (callerSignal?.aborted) {
+            throw createCallerAbortError(callerSignal);
+          }
 
+          const abortController = new AbortController();
           const timeoutId = init?.timeoutInMs
             ? setTimeout(() => {
                 abortController.abort();
-              }, init?.timeoutInMs)
+              }, init.timeoutInMs)
             : undefined;
 
-          init?.signal?.addEventListener("abort", () => {
-            abortController.abort();
-          });
-
-          const [response, span] = await doFetchRequest(
-            input,
-            { ...(init ?? {}), signal: abortController.signal },
-            attempt
-          );
+          const onCallerAbort = () => {
+            abortController.abort(callerSignal?.reason);
+          };
 
-          if (timeoutId) {
-            clearTimeout(timeoutId);
+          if (callerSignal) {
+            if (callerSignal.aborted) {
+              abortController.abort(callerSignal.reason);
+            } else {
+              callerSignal.addEventListener("abort", onCallerAbort, { once: true });
+            }
           }
 
-          if (response.ok) {
-            span.setAttributes(createFetchResponseAttributes(response));
+          let response: Response;
+          let requestSpan: Span;
 
-            span.end();
+          try {
+            [response, requestSpan] = await doFetchRequest(
+              input,
+              { ...(init ?? {}), signal: abortController.signal },
+              attempt
+            );
+          } finally {
+            if (timeoutId) {
+              clearTimeout(timeoutId);
+            }
 
+            if (callerSignal) {
+              callerSignal.removeEventListener("abort", onCallerAbort);
+            }
+          }
+
+          if (response.ok) {
+            requestSpan.setAttributes(createFetchResponseAttributes(response));
+            requestSpan.end();
             return response;
           }
 
@@ -199,67 +247,65 @@ async function retryFetch(
             attempt
           );
 
-          if (!nextRetry) {
-            span.setAttributes(createFetchResponseAttributes(response));
-
-            span.end();
-
-            return response;
-          }
-
-          if (attempt >= MAX_ATTEMPTS) {
-            span.setAttributes(createFetchResponseAttributes(response));
-
-            span.end();
-
+          if (!nextRetry || attempt >= MAX_ATTEMPTS) {
+            requestSpan.setAttributes(createFetchResponseAttributes(response));
+            requestSpan.end();
             return response;
           }
 
           if (nextRetry.type === "delay") {
             const continueDate = new Date(Date.now() + nextRetry.value);
-            span.setAttribute(SemanticInternalAttributes.RETRY_AT, continueDate.toISOString());
-            span.setAttribute(SemanticInternalAttributes.RETRY_COUNT, attempt);
-            span.setAttribute(SemanticInternalAttributes.RETRY_DELAY, `${nextRetry.value}ms`);
-
-            span.end();
-
+            requestSpan.setAttribute(
+              SemanticInternalAttributes.RETRY_AT,
+              continueDate.toISOString()
+            );
+            requestSpan.setAttribute(SemanticInternalAttributes.RETRY_COUNT, attempt);
+            requestSpan.setAttribute(
+              SemanticInternalAttributes.RETRY_DELAY,
+              `${nextRetry.value}ms`
+            );
+            requestSpan.end();
             await wait.until({ date: continueDate });
           } else {
             const now = Date.now();
             const nextRetryDate = new Date(nextRetry.value);
             const isInFuture = nextRetryDate.getTime() > now;
 
-            span.setAttribute(
+            requestSpan.setAttribute(
               SemanticInternalAttributes.RETRY_AT,
-              new Date(nextRetry.value).toISOString()
+              nextRetryDate.toISOString()
             );
-            span.setAttribute(SemanticInternalAttributes.RETRY_COUNT, attempt);
+            requestSpan.setAttribute(SemanticInternalAttributes.RETRY_COUNT, attempt);
 
             if (isInFuture) {
-              span.setAttribute(
+              requestSpan.setAttribute(
                 SemanticInternalAttributes.RETRY_DELAY,
                 `${nextRetry.value - now}ms`
               );
             }
 
-            span.end();
-
-            await wait.until({ date: new Date(nextRetry.value) });
+            requestSpan.end();
+            await wait.until({ date: nextRetryDate });
           }
         } catch (e) {
+          if (e instanceof Error && e.name === "AbortError" && callerSignal?.aborted) {
+            throw e;
+          }
+
           if (e instanceof FetchErrorWithSpan && e.originalError instanceof Error) {
             if (e.originalError.name === "AbortError") {
+              // Caller-initiated aborts must not be retried as timeouts.
+              if (callerSignal?.aborted) {
+                e.span.end();
+                throw e.originalError;
+              }
+
               const nextRetryDelay = calculateNextRetryDelay(
                 resolveDefaults(init?.retry, "timeout", defaultFetchRetryOptions.timeout),
                 attempt
               );
 
-              if (!nextRetryDelay) {
-                e.span.end();
-                throw e;
-              }
-
-              if (attempt >= MAX_ATTEMPTS) {
+              if (!nextRetryDelay || attempt >= MAX_ATTEMPTS) {
                 e.span.end();
                 throw e;
               }
@@ -268,12 +314,9 @@ async function retryFetch(
               e.span.setAttribute(SemanticInternalAttributes.RETRY_AT, continueDate.toISOString());
               e.span.setAttribute(SemanticInternalAttributes.RETRY_COUNT, attempt);
               e.span.setAttribute(SemanticInternalAttributes.RETRY_DELAY, `${nextRetryDelay}ms`);
-
               e.span.end();
-
               await wait.until({ date: continueDate });
-
-              continue; // Move to the next attempt
+              continue;
             } else if (
               e.originalError.name === "TypeError" &&
               "cause" in e.originalError &&
@@ -288,12 +331,7 @@ async function retryFetch(
                 attempt
               );
 
-              if (!nextRetryDelay) {
-                e.span.end();
-                throw e;
-              }
-
-              if (attempt >= MAX_ATTEMPTS) {
+              if (!nextRetryDelay || attempt >= MAX_ATTEMPTS) {
                 e.span.end();
                 throw e;
               }
@@ -304,12 +342,9 @@ async function retryFetch(
               );
               e.span.setAttribute(SemanticInternalAttributes.RETRY_COUNT, attempt);
               e.span.setAttribute(SemanticInternalAttributes.RETRY_DELAY, `${nextRetryDelay}ms`);
-
               e.span.end();
-
               await wait.until({ date: new Date(Date.now() + nextRetryDelay) });
-
-              continue; // Move to the next attempt
+              continue;
             }
           }