diff --git a/README.md b/README.md index 91c1590..3859dc7 100644 --- a/README.md +++ b/README.md @@ -72,7 +72,7 @@ Supported selection policies: - `{ type: 'priority', priorities?: { ... } }` (default): highest priority first. Missing priorities are `0`; ties use registration order. After cooldown, a preferred provider can receive one recovery probe while other requests use alternatives. - `{ type: 'sticky' }`: keep the successful provider until it fails or rejects circuit admission. Start with the first registered provider, then try other providers in registration order. The first successful switch from a selection generation wins; late results cannot overwrite a newer choice. -Provider `protection` inherits shared settings. Nested `backoff` and `circuit` objects replace the inherited object entirely; `circuit: false` disables an inherited breaker. Provider IDs must be non-empty and unique. Explicit priority keys must identify registered providers, and values must be finite. Router concurrency must be a positive integer or `Infinity`, and `maxPending` a non-negative integer. +Provider `protection` inherits shared settings. Nested `backoff` and `circuit` objects replace the inherited object entirely; `circuit: false` disables an inherited breaker. Provider IDs must be non-empty and unique. Explicit priority keys must identify registered providers, and values must be finite. Both `protect()` and router require concurrency to be a positive integer or `Infinity`, and `maxPending` a non-negative integer or `Infinity`. A router's `errorPolicy`, or its replacement in `execute` options, returns `retry`, `failover` and `countForCircuit`. All three default to `true`. `failover` permits switching after local retries finish or are disabled. Circuit rejection skips a provider without running the policy. An error with `failover: false` rejects directly, and a policy exception rejects with `ErrorPolicyError` without switching. @@ -132,7 +132,7 @@ To limit the entire call, including queue wait and backoff, pass an external sig `concurrency` defaults to `Infinity`: no limit. A slot stays occupied for the entire operation, including retries and backoff. The limit counts active calls, not callbacks left running after an immediate timeout. -`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. +`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. Use `Infinity` for an unbounded queue. ## Circuit breaker diff --git a/src/protect.test.ts b/src/protect.test.ts index 30dcc75..a1d145c 100644 --- a/src/protect.test.ts +++ b/src/protect.test.ts @@ -29,6 +29,14 @@ afterEach(() => { vi.restoreAllMocks() }) +it.each([0, -1, 1.5, NaN, -Infinity])('rejects invalid concurrency %s', (concurrency) => { + expect(() => protect({ concurrency })).toThrow(RangeError) +}) + +it.each([-1, 1.5, NaN, -Infinity])('rejects invalid pending limit %s', (maxPending) => { + expect(() => protect({ maxPending })).toThrow(RangeError) +}) + it('returns operation results and does not retry by default', async () => { const run = protect() const error = new Error('failed') diff --git a/src/router.test.ts b/src/router.test.ts index 6ee0e30..4c71da0 100644 --- a/src/router.test.ts +++ b/src/router.test.ts @@ -72,6 +72,21 @@ it('uses registration order by default and breaks equal priorities with that ord await expect(tied.execute((resource) => resource)).resolves.toBe('a') }) +it.each(['toString', 'constructor', '__proto__'])( + 'ignores inherited priorities for provider %s', + async (id) => { + const router = createRouter({ + providers: [ + { id, resource: id }, + { id: 'preferred', resource: 'preferred' }, + ], + selection: { type: 'priority', priorities: { preferred: 10 } }, + }) + + await expect(router.execute((resource) => resource)).resolves.toBe('preferred') + }, +) + it('inherits protection, replaces nested settings and can disable an inherited circuit', async () => { vi.spyOn(Math, 'random').mockReturnValue(0.5) @@ -561,6 +576,18 @@ it('waits for a timed-out callback before failing over and releasing its slot', expect(visits).toEqual(['a', 'b']) }) +it('supports an unbounded pending limit', async () => { + const router = createRouter({ providers, concurrency: 1, maxPending: Infinity }) + const blocker = deferred() + const active = router.execute(() => blocker.promise) + const first = router.execute(() => 'first') + const second = router.execute(() => 'second') + + blocker.resolve() + await active + await expect(Promise.all([first, second])).resolves.toEqual(['first', 'second']) +}) + describe('configuration', () => { it.each<{ name: string; options: Partial> }>([ { name: 'empty providers', options: { providers: [] } }, diff --git a/src/router.ts b/src/router.ts index e4efe43..cb80105 100644 --- a/src/router.ts +++ b/src/router.ts @@ -8,6 +8,7 @@ export function createRouter( options: RouterOptions, ): Router { const selection = options.selection ?? { type: 'priority' } + const priorities = selection.type === 'priority' ? selection.priorities : undefined const ids = new Set() const providers = options.providers.map((provider) => { if (provider.id.length === 0 || ids.has(provider.id)) { @@ -20,7 +21,10 @@ export function createRouter( id: provider.id, resource: provider.resource, protection: new Protection(provider.protection, options.protection), - priority: selection.type === 'priority' ? (selection.priorities?.[provider.id] ?? 0) : 0, + priority: + priorities !== undefined && Object.hasOwn(priorities, provider.id) + ? (priorities[provider.id] ?? 0) + : 0, } }) const first = providers[0] @@ -30,7 +34,7 @@ export function createRouter( } if (selection.type === 'priority') { - for (const [id, priority] of Object.entries(selection.priorities ?? {})) { + for (const [id, priority] of Object.entries(priorities ?? {})) { if (!ids.has(id) || !Number.isFinite(priority)) { throw new TypeError('Priorities must reference registered providers and be finite') } @@ -41,21 +45,10 @@ export function createRouter( throw new TypeError('Unknown selection policy') } - const concurrency = options.concurrency ?? Infinity - const maxPending = options.maxPending ?? 0 - - if (concurrency !== Infinity && (!Number.isInteger(concurrency) || concurrency < 1)) { - throw new RangeError('concurrency must be a positive integer or Infinity') - } - - if (!Number.isInteger(maxPending) || maxPending < 0) { - throw new RangeError('maxPending must be a non-negative integer') - } - const sticky = selection.type === 'sticky' ? new StickySelection(first, providers.slice(1)) : undefined const defaultErrorPolicy = options.errorPolicy - const scheduler = new Scheduler(concurrency, maxPending) + const scheduler = new Scheduler(options.concurrency ?? Infinity, options.maxPending ?? 0) function run( operation: (resource: Resource, context: RunContext) => Result, diff --git a/src/scheduler.ts b/src/scheduler.ts index d8f8531..0754111 100644 --- a/src/scheduler.ts +++ b/src/scheduler.ts @@ -12,7 +12,15 @@ export class Scheduler { constructor( private readonly concurrency: number, private readonly maxPending: number, - ) {} + ) { + if (concurrency !== Infinity && (!Number.isInteger(concurrency) || concurrency < 1)) { + throw new RangeError('concurrency must be a positive integer or Infinity') + } + + if (maxPending !== Infinity && (!Number.isInteger(maxPending) || maxPending < 0)) { + throw new RangeError('maxPending must be a non-negative integer or Infinity') + } + } rejectPending(error: unknown): void { for (const entry of this.pending) {