From f2aa4f14f54fa9ec6427bdb23925ac2af036c039 Mon Sep 17 00:00:00 2001 From: Ethan Elsop Date: Wed, 22 Jul 2026 19:41:16 -0400 Subject: [PATCH] Add optional request governor for concurrency and rate limiting (1.2.0) Introduces a client-side governor so consumers can stay within an upstream API's concurrency and rate limits during large or bursty workloads such as long paginated pulls. New FetchEnhConfig options (all optional, also accepted by setConfig): - concurrency: cap logical requests in flight via a fair FIFO semaphore, applied once per logical request so a slot is not held during retry backoff. - maxRps: cap requests started per second (minimum spacing between starts). - minIntervalMs: explicit minimum spacing; takes precedence over maxRps. Both gates honour a request's AbortSignal (a cancelled queued request rejects promptly and never runs), and the governor is a zero-overhead pass-through when no option is set. Adds a governor test suite; full suite 261 passing. --- CHANGELOG.md | 27 ++++- package-lock.json | 4 +- package.json | 2 +- src/core/governor.ts | 126 +++++++++++++++++++++++ src/index.ts | 72 +++++++++---- src/types/config.ts | 17 +++ tests/fetchEnh.governor.test.ts | 176 ++++++++++++++++++++++++++++++++ 7 files changed, 398 insertions(+), 26 deletions(-) create mode 100644 src/core/governor.ts create mode 100644 tests/fetchEnh.governor.test.ts 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); + }); +});