diff --git a/CHANGELOG.md b/CHANGELOG.md index 130b902..cf7a3bd 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -9,6 +9,30 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 _No changes yet._ +## [1.2.0] — 2026-07-22 + +Adds an optional client-side request governor for staying within an upstream +API's rate and concurrency limits during large or bursty workloads (e.g. +long paginated pulls). Purely additive — no behavioural change when the new +options are omitted. + +### Added + +- **`FetchEnhConfig.concurrency`** — caps the number of logical requests in + flight at once with a fair FIFO semaphore. Applied **once per logical + request**, not per retry attempt, so a request does not hold a slot while it + sleeps between retries. +- **`FetchEnhConfig.maxRps`** — caps requests started per second, enforced as a + minimum spacing (`1000 / maxRps` ms) between request starts. +- **`FetchEnhConfig.minIntervalMs`** — explicit minimum spacing (ms) between + request starts; takes precedence over `maxRps`. +- All three are also accepted by **`setConfig(...)`**, which rebuilds the + governor with the merged values. +- Both gates honour a request's `AbortSignal`: a cancelled request waiting in + the concurrency queue rejects promptly with an `AbortError` and never runs. +- When none of the three options is set, the governor is a zero-overhead + pass-through. + ## [1.1.0] — 2026-07-22 Data-integrity and resilience release. One behavioural change to pagination @@ -397,6 +421,7 @@ PKCE refresh recovery, full type-system rigor across primitive bodies and JSON encoding, and `duplex: 'half'` propagation for `ReadableStream` bodies — is captured here as the canonical first set of release notes. -[Unreleased]: https://github.com/erelsop/fetch-enh/compare/v1.1.0...HEAD +[Unreleased]: https://github.com/erelsop/fetch-enh/compare/v1.2.0...HEAD +[1.2.0]: https://github.com/erelsop/fetch-enh/compare/v1.1.0...v1.2.0 [1.1.0]: https://github.com/erelsop/fetch-enh/compare/v1.0.0...v1.1.0 [1.0.0]: https://github.com/erelsop/fetch-enh/releases/tag/v1.0.0 diff --git a/package-lock.json b/package-lock.json index 025e233..0194932 100644 --- a/package-lock.json +++ b/package-lock.json @@ -1,12 +1,12 @@ { "name": "@erelsop/fetch-enh", - "version": "1.1.0", + "version": "1.2.0", "lockfileVersion": 3, "requires": true, "packages": { "": { "name": "@erelsop/fetch-enh", - "version": "1.1.0", + "version": "1.2.0", "license": "MIT", "devDependencies": { "@types/jest": "^29.5.6", diff --git a/package.json b/package.json index 4b71dad..bb24539 100644 --- a/package.json +++ b/package.json @@ -1,6 +1,6 @@ { "name": "@erelsop/fetch-enh", - "version": "1.1.0", + "version": "1.2.0", "description": "An enhanced fetch utility with built-in retry logic, authentication strategies, request/response interceptors, and intelligent error handling.", "main": "dist/index.js", "types": "dist/index.d.ts", diff --git a/src/core/governor.ts b/src/core/governor.ts new file mode 100644 index 0000000..1b36c30 --- /dev/null +++ b/src/core/governor.ts @@ -0,0 +1,126 @@ +import { sleep } from './retryEngine'; + +/** + * Options controlling the {@link RequestGovernor}. All fields are optional; an + * empty config produces a pass-through governor with no limiting. + */ +export interface GovernorOptions { + /** Maximum number of logical requests allowed in flight at once. */ + concurrency?: number; + /** + * Maximum requests started per second. Converted internally to a minimum + * spacing between request starts (`1000 / maxRps` ms). Ignored if + * `minIntervalMs` is also set. + */ + maxRps?: number; + /** Minimum spacing (ms) between successive request starts. Takes precedence over `maxRps`. */ + minIntervalMs?: number; +} + +/** + * Serialises access to the network according to a concurrency ceiling and/or a + * minimum spacing between request starts. Applied once per *logical* request + * (not per retry attempt), so a request does not hold a concurrency slot while + * it sleeps between retries. + * + * The concurrency gate is a fair FIFO semaphore. The rate gate is a virtual + * "next allowed start" clock, so spacing holds across bursts without needing a + * background timer. Both gates honour an optional `AbortSignal` so a cancelled + * request does not sit in the queue forever. + * + * @internal + */ +export class RequestGovernor { + private readonly _concurrency?: number; + private readonly _minIntervalMs?: number; + + private _active = 0; + private _waiters: Array<{ resolve: () => void; reject: (e: unknown) => void; onAbort?: () => void; signal?: AbortSignal }> = []; + private _nextAllowedStart = 0; + + constructor(opts: GovernorOptions = {}) { + if (opts.concurrency != null && opts.concurrency > 0) { + this._concurrency = opts.concurrency; + } + if (opts.minIntervalMs != null && opts.minIntervalMs > 0) { + this._minIntervalMs = opts.minIntervalMs; + } else if (opts.maxRps != null && opts.maxRps > 0) { + this._minIntervalMs = 1000 / opts.maxRps; + } + } + + /** True when this governor imposes no limits (fast path). */ + get isNoop(): boolean { + return this._concurrency === undefined && this._minIntervalMs === undefined; + } + + /** + * Runs `fn` under the governor's limits. Acquires a concurrency slot (waiting + * in FIFO order if at capacity), then waits out any required inter-request + * spacing, then invokes `fn`. The slot is always released when `fn` settles. + */ + async run(fn: () => Promise, signal?: AbortSignal): Promise { + if (this.isNoop) return fn(); + await this._acquireSlot(signal); + try { + await this._awaitRateWindow(signal); + return await fn(); + } finally { + this._releaseSlot(); + } + } + + private async _acquireSlot(signal?: AbortSignal): Promise { + if (this._concurrency === undefined) return; + if (this._active < this._concurrency) { + this._active++; + return; + } + // At capacity — queue until a slot frees up (or the caller aborts). + await new Promise((resolve, reject) => { + if (signal?.aborted) { + reject(makeAbortError()); + return; + } + const waiter = { resolve, reject, signal, onAbort: undefined as (() => void) | undefined }; + if (signal) { + waiter.onAbort = () => { + const idx = this._waiters.indexOf(waiter); + if (idx >= 0) this._waiters.splice(idx, 1); + reject(makeAbortError()); + }; + signal.addEventListener('abort', waiter.onAbort, { once: true }); + } + this._waiters.push(waiter); + }); + // A waiter is only resolved by _releaseSlot, which hands over the slot + // without decrementing — so _active already accounts for this holder. + } + + private _releaseSlot(): void { + if (this._concurrency === undefined) return; + const next = this._waiters.shift(); + if (next) { + // Hand the slot directly to the next waiter (no decrement/increment race). + if (next.onAbort && next.signal) next.signal.removeEventListener('abort', next.onAbort); + next.resolve(); + } else { + this._active--; + } + } + + private async _awaitRateWindow(signal?: AbortSignal): Promise { + if (this._minIntervalMs === undefined) return; + const now = Date.now(); + const start = Math.max(now, this._nextAllowedStart); + this._nextAllowedStart = start + this._minIntervalMs; + const wait = start - now; + if (wait > 0) await sleep(wait, signal); + } +} + +function makeAbortError(): Error { + const err: any = new Error('The operation was aborted.'); + err.name = 'AbortError'; + return err; +} diff --git a/src/index.ts b/src/index.ts index ad4d076..ebff8b2 100644 --- a/src/index.ts +++ b/src/index.ts @@ -45,6 +45,7 @@ import { classifyRetry, } from './core/retryEngine'; import { DeduplicationCache } from './core/deduplication'; +import { RequestGovernor, type GovernorOptions } from './core/governor'; import { paginate, paginateCursor, paginateIter, paginateCursorIter } from './core/pagination'; /** @@ -98,6 +99,8 @@ class FetchEnh { private _interceptors = new InterceptorPipeline(); private _auth = new AuthPipeline(); private _dedupeCache = new DeduplicationCache(); + private _governorOpts: GovernorOptions = {}; + private _governor = new RequestGovernor(); constructor({ baseURL = '', @@ -107,6 +110,9 @@ class FetchEnh { queryStyle, dedupe, dedupeKey, + concurrency, + maxRps, + minIntervalMs, onRetry, onComplete, }: FetchEnhConfig = {}) { @@ -126,6 +132,8 @@ class FetchEnh { if (dedupeKey) this._dedupeKey = dedupeKey; if (onRetry) this._onRetry = onRetry; if (onComplete) this._onComplete = onComplete; + this._governorOpts = { concurrency, maxRps, minIntervalMs }; + this._governor = new RequestGovernor(this._governorOpts); } // ── Interceptor delegation ──────────────────────────────────────────────── @@ -487,31 +495,39 @@ class FetchEnh { // Wrap interceptors + _fetchAndParse in a synchronously-created promise so that // dedup tracking can happen before the first `await`. If we awaited interceptors // inline here, a concurrent identical request could slip past the dedup check - // during the async gap and produce a duplicate in-flight request. + // during the async gap and produce a duplicate in-flight request. `governor.run` + // returns its promise synchronously, so dedup tracking below still runs before + // any await; deduplicated callers share the single governed request. + // + // The governor gates each *logical* request (concurrency + rate spacing) once, + // outside the retry loop, so a request does not hold a concurrency slot while + // it sleeps between retries. When no limits are configured it is a pass-through. // // Request interceptors run once per logical call (here, inside the IIFE). // Auth strategies run once per attempt inside _fetchAndParse so that a token // refresh during a retry window takes effect immediately. - const startTime = Date.now(); - const promise = (async (): Promise => { - const interceptedRequest = await this._applyRequestInterceptors(request); - return this._fetchAndParse( - interceptedRequest, - responseType, - retries, - timeout, - signal, - 1, - startTime, - { - method: methodUpper, - bodyFactory, - bodyReplayable, - rawBody, - }, - callRetryConfig - ) as Promise; - })(); + const promise = this._governor.run(() => { + const startTime = Date.now(); + return (async (): Promise => { + const interceptedRequest = await this._applyRequestInterceptors(request); + return this._fetchAndParse( + interceptedRequest, + responseType, + retries, + timeout, + signal, + 1, + startTime, + { + method: methodUpper, + bodyFactory, + bodyReplayable, + rawBody, + }, + callRetryConfig + ) as Promise; + })(); + }, signal); if (shouldDedupe) { this._dedupeCache.track(dedupeKey, promise); @@ -899,7 +915,8 @@ class FetchEnh { setConfig(config: FetchEnhConfig): void { const knownKeys = new Set([ 'baseURL', 'defaultHeaders', 'defaultTimeout', 'defaultRetries', - 'queryStyle', 'dedupe', 'dedupeKey', 'onRetry', 'onComplete', + 'queryStyle', 'dedupe', 'dedupeKey', 'concurrency', 'maxRps', + 'minIntervalMs', 'onRetry', 'onComplete', ]); for (const key of Object.keys(config)) { if (!knownKeys.has(key)) { @@ -937,6 +954,17 @@ class FetchEnh { if ('onComplete' in config) { this._onComplete = config.onComplete; } + // Rebuild the governor if any of its inputs changed. A fresh governor starts + // with an empty queue and reset rate clock; avoid changing these mid-flight + // while requests are queued. + if ('concurrency' in config || 'maxRps' in config || 'minIntervalMs' in config) { + this._governorOpts = { + concurrency: 'concurrency' in config ? config.concurrency : this._governorOpts.concurrency, + maxRps: 'maxRps' in config ? config.maxRps : this._governorOpts.maxRps, + minIntervalMs: 'minIntervalMs' in config ? config.minIntervalMs : this._governorOpts.minIntervalMs, + }; + this._governor = new RequestGovernor(this._governorOpts); + } } /** diff --git a/src/types/config.ts b/src/types/config.ts index 2d35b3b..0677949 100644 --- a/src/types/config.ts +++ b/src/types/config.ts @@ -9,6 +9,23 @@ export interface FetchEnhConfig { }; readonly dedupe?: boolean; readonly dedupeKey?: (params: { method: string; url: string; body?: unknown }) => string; + /** + * Maximum number of logical requests allowed in flight at once. Applied once + * per logical request (not per retry attempt), so a request waiting out a + * retry backoff does not occupy a slot. Omit for unlimited concurrency. + */ + readonly concurrency?: number; + /** + * Maximum requests started per second, enforced as a minimum spacing between + * request starts (`1000 / maxRps` ms). Useful for staying under an API's rate + * limit on long paginated pulls. Ignored if `minIntervalMs` is also set. + */ + readonly maxRps?: number; + /** + * Minimum spacing (ms) between successive request starts. Takes precedence + * over `maxRps`. Omit for no spacing. + */ + readonly minIntervalMs?: number; readonly onRetry?: (info: { attempt: number; delay: number; method: string; url: string; reason: 'status' | 'network'; status?: number }) => void; readonly onComplete?: (info: { method: string; url: string; status?: number; ok: boolean; attempts: number; elapsedMs: number }) => void; } diff --git a/tests/fetchEnh.governor.test.ts b/tests/fetchEnh.governor.test.ts new file mode 100644 index 0000000..7257e91 --- /dev/null +++ b/tests/fetchEnh.governor.test.ts @@ -0,0 +1,176 @@ +/** + * fetchEnh.governor.test.ts + * + * Tests for the F3 request governor: concurrency ceiling, rate spacing + * (maxRps / minIntervalMs), abort-while-queued, and the no-op fast path. + * The RequestGovernor unit is exercised directly (deterministic), and the + * FetchEnh integration is smoke-tested via fetch-mock. + */ +import FetchEnh from '../src'; +import { RequestGovernor } from '../src/core/governor'; +import fetchMock from 'jest-fetch-mock'; + +beforeEach(() => { + fetchMock.resetMocks(); +}); + +const deferred = () => { + let resolve!: (v: T) => void; + let reject!: (e: unknown) => void; + const promise = new Promise((res, rej) => { resolve = res; reject = rej; }); + return { promise, resolve, reject }; +}; + +describe('RequestGovernor — concurrency', () => { + test('never exceeds the concurrency ceiling and drains all tasks', async () => { + const gov = new RequestGovernor({ concurrency: 2 }); + let active = 0; + let peak = 0; + const gates = [deferred(), deferred(), deferred(), deferred(), deferred()]; + + const runs = gates.map((g, i) => + gov.run(async () => { + active++; + peak = Math.max(peak, active); + await g.promise; + active--; + return i; + }), + ); + + // Let the scheduler settle: only 2 should be active. + await Promise.resolve(); + await Promise.resolve(); + expect(active).toBe(2); + + // Release one at a time; a queued task should take each freed slot. + for (const g of gates) { + g.resolve(undefined); + await Promise.resolve(); + await Promise.resolve(); + } + + const results = await Promise.all(runs); + expect(results).toEqual([0, 1, 2, 3, 4]); + expect(peak).toBe(2); + expect(active).toBe(0); + }); + + test('queued tasks are served in FIFO order', async () => { + const gov = new RequestGovernor({ concurrency: 1 }); + const order: number[] = []; + const g0 = deferred(); + + // First task occupies the single slot until we release g0. + const first = gov.run(async () => { order.push(0); await g0.promise; }); + await Promise.resolve(); + + // These three queue behind it. + const rest = [1, 2, 3].map((n) => gov.run(async () => { order.push(n); })); + + g0.resolve(undefined); + await Promise.all([first, ...rest]); + expect(order).toEqual([0, 1, 2, 3]); + }); + + test('a rejecting task still releases its slot', async () => { + const gov = new RequestGovernor({ concurrency: 1 }); + await expect(gov.run(async () => { throw new Error('boom'); })).rejects.toThrow('boom'); + // Slot must be free for the next task. + await expect(gov.run(async () => 'ok')).resolves.toBe('ok'); + }); +}); + +describe('RequestGovernor — abort while queued', () => { + test('an aborted queued task rejects and frees no real slot', async () => { + const gov = new RequestGovernor({ concurrency: 1 }); + const g0 = deferred(); + const first = gov.run(async () => { await g0.promise; }); + await Promise.resolve(); + + const ac = new AbortController(); + const queued = gov.run(async () => 'should-not-run', ac.signal); + ac.abort(); + + await expect(queued).rejects.toMatchObject({ name: 'AbortError' }); + + // The first task can still finish and a fresh task can acquire the slot. + g0.resolve(undefined); + await first; + await expect(gov.run(async () => 'ok')).resolves.toBe('ok'); + }); +}); + +describe('RequestGovernor — rate spacing', () => { + test('minIntervalMs spaces out successive starts', async () => { + const gov = new RequestGovernor({ minIntervalMs: 40 }); + const starts: number[] = []; + const t0 = Date.now(); + await Promise.all( + [0, 1, 2].map(() => gov.run(async () => { starts.push(Date.now() - t0); })), + ); + starts.sort((a, b) => a - b); + // Second start ≥ ~40ms after first, third ≥ ~80ms (allow scheduler slack). + expect(starts[1]).toBeGreaterThanOrEqual(30); + expect(starts[2]).toBeGreaterThanOrEqual(70); + }); + + test('maxRps is converted to a minimum interval', async () => { + const gov = new RequestGovernor({ maxRps: 25 }); // 40ms spacing + const starts: number[] = []; + const t0 = Date.now(); + await Promise.all( + [0, 1].map(() => gov.run(async () => { starts.push(Date.now() - t0); })), + ); + starts.sort((a, b) => a - b); + expect(starts[1]).toBeGreaterThanOrEqual(30); + }); +}); + +describe('RequestGovernor — no-op', () => { + test('isNoop is true with no limits and runs fn directly', async () => { + const gov = new RequestGovernor(); + expect(gov.isNoop).toBe(true); + await expect(gov.run(async () => 42)).resolves.toBe(42); + }); +}); + +describe('FetchEnh integration', () => { + test('concurrency caps simultaneous in-flight requests', async () => { + let active = 0; + let peak = 0; + fetchMock.mockResponse(async () => { + active++; + peak = Math.max(peak, active); + await new Promise((r) => setTimeout(r, 15)); + active--; + return { body: JSON.stringify({ ok: true }), init: { status: 200, headers: { 'content-type': 'application/json' } } } as any; + }); + + const api = new FetchEnh({ baseURL: 'https://api.test', concurrency: 3 }); + await Promise.all( + Array.from({ length: 9 }, (_, i) => api.get({ endpoint: `/r/${i}`, responseType: 'json' })), + ); + expect(peak).toBeLessThanOrEqual(3); + expect(fetchMock).toHaveBeenCalledTimes(9); + }); + + test('setConfig can install a governor after construction', async () => { + let active = 0; + let peak = 0; + fetchMock.mockResponse(async () => { + active++; + peak = Math.max(peak, active); + await new Promise((r) => setTimeout(r, 10)); + active--; + return { body: '{}', init: { status: 200, headers: { 'content-type': 'application/json' } } } as any; + }); + + const api = new FetchEnh({ baseURL: 'https://api.test' }); + api.setConfig({ concurrency: 2 }); + await Promise.all( + Array.from({ length: 6 }, (_, i) => api.get({ endpoint: `/r/${i}`, responseType: 'json' })), + ); + expect(peak).toBeLessThanOrEqual(2); + }); +});