fix(core): add per-request timeout to Requester to catch silent fetch hangs

packages/core/modules/requester/requester.js

@@ -3,6 +3,8 @@ const { Delegate } = require('../../core');
 const { FetchError } = require('../../errors');
 const { get } = require('../../assertions');
 
+const DEFAULT_REQUEST_TIMEOUT_MS = 60_000;
+
 class Requester extends Delegate {
     constructor(params) {
         super(params);
@@ -13,6 +15,30 @@ class Requester extends Delegate {
         this.delegateTypes.push(this.DLGT_INVALID_AUTH);
         this.agent = get(params, 'agent', null);
 
+        // Per-attempt HTTP timeout. Without this the framework called fetch()
+        // with no AbortController and no timeout — a silently-hung TCP
+        // connection (server accepts but never responds) blocked the calling
+        // promise forever, cascading into stalled batches, stalled syncs,
+        // and worker-lambda timeouts.
+        //
+        // Configuration precedence:
+        //   1. Instance param:   new Requester({ requestTimeoutMs: 30_000 })
+        //   2. Class static:     static requestTimeoutMs = 30_000
+        //   3. Default:          DEFAULT_REQUEST_TIMEOUT_MS (60s)
+        //
+        // Pass 0 (or null) to disable the timeout entirely — reserved for
+        // test doubles and documented long-running endpoints.
+        // Intentionally NOT using `get(params, ...)` here — the Frigg
+        // `get` helper throws RequiredPropertyError if the key is missing
+        // and no default is provided, which would collide with the fall-
+        // through to the class-level static override.
+        const instanceTimeout = params?.requestTimeoutMs;
+        this.requestTimeoutMs =
+            instanceTimeout !== undefined && instanceTimeout !== null
+                ? instanceTimeout
+                : this.constructor.requestTimeoutMs ??
+                  DEFAULT_REQUEST_TIMEOUT_MS;
+
         // Allow passing in the fetch function
         // Instance methods can use this.fetch without differentiating
         this.fetch = get(params, 'fetch', fetch);
@@ -48,20 +74,54 @@ class Requester extends Delegate {
 
         if (this.agent) options.agent = this.agent;
 
+        // Per-attempt timeout — fresh AbortController per call so the retry
+        // recursion (with its own backoff sleeps) always gets a clean
+        // signal. Timer is cleared in the finally block regardless of
+        // outcome.
+        const timeoutMs = this.requestTimeoutMs;
+        const controller = timeoutMs > 0 ? new AbortController() : null;
+        const timeoutHandle = controller
+            ? setTimeout(() => controller.abort(), timeoutMs)
+            : null;
+        const fetchOptions = controller
+            ? { ...options, signal: controller.signal }
+            : options;
+
         let response;
         try {
-            response = await this.fetch(encodedUrl, options);
+            response = await this.fetch(encodedUrl, fetchOptions);
         } catch (e) {
-            if (e.code === 'ECONNRESET' && i < this.backOff.length) {
+            // AbortController fires AbortError (name) / ETIMEDOUT-shaped
+            // errors (type on node-fetch) when we hit the timeout. No
+            // retry on timeout: a slow endpoint is a downstream problem,
+            // and each retry would wait another `timeoutMs` before giving
+            // up — amplifying the hang into a per-record multi-minute
+            // stall at batch scale.
+            const isTimeout =
+                e?.name === 'AbortError' || e?.type === 'aborted';
+            if (e?.code === 'ECONNRESET' && i < this.backOff.length) {
                 const delay = this.backOff[i] * 1000;
                 await new Promise((resolve) => setTimeout(resolve, delay));
                 return this._request(url, options, i + 1);
             }
-            throw await FetchError.create({
+            const fetchError = await FetchError.create({
                 resource: encodedUrl,
                 init: options,
-                responseBody: e,
+                responseBody: isTimeout
+                    ? `Request timed out after ${timeoutMs}ms`
+                    : e,
             });
+            if (isTimeout) {
+                // Flag + machine-readable fields so callers can
+                // distinguish a timeout from a generic network error
+                // without parsing the message (which FetchError
+                // sanitizes outside of STAGE=dev).
+                fetchError.isTimeout = true;
+                fetchError.timeoutMs = timeoutMs;
+            }
+            throw fetchError;
+        } finally {
+            if (timeoutHandle) clearTimeout(timeoutHandle);
         }
         const { status } = response;
 

packages/core/modules/requester/requester.test.js

@@ -1,28 +1,276 @@
 const { Requester } = require('./requester');
 
-describe('429 and 5xx testing', () => {
-    let backOffArray = [1, 1, 1];
-    let requester = new Requester({ backOff: backOffArray });
-    let sum = backOffArray.reduce((a, b) => {
-        return a + b;
-    }, 0);
-    it.skip("should retry with 'exponential' back off due to 429", async () => {
-        let startTime = await Date.now();
-        let res = await requester._get({
-            url: 'https://70e18ff0-1967-4fb5-8f96-10477ab6bb9e.mock.pstmn.io//429',
-        });
-        let endTime = await Date.now();
-        let difference = endTime - startTime;
-        expect(difference).toBeGreaterThan(sum * 1000);
+/**
+ * Requester is abstract: subclasses provide `addAuthHeaders`. For the
+ * timeout / retry tests we don't care about auth headers, so this
+ * subclass just passes them through.
+ */
+class TestRequester extends Requester {
+    async addAuthHeaders(headers) {
+        return headers;
+    }
+}
+
+describe('Requester', () => {
+    describe('429 and 5xx testing', () => {
+        let backOffArray = [1, 1, 1];
+        let requester = new TestRequester({ backOff: backOffArray });
+        let sum = backOffArray.reduce((a, b) => {
+            return a + b;
+        }, 0);
+        it.skip("should retry with 'exponential' back off due to 429", async () => {
+            let startTime = await Date.now();
+            let res = await requester._get({
+                url: 'https://70e18ff0-1967-4fb5-8f96-10477ab6bb9e.mock.pstmn.io//429',
+            });
+            let endTime = await Date.now();
+            let difference = endTime - startTime;
+            expect(difference).toBeGreaterThan(sum * 1000);
+        });
+
+        it.skip("should retry with 'exponential' back off due to 500", async () => {
+            let startTime = await Date.now();
+            let res = await requester._get({
+                url: 'https://70e18ff0-1967-4fb5-8f96-10477ab6bb9e.mock.pstmn.io//5xx',
+            });
+            let endTime = await Date.now();
+            let difference = endTime - startTime;
+            expect(difference).toBeGreaterThan(sum * 1000);
+        });
+    });
+
+    describe('requestTimeoutMs configuration', () => {
+        it('defaults to 60000ms', () => {
+            const requester = new TestRequester({});
+            expect(requester.requestTimeoutMs).toBe(60_000);
+        });
+
+        it('honors an instance-level override', () => {
+            const requester = new TestRequester({ requestTimeoutMs: 30_000 });
+            expect(requester.requestTimeoutMs).toBe(30_000);
+        });
+
+        it('honors a class-level static override', () => {
+            class TighterRequester extends Requester {
+                static requestTimeoutMs = 15_000;
+            }
+            const requester = new TighterRequester({});
+            expect(requester.requestTimeoutMs).toBe(15_000);
+        });
+
+        it('prefers instance param over class static', () => {
+            class TighterRequester extends Requester {
+                static requestTimeoutMs = 15_000;
+            }
+            const requester = new TighterRequester({ requestTimeoutMs: 5_000 });
+            expect(requester.requestTimeoutMs).toBe(5_000);
+        });
+
+        it('accepts 0 to disable the timeout', () => {
+            const requester = new TestRequester({ requestTimeoutMs: 0 });
+            expect(requester.requestTimeoutMs).toBe(0);
+        });
+
+        it('accepts null to fall back to the class/default', () => {
+            class TighterRequester extends Requester {
+                static requestTimeoutMs = 15_000;
+            }
+            const requester = new TighterRequester({ requestTimeoutMs: null });
+            expect(requester.requestTimeoutMs).toBe(15_000);
+        });
     });
 
-    it.skip("should retry with 'exponential' back off due to 500", async () => {
-        let startTime = await Date.now();
-        let res = await requester._get({
-            url: 'https://70e18ff0-1967-4fb5-8f96-10477ab6bb9e.mock.pstmn.io//5xx',
+    describe('AbortController timeout behavior', () => {
+        /**
+         * Fetch that honors the AbortSignal: rejects with an AbortError when
+         * the signal fires, otherwise never resolves (simulating a hung
+         * upstream). Lets jest's fake-timer machinery drive the abort.
+         */
+        function hangingFetch() {
+            return jest.fn((_url, options) => {
+                return new Promise((_resolve, reject) => {
+                    if (!options?.signal) return; // disabled timeout path
+                    options.signal.addEventListener('abort', () => {
+                        const err = new Error('The user aborted a request.');
+                        err.name = 'AbortError';
+                        reject(err);
+                    });
+                });
+            });
+        }
+
+        function okFetch(body = { ok: true }) {
+            return jest.fn().mockResolvedValue({
+                status: 200,
+                headers: { get: () => 'application/json' },
+                json: async () => body,
+            });
+        }
+
+        it('aborts and throws a FetchError when fetch never resolves', async () => {
+            jest.useFakeTimers();
+            try {
+                const fetchMock = hangingFetch();
+                const requester = new TestRequester({
+                    requestTimeoutMs: 100,
+                    fetch: fetchMock,
+                });
+                const p = requester._get({ url: 'https://example.com/slow' });
+                p.catch(() => {}); // prevent unhandled-rejection noise
+                await jest.advanceTimersByTimeAsync(101);
+                await expect(p).rejects.toMatchObject({
+                    isTimeout: true,
+                    timeoutMs: 100,
+                });
+                expect(fetchMock).toHaveBeenCalledTimes(1);
+            } finally {
+                jest.useRealTimers();
+            }
+        });
+
+        it('does not retry on timeout (single attempt only)', async () => {
+            jest.useFakeTimers();
+            try {
+                const fetchMock = hangingFetch();
+                const requester = new TestRequester({
+                    requestTimeoutMs: 50,
+                    backOff: [1, 1, 1],
+                    fetch: fetchMock,
+                });
+                const p = requester._get({ url: 'https://example.com/hang' });
+                p.catch(() => {});
+                await jest.advanceTimersByTimeAsync(60);
+                await expect(p).rejects.toMatchObject({ isTimeout: true });
+                expect(fetchMock).toHaveBeenCalledTimes(1);
+            } finally {
+                jest.useRealTimers();
+            }
+        });
+
+        it('fires the timeout when fetch hangs past requestTimeoutMs', async () => {
+            jest.useFakeTimers();
+            try {
+                const requester = new TestRequester({
+                    requestTimeoutMs: 200,
+                    fetch: hangingFetch(),
+                });
+                const p = requester._get({
+                    url: 'https://example.com/slow',
+                });
+                p.catch(() => {});
+                // Advance just short of the timeout — still pending.
+                await jest.advanceTimersByTimeAsync(150);
+                // ...then past it — should now reject.
+                await jest.advanceTimersByTimeAsync(100);
+                await expect(p).rejects.toMatchObject({
+                    isTimeout: true,
+                    timeoutMs: 200,
+                });
+            } finally {
+                jest.useRealTimers();
+            }
+        });
+
+        it('does not wrap fetch in a timeout when requestTimeoutMs is 0', async () => {
+            // When disabled, no AbortController is wired in — the fetch
+            // options should not carry a `signal`.
+            const fetchMock = jest.fn(async (_url, options) => {
+                expect(options.signal).toBeUndefined();
+                return {
+                    status: 200,
+                    headers: { get: () => 'application/json' },
+                    json: async () => ({ ok: true }),
+                };
+            });
+            const requester = new TestRequester({
+                requestTimeoutMs: 0,
+                fetch: fetchMock,
+            });
+            const result = await requester._get({
+                url: 'https://example.com/ok',
+            });
+            expect(result).toEqual({ ok: true });
+            expect(fetchMock).toHaveBeenCalledTimes(1);
+        });
+
+        it('passes a signal to fetch when timeout is enabled', async () => {
+            const fetchMock = jest.fn(async (_url, options) => {
+                expect(options.signal).toBeDefined();
+                expect(typeof options.signal.addEventListener).toBe('function');
+                return {
+                    status: 200,
+                    headers: { get: () => 'application/json' },
+                    json: async () => ({ ok: true }),
+                };
+            });
+            const requester = new TestRequester({
+                requestTimeoutMs: 30_000,
+                fetch: fetchMock,
+            });
+            await requester._get({ url: 'https://example.com/ok' });
+            expect(fetchMock).toHaveBeenCalledTimes(1);
+        });
+
+        it('does not time out a fast, successful response', async () => {
+            const fetchMock = okFetch({ data: 'fresh' });
+            const requester = new TestRequester({
+                requestTimeoutMs: 5_000,
+                fetch: fetchMock,
+            });
+            const result = await requester._get({
+                url: 'https://example.com/fast',
+            });
+            expect(result).toEqual({ data: 'fresh' });
+        });
+
+        it('clears the timer once the fetch resolves so long-running processes do not leak timers', async () => {
+            const fetchMock = okFetch();
+            const requester = new TestRequester({
+                requestTimeoutMs: 5_000,
+                fetch: fetchMock,
+            });
+            // If we did not clear the timer, the Node event loop would keep
+            // a reference and this assertion is a smoke test for that: the
+            // call should resolve in real time, not after 5s.
+            const start = Date.now();
+            await requester._get({ url: 'https://example.com/ok' });
+            expect(Date.now() - start).toBeLessThan(500);
+        });
+    });
+
+    describe('ECONNRESET retry (regression guard)', () => {
+        it('still retries on ECONNRESET following the backOff schedule', async () => {
+            jest.useFakeTimers();
+            try {
+                let attempts = 0;
+                const fetchMock = jest.fn(async () => {
+                    attempts++;
+                    if (attempts <= 2) {
+                        const err = new Error('socket hang up');
+                        err.code = 'ECONNRESET';
+                        throw err;
+                    }
+                    return {
+                        status: 200,
+                        headers: { get: () => 'application/json' },
+                        json: async () => ({ ok: true }),
+                    };
+                });
+                const requester = new TestRequester({
+                    fetch: fetchMock,
+                    backOff: [0, 0, 0],
+                    requestTimeoutMs: 10_000,
+                });
+
+                const p = requester._get({ url: 'https://example.com/flaky' });
+                // Backoffs are 0s, so just flush queued timeouts/microtasks.
+                await jest.runAllTimersAsync();
+                const result = await p;
+                expect(result).toEqual({ ok: true });
+                expect(fetchMock).toHaveBeenCalledTimes(3);
+            } finally {
+                jest.useRealTimers();
+            }
         });
-        let endTime = await Date.now();
-        let difference = endTime - startTime;
-        expect(difference).toBeGreaterThan(sum * 1000);
     });
 });