From 5cb62ec4bb123b863bedc36ef434ce12db92d075 Mon Sep 17 00:00:00 2001 From: Muhammad Umer Aamir <93091750+umeraamir69@users.noreply.github.com> Date: Sat, 14 Nov 2026 01:27:45 -0500 Subject: [PATCH] fix(sdk): honor caller abort signals in retry.fetch Stop treating caller AbortErrors as timeouts, listen to Request signals, and reject immediately when the signal is already aborted. --- packages/trigger-sdk/src/v3/retry.test.ts | 83 ++++++++++- packages/trigger-sdk/src/v3/retry.ts | 159 +++++++++++++--------- 2 files changed, 179 insertions(+), 63 deletions(-) diff --git a/packages/trigger-sdk/src/v3/retry.test.ts b/packages/trigger-sdk/src/v3/retry.test.ts index 3b22c2cecc2..35c78c0558b 100644 --- a/packages/trigger-sdk/src/v3/retry.test.ts +++ b/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((_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); + }); +}); diff --git a/packages/trigger-sdk/src/v3/retry.ts b/packages/trigger-sdk/src/v3/retry.ts index 3f509c52975..8d31cca7cdc 100644 --- a/packages/trigger-sdk/src/v3/retry.ts +++ b/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; } }