From 2b7d0fb5ba6dd60900937c9f7f482d06eb0e6b3f Mon Sep 17 00:00:00 2001 From: unsady <38117114+unsady@users.noreply.github.com> Date: Sun, 11 Oct 2026 16:08:44 +0200 Subject: [PATCH 1/2] feat: add circuit breaker --- README.md | 18 +++- src/circuit.ts | 79 +++++++++++++++ src/errors.ts | 7 ++ src/index.ts | 3 +- src/protect.test.ts | 241 +++++++++++++++++++++++++++++++++++++++++++- src/protect.ts | 64 +++++++++++- src/types.ts | 6 ++ 7 files changed, 410 insertions(+), 8 deletions(-) create mode 100644 src/circuit.ts diff --git a/README.md b/README.md index be0e97d..3adced3 100644 --- a/README.md +++ b/README.md @@ -22,6 +22,10 @@ const run = protect({ timeoutMode: 'immediate', concurrency: 8, maxPending: 100, + circuit: { + threshold: 5, + cooldown: 30_000, + }, onError: (error) => { console.error('Operation failed', error) }, @@ -45,7 +49,7 @@ Without `backoff`, retries start immediately. To add a delay, provide `backoff.t 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. -`onError(error)` is called once when a call rejects, including cancellation, timeout and queue overflow. If the hook throws, its error replaces the original rejection. +`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. ## Timeout @@ -64,6 +68,18 @@ To limit the entire call, including queue wait and backoff, pass an external sig `maxPending` defaults to `0`: no queue. When all concurrency slots are occupied, new calls reject with `QueueFullError` without invoking their callbacks. Set `maxPending > 0` to enable a queue; the limit counts only waiting calls, excluding active ones. +## 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. + +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. + +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. + +Circuit state is shared by all calls on a runner and independent between runners. Use a separate runner for each protected resource. + ## Cancellation ```ts diff --git a/src/circuit.ts b/src/circuit.ts new file mode 100644 index 0000000..8c0b89e --- /dev/null +++ b/src/circuit.ts @@ -0,0 +1,79 @@ +import { CircuitOpenError } from './errors.js' +import type { CircuitOptions } from './types.js' + +export class Circuit { + private state: 'closed' | 'open' | 'half-open' = 'closed' + private failures = 0 + private generation = 0 + private until = 0 + private probing = false + + constructor(private readonly options: CircuitOptions) {} + + check(): void { + if (this.state === 'open') { + if (Date.now() < this.until) { + throw new CircuitOpenError() + } + + this.state = 'half-open' + } + + if (this.state === 'half-open' && this.probing) { + throw new CircuitOpenError() + } + } + + enter(): { generation: number; probe: boolean } { + this.check() + + const probe = this.state === 'half-open' + + if (probe) { + this.probing = true + } + + return { generation: this.generation, probe } + } + + succeed(generation: number): void { + if (generation !== this.generation) { + return + } + + if (this.state === 'half-open') { + this.state = 'closed' + this.probing = false + this.generation += 1 + } + + this.failures = 0 + } + + fail(generation: number): boolean { + if (generation !== this.generation) { + return false + } + + this.failures += 1 + + if (this.state !== 'half-open' && this.failures < this.options.threshold) { + return false + } + + this.state = 'open' + this.until = Date.now() + this.options.cooldown + this.generation += 1 + this.probing = false + return true + } + + cancel(generation: number): void { + if (generation !== this.generation || this.state !== 'half-open') { + return + } + + this.probing = false + this.generation += 1 + } +} diff --git a/src/errors.ts b/src/errors.ts index 10da98c..4f912f0 100644 --- a/src/errors.ts +++ b/src/errors.ts @@ -1,3 +1,10 @@ +export class CircuitOpenError extends Error { + constructor() { + super('The circuit is open') + this.name = 'CircuitOpenError' + } +} + export class QueueFullError extends Error { constructor() { super('The pending operation limit has been reached') diff --git a/src/index.ts b/src/index.ts index 8e860e6..545bcee 100644 --- a/src/index.ts +++ b/src/index.ts @@ -1,7 +1,8 @@ -export { QueueFullError, TimeoutError } from './errors.js' +export { CircuitOpenError, QueueFullError, TimeoutError } from './errors.js' export { protect } from './protect.js' export type { BackoffOptions, + CircuitOptions, ProtectOptions, RunContext, Runner, diff --git a/src/protect.test.ts b/src/protect.test.ts index 5fa78e0..3d13a37 100644 --- a/src/protect.test.ts +++ b/src/protect.test.ts @@ -1,6 +1,12 @@ -import { afterEach, beforeEach, expect, it, vi } from 'vitest' +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest' -import { protect, QueueFullError, TimeoutError, type RunContext } from './index.js' +import { + CircuitOpenError, + protect, + QueueFullError, + TimeoutError, + type RunContext, +} from './index.js' function deferred() { let resolve!: () => void @@ -294,6 +300,237 @@ it('waits for timed-out callbacks before retrying or releasing the slot in wait expect(onError).toHaveBeenCalledTimes(1) }) +describe('circuit', () => { + it('counts final failures after retries and resets the count on success', async () => { + const run = protect({ attempts: 2, circuit: { threshold: 2, cooldown: 100 } }) + const error = new Error('unavailable') + const fail = vi.fn<() => Promise>().mockRejectedValue(error) + const recover = vi + .fn<() => Promise>() + .mockRejectedValueOnce(error) + .mockResolvedValue('saved') + const skipped = vi.fn<() => void>() + + await expect(run(fail)).rejects.toBe(error) + await expect(run(recover)).resolves.toBe('saved') + await expect(run(fail)).rejects.toBe(error) + await expect(run(fail)).rejects.toBe(error) + await expect(run(skipped)).rejects.toBeInstanceOf(CircuitOpenError) + expect(fail).toHaveBeenCalledTimes(6) + expect(skipped).not.toHaveBeenCalled() + + await expect( + protect({ circuit: { threshold: 2, cooldown: 100 } })(() => 'saved'), + ).resolves.toBe('saved') + }) + + it('rejects queued calls when opening without cancelling active calls', async () => { + const run = protect({ + concurrency: 2, + maxPending: 1, + circuit: { threshold: 1, cooldown: 100 }, + }) + const active = deferred() + const failure = deferred() + const error = new Error('unavailable') + let received: AbortSignal | undefined + const running = run(({ signal }) => { + received = signal + return active.promise + }) + const failed = run(async () => { + await failure.promise + throw error + }).catch((error: unknown) => error) + const skipped = vi.fn<() => void>() + const waiting = run(skipped).catch((error: unknown) => error) + + failure.resolve() + await expect(failed).resolves.toBe(error) + await expect(waiting).resolves.toBeInstanceOf(CircuitOpenError) + expect(received?.aborted).toBe(false) + + active.resolve() + await running + await expect(run(skipped)).rejects.toBeInstanceOf(CircuitOpenError) + expect(skipped).not.toHaveBeenCalled() + }) + + it('admits one probe after cooldown, reopens on failure and closes on success', async () => { + const run = protect({ circuit: { threshold: 1, cooldown: 100 } }) + const error = new Error('unavailable') + + function fail(): never { + throw error + } + + const skipped = vi.fn<() => void>() + + await expect(run(fail)).rejects.toBe(error) + await vi.advanceTimersByTimeAsync(99) + await expect(run(skipped)).rejects.toBeInstanceOf(CircuitOpenError) + await vi.advanceTimersByTimeAsync(1) + + const blocker = deferred() + const probe = run(async () => { + await blocker.promise + throw error + }).catch((error: unknown) => error) + + await expect(run(skipped)).rejects.toBeInstanceOf(CircuitOpenError) + blocker.resolve() + await expect(probe).resolves.toBe(error) + + await vi.advanceTimersByTimeAsync(99) + await expect(run(skipped)).rejects.toBeInstanceOf(CircuitOpenError) + await vi.advanceTimersByTimeAsync(1) + await expect(run(() => 'recovered')).resolves.toBe('recovered') + await expect(run(() => 'saved')).resolves.toBe('saved') + expect(skipped).not.toHaveBeenCalled() + }) + + it('does not count cancellation or queue overflow, or reset failures because of them', async () => { + const run = protect({ concurrency: 1, circuit: { threshold: 2, cooldown: 100 } }) + const error = new Error('unavailable') + + function fail(): never { + throw error + } + + await expect(run(fail)).rejects.toBe(error) + + const blocker = deferred() + const controller = new AbortController() + const reason = new Error('cancelled') + const cancelled = run(() => blocker.promise, { signal: controller.signal }).catch( + (error: unknown) => error, + ) + const skipped = vi.fn<() => void>() + + await expect(run(skipped)).rejects.toBeInstanceOf(QueueFullError) + controller.abort(reason) + await expect(cancelled).resolves.toBe(reason) + + blocker.resolve() + await vi.advanceTimersByTimeAsync(0) + await expect(run(fail)).rejects.toBe(error) + await expect(run(skipped)).rejects.toBeInstanceOf(CircuitOpenError) + 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 + }) + 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) + + 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) + + 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) + expect(skipped).not.toHaveBeenCalled() + }) + + it('ignores old failures and cancelled probe results across recovery', async () => { + const run = protect({ circuit: { threshold: 1, cooldown: 100 } }) + const error = new Error('unavailable') + const old = deferred() + const failed = run(async () => { + await old.promise + throw error + }).catch((error: unknown) => error) + + await expect( + run(() => { + throw error + }), + ).rejects.toBe(error) + await vi.advanceTimersByTimeAsync(100) + + const controller = new AbortController() + const reason = new Error('cancelled') + const abandoned = deferred() + const cancelled = run(() => abandoned.promise, { signal: controller.signal }).catch( + (error: unknown) => error, + ) + + controller.abort(reason) + await expect(cancelled).resolves.toBe(reason) + + const replacement = deferred() + const probe = run(() => replacement.promise) + + abandoned.resolve() + await vi.advanceTimersByTimeAsync(0) + const skipped = vi.fn<() => void>() + + await expect(run(skipped)).rejects.toBeInstanceOf(CircuitOpenError) + replacement.resolve() + await probe + + old.resolve() + await expect(failed).resolves.toBe(error) + await expect(run(() => 'saved')).resolves.toBe('saved') + expect(skipped).not.toHaveBeenCalled() + }) + + it('reserves the probe only when starting and rejects other queued candidates', async () => { + const run = protect({ + concurrency: 1, + maxPending: 2, + circuit: { threshold: 1, cooldown: 100 }, + }) + const error = new Error('unavailable') + + await expect( + run(() => { + throw error + }), + ).rejects.toBe(error) + await vi.advanceTimersByTimeAsync(100) + + const controller = new AbortController() + const reason = new Error('cancelled') + const abandoned = deferred() + const cancelled = run(() => abandoned.promise, { signal: controller.signal }).catch( + (error: unknown) => error, + ) + + controller.abort(reason) + await expect(cancelled).resolves.toBe(reason) + + const replacement = deferred() + const probe = run(() => replacement.promise) + const skipped = vi.fn<() => void>() + const waiting = run(skipped).catch((error: unknown) => error) + + abandoned.resolve() + await expect(waiting).resolves.toBeInstanceOf(CircuitOpenError) + expect(skipped).not.toHaveBeenCalled() + + replacement.resolve() + await probe + await expect(run(() => 'saved')).resolves.toBe('saved') + }) +}) + it('cancels backoff with the external signal and frees the slot without retrying', async () => { const run = protect({ concurrency: 1, diff --git a/src/protect.ts b/src/protect.ts index 44f0c58..37ca92d 100644 --- a/src/protect.ts +++ b/src/protect.ts @@ -1,9 +1,11 @@ -import { QueueFullError } from './errors.js' +import { Circuit } from './circuit.js' +import { CircuitOpenError, QueueFullError } from './errors.js' import { execute } from './retry.js' import type { ProtectOptions, RunContext, Runner, RunOptions } from './types.js' interface PendingOperation { start: () => void + reject: (error: unknown) => void } export function protect(options: ProtectOptions = {}): Runner { @@ -14,9 +16,18 @@ export function protect(options: ProtectOptions = {}): Runner { : { ...options.backoff, jitter: options.backoff.jitter ?? 0 } const concurrency = options.concurrency ?? Infinity const maxPending = options.maxPending ?? 0 + const circuit = options.circuit === undefined ? undefined : new Circuit({ ...options.circuit }) const pending = new Set() let active = 0 + function rejectQueue(): void { + const error = new CircuitOpenError() + + for (const entry of pending) { + entry.reject(error) + } + } + function drain(): void { while (active < concurrency) { const next = pending.values().next().value @@ -41,16 +52,41 @@ export function protect(options: ProtectOptions = {}): Runner { return } - function cancelPending(): void { + try { + circuit?.check() + } catch (error) { + reject(error) + return + } + + function rejectPending(error: unknown): void { pending.delete(entry) signal?.removeEventListener('abort', cancelPending) - reject(signal?.reason) + reject(error) + } + + function cancelPending(): void { + rejectPending(signal?.reason) } function start(): void { signal?.removeEventListener('abort', cancelPending) + + let admission: { generation: number; probe: boolean } | undefined + + try { + admission = circuit?.enter() + } catch (error) { + rejectPending(error) + return + } + active += 1 + if (admission?.probe) { + rejectQueue() + } + const controller = new AbortController() function forwardAbort(): void { @@ -58,6 +94,10 @@ export function protect(options: ProtectOptions = {}): Runner { } function onAbort(): void { + if (admission !== undefined) { + circuit?.cancel(admission.generation) + } + reject(controller.signal.reason) } @@ -82,17 +122,33 @@ export function protect(options: ProtectOptions = {}): Runner { timeoutMode, }).then( (value) => { + if (admission !== undefined && !controller.signal.aborted) { + circuit?.succeed(admission.generation) + } + 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() + } + } + reject(error) finish() }, ) } - const entry: PendingOperation = { start } + const entry: PendingOperation = { start, reject: rejectPending } if (active < concurrency && pending.size === 0) { start() diff --git a/src/types.ts b/src/types.ts index cc7d2b4..f21e4d4 100644 --- a/src/types.ts +++ b/src/types.ts @@ -4,6 +4,11 @@ export interface BackoffOptions { jitter?: number } +export interface CircuitOptions { + threshold: number + cooldown: number +} + export type TimeoutMode = 'immediate' | 'wait' export interface ProtectOptions { @@ -13,6 +18,7 @@ export interface ProtectOptions { timeoutMode?: TimeoutMode concurrency?: number maxPending?: number + circuit?: CircuitOptions onError?: (error: unknown) => void } From 86fac7031f5156df4dd04da9c322a7698f4ccb98 Mon Sep 17 00:00:00 2001 From: unsady <38117114+unsady@users.noreply.github.com> Date: Sun, 11 Oct 2026 17:16:54 +0200 Subject: [PATCH 2/2] feat: add per-call fallback --- README.md | 21 ++++++ src/fallback.ts | 39 ++++++++++ src/protect.test.ts | 176 +++++++++++++++++++++++++++++++++++++++++++- src/protect.ts | 18 +++-- src/types.ts | 9 ++- 5 files changed, 252 insertions(+), 11 deletions(-) create mode 100644 src/fallback.ts diff --git a/README.md b/README.md index 3adced3..6f2cf5c 100644 --- a/README.md +++ b/README.md @@ -80,6 +80,27 @@ External cancellation, `QueueFullError` and `CircuitOpenError` neither count as Circuit state is shared by all calls on a runner and independent between runners. Use a separate runner for each protected resource. +## Fallback + +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 })), +}) +``` + +`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 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. + +External cancellation also rejects a call while fallback is running. To stop the backup operation itself, forward the same external signal to its runner. + ## Cancellation ```ts diff --git a/src/fallback.ts b/src/fallback.ts new file mode 100644 index 0000000..7890b1f --- /dev/null +++ b/src/fallback.ts @@ -0,0 +1,39 @@ +export function recover( + fallback: (error: unknown) => Result, + error: unknown, + signal: AbortSignal | undefined, +): Promise> { + return new Promise((resolve, reject) => { + if (signal?.aborted) { + reject(signal.reason) + return + } + + function finish(): void { + signal?.removeEventListener('abort', onAbort) + } + + function onAbort(): void { + finish() + reject(signal?.reason) + } + + signal?.addEventListener('abort', onAbort, { once: true }) + + try { + void Promise.resolve(fallback(error)).then( + (value) => { + finish() + resolve(value) + }, + (error: unknown) => { + finish() + reject(error) + }, + ) + } catch (error) { + finish() + reject(error) + } + }) +} diff --git a/src/protect.test.ts b/src/protect.test.ts index 3d13a37..0299961 100644 --- a/src/protect.test.ts +++ b/src/protect.test.ts @@ -1,4 +1,4 @@ -import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest' +import { afterEach, beforeEach, describe, expect, expectTypeOf, it, vi } from 'vitest' import { CircuitOpenError, @@ -300,6 +300,180 @@ it('waits for timed-out callbacks before retrying or releasing the slot in wait expect(onError).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 error = new Error('primary failed') + const operation = vi.fn<() => Promise>().mockRejectedValue(error) + const fallback = vi.fn<(error: unknown) => Promise>(async () => 42) + + await expect(run(() => 'primary', { fallback })).resolves.toBe('primary') + expect(fallback).not.toHaveBeenCalled() + + const result = run(operation, { fallback }) + + expectTypeOf(result).toEqualTypeOf>() + await expect(result).resolves.toBe(42) + expect(operation).toHaveBeenCalledTimes(3) + expect(fallback.mock.calls).toEqual([[error]]) + expect(onError).not.toHaveBeenCalled() + 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 error = new Error('backup failed') + const fallback = vi.fn<(error: unknown) => Promise>(() => { + if (mode === 'throw') { + throw error + } + + return Promise.reject(error) + }) + + await expect( + run( + () => { + throw new Error('primary failed') + }, + { 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 backup = protect({ attempts: 2, circuit: { threshold: 1, cooldown: 100 } }) + const error = new Error('primary failed') + const operation = vi + .fn<() => Promise>() + .mockRejectedValueOnce(error) + .mockResolvedValue('primary') + const fallback = vi.fn<(error: unknown) => Promise>(() => backup(() => 'backup')) + + 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[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 primary = deferred() + const backup = deferred() + const fallback = vi.fn<(error: unknown) => Promise>(async () => { + await backup.promise + return 42 + }) + const result = run( + async () => { + await primary.promise + throw new Error('late') + }, + { fallback }, + ) + + await vi.advanceTimersByTimeAsync(100) + expect(fallback.mock.calls[0]?.[0]).toBeInstanceOf(TimeoutError) + await expect(run(() => 'primary')).resolves.toBe('primary') + await vi.advanceTimersByTimeAsync(1_000) + + backup.resolve() + await expect(result).resolves.toBe(42) + expect(fallback).toHaveBeenCalledTimes(1) + + primary.resolve() + await vi.advanceTimersByTimeAsync(0) + }) + + it('handles queue overflow without invoking the primary callback', async () => { + const run = protect({ concurrency: 1 }) + const blocker = deferred() + const running = run(() => blocker.promise) + const skipped = vi.fn<() => string>(() => 'primary') + const fallback = vi.fn<(error: unknown) => string>(() => 'backup') + const result = run(skipped, { fallback }) + + expectTypeOf(result).toEqualTypeOf>() + await expect(result).resolves.toBe('backup') + expect(fallback.mock.calls[0]?.[0]).toBeInstanceOf(QueueFullError) + expect(skipped).not.toHaveBeenCalled() + + blocker.resolve() + await running + }) + + it('does not recover already aborted, queued or active cancellations', async () => { + const run = protect({ concurrency: 1, maxPending: 1 }) + const reason = new Error('cancelled') + const skipped = vi.fn<() => void>() + const fallback = vi.fn<(error: unknown) => string>(() => 'backup') + + await expect(run(skipped, { signal: AbortSignal.abort(reason), fallback })).rejects.toBe(reason) + + const blocker = deferred() + const active = new AbortController() + const queued = new AbortController() + const running = run(() => blocker.promise, { signal: active.signal, fallback }).catch( + (error: unknown) => error, + ) + const waiting = run(skipped, { signal: queued.signal, fallback }).catch( + (error: unknown) => error, + ) + + queued.abort(reason) + active.abort(reason) + await expect(waiting).resolves.toBe(reason) + await expect(running).resolves.toBe(reason) + expect(fallback).not.toHaveBeenCalled() + expect(skipped).not.toHaveBeenCalled() + + blocker.resolve() + await vi.advanceTimersByTimeAsync(0) + }) + + it('cancels an active fallback and ignores its late rejection', async () => { + const onError = vi.fn<(error: unknown) => void>() + const run = protect({ onError }) + 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') + }) + const cancelled = run( + () => { + throw new Error('primary failed') + }, + { signal: controller.signal, fallback }, + ).catch((error: unknown) => error) + + await vi.advanceTimersByTimeAsync(0) + expect(fallback).toHaveBeenCalledTimes(1) + controller.abort(reason) + await expect(cancelled).resolves.toBe(reason) + + blocker.resolve() + await vi.advanceTimersByTimeAsync(0) + expect(onError.mock.calls).toEqual([[reason]]) + }) +}) + describe('circuit', () => { it('counts final failures after retries and resets the count on success', async () => { const run = protect({ attempts: 2, circuit: { threshold: 2, cooldown: 100 } }) diff --git a/src/protect.ts b/src/protect.ts index 37ca92d..aa5b991 100644 --- a/src/protect.ts +++ b/src/protect.ts @@ -1,5 +1,6 @@ import { Circuit } from './circuit.js' import { CircuitOpenError, QueueFullError } from './errors.js' +import { recover } from './fallback.js' import { execute } from './retry.js' import type { ProtectOptions, RunContext, Runner, RunOptions } from './types.js' @@ -41,11 +42,11 @@ export function protect(options: ProtectOptions = {}): Runner { } } - function run( + function run( operation: (context: RunContext) => Result, - runOptions: RunOptions = {}, - ): Promise> { - const { signal } = runOptions + runOptions: RunOptions = {}, + ): Promise | Awaited> { + const { signal, fallback } = runOptions const result = new Promise>((resolve, reject) => { if (signal?.aborted) { reject(signal.reason) @@ -160,11 +161,16 @@ export function protect(options: ProtectOptions = {}): Runner { } }) + const recovered = + fallback === undefined + ? result + : result.catch((error: unknown) => recover(fallback, error, signal)) + if (onError === undefined) { - return result + return recovered } - return result.catch((error: unknown) => { + return recovered.catch((error: unknown) => { onError(error) throw error }) diff --git a/src/types.ts b/src/types.ts index f21e4d4..dd9c819 100644 --- a/src/types.ts +++ b/src/types.ts @@ -26,13 +26,14 @@ export interface RunContext { signal: AbortSignal } -export interface RunOptions { +export interface RunOptions { signal?: AbortSignal + fallback?: (error: unknown) => Result } export interface Runner { - ( + ( operation: (context: RunContext) => Result, - options?: RunOptions, - ): Promise> + options?: RunOptions, + ): Promise | Awaited> }