From aa88a09d94ea208522729ce75a9b22dc3c8d1cb9 Mon Sep 17 00:00:00 2001 From: CD Cabrera Date: Wed, 19 Aug 2026 11:14:12 -0400 Subject: [PATCH 1/3] fix(deferTask): pf-4402 continueOnError, schedule padding --- .../__snapshots__/server.task.test.ts.snap | 34 ++++++- src/__tests__/server.task.test.ts | 56 ++++++++++- src/server.task.ts | 97 +++++++++++++------ 3 files changed, 150 insertions(+), 37 deletions(-) diff --git a/src/__tests__/__snapshots__/server.task.test.ts.snap b/src/__tests__/__snapshots__/server.task.test.ts.snap index d45d528c..a339fd21 100644 --- a/src/__tests__/__snapshots__/server.task.test.ts.snap +++ b/src/__tests__/__snapshots__/server.task.test.ts.snap @@ -6,7 +6,7 @@ exports[`deferTask should cancel a task 1`] = ` "type": "isRunning", "value": { "controller": AbortController {}, - "count": 1, + "count": 3, "isRunning": false, "promise": Promise {}, }, @@ -15,7 +15,7 @@ exports[`deferTask should cancel a task 1`] = ` "type": "start", "value": { "controller": AbortController {}, - "count": 1, + "count": 3, "isRunning": false, "promise": Promise {}, }, @@ -24,7 +24,35 @@ exports[`deferTask should cancel a task 1`] = ` "type": "run", "value": { "controller": AbortController {}, - "count": 1, + "count": 3, + "isRunning": false, + "promise": Promise {}, + }, + }, + { + "type": "run", + "value": { + "controller": AbortController {}, + "count": 3, + "isRunning": false, + "promise": Promise {}, + }, + }, + { + "type": "run", + "value": { + "controller": AbortController {}, + "count": 3, + "isRunning": false, + "promise": Promise {}, + }, + }, + { + "type": "run:cancel", + "value": { + "controller": AbortController {}, + "count": 3, + "error": [Error: Task canceled], "isRunning": false, "promise": Promise {}, }, diff --git a/src/__tests__/server.task.test.ts b/src/__tests__/server.task.test.ts index 976e2f0b..0347cc80 100644 --- a/src/__tests__/server.task.test.ts +++ b/src/__tests__/server.task.test.ts @@ -1,4 +1,5 @@ import { deferTask, delay } from '../server.task'; +import { log } from '../logger'; describe('deferTask', () => { beforeEach(() => { @@ -71,21 +72,72 @@ describe('deferTask', () => { it('should cancel a task', async () => { const mockDebug = jest.fn(); const mockFunc = jest.fn().mockReturnValue('lorem ipsum'); - const handle = deferTask(mockFunc, { debug: mockDebug, repeat: 3, cancelMs: 100, intervalMs: 110 })(); + const handle = deferTask(mockFunc, { debug: mockDebug, repeat: 5, cancelMs: 250, intervalMs: 100 })(); expect(handle.isRunning()).toBe(false); await Promise.allSettled([ handle.start(), - jest.advanceTimersByTimeAsync(85) + jest.advanceTimersByTimeAsync(300) ]); expect(mockDebug.mock.calls.map(arr => ({ type: arr[0].type, value: arr[0].value() }))).toMatchSnapshot(); expect(handle.isRunning()).toBe(false); + expect(mockFunc).toHaveBeenCalledTimes(3); + }); + + it('should disable cutoff and warn if cancelMs <= intervalMs', async () => { + const warnSpy = jest.spyOn(log, 'warn').mockImplementation(() => {}); + const mockFunc = jest.fn().mockReturnValue('ok'); + const handle = deferTask(mockFunc, { repeat: 1, cancelMs: 100, intervalMs: 100 })(); + const result = await handle.start(); + + expect(result).toBe('ok'); + expect(warnSpy).toHaveBeenCalledWith( + expect.stringContaining('cancelMs (100) must be greater than intervalMs (100)') + ); expect(mockFunc).toHaveBeenCalledTimes(1); }); + it('should repeat indefinitely when repeat is undefined until stopped', async () => { + const mockFunc = jest.fn().mockReturnValue('polling'); + const handle = deferTask(mockFunc, { repeat: Infinity, intervalMs: 100 })(); + + // Starts execution: Call 1 runs immediately at t = 0ms + handle.start(); + + // Advance timers for 2 intervals -> Call 2 (t=100ms) and Call 3 (t=200ms) + await jest.advanceTimersByTimeAsync(100); + await jest.advanceTimersByTimeAsync(100); + + expect(mockFunc).toHaveBeenCalledTimes(3); + expect(handle.isRunning()).toBe(true); + + // Stop the loop + await handle.stop(); + expect(handle.isRunning()).toBe(false); + }); + + it('should continue loop on error when continueOnError is true', async () => { + const errorSpy = jest.spyOn(log, 'error').mockImplementation(() => {}); + const mockFunc = jest + .fn() + .mockRejectedValueOnce(new Error('transient network failure')) + .mockResolvedValue('recovered'); + const handle = deferTask(mockFunc, { repeat: 2, intervalMs: 100, continueOnError: true })(); + const runPromise = handle.start(); + + // Advance past iteration 1 error delay to iteration 2 + await jest.advanceTimersByTimeAsync(100); + + const result = await runPromise; + + expect(result).toBe('recovered'); + expect(mockFunc).toHaveBeenCalledTimes(2); + expect(errorSpy).toHaveBeenCalledWith('Defer task error', expect.any(Error)); + }); + it('should enforce a timeout', async () => { const mockDebug = jest.fn(); const mockFunc = jest.fn().mockImplementation(() => new Promise(resolve => setTimeout(resolve, 500))); diff --git a/src/server.task.ts b/src/server.task.ts index 1d51a360..3f7975f0 100644 --- a/src/server.task.ts +++ b/src/server.task.ts @@ -35,16 +35,24 @@ interface DeferTaskHandle { * Options for the deferred task. * * @property [cancelMs] - Hard ms cutoff for cancellation. `undefined` - * disables the cutoff. (default `undefined`) + * disables the cutoff. Must be greater than `intervalMs` when set + * otherwise a warning is logged and the cutoff is disabled (default `undefined`) + * @property [continueOnError] - When `true`, a run error (including a per-run timeout) is + * logged and the repeat loop continues instead of rejecting out of `start()`. + * Defaults to `false`. `stop()` and `cancelMs` still terminate the loop. * @property {DeferTaskDebugHandler} [debug] - Debug callback for lifecycle events. * See {@link deferTask}. * @property [intervalMs] - Max time for both per-execution timeout AND - * the randomized base delay between repetitions. (default `1000`) - * @property [repeat] - Number of loops. (default `1`) + * the randomized base delay between repetitions. The per-execution timeout is + * derived as `intervalMs * 1.5` so a run is never killed by the same value used + * for scheduling. (default `1000`) + * @property [repeat] - Number of loops. Pass `Infinity` explicitly to loop + * indefinitely. (default `1`) * @property [errorMessage] - Custom error for timeouts. (default `'Task timed out'`) */ interface DeferTaskOptions { cancelMs?: number; + continueOnError?: boolean; debug?: DeferTaskDebugHandler; intervalMs?: number; repeat?: number | undefined; @@ -92,22 +100,28 @@ const delay = ({ ms, signal }: { ms: number; signal?: AbortSignal | undefined }) * exposing `start()`, `stop()`, and `isRunning()`. * * Options: - * - `repeat`: number of executions. Defaults to `1`. Pass `undefined` to repeat + * - `repeat`: number of executions. Defaults to `1`. Pass `Infinity` to repeat * **indefinitely** until `stop()` is called, `cancelMs` fires, or the task throws. - * - `intervalMs`: per-execution timeout. Defaults to `1000` ms. Exceeding it rejects + * - `intervalMs`: randomized base delay between repetitions. Defaults to `1000` ms. + * The per-execution timeout is derived as `intervalMs * 1.5`; exceeding it rejects * `start()` with `errorMessage` and emits a `run:error` debug event. + * - `continueOnError`: when `true`, a run error (including a per-run timeout) is + * logged and the repeat loop continues instead of rejecting out of `start()`. + * Defaults to `false`. `stop()` and `cancelMs` still terminate the loop. * - `cancelMs`: hard cutoff across the entire `start()` lifetime. `undefined` (default) - * disables the cutoff. When it fires, `start()` rejects with `'Task canceled'` and - * emits a `run:cancel` debug event. + * disables the cutoff. Must be greater than `intervalMs` otherwise a warning is + * logged and the cutoff is disabled. When it fires, `start()` rejects with + * `'Task canceled'` and emits a `run:cancel` debug event. * - `errorMessage`: message used for the per-execution timeout rejection. * Defaults to `'Task timed out'`. * - `debug`: callback invoked for lifecycle events. Emitted `type` values: * `start`, `run`, `run:stopped`, `run:error`, `run:cancel`, `stop`, `stop:error`, * `isRunning`. `info.value` is a thunk returning a snapshot of internal state. * - * @note Repeating loops should yield between iterations (this implementation uses - * `await delay(intervalMs)`). This interval serves as both the per-execution - * timeout and as part of a randomized base delay before the next loop. Do not + * @note Repeating loops yield between iterations (this implementation uses + * `await delay(intervalMs)`). This interval is part of a randomized base delay + * before the next loop, while the per-execution timeout is derived as + * `intervalMs * 1.5` so it is always greater than the scheduling delay. Do not * recurse or loop back immediately after a fast synchronous execution when repeat * is unlimited (resource exhaustion). * @@ -124,35 +138,49 @@ const delay = ({ ms, signal }: { ms: number; signal?: AbortSignal | undefined }) * invocation produces an independent handle with its own running state. * * @example Basic use - * const handle = deferTask(pollFunc, { repeat: undefined, intervalMs: 5000 })(passedArgsToPollFunc); - * // Start the task - * void handle.start(); + * const handle = deferTask(pollFunc, { repeat: Infinity, intervalMs: 5000 })(passedArgsToPollFunc); + * // Start the task. Attach a rejection handler, task errors re-reject out of + * // `start()` and an unhandled rejection can crash the process. + * handle.start().catch(error => log.error('Task error', error)); * // Stop the task * await handle.stop(); * * @example Application pattern * // Function to poll * const pollFunc = async (passedArgsToPollFunc: string) => {} - * // Create a handle for the task - * pollFunc.deferTask = deferTask(pollFunc, { repeat: undefined, intervalMs: 5000 }); + * // Create a handle for the task. `continueOnError` keeps the loop alive + * // when a single run fails (errors are logged instead of re-rejecting). + * pollFunc.deferTask = deferTask(pollFunc, { repeat: undefined, intervalMs: 5000, continueOnError: true }); * - * // Start the task. - * void pollFunc.deferTask.start(passedArgsToPollFunc); + * // Start the task. Attach a rejection handler, an unhandled rejection can crash the process. + * pollFunc.deferTask.start(passedArgsToPollFunc).catch(error => log.error('Task error', error)); * // Stop the task * await pollFunc.deferTask.stop(); */ const deferTask = ( func: ((...args: TArgs) => TReturn | Promise) | Promise, - { + options: DeferTaskOptions = {} +) => { + const { cancelMs, + continueOnError = false, debug = () => {}, - repeat = 1, + repeat, intervalMs, errorMessage = 'Task timed out' - }: DeferTaskOptions = {} -) => { - const updatedRepeat = typeof repeat === 'number' ? repeat : undefined; + } = options || {}; + + const validRepeat = typeof repeat === 'number' && repeat > 0 ? repeat : 1; + const updatedRepeat = Number.isFinite(validRepeat) ? validRepeat : undefined; const updatedIntervalMs = intervalMs ?? 1000; + const runTimeoutMs = updatedIntervalMs * 1.5; + let updatedCancelMs = cancelMs; + + if (updatedCancelMs !== undefined && updatedCancelMs <= updatedIntervalMs) { + log.warn(`Defer task cancelMs (${updatedCancelMs}) must be greater than intervalMs (${updatedIntervalMs}). Cutoff disabled.`); + updatedCancelMs = undefined; + } + const updatedFunc = async (...args: TArgs) => (!isAsync(func) && isPromise(func) ? func as Promise : (func as (...args: TArgs) => TReturn | Promise)(...args)); @@ -188,7 +216,7 @@ const deferTask = ( return undefined; }, { - timeout: updatedIntervalMs, + timeout: runTimeoutMs, errorMessage }); @@ -198,6 +226,17 @@ const deferTask = ( }); const result = await startFunc.catch(error => { + if (continueOnError && (updatedRepeat === undefined || state.count < updatedRepeat)) { + debug({ + type: 'run:error', + value: () => ({ ...state, error }) + }); + + log.error('Defer task error', error); + + return undefined; + } + state.isRunning = false; debug({ @@ -248,15 +287,9 @@ const deferTask = ( value: () => ({ ...state }) }); - if (cancelMs !== undefined) { - updatedTask = timeoutFunction(() => { - const response = task(); - - state.isRunning = false; - - return response; - }, { - timeout: cancelMs, + if (updatedCancelMs !== undefined) { + updatedTask = timeoutFunction(() => task(), { + timeout: updatedCancelMs, errorMessage: 'Task canceled' }).catch(error => { state.isRunning = false; From 6bab9e9d5be7eebe058f413818a4c58c64e8f1c2 Mon Sep 17 00:00:00 2001 From: CD Cabrera Date: Wed, 19 Aug 2026 15:35:51 -0400 Subject: [PATCH 2/3] fix: review update --- src/server.stats.ts | 14 ++++++-------- 1 file changed, 6 insertions(+), 8 deletions(-) diff --git a/src/server.stats.ts b/src/server.stats.ts index 36507280..1536858c 100644 --- a/src/server.stats.ts +++ b/src/server.stats.ts @@ -51,12 +51,11 @@ const healthReport = (statsOptions: StatsSession) => { /** * Task for `healthReport`. - * - * @note `undefined` repeat means the task will run indefinitely. */ healthReport.deferTask = deferTask(healthReport, { intervalMs: DEFAULT_OPTIONS.stats.reportIntervalMs.health, - repeat: undefined + repeat: Infinity, + continueOnError: true }); /** @@ -101,12 +100,11 @@ const transportReport = ( /** * Task for `transportReport`. - * - * @note `undefined` repeat means the task will run indefinitely. */ transportReport.deferTask = deferTask(transportReport, { intervalMs: DEFAULT_OPTIONS.stats.reportIntervalMs.transport, - repeat: undefined + repeat: Infinity, + continueOnError: true }); /** @@ -147,10 +145,10 @@ const createServerStats = (statsOptions = getStatsOptions(), options = getOption const httpPort = options.isHttp ? httpHandle?.port : undefined; const stats = statsReport({ httpPort }, statsOptions); - // Start the health report. Defining repeat as undefined keeps the loop infinite. + // Start the health report. healthTask = healthReport.deferTask(statsOptions); - // Start the transport report. Defining repeat as undefined keeps the loop infinite. + // Start the transport report. transportTask = transportReport.deferTask({ httpPort }, statsOptions); void healthTask.start(); From 8f755ff2ca96a919b65829197b4cafaf1d637b47 Mon Sep 17 00:00:00 2001 From: CD Cabrera Date: Wed, 19 Aug 2026 15:40:54 -0400 Subject: [PATCH 3/3] fix: review update --- src/__tests__/server.task.test.ts | 2 +- src/server.task.ts | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/src/__tests__/server.task.test.ts b/src/__tests__/server.task.test.ts index 0347cc80..b2a75849 100644 --- a/src/__tests__/server.task.test.ts +++ b/src/__tests__/server.task.test.ts @@ -100,7 +100,7 @@ describe('deferTask', () => { expect(mockFunc).toHaveBeenCalledTimes(1); }); - it('should repeat indefinitely when repeat is undefined until stopped', async () => { + it('should repeat indefinitely when repeat is Infinity until stopped', async () => { const mockFunc = jest.fn().mockReturnValue('polling'); const handle = deferTask(mockFunc, { repeat: Infinity, intervalMs: 100 })(); diff --git a/src/server.task.ts b/src/server.task.ts index 3f7975f0..68feb2b7 100644 --- a/src/server.task.ts +++ b/src/server.task.ts @@ -150,7 +150,7 @@ const delay = ({ ms, signal }: { ms: number; signal?: AbortSignal | undefined }) * const pollFunc = async (passedArgsToPollFunc: string) => {} * // Create a handle for the task. `continueOnError` keeps the loop alive * // when a single run fails (errors are logged instead of re-rejecting). - * pollFunc.deferTask = deferTask(pollFunc, { repeat: undefined, intervalMs: 5000, continueOnError: true }); + * pollFunc.deferTask = deferTask(pollFunc, { repeat: Infinity, intervalMs: 5000, continueOnError: true }); * * // Start the task. Attach a rejection handler, an unhandled rejection can crash the process. * pollFunc.deferTask.start(passedArgsToPollFunc).catch(error => log.error('Task error', error));