diff --git a/docs/.vitepress/config.mts b/docs/.vitepress/config.mts index 2ed240f..89d1857 100644 --- a/docs/.vitepress/config.mts +++ b/docs/.vitepress/config.mts @@ -66,6 +66,7 @@ export default defineConfig({ { text: 'Trace', link: '/execution/trace' }, { text: 'Cache', link: '/execution/cache' }, { text: 'Memoize', link: '/execution/memoize' }, + { text: 'Retry', link: '/execution/retry' }, { text: 'Timer', link: '/execution/timer' } ] }, diff --git a/docs/execution/retry.md b/docs/execution/retry.md new file mode 100644 index 0000000..18e4273 --- /dev/null +++ b/docs/execution/retry.md @@ -0,0 +1,169 @@ +# Retry + +`retry(fn, options?)` wraps a function with bounded retries. `@retry(options?)` applies the same policy to a method. +Both accept synchronous throws, rejected promises, and successful results. Every invocation returns a Promise, +including when the original function is synchronous. Declare decorated methods `async` so their TypeScript return +types match runtime behavior. + +## Usage + +```ts +import { retry } from 'execution-engine'; + +const getInvoice = retry( + async (invoiceId: string) => { + const response = await fetch(`/invoices/${invoiceId}`); + if (!response.ok) throw new Error(`HTTP ${response.status}`); + return response.json(); + }, + { + attempts: 3, + backoff: 'exponential', + initialDelay: 200, + maxDelay: 2_000, + jitter: true, + shouldRetry: ({ error }) => error instanceof Error && /^HTTP 5\d\d$/.test(error.message), + onRetry: ({ attempt, delay }) => console.log(`Attempt ${attempt} failed; retrying after ${delay} ms`) + } +); + +const invoice = await getInvoice('inv-123'); +``` + +The hooks receive a context object, rather than positional `error, attempt` arguments. Attempts count executions, +so `attempts: 3` permits at most two retries. The first execution starts without a delay. + +### Decorator + +```ts +import { retry } from 'execution-engine'; + +class BillingService { + constructor(private readonly baseUrl: string) {} + + @retry({ attempts: 3, backoff: 'fixed', initialDelay: 200 }) + async getInvoice(invoiceId: string): Promise { + const response = await fetch(`${this.baseUrl}/invoices/${invoiceId}`); + if (!response.ok) throw new Error(`HTTP ${response.status}`); + return response.json(); + } +} +``` + +Enable `experimentalDecorators`. Both forms preserve the receiver and forward the original arguments to every +attempt. Attempt counts, results and cancellation are isolated between concurrent calls and instances. The policy +is copied when the wrapper or decorator is created. + +### Cancellation per call + +The function wrapper exposes `withOptions({ signal }, ...args)`: + +```ts +const controller = new AbortController(); +const pending = getInvoice.withOptions({ signal: controller.signal }, 'inv-123'); +controller.abort(); +await pending; // Rejects with controller.signal.reason (AbortError by default). +``` + +For a function that needs a receiver, use +`wrapped.withOptions.call(receiver, { signal }, ...args)`. The signal is call configuration and is not appended +to the original function's arguments. + +For decorators, `getSignal(args)` selects a signal from the method's own arguments. This also lets the method +pass it to an operation that supports cancellation: + +```ts +class BillingService { + @retry({ getSignal: (args) => args[1] as AbortSignal | undefined }) + async getInvoice(invoiceId: string, signal?: AbortSignal): Promise { + const response = await fetch(`/invoices/${invoiceId}`, { signal }); + if (!response.ok) throw new Error(`HTTP ${response.status}`); + return response.json(); + } +} +``` + +`getSignal` is also available to function wrappers; an explicit `withOptions` signal takes precedence. +An already aborted call does not execute the function. Cancellation clears a pending delay and rejects promptly +while an operation, predicate, or hook is pending. It removes its abort listeners and starts no further attempts. +Retry cannot interrupt the underlying work itself; pass the signal to that work when cancellation is supported. + +## API + +### `RetryOptions` + +| Option | Default | Behavior | +| --- | --- | --- | +| `attempts` | `3` | Positive safe integer; includes the first execution. | +| `backoff` | `'exponential'` | `'fixed'`, `'exponential'`, a fixed delay in milliseconds, or `(context: RetryContext) => number`. | +| `initialDelay` | `100` | Milliseconds for fixed backoff or the first exponential delay. Ignored by numeric and custom backoff. | +| `maxDelay` | `2_147_483_647` | Caps every strategy before jitter; matches the largest supported timer delay. | +| `jitter` | `false` | Full jitter: an integer between zero and the floored capped delay, inclusive. | +| `shouldRetry` | Retry errors | `(context: RetryContext) => boolean \| Promise`. Can also request retries of results. | +| `onRetry` | — | `(context: RetryContext) => void \| Promise`, awaited before the delay. | +| `getSignal` | — | `(args: readonly unknown[]) => AbortSignal \| undefined`, resolved once per call. | +| `engine` | Inferred when available | `TraceableEngine` for recording the operation and its attempts. | + +Exponential backoff uses `initialDelay * 2 ** (attempt - 1)`. Numeric delay options and custom strategy results must +be finite numbers between zero and `2_147_483_647`; invalid values throw `RangeError`. Policy options are validated +when creating the wrapper; custom delays are validated when evaluated. A zero delay still yields through a timer. + +```ts +import type { RetryContext, RetryBackoff } from 'execution-engine'; + +const backoff: RetryBackoff = ({ attempt }) => attempt * 250; + +// The predicate can retry successful results as well as errors. +const poll = retry(() => fetch('/job/status'), { + backoff, + shouldRetry: async ({ error, result }) => error !== undefined || result?.status === 202 +}); +``` + +### `RetryContext` + +```ts +interface RetryContext { + attempt: number; + error?: unknown; + result?: Result; + delay: number; +} +``` + +`attempt` is the completed, one-based execution. A failure context contains `error` (even if the thrown value +was `undefined`); a success context contains `result`. A custom backoff receives `delay: 0`; `shouldRetry` and +`onRetry` receive the computed, capped and jittered delay. Custom backoff runs before the predicate. Neither +hook nor the strategy runs after the final allowed execution, or after an accepted result with no predicate. + +Exhaustion rejects with the exact last thrown value, preserving an Error's identity and stack. A predicate returning +`false` stops immediately and returns the result or rejects with the current error. When result retries exhaust +the attempt budget, the final result is returned. Errors from strategies, predicates and hooks stop the call +immediately and propagate without being retried. + +`RetryFunction` describes the Promise-returning wrapper and its `withOptions` method. +`RetryCallOptions` contains the optional per-call `signal`. All retry functions and types are exported from the +package entry point. + +## Trace integration + +Pass `{ engine }`, invoke the wrapper with a `TraceableEngine` receiver, or decorate a method whose receiver has +a `TraceableEngine` at `this.engine`. Retry records one parent operation labeled `retry ` and one +child per execution labeled `attempt 1`, `attempt 2`, and so on. Nested calls retain the engine's current parent. + +```ts +import { retry, TraceableEngine } from 'execution-engine'; + +const engine = new TraceableEngine(); +const getStatus = retry(() => fetch('/status'), { engine }); +await getStatus(); +const graph = engine.getTrace(); +``` + +Attempt nodes record inputs, outputs or serialized failure reasons, and timing. Retried attempts have a narrative +containing the planned delay and whether the retry came from an error or result. The parent records the final +output or error and a success, failure, or cancellation narrative. Aborted delays do not create another attempt. +Trace errors use the engine's existing serialization; the rejected error remains the original value. + +Without an engine, retry executes without creating a graph. Retry trace nodes explicitly record full execution +details; any enclosing node's extraction settings still apply to that enclosing node. diff --git a/docs/guide/which-api-to-use.md b/docs/guide/which-api-to-use.md index 7188db8..a5c7ae0 100644 --- a/docs/guide/which-api-to-use.md +++ b/docs/guide/which-api-to-use.md @@ -54,13 +54,14 @@ Same graph, different ergonomics: ## Decorator or plain function? -Tracing, caching and memoization each have a decorator and a plain-function API, backed by the same implementation. +Tracing, caching, memoization and retry each have a decorator and a plain-function API, backed by the same implementation. | Decorator | Plain function | | --- | --- | | [`@trace`](../execution/trace.md#trace-decorator) | [`executionTrace`](../execution/trace.md#executiontrace) | | [`@cache`](../execution/cache.md#cache-decorator) | [`executeCache`](../execution/cache.md#executecache) | | [`@memoize`](../execution/memoize.md#memoize-decorator) | [`executeMemoize`](../execution/memoize.md#executememoize) | +| [`@retry`](../execution/retry.md#decorator) | [`retry`](../execution/retry.md) | Use the plain function when you are not writing classes, cannot enable `experimentalDecorators`, or need to wrap someone else's function. `executeCache` and `executeMemoize` additionally require a `functionId`, since there is no diff --git a/docs/reference/types.md b/docs/reference/types.md index 4f7d363..8686904 100644 --- a/docs/reference/types.md +++ b/docs/reference/types.md @@ -215,6 +215,11 @@ The memoization module also exports its bounds as constants: `memoizationDefault immediately on fulfillment or rejection. Synchronous results keep timer-based expiration, including a zero-delay timer with `ttl: 0`. +## Retry + +[`RetryOptions`, `RetryContext`, `RetryBackoff`, `RetryFunction`, and `RetryCallOptions`](../execution/retry.md#api) +describe bounded retries, hooks, backoff strategies and per-call cancellation. + ## Type guards | Guard | Narrows to | diff --git a/src/execution/retry.decorator.spec.ts b/src/execution/retry.decorator.spec.ts new file mode 100644 index 0000000..ea6ec02 --- /dev/null +++ b/src/execution/retry.decorator.spec.ts @@ -0,0 +1,103 @@ +import { retry } from './retry'; +import { engine, EngineTask, run } from '../engine/executionEngineDecorators'; +import { TraceableEngine } from '../engine/traceableEngine'; + +describe('@retry', () => { + beforeEach(() => jest.useFakeTimers()); + afterEach(() => jest.useRealTimers()); + + it('preserves arguments, instance state and independent calls', async () => { + class Service { + calls = 0; + constructor(readonly prefix: string) {} + + @retry({ backoff: 0 }) + async get(id: number): Promise { + this.calls++; + if (this.calls < 2) { + throw new Error('transient'); + } + return `${this.prefix}-${id}`; + } + } + const first = new Service('first'); + const second = new Service('second'); + const results = Promise.all([first.get(1), second.get(2)]); + await jest.runAllTimersAsync(); + await expect(results).resolves.toEqual(['first-1', 'second-2']); + expect(first.calls).toBe(2); + expect(second.calls).toBe(2); + }); + + it('selects cancellation from each method call without changing the policy', async () => { + class Service { + calls = 0; + + @retry({ getSignal: (args) => args[1] as AbortSignal, backoff: 10 }) + async get(id: string, signal?: AbortSignal): Promise { + this.calls++; + if (signal) { + throw new Error('retry'); + } + return id; + } + } + const service = new Service(); + const controller = new AbortController(); + const cancelled = service.get('first', controller.signal); + const rejected = expect(cancelled).rejects.toBe('stop'); + await jest.advanceTimersByTimeAsync(0); + controller.abort('stop'); + await rejected; + await expect(service.get('second')).resolves.toBe('second'); + expect(service.calls).toBe(2); + expect(jest.getTimerCount()).toBe(0); + }); + + it('supports symbol methods and discovers this.engine for tracing', async () => { + const method = Symbol('get'); + class Service { + engine = new TraceableEngine(); + calls = 0; + + @retry({ backoff: 0 }) + async [method](): Promise { + this.calls++; + if (this.calls < 2) { + throw new Error('temporary'); + } + return 42; + } + } + const service = new Service(); + const result = service[method](); + await jest.runAllTimersAsync(); + await expect(result).resolves.toBe(42); + expect(service.engine.getTraceNodes()).toHaveLength(3); + expect(service.engine.getTraceNodes().filter(({ data }) => data.parent)).toHaveLength(2); + }); + + it('composes with @run and @engine without losing the Promise result', async () => { + @engine() + class Service extends EngineTask { + calls = 0; + + @run({ trace: { label: 'service' } }) + @retry({ backoff: 0 }) + async get(): Promise { + this.calls++; + if (this.calls === 1) { + throw new Error('temporary'); + } + return 42; + } + } + const service = new Service(); + const result = service.get(); + await jest.runAllTimersAsync(); + await expect(result).resolves.toBe(42); + const nodes = service.engine.getTraceNodes().map(({ data }) => data); + expect(nodes).toHaveLength(4); + expect(nodes.find((node) => node.label.startsWith('retry ')).parent).toBe(nodes.find((node) => node.label === 'service').id); + }); +}); diff --git a/src/execution/retry.spec.ts b/src/execution/retry.spec.ts new file mode 100644 index 0000000..ed123a9 --- /dev/null +++ b/src/execution/retry.spec.ts @@ -0,0 +1,369 @@ +import { retry, RetryContext } from './retry'; +import { TraceableEngine } from '../engine/traceableEngine'; + +describe('retry', () => { + beforeEach(() => jest.useFakeTimers()); + afterEach(() => { + jest.restoreAllMocks(); + jest.useRealTimers(); + }); + + it('preserves typed arguments, receiver and falsy sync results', async () => { + const receiver = { + prefix: 'invoice', + get: retry(function (this: { prefix: string }, id: number, enabled: boolean): string { + return enabled ? `${this.prefix}-${id}` : ''; + }) + }; + const result: Promise = receiver.get(42, true); + await expect(result).resolves.toBe('invoice-42'); + await expect(receiver.get(0, false)).resolves.toBe(''); + expect(jest.getTimerCount()).toBe(0); + }); + + it('retries synchronous throws using the default three attempts and exponential delays', async () => { + const error = new Error('temporary'); + const operation = jest + .fn() + .mockImplementationOnce(() => { + throw error; + }) + .mockImplementationOnce(() => { + throw error; + }) + .mockReturnValue(7); + const onRetry = jest.fn(); + const result = retry(operation, { onRetry })(); + await jest.advanceTimersByTimeAsync(99); + expect(operation).toHaveBeenCalledTimes(1); + await jest.advanceTimersByTimeAsync(1); + expect(operation).toHaveBeenCalledTimes(2); + await jest.advanceTimersByTimeAsync(199); + expect(operation).toHaveBeenCalledTimes(2); + await jest.advanceTimersByTimeAsync(1); + await expect(result).resolves.toBe(7); + expect(onRetry.mock.calls.map(([context]) => context.delay)).toEqual([100, 200]); + expect(jest.getTimerCount()).toBe(0); + }); + + it.each(['async', 'promise', 'sync'] as const)('preserves the final %s error identity and stack', async (kind) => { + const errors = [new Error('first'), new Error('last')]; + const stack = errors[1].stack; + let count = 0; + const operation = + kind === 'async' + ? async (): Promise => { + throw errors[count++]; + } + : kind === 'promise' + ? (): Promise => Promise.reject(errors[count++]) + : (): never => { + throw errors[count++]; + }; + const result = retry(operation, { attempts: 2, initialDelay: 10 })(); + const rejected = expect(result).rejects.toBe(errors[1]); + await jest.runAllTimersAsync(); + await rejected; + expect(count).toBe(2); + expect(errors[1].stack).toBe(stack); + }); + + it.each([undefined, null, 'failure', 0])('preserves thrown values including %s', async (error) => { + const result = retry( + () => { + throw error; + }, + { attempts: 1 } + )(); + await expect(result).rejects.toBe(error); + }); + + it('stops immediately for a non-retryable error', async () => { + const error = new Error('permanent'); + const operation = jest.fn().mockRejectedValue(error); + const shouldRetry = jest.fn(async (context: RetryContext) => context.error !== error); + const onRetry = jest.fn(); + await expect(retry(operation, { shouldRetry, onRetry })()).rejects.toBe(error); + expect(operation).toHaveBeenCalledTimes(1); + expect(shouldRetry).toHaveBeenCalledWith({ attempt: 1, error, delay: 100 }); + expect(onRetry).not.toHaveBeenCalled(); + expect(jest.getTimerCount()).toBe(0); + }); + + it.each([ + { backoff: 'fixed' as const, initialDelay: 20, expected: [20, 20, 20] }, + { backoff: 15, expected: [15, 15, 15] }, + { backoff: 'exponential' as const, initialDelay: 20, maxDelay: 30, expected: [20, 30, 30] }, + { backoff: ({ attempt }: RetryContext): number => attempt * 10, maxDelay: 25, expected: [10, 20, 25] } + ])('uses capped backoff $expected', async ({ expected, ...options }) => { + const operation = jest.fn().mockRejectedValue('failure'); + const onRetry = jest.fn(); + const result = retry(operation, { ...options, attempts: 4, onRetry })(); + const rejected = expect(result).rejects.toBe('failure'); + await jest.runAllTimersAsync(); + await rejected; + expect(onRetry.mock.calls.map(([context]) => context.delay)).toEqual(expected); + expect(onRetry).toHaveBeenCalledTimes(3); + }); + + it('applies full jitter after the maximum delay', async () => { + jest.spyOn(Math, 'random').mockReturnValue(0.5); + const onRetry = jest.fn(); + const operation = jest.fn().mockRejectedValueOnce('failure').mockResolvedValue(false); + const result = retry(operation, { backoff: 500, maxDelay: 100, jitter: true, onRetry })(); + await jest.runAllTimersAsync(); + await expect(result).resolves.toBe(false); + expect(onRetry).toHaveBeenCalledWith({ attempt: 1, error: 'failure', delay: 50 }); + }); + + it('allows async result predicates and returns the final result on exhaustion', async () => { + const operation = jest.fn().mockReturnValueOnce(0).mockReturnValueOnce(1).mockReturnValue(2); + const shouldRetry = jest.fn(async ({ result }: RetryContext) => result < 3); + const onRetry = jest.fn(); + const result = retry(operation, { shouldRetry, onRetry, backoff: 0 })(); + await jest.runAllTimersAsync(); + await expect(result).resolves.toBe(2); + expect(shouldRetry).toHaveBeenCalledTimes(2); + expect(shouldRetry.mock.calls[0][0]).toEqual({ attempt: 1, result: 0, delay: 0 }); + expect(onRetry).toHaveBeenCalledTimes(2); + }); + + it('returns a result immediately when the predicate accepts it', async () => { + const operation = jest.fn().mockReturnValue(undefined); + await expect(retry(operation, { shouldRetry: () => false })()).resolves.toBeUndefined(); + expect(operation).toHaveBeenCalledTimes(1); + }); + + it('awaits onRetry before starting the delay', async () => { + let release: () => void; + const hook = new Promise((resolve) => { + release = resolve; + }); + const operation = jest.fn().mockRejectedValueOnce('failure').mockResolvedValue(1); + const result = retry(operation, { onRetry: () => hook })(); + await jest.advanceTimersByTimeAsync(1000); + expect(operation).toHaveBeenCalledTimes(1); + expect(jest.getTimerCount()).toBe(0); + release(); + await jest.advanceTimersByTimeAsync(99); + expect(operation).toHaveBeenCalledTimes(1); + await jest.advanceTimersByTimeAsync(1); + await expect(result).resolves.toBe(1); + }); + + it.each(['shouldRetry', 'onRetry', 'backoff'] as const)('propagates errors from %s without retrying the hook', async (hook) => { + const error = new Error('hook failed'); + const operation = jest.fn().mockRejectedValue('original'); + const callback = jest.fn((): never => { + throw error; + }); + await expect(retry(operation, { [hook]: callback })()).rejects.toBe(error); + expect(operation).toHaveBeenCalledTimes(1); + expect(callback).toHaveBeenCalledTimes(1); + expect(jest.getTimerCount()).toBe(0); + }); + + it.each([0, -1, 1.5, NaN, Infinity, Number.MAX_SAFE_INTEGER + 1])('rejects invalid attempts %s at construction', (attempts) => { + expect(() => retry(() => 1, { attempts })).toThrow(RangeError); + }); + + it.each([-1, NaN, Infinity, 2_147_483_648])('rejects invalid delays %s', (delay) => { + for (const option of ['initialDelay', 'maxDelay', 'backoff']) { + expect(() => retry(() => 1, { [option]: delay })).toThrow(RangeError); + } + }); + + it('rejects invalid custom delays before scheduling a timer', async () => { + await expect( + retry( + () => { + throw 'failure'; + }, + { backoff: () => -1 } + )() + ).rejects.toThrow(RangeError); + expect(jest.getTimerCount()).toBe(0); + }); + + it('rejects an already aborted invocation without executing', async () => { + const controller = new AbortController(); + controller.abort(); + const operation = jest.fn(); + await expect(retry(operation).withOptions({ signal: controller.signal })).rejects.toMatchObject({ name: 'AbortError' }); + expect(operation).not.toHaveBeenCalled(); + }); + + it('cancels pending delays without affecting concurrent calls', async () => { + const controller = new AbortController(); + const cancelled = new Error('cancelled'); + const counts = new Map(); + const operation = jest.fn((id: string) => { + const count = (counts.get(id) ?? 0) + 1; + counts.set(id, count); + if (count < 2) { + throw new Error('retry'); + } + return id; + }); + const wrapped = retry(operation); + const first = wrapped.withOptions({ signal: controller.signal }, 'first'); + const rejected = expect(first).rejects.toBe(cancelled); + const second = wrapped('second'); + await jest.advanceTimersByTimeAsync(50); + controller.abort(cancelled); + await rejected; + expect(jest.getTimerCount()).toBe(1); + await jest.advanceTimersByTimeAsync(50); + await expect(second).resolves.toBe('second'); + expect(counts.get('first')).toBe(1); + expect(counts.get('second')).toBe(2); + expect(jest.getTimerCount()).toBe(0); + }); + + it.each(['operation', 'shouldRetry', 'onRetry'] as const)('cancels while %s is pending and removes listeners', async (stage) => { + const controller = new AbortController(); + const add = jest.spyOn(controller.signal, 'addEventListener'); + const remove = jest.spyOn(controller.signal, 'removeEventListener'); + const pending = new Promise(() => {}); + const operation = jest.fn(() => (stage === 'operation' ? pending : Promise.reject('failure'))); + const options = + stage === 'shouldRetry' + ? { shouldRetry: (): Promise => pending } + : stage === 'onRetry' + ? { onRetry: (): Promise => pending } + : {}; + const result = retry(operation, options).withOptions({ signal: controller.signal }); + const rejected = expect(result).rejects.toBe('cancelled'); + await jest.advanceTimersByTimeAsync(0); + controller.abort('cancelled'); + await rejected; + expect(operation).toHaveBeenCalledTimes(1); + expect(remove.mock.calls.length).toBe(add.mock.calls.length); + expect(jest.getTimerCount()).toBe(0); + }); + + it('removes abort listeners on success and preserves a receiver with call options', async () => { + const controller = new AbortController(); + const add = jest.spyOn(controller.signal, 'addEventListener'); + const remove = jest.spyOn(controller.signal, 'removeEventListener'); + const wrapped = retry(function (this: { id: string }, suffix: string): string { + return this.id + suffix; + }); + await expect(wrapped.withOptions.call({ id: 'a' }, { signal: controller.signal }, 'b')).resolves.toBe('ab'); + expect(remove.mock.calls.length).toBe(add.mock.calls.length); + }); + + it('records attempts, retry delays, failure reasons and the final outcome in the engine', async () => { + const engine = new TraceableEngine(); + const operation = jest.fn().mockRejectedValueOnce(new Error('temporary')).mockResolvedValue(42); + const result = retry(operation, { engine, initialDelay: 10 })(); + await jest.runAllTimersAsync(); + await expect(result).resolves.toBe(42); + const nodes = engine.getTraceNodes(); + const parent = nodes.find(({ data }) => data.label.startsWith('retry ')).data; + const children = nodes.filter(({ data }) => data.parent === parent.id).map(({ data }) => data); + expect(children).toHaveLength(2); + expect(children[0]).toMatchObject({ + label: 'attempt 1', + errors: [{ message: 'temporary' }], + narratives: ['Retry after 10 ms (error).'] + }); + expect(children[1]).toMatchObject({ label: 'attempt 2', outputs: 42 }); + expect(parent).toMatchObject({ outputs: 42, abstract: false }); + expect(parent.errors).toBeUndefined(); + expect(parent.narratives).toEqual(['Retry succeeded.']); + expect(engine.getTrace().filter((item) => item.group === 'edges')).toHaveLength(1); + }); + + it('infers the engine receiver and preserves nesting inside run', async () => { + const engine = new TraceableEngine(); + const result = await engine.run( + retry(function (this: TraceableEngine): number { + expect(this).toBe(engine); + return 1; + }), + [], + { trace: { label: 'outer' } } + ); + expect(result.outputs).toBe(1); + const nodes = engine.getTraceNodes().map(({ data }) => data); + const outer = nodes.find((node) => node.label === 'outer'); + const parent = nodes.find((node) => node.label.startsWith('retry ')); + expect(parent.parent).toBe(outer.id); + expect(nodes.find((node) => node.label === 'attempt 1').parent).toBe(parent.id); + }); + + it('traces cancellation and final failure without replacing the rejection', async () => { + const engine = new TraceableEngine(); + const controller = new AbortController(); + const result = retry( + () => { + throw new Error('temporary'); + }, + { engine } + ).withOptions({ signal: controller.signal }); + const rejected = expect(result).rejects.toBe('cancelled'); + await jest.advanceTimersByTimeAsync(10); + controller.abort('cancelled'); + await rejected; + const parent = engine.getTraceNodes().find(({ data }) => data.label.startsWith('retry ')).data; + expect(parent).toMatchObject({ errors: ['cancelled'], narratives: ['Retry cancelled.'], abstract: false }); + }); + + it('records the last failure on the trace parent after exhausting attempts', async () => { + const engine = new TraceableEngine(); + const error = new Error('final failure'); + const result = retry(() => { + throw error; + }, { engine, backoff: 0, attempts: 2 })(); + const rejected = expect(result).rejects.toBe(error); + await jest.runAllTimersAsync(); + await rejected; + const nodes = engine.getTraceNodes().map(({ data }) => data); + expect(nodes).toHaveLength(3); + expect(nodes.find((node) => node.label.startsWith('retry '))).toMatchObject({ + errors: [{ message: 'final failure' }], narratives: ['Retry failed.'], abstract: false + }); + }); + + it('passes the failure and receiver to custom backoff and both hooks', async () => { + const error = new Error('temporary'); + const receiver = { id: 'receiver' }; + const backoff = jest.fn(function (this: typeof receiver, context: RetryContext): number { + expect(this).toBe(receiver); + expect(context).toEqual({ attempt: 1, error, delay: 0 }); + return 20; + }); + const shouldRetry = jest.fn(function (this: typeof receiver, context: RetryContext): boolean { + expect(this).toBe(receiver); + return context.error === error; + }); + const onRetry = jest.fn(function (this: typeof receiver): void { + expect(this).toBe(receiver); + }); + const operation = jest.fn().mockRejectedValueOnce(error).mockResolvedValue(42); + const result = retry(operation, { backoff, shouldRetry, onRetry, attempts: 2 }).call(receiver); + await jest.runAllTimersAsync(); + await expect(result).resolves.toBe(42); + expect(onRetry).toHaveBeenCalledWith({ attempt: 1, error, delay: 20 }); + }); + + it('uses the explicit per-call signal instead of the argument selector', async () => { + const controller = new AbortController(); + controller.abort('selector signal'); + const getSignal = jest.fn(() => controller.signal); + const operation = jest.fn().mockReturnValue(42); + const wrapped = retry(operation, { getSignal }); + await expect(wrapped()).rejects.toBe('selector signal'); + await expect(wrapped.withOptions({ signal: new AbortController().signal })).resolves.toBe(42); + expect(getSignal).toHaveBeenCalledTimes(1); + expect(operation).toHaveBeenCalledTimes(1); + }); + + it('retries with real timers', async () => { + jest.useRealTimers(); + const operation = jest.fn().mockRejectedValueOnce('temporary').mockResolvedValue(42); + await expect(retry(operation, { initialDelay: 1 })()).resolves.toBe(42); + expect(operation).toHaveBeenCalledTimes(2); + }); +}); diff --git a/src/execution/retry.ts b/src/execution/retry.ts new file mode 100644 index 0000000..fc3aac3 --- /dev/null +++ b/src/execution/retry.ts @@ -0,0 +1,287 @@ +import { EngineNodeData } from '../common/models/engineNodeData.model'; +import { TraceableEngine } from '../engine/traceableEngine'; + +/** Context for the completed, one-based attempt and its proposed next delay in milliseconds. */ +export interface RetryContext { + attempt: number; + error?: unknown; + result?: Result; + delay: number; +} + +/** A number is a fixed delay in milliseconds. Custom strategies receive the completed attempt. */ +export type RetryBackoff = number | 'fixed' | 'exponential' | ((context: RetryContext) => number); + +export interface RetryOptions { + /** Total executions including the first. Defaults to 3; must be a positive safe integer. */ + attempts?: number; + /** Defaults to exponential: initialDelay * 2 ** (attempt - 1). */ + backoff?: RetryBackoff; + /** First delay in milliseconds, also used for fixed backoff. Defaults to 100. */ + initialDelay?: number; + /** Cap in milliseconds, applied before jitter. Defaults to 2_147_483_647 (the timer limit). */ + maxDelay?: number; + /** Full jitter: choose an integer from 0 through the capped delay. Defaults to false. */ + jitter?: boolean; + /** Defaults to retrying thrown/rejected errors only. May also request retries of successful results. */ + shouldRetry?: (context: RetryContext) => boolean | Promise; + /** Runs before each retry delay, never after the last attempt. Hook errors stop retries. */ + onRetry?: (context: RetryContext) => void | Promise; + /** Resolve cancellation from the current call's arguments; useful for decorators. */ + getSignal?: (args: readonly unknown[]) => AbortSignal | undefined; + /** Record a parent operation and child attempts. Also inferred from this or this.engine. */ + engine?: TraceableEngine; +} + +/** Cancellation belongs to an invocation rather than the shared retry policy. */ +export interface RetryCallOptions { + signal?: AbortSignal; +} + +/** Retains the original arguments and receiver; all calls return a Promise, including sync functions. */ +export interface RetryFunction { + (this: Receiver, ...args: Args): Promise; + withOptions(this: Receiver, options: RetryCallOptions, ...args: Args): Promise; +} + +const timerLimit = 2_147_483_647; + +function validateDelay(delay: number, name: string): number { + if (!Number.isFinite(delay) || delay < 0 || delay > timerLimit) { + throw new RangeError(`${name} must be between 0 and ${timerLimit} milliseconds`); + } + return delay; +} + +/** Waits for work or cancellation, always removing the abort listener. */ +function abortable(work: Promise, signal?: AbortSignal): Promise { + if (!signal) { + return work; + } + return new Promise((resolve, reject) => { + const onAbort = (): void => { + signal.removeEventListener('abort', onAbort); + reject(signal.reason); + }; + signal.addEventListener('abort', onAbort, { once: true }); + work.then( + (result) => { + signal.removeEventListener('abort', onAbort); + resolve(result); + }, + (error) => { + signal.removeEventListener('abort', onAbort); + reject(error); + } + ); + if (signal.aborted) { + onAbort(); + } + }); +} + +async function wait(delay: number, signal?: AbortSignal): Promise { + let timer: ReturnType; + try { + await abortable( + new Promise((resolve) => { + timer = setTimeout(resolve, delay); + }), + signal + ); + } finally { + clearTimeout(timer); + } +} + +function createRetry( + blockFunction: (this: Receiver, ...args: Args) => Output, + options: RetryOptions> +): RetryFunction, Receiver> { + // Copy the policy so later changes to the caller's options cannot alter an in-flight call. + const policy = { ...options }; + const attempts = policy.attempts ?? 3; + if (!Number.isSafeInteger(attempts) || attempts < 1) { + throw new RangeError('attempts must be a positive safe integer'); + } + const initialDelay = validateDelay(policy.initialDelay ?? 100, 'initialDelay'); + const maxDelay = validateDelay(policy.maxDelay ?? timerLimit, 'maxDelay'); + const backoff = policy.backoff ?? 'exponential'; + if (typeof backoff === 'number') { + validateDelay(backoff, 'backoff'); + } else if (typeof backoff !== 'function' && backoff !== 'fixed' && backoff !== 'exponential') { + throw new TypeError('backoff must be fixed, exponential, a delay, or a function'); + } + + async function invoke(receiver: Receiver, args: Args, callOptions: RetryCallOptions = {}): Promise> { + const signal = callOptions.signal ?? policy.getSignal?.call(receiver, args); + const scope = receiver as { engine?: TraceableEngine }; + const engine = + policy.engine ?? + (receiver instanceof TraceableEngine ? receiver : scope?.engine instanceof TraceableEngine ? scope.engine : undefined); + + async function executeAttempts(parent?: EngineNodeData): Promise> { + let succeeded = false; + try { + for (let attempt = 1; attempt <= attempts; attempt++) { + signal?.throwIfAborted(); + let attemptNode: EngineNodeData; + let result: Awaited; + let error: unknown; + let failed = false; + const execute = async (): Promise> => { + return abortable( + Promise.resolve().then(() => { + signal?.throwIfAborted(); + return blockFunction.apply(receiver, args); + }), + signal + ); + }; + try { + if (engine && parent) { + const trace = await engine.run( + async (node: EngineNodeData) => { + attemptNode = node; + node.inputs = args; + return execute(); + }, + [], + { + trace: { parent: parent.id, label: `attempt ${attempt}` }, + config: { errors: 'throw', traceExecution: true } + } + ); + result = trace.outputs as Awaited; + } else { + result = await execute(); + } + } catch (caught) { + failed = true; + error = caught; + } + signal?.throwIfAborted(); + if (attempt === attempts || (!failed && !policy.shouldRetry)) { + if (failed) { + throw error; + } + succeeded = true; + return result; + } + const context: RetryContext> = { + attempt, + ...(failed ? { error } : { result }), + delay: 0 + }; + let delay = + typeof backoff === 'function' + ? validateDelay(backoff.call(receiver, context), 'custom backoff') + : typeof backoff === 'number' + ? backoff + : backoff === 'fixed' || initialDelay === 0 + ? initialDelay + : Math.min(initialDelay * 2 ** (attempt - 1), timerLimit); + delay = Math.min(delay, maxDelay); + context.delay = policy.jitter ? Math.floor(Math.random() * (Math.floor(delay) + 1)) : delay; + const shouldRetry = policy.shouldRetry + ? await abortable( + Promise.resolve().then(() => { + signal?.throwIfAborted(); + return policy.shouldRetry.call(receiver, context); + }), + signal + ) + : failed; + signal?.throwIfAborted(); + if (!shouldRetry) { + if (failed) { + throw error; + } + succeeded = true; + return result; + } + if (engine && attemptNode) { + engine.pushNarratives(attemptNode.id, `Retry after ${context.delay} ms (${failed ? 'error' : 'result'}).`); + } + if (policy.onRetry) { + await abortable( + Promise.resolve().then(() => { + signal?.throwIfAborted(); + return policy.onRetry.call(receiver, context); + }), + signal + ); + } + signal?.throwIfAborted(); + await wait(context.delay, signal); + } + } finally { + if (parent) { + parent.narratives = [signal?.aborted ? 'Retry cancelled.' : succeeded ? 'Retry succeeded.' : 'Retry failed.']; + if (succeeded) { + // Clear errors inherited by the engine's placeholder parent from a failed child. + parent.errors = undefined; + } + } + } + } + + if (engine) { + const trace = await engine.run( + async (parent: EngineNodeData) => { + parent.inputs = args; + return executeAttempts(parent); + }, + [], + { + trace: { label: `retry ${blockFunction.name || 'anonymous'}` }, + config: { errors: 'throw', traceExecution: true } + } + ); + return trace.outputs as Awaited; + } + return executeAttempts(); + } + + const wrapped: RetryFunction, Receiver> = async function (this: Receiver, ...args: Args): Promise> { + return invoke(this, args); + } as RetryFunction, Receiver>; + wrapped.withOptions = function (this: Receiver, callOptions: RetryCallOptions, ...args: Args): Promise> { + return invoke((this as unknown) === wrapped ? undefined : this, args, callOptions); + }; + return wrapped; +} + +/** + * Wrap a sync/async function in a bounded retry policy. Always returns a Promise and preserves this and arguments. + * Exhaustion rejects with the exact final thrown value, including its original stack. Successful results are only + * retried when shouldRetry requests it; exhausting result retries returns the final result. Hook/strategy errors + * stop immediately. Cancellation rejects with signal.reason, clears pending delays, and does not interrupt the + * underlying operation; pass the same signal to that operation when it supports cancellation. + * + * Use wrapped.withOptions({ signal }, ...args) for per-call cancellation. For a receiver, use + * wrapped.withOptions.call(receiver, { signal }, ...args). Decorators can use getSignal(args). + */ +export function retry( + blockFunction: (this: Receiver, ...args: Args) => Output, + options?: RetryOptions> +): RetryFunction, Receiver>; +/** + * Apply retry to a method. The runtime result is always a Promise; declare decorated methods async. + * The policy and attempt state are isolated per invocation. Uses the same options as the function form. + */ +export function retry(options?: RetryOptions): MethodDecorator; +export function retry( + functionOrOptions: ((this: Receiver, ...args: Args) => Output) | RetryOptions> = {}, + options: RetryOptions> = {} +): RetryFunction, Receiver> | MethodDecorator { + if (typeof functionOrOptions === 'function') { + return createRetry(functionOrOptions, options); + } + return (target: object, propertyKey: string | symbol, descriptor: PropertyDescriptor): void => { + if (typeof descriptor.value !== 'function') { + throw new TypeError('retry can only decorate methods'); + } + descriptor.value = createRetry(descriptor.value, functionOrOptions); + }; +} diff --git a/src/index.ts b/src/index.ts index 3a5eb6c..1f6d870 100644 --- a/src/index.ts +++ b/src/index.ts @@ -19,6 +19,7 @@ export * from './execution/cache'; export * from './execution/execute'; export * from './execution/memoize.decorator'; export * from './execution/memoize'; +export * from './execution/retry'; export * from './execution/trace.decorator'; export * from './execution/trace'; export * from './timer/executionTimer';