From a8290ff2f0b0d3fa7b79b4bc66a1efdf89f45786 Mon Sep 17 00:00:00 2001 From: unsady <38117114+unsady@users.noreply.github.com> Date: Sun, 11 Oct 2026 19:08:41 +0200 Subject: [PATCH] feat!: revise protect error handling --- README.md | 53 ++++++++--- src/errors.ts | 7 ++ src/fallback.ts | 16 ++-- src/index.ts | 4 +- src/protect.test.ts | 218 ++++++++++++++++++++++++++++++++++---------- src/protect.ts | 66 +++++++++----- src/retry.ts | 49 ++++++++-- src/types.ts | 12 ++- 8 files changed, 321 insertions(+), 104 deletions(-) diff --git a/README.md b/README.md index 6f2cf5c..e11dd26 100644 --- a/README.md +++ b/README.md @@ -26,13 +26,10 @@ const run = protect({ threshold: 5, cooldown: 30_000, }, - onError: (error) => { - console.error('Operation failed', error) - }, }) const url = 'https://api.example.com/users/1' -const response = await run(({ signal }) => fetch(url, { signal })) +const response = await run(() => fetch(url)) ``` `protect()` creates a runner. Pass a callback to run an operation when a concurrency slot is available. Each call returns a promise with the operation's result or error. @@ -47,9 +44,36 @@ Without `backoff`, retries start immediately. To add a delay, provide `backoff.t `backoff.jitter` is an optional fraction from `0` to `1` that randomly reduces the base delay. With a 1,000 millisecond base delay, `jitter: 0.2` produces delays between 800 and 1,000 milliseconds. It defaults to `0`: no jitter. Use `1` for full jitter between zero and the base delay. -Thrown errors and rejected promises trigger retries. `fetch()` resolves for HTTP error responses; throw in your callback to retry them. Only retry operations that are safe to repeat. +By default, thrown errors and rejected promises trigger retries. `fetch()` resolves for HTTP error responses; throw in your callback to retry them. Only retry operations that are safe to repeat. + +## Error policy + +Use `errorPolicy` to decide whether an operation error permits another attempt and whether its final failure counts toward the circuit: + +```ts +const run = protect({ + attempts: 3, + errorPolicy: (error) => ({ + retry: isTransient(error), + countForCircuit: isServiceFailure(error), + }), +}) + +await run(writeOperation, { + errorPolicy: (error) => ({ + retry: false, + countForCircuit: isServiceFailure(error), + }), +}) +``` + +The predicates are application-defined. A synchronous policy runs after each failed attempt, including timeout, before retrying. `retry` permits another attempt within the configured limit; `countForCircuit` applies only to the final error. Without a policy, both are `true`. + +A policy in the run options replaces the runner's policy for that call. Errors excluded from the circuit are neutral: they neither increase nor reset its failure count. External cancellation and queue or circuit admission errors bypass the policy. + +If the policy throws, the call rejects with `ErrorPolicyError`, an `AggregateError` whose `errors` contains the operation error followed by the policy error, and whose `cause` is the operation error. No retry or fallback runs. This is a neutral circuit outcome and releases a probe reservation. -`onError(error)` is called once when a call rejects, including cancellation, timeout, queue overflow and circuit rejection. If the hook throws, its error replaces the original rejection. +Handle final call errors with `try`/`catch` or `.catch()`. ## Timeout @@ -70,13 +94,13 @@ To limit the entire call, including queue wait and backoff, pass an external sig ## Circuit breaker -`circuit: { threshold: 5, cooldown: 30_000 }` opens the circuit after five consecutive failed calls, counted after all retries. Success resets the count. Both settings are required; `cooldown` is in milliseconds. Without `circuit`, the breaker is disabled. +`circuit: { threshold: 5, cooldown: 30_000 }` opens the circuit after five consecutive counted failures, evaluated when each call ends. Success resets the count. Both settings are required; `cooldown` is in milliseconds. Without `circuit`, the breaker is disabled. While open, new and queued calls reject with `CircuitOpenError` without invoking their callbacks. Already active calls are not cancelled, but their later results do not affect the current circuit state. -After cooldown, one probe call is allowed, using the runner's normal retry and timeout settings. Other calls reject until the probe finishes. Success closes the circuit; failure opens it for another full cooldown. Cancelling a probe allows another probe without closing the circuit. +After cooldown, one probe call is allowed, using the runner's normal retry and timeout settings. Other calls reject until the probe finishes. Success closes the circuit; failure opens it for another full cooldown. A neutral probe outcome, including cancellation or an error excluded by the policy, allows another probe immediately without closing the circuit or starting another cooldown. -External cancellation, `QueueFullError` and `CircuitOpenError` neither count as failures nor reset the count. Final timeouts count as failures. With `timeoutMode: 'wait'`, a callback that never settles can also block the probe indefinitely. +External cancellation, `QueueFullError` and `CircuitOpenError` neither count as failures nor reset the count. Final timeouts count as failures unless excluded by the error policy. With `timeoutMode: 'wait'`, a callback that never settles can also block the probe indefinitely. Circuit state is shared by all calls on a runner and independent between runners. Use a separate runner for each protected resource. @@ -85,21 +109,20 @@ Circuit state is shared by all calls on a runner and independent between runners Pass `fallback` in the run options to use a backup operation after the primary call fails: ```ts -const backup = protect({ attempts: 2, timeout: 5_000 }) const backupUrl = 'https://backup.example.com/users/1' -const response = await run(({ signal }) => fetch(url, { signal }), { - fallback: () => backup(({ signal }) => fetch(backupUrl, { signal })), +const response = await run(() => fetch(url), { + fallback: () => fetch(backupUrl), }) ``` -`fallback(error)` runs once after retries are exhausted, or immediately when the circuit or queue rejects the call. It can return a value or promise; the returned value becomes the result of `run`. It is not invoked when the primary succeeds or the external signal is aborted. +`fallback(error, { signal })` runs once after the operation's final error, including when retries are disabled by the policy, or immediately when the circuit or queue rejects the call. It can return a value or promise; the returned value becomes the result of `run`. It is not invoked when the primary succeeds, the external signal is aborted or the error policy throws. Disabling retry does not disable fallback; its callback must decide whether recovery is safe. Fallback does not use the primary runner's retries, timeout or concurrency slot. Use a separate runner to protect the backup operation. A successful fallback does not close the primary circuit or reset its failure count. -If fallback throws or rejects, its error becomes the call's rejection. `onError` reports only the final rejection; it is not called when fallback succeeds. +If fallback throws or rejects, its error becomes the call's rejection. -External cancellation also rejects a call while fallback is running. To stop the backup operation itself, forward the same external signal to its runner. +Fallback receives the external signal, or a non-aborted signal when none was supplied. This is the call's signal, not an attempt signal aborted by timeout. External cancellation also rejects a call while fallback is running. Forward the fallback context's signal to its operation or backup runner for cooperative cancellation. ## Cancellation diff --git a/src/errors.ts b/src/errors.ts index 4f912f0..a9aef53 100644 --- a/src/errors.ts +++ b/src/errors.ts @@ -5,6 +5,13 @@ export class CircuitOpenError extends Error { } } +export class ErrorPolicyError extends AggregateError { + constructor(error: unknown, policyError: unknown) { + super([error, policyError], 'The error policy threw', { cause: error }) + this.name = 'ErrorPolicyError' + } +} + export class QueueFullError extends Error { constructor() { super('The pending operation limit has been reached') diff --git a/src/fallback.ts b/src/fallback.ts index 7890b1f..8000d6a 100644 --- a/src/fallback.ts +++ b/src/fallback.ts @@ -1,27 +1,29 @@ +import type { RunContext } from './types.js' + export function recover( - fallback: (error: unknown) => Result, + fallback: (error: unknown, context: RunContext) => Result, error: unknown, - signal: AbortSignal | undefined, + signal: AbortSignal = new AbortController().signal, ): Promise> { return new Promise((resolve, reject) => { - if (signal?.aborted) { + if (signal.aborted) { reject(signal.reason) return } function finish(): void { - signal?.removeEventListener('abort', onAbort) + signal.removeEventListener('abort', onAbort) } function onAbort(): void { finish() - reject(signal?.reason) + reject(signal.reason) } - signal?.addEventListener('abort', onAbort, { once: true }) + signal.addEventListener('abort', onAbort, { once: true }) try { - void Promise.resolve(fallback(error)).then( + void Promise.resolve(fallback(error, { signal })).then( (value) => { finish() resolve(value) diff --git a/src/index.ts b/src/index.ts index 545bcee..bab1af6 100644 --- a/src/index.ts +++ b/src/index.ts @@ -1,8 +1,10 @@ -export { CircuitOpenError, QueueFullError, TimeoutError } from './errors.js' +export { CircuitOpenError, ErrorPolicyError, QueueFullError, TimeoutError } from './errors.js' export { protect } from './protect.js' export type { BackoffOptions, CircuitOptions, + ErrorDecision, + ErrorPolicy, ProtectOptions, RunContext, Runner, diff --git a/src/protect.test.ts b/src/protect.test.ts index 0299961..ff41712 100644 --- a/src/protect.test.ts +++ b/src/protect.test.ts @@ -2,9 +2,11 @@ import { afterEach, beforeEach, describe, expect, expectTypeOf, it, vi } from 'v import { CircuitOpenError, + ErrorPolicyError, protect, QueueFullError, TimeoutError, + type ErrorPolicy, type RunContext, } from './index.js' @@ -52,18 +54,16 @@ it('retries immediately without backoff and returns the successful result', asyn expect(operation).toHaveBeenCalledTimes(3) }) -it('stops at the attempt limit and reports the final error once', async () => { +it('stops at the attempt limit with exponential backoff', async () => { const error = new Error('unavailable') const times: number[] = [] const operation = vi.fn<() => Promise>(() => { times.push(Date.now()) return Promise.reject(error) }) - const onError = vi.fn<(error: unknown) => void>() const run = protect({ attempts: 4, backoff: { type: 'exponential', delay: 10 }, - onError, }) const rejected = run(operation).catch((error: unknown) => error) @@ -72,7 +72,6 @@ it('stops at the attempt limit and reports the final error once', async () => { expect(operation).toHaveBeenCalledTimes(4) expect(times).toEqual([0, 10, 30, 70]) - expect(onError.mock.calls).toEqual([[error]]) }) it.each<{ type: 'fixed' | 'exponential'; secondDelay: number }>([ @@ -105,7 +104,8 @@ it.each<{ type: 'fixed' | 'exponential'; secondDelay: number }>([ }) it('rejects excess calls without queuing by default', async () => { - const run = protect({ concurrency: 1 }) + const policy = vi.fn(() => ({ retry: true, countForCircuit: true })) + const run = protect({ concurrency: 1, errorPolicy: policy }) const blocker = deferred() const running = run(() => blocker.promise) const overflow = vi.fn<() => string>(() => 'saved') @@ -116,6 +116,7 @@ it('rejects excess calls without queuing by default', async () => { blocker.resolve() await running await expect(run(overflow)).resolves.toBe('saved') + expect(policy).not.toHaveBeenCalled() }) it('shares concurrency and pending limits across operations and releases a slot after failure', async () => { @@ -145,7 +146,8 @@ it('shares concurrency and pending limits across operations and releases a slot }) it('skips already aborted operations and frees a pending slot on cancellation', async () => { - const run = protect({ concurrency: 1, maxPending: 1 }) + const policy = vi.fn(() => ({ retry: true, countForCircuit: true })) + const run = protect({ concurrency: 1, maxPending: 1, errorPolicy: policy }) const controller = new AbortController() const reason = new Error('cancelled') const skipped = vi.fn<() => void>() @@ -165,10 +167,12 @@ it('skips already aborted operations and frees a pending slot on cancellation', await running await expect(replacement).resolves.toBe('replacement') expect(skipped).not.toHaveBeenCalled() + expect(policy).not.toHaveBeenCalled() }) it('forwards cancellation without releasing the slot until the operation settles or retrying it', async () => { - const run = protect({ concurrency: 1, maxPending: 1, attempts: 3 }) + const policy = vi.fn(() => ({ retry: true, countForCircuit: true })) + const run = protect({ concurrency: 1, maxPending: 1, attempts: 3, errorPolicy: policy }) const controller = new AbortController() const blocker = deferred() const reason = new Error('cancelled') @@ -190,6 +194,7 @@ it('forwards cancellation without releasing the slot until the operation settles blocker.resolve() await expect(waiting).resolves.toBe('next') expect(operation).toHaveBeenCalledTimes(1) + expect(policy).not.toHaveBeenCalled() }) it('gives each attempt its own timeout, excluding backoff and queue wait', async () => { @@ -238,13 +243,11 @@ it('gives each attempt its own timeout, excluding backoff and queue wait', async }) it('releases the slot on the final timeout and ignores a late callback rejection', async () => { - const onError = vi.fn<(error: unknown) => void>() const run = protect({ concurrency: 1, maxPending: 1, timeout: 100, timeoutMode: 'immediate', - onError, }) const blocker = deferred() const timedOut = run(async () => { @@ -259,18 +262,15 @@ it('releases the slot on the final timeout and ignores a late callback rejection blocker.resolve() await vi.advanceTimersByTimeAsync(0) - expect(onError).toHaveBeenCalledTimes(1) }) it('waits for timed-out callbacks before retrying or releasing the slot in wait mode', async () => { - const onError = vi.fn<(error: unknown) => void>() const run = protect({ concurrency: 1, maxPending: 1, attempts: 2, timeout: 100, timeoutMode: 'wait', - onError, }) const first = deferred() const second = deferred() @@ -291,19 +291,135 @@ it('waits for timed-out callbacks before retrying or releasing the slot in wait await vi.advanceTimersByTimeAsync(100) expect(signals[1]?.reason).toBeInstanceOf(TimeoutError) expect(operation).toHaveBeenCalledTimes(2) - expect(onError).not.toHaveBeenCalled() expect(next).not.toHaveBeenCalled() second.resolve() await expect(timedOut).resolves.toBeInstanceOf(TimeoutError) await expect(waiting).resolves.toBe('next') - expect(onError).toHaveBeenCalledTimes(1) +}) + +describe('error policy', () => { + it('replaces the runner policy per call and classifies the final error independently', async () => { + const error = new Error('provider failed') + const excluded = new Error('invalid input') + const policy = vi.fn(() => ({ retry: false, countForCircuit: true })) + const override = vi.fn((failure) => ({ + retry: failure === error, + countForCircuit: failure === error, + })) + const run = protect({ + attempts: 3, + circuit: { threshold: 1, cooldown: 100 }, + errorPolicy: policy, + }) + const operation = vi + .fn<() => Promise>() + .mockRejectedValueOnce(error) + .mockRejectedValue(excluded) + + await expect(run(operation, { errorPolicy: override })).rejects.toBe(excluded) + expect(operation).toHaveBeenCalledTimes(2) + expect(override.mock.calls).toEqual([[error], [excluded]]) + expect(policy).not.toHaveBeenCalled() + + await expect(run(operation)).rejects.toBe(excluded) + expect(operation).toHaveBeenCalledTimes(3) + expect(policy.mock.calls).toEqual([[excluded]]) + await expect(run(operation)).rejects.toBeInstanceOf(CircuitOpenError) + expect(policy).toHaveBeenCalledTimes(1) + }) + + it('preserves counted failures across neutral errors and permits an immediate replacement probe', async () => { + const error = new Error('provider failed') + const excluded = new Error('invalid input') + const run = protect({ + circuit: { threshold: 2, cooldown: 100 }, + errorPolicy: (failure) => ({ retry: false, countForCircuit: failure === error }), + }) + + function fail(): never { + throw error + } + + function neutral(): never { + throw excluded + } + + await expect(run(fail)).rejects.toBe(error) + await expect(run(neutral)).rejects.toBe(excluded) + await expect(run(fail)).rejects.toBe(error) + await expect(run(() => 'skipped')).rejects.toBeInstanceOf(CircuitOpenError) + await vi.advanceTimersByTimeAsync(100) + await expect(run(neutral)).rejects.toBe(excluded) + + const blocker = deferred() + const probe = run(() => blocker.promise) + + await expect(run(() => 'skipped')).rejects.toBeInstanceOf(CircuitOpenError) + blocker.resolve() + await probe + await expect(run(() => 'saved')).resolves.toBe('saved') + }) + + it('preserves both errors when the policy throws, skips fallback and releases the probe and slot', async () => { + const error = new Error('provider failed') + const policyError = new Error('policy failed') + const run = protect({ + concurrency: 1, + attempts: 3, + circuit: { threshold: 1, cooldown: 100 }, + }) + const operation = vi.fn<() => never>(() => { + throw error + }) + + await expect(run(operation)).rejects.toBe(error) + await vi.advanceTimersByTimeAsync(100) + operation.mockClear() + + const policy = vi.fn(() => { + throw policyError + }) + const fallback = vi.fn<() => string>(() => 'backup') + const failure = await run(operation, { errorPolicy: policy, fallback }).catch( + (error: unknown) => error, + ) + + expect(failure).toBeInstanceOf(ErrorPolicyError) + expect(failure).toMatchObject({ errors: [error, policyError], cause: error }) + expect(operation).toHaveBeenCalledTimes(1) + expect(policy).toHaveBeenCalledTimes(1) + expect(fallback).not.toHaveBeenCalled() + + const blocker = deferred() + const probe = run(() => blocker.promise) + + await expect(run(() => 'skipped')).rejects.toBeInstanceOf(CircuitOpenError) + blocker.resolve() + await probe + await expect(run(() => 'saved')).resolves.toBe('saved') + }) + + it('allows fallback even when retries are disabled', async () => { + const error = new Error('invalid input') + const run = protect({ + attempts: 3, + errorPolicy: () => ({ retry: false, countForCircuit: false }), + }) + const operation = vi.fn<() => never>(() => { + throw error + }) + const fallback = vi.fn<() => string>(() => 'cached') + + await expect(run(operation, { fallback })).resolves.toBe('cached') + expect(operation).toHaveBeenCalledTimes(1) + expect(fallback).toHaveBeenCalledTimes(1) + }) }) describe('fallback', () => { it('runs only after final failure and returns an awaited, correctly typed result', async () => { - const onError = vi.fn<(error: unknown) => void>() - const run = protect({ attempts: 3, onError }) + const run = protect({ attempts: 3 }) const error = new Error('primary failed') const operation = vi.fn<() => Promise>().mockRejectedValue(error) const fallback = vi.fn<(error: unknown) => Promise>(async () => 42) @@ -316,16 +432,14 @@ describe('fallback', () => { expectTypeOf(result).toEqualTypeOf>() await expect(result).resolves.toBe(42) expect(operation).toHaveBeenCalledTimes(3) - expect(fallback.mock.calls).toEqual([[error]]) - expect(onError).not.toHaveBeenCalled() + expect(fallback.mock.calls[0]?.[0]).toBe(error) expectTypeOf(run(() => 'primary')).toEqualTypeOf>() }) it.each(['throw', 'reject'] as const)( 'reports a fallback %s without retrying it', async (mode) => { - const onError = vi.fn<(error: unknown) => void>() - const run = protect({ attempts: 3, onError }) + const run = protect({ attempts: 3 }) const error = new Error('backup failed') const fallback = vi.fn<(error: unknown) => Promise>(() => { if (mode === 'throw') { @@ -344,13 +458,11 @@ describe('fallback', () => { ), ).rejects.toBe(error) expect(fallback).toHaveBeenCalledTimes(1) - expect(onError.mock.calls).toEqual([[error]]) }, ) it('can use a backup runner without closing the primary circuit', async () => { - const onError = vi.fn<(error: unknown) => void>() - const run = protect({ circuit: { threshold: 1, cooldown: 100 }, onError }) + const run = protect({ circuit: { threshold: 1, cooldown: 100 } }) const backup = protect({ attempts: 2, circuit: { threshold: 1, cooldown: 100 } }) const error = new Error('primary failed') const operation = vi @@ -362,23 +474,33 @@ describe('fallback', () => { await expect(run(operation, { fallback })).resolves.toBe('backup') await expect(run(operation, { fallback })).resolves.toBe('backup') expect(operation).toHaveBeenCalledTimes(1) - expect(fallback.mock.calls[0]).toEqual([error]) + expect(fallback.mock.calls[0]?.[0]).toBe(error) expect(fallback.mock.calls[1]?.[0]).toBeInstanceOf(CircuitOpenError) await vi.advanceTimersByTimeAsync(100) await expect(run(operation, { fallback })).resolves.toBe('primary') expect(fallback).toHaveBeenCalledTimes(2) - expect(onError).not.toHaveBeenCalled() }) it('handles final timeout outside the primary concurrency slot and timeout budget', async () => { - const run = protect({ concurrency: 1, timeout: 100 }) + const policy = vi.fn(() => ({ retry: false, countForCircuit: false })) + const run = protect({ + concurrency: 1, + attempts: 3, + timeout: 100, + circuit: { threshold: 1, cooldown: 100 }, + errorPolicy: policy, + }) const primary = deferred() const backup = deferred() - const fallback = vi.fn<(error: unknown) => Promise>(async () => { - await backup.promise - return 42 - }) + let received: AbortSignal | undefined + const fallback = vi.fn<(error: unknown, context: RunContext) => Promise>( + async (_error, { signal }) => { + received = signal + await backup.promise + return 42 + }, + ) const result = run( async () => { await primary.promise @@ -389,6 +511,10 @@ describe('fallback', () => { await vi.advanceTimersByTimeAsync(100) expect(fallback.mock.calls[0]?.[0]).toBeInstanceOf(TimeoutError) + expect(policy.mock.calls[0]?.[0]).toBeInstanceOf(TimeoutError) + expect(policy).toHaveBeenCalledTimes(1) + expect(received).toBeInstanceOf(AbortSignal) + expect(received?.aborted).toBe(false) await expect(run(() => 'primary')).resolves.toBe('primary') await vi.advanceTimersByTimeAsync(1_000) @@ -447,15 +573,18 @@ describe('fallback', () => { }) it('cancels an active fallback and ignores its late rejection', async () => { - const onError = vi.fn<(error: unknown) => void>() - const run = protect({ onError }) + const run = protect() const controller = new AbortController() const blocker = deferred() const reason = new Error('cancelled') - const fallback = vi.fn<(error: unknown) => Promise>(async () => { - await blocker.promise - throw new Error('late') - }) + let received: AbortSignal | undefined + const fallback = vi.fn<(error: unknown, context: RunContext) => Promise>( + async (_error, { signal }) => { + received = signal + await blocker.promise + throw new Error('late') + }, + ) const cancelled = run( () => { throw new Error('primary failed') @@ -465,12 +594,13 @@ describe('fallback', () => { await vi.advanceTimersByTimeAsync(0) expect(fallback).toHaveBeenCalledTimes(1) + expect(received).toBe(controller.signal) controller.abort(reason) await expect(cancelled).resolves.toBe(reason) + expect(received?.reason).toBe(reason) blocker.resolve() await vi.advanceTimersByTimeAsync(0) - expect(onError.mock.calls).toEqual([[reason]]) }) }) @@ -592,16 +722,11 @@ describe('circuit', () => { expect(skipped).not.toHaveBeenCalled() }) - it('counts only the final timeout and ignores detached callbacks and hook errors', async () => { - const hookError = new Error('hook failed') - const onError = vi.fn<(error: unknown) => void>(() => { - throw hookError - }) + it('counts only the final timeout and ignores detached callbacks', async () => { const run = protect({ attempts: 2, timeout: 100, circuit: { threshold: 1, cooldown: 100 }, - onError, }) const blocker = deferred() const timedOut = run(() => blocker.promise).catch((error: unknown) => error) @@ -609,16 +734,13 @@ describe('circuit', () => { await vi.advanceTimersByTimeAsync(100) await expect(run(() => 'available')).resolves.toBe('available') await vi.advanceTimersByTimeAsync(100) - await expect(timedOut).resolves.toBe(hookError) - expect(onError).toHaveBeenCalledTimes(1) - expect(onError.mock.calls[0]?.[0]).toBeInstanceOf(TimeoutError) + await expect(timedOut).resolves.toBeInstanceOf(TimeoutError) blocker.resolve() await vi.advanceTimersByTimeAsync(0) const skipped = vi.fn<() => void>() - await expect(run(skipped)).rejects.toBe(hookError) - expect(onError.mock.calls[1]?.[0]).toBeInstanceOf(CircuitOpenError) + await expect(run(skipped)).rejects.toBeInstanceOf(CircuitOpenError) expect(skipped).not.toHaveBeenCalled() }) diff --git a/src/protect.ts b/src/protect.ts index aa5b991..7fdb85f 100644 --- a/src/protect.ts +++ b/src/protect.ts @@ -10,7 +10,12 @@ interface PendingOperation { } export function protect(options: ProtectOptions = {}): Runner { - const { attempts = 1, timeout, timeoutMode = 'immediate', onError } = options + const { + attempts = 1, + timeout, + timeoutMode = 'immediate', + errorPolicy: defaultErrorPolicy, + } = options const backoff = options.backoff === undefined ? undefined @@ -47,6 +52,9 @@ export function protect(options: ProtectOptions = {}): Runner { runOptions: RunOptions = {}, ): Promise | Awaited> { const { signal, fallback } = runOptions + const errorPolicy = runOptions.errorPolicy ?? defaultErrorPolicy + let recoverable = true + const result = new Promise>((resolve, reject) => { if (signal?.aborted) { reject(signal.reason) @@ -121,26 +129,38 @@ export function protect(options: ProtectOptions = {}): Runner { backoff, timeout, timeoutMode, + errorPolicy, }).then( - (value) => { - if (admission !== undefined && !controller.signal.aborted) { - circuit?.succeed(admission.generation) + (outcome) => { + if (outcome.ok) { + if (admission !== undefined && !controller.signal.aborted) { + circuit?.succeed(admission.generation) + } + + resolve(outcome.value) + } else { + if (admission !== undefined) { + if ( + controller.signal.aborted || + !outcome.countForCircuit || + outcome.error instanceof QueueFullError || + outcome.error instanceof CircuitOpenError + ) { + circuit?.cancel(admission.generation) + } else if (circuit?.fail(admission.generation)) { + rejectQueue() + } + } + + recoverable = outcome.recoverable + reject(outcome.error) } - resolve(value) finish() }, (error: unknown) => { if (admission !== undefined) { - if ( - controller.signal.aborted || - error instanceof QueueFullError || - error instanceof CircuitOpenError - ) { - circuit?.cancel(admission.generation) - } else if (circuit?.fail(admission.generation)) { - rejectQueue() - } + circuit?.cancel(admission.generation) } reject(error) @@ -161,18 +181,16 @@ export function protect(options: ProtectOptions = {}): Runner { } }) - const recovered = - fallback === undefined - ? result - : result.catch((error: unknown) => recover(fallback, error, signal)) - - if (onError === undefined) { - return recovered + if (fallback === undefined) { + return result } - return recovered.catch((error: unknown) => { - onError(error) - throw error + return result.catch((error: unknown) => { + if (!recoverable) { + throw error + } + + return recover(fallback, error, signal) }) } diff --git a/src/retry.ts b/src/retry.ts index bb98bff..ae37412 100644 --- a/src/retry.ts +++ b/src/retry.ts @@ -1,13 +1,24 @@ -import { TimeoutError } from './errors.js' -import type { BackoffOptions, RunContext, TimeoutMode } from './types.js' +import { ErrorPolicyError, TimeoutError } from './errors.js' +import type { + BackoffOptions, + ErrorDecision, + ErrorPolicy, + RunContext, + TimeoutMode, +} from './types.js' interface ExecutionOptions { attempts: number backoff: Required | undefined timeout: number | undefined timeoutMode: TimeoutMode + errorPolicy: ErrorPolicy | undefined } +type ExecutionResult = + | { ok: true; value: Awaited } + | { ok: false; error: unknown; countForCircuit: boolean; recoverable: boolean } + function wait(duration: number, signal: AbortSignal): Promise { return new Promise((resolve, reject) => { signal.throwIfAborted() @@ -80,20 +91,44 @@ async function runAttempt( export async function execute( operation: (context: RunContext) => Result, signal: AbortSignal, - { attempts, backoff, timeout, timeoutMode }: ExecutionOptions, -): Promise> { + { attempts, backoff, timeout, timeoutMode, errorPolicy }: ExecutionOptions, +): Promise> { let attempt = 1 while (true) { signal.throwIfAborted() try { - return await runAttempt(operation, signal, timeout, timeoutMode) + const value = await runAttempt(operation, signal, timeout, timeoutMode) + + return { ok: true, value } } catch (error) { signal.throwIfAborted() - if (attempt >= attempts) { - throw error + let decision: ErrorDecision + + try { + decision = errorPolicy?.(error) ?? { retry: true, countForCircuit: true } + } catch (policyError) { + signal.throwIfAborted() + + return { + ok: false, + error: new ErrorPolicyError(error, policyError), + countForCircuit: false, + recoverable: false, + } + } + + signal.throwIfAborted() + + if (!decision.retry || attempt >= attempts) { + return { + ok: false, + error, + countForCircuit: decision.countForCircuit, + recoverable: true, + } } if (backoff !== undefined) { diff --git a/src/types.ts b/src/types.ts index dd9c819..231a887 100644 --- a/src/types.ts +++ b/src/types.ts @@ -11,6 +11,13 @@ export interface CircuitOptions { export type TimeoutMode = 'immediate' | 'wait' +export interface ErrorDecision { + retry: boolean + countForCircuit: boolean +} + +export type ErrorPolicy = (error: unknown) => ErrorDecision + export interface ProtectOptions { attempts?: number backoff?: BackoffOptions @@ -19,7 +26,7 @@ export interface ProtectOptions { concurrency?: number maxPending?: number circuit?: CircuitOptions - onError?: (error: unknown) => void + errorPolicy?: ErrorPolicy } export interface RunContext { @@ -28,7 +35,8 @@ export interface RunContext { export interface RunOptions { signal?: AbortSignal - fallback?: (error: unknown) => Result + errorPolicy?: ErrorPolicy + fallback?: (error: unknown, context: RunContext) => Result } export interface Runner {