Add optional request governor for concurrency and rate limiting (1.2.0)

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

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",

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<T>(fn: () => Promise<T>, signal?: AbortSignal): Promise<T> {
+    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<void> {
+    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<void>((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<void> {
+    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;
+}

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<T> => {
-      const interceptedRequest = await this._applyRequestInterceptors(request);
-      return this._fetchAndParse(
-        interceptedRequest,
-        responseType,
-        retries,
-        timeout,
-        signal,
-        1,
-        startTime,
-        {
-          method: methodUpper,
-          bodyFactory,
-          bodyReplayable,
-          rawBody,
-        },
-        callRetryConfig
-      ) as Promise<T>;
-    })();
+    const promise = this._governor.run<T>(() => {
+      const startTime = Date.now();
+      return (async (): Promise<T> => {
+        const interceptedRequest = await this._applyRequestInterceptors(request);
+        return this._fetchAndParse(
+          interceptedRequest,
+          responseType,
+          retries,
+          timeout,
+          signal,
+          1,
+          startTime,
+          {
+            method: methodUpper,
+            bodyFactory,
+            bodyReplayable,
+            rawBody,
+          },
+          callRetryConfig
+        ) as Promise<T>;
+      })();
+    }, signal);
 
     if (shouldDedupe) {
       this._dedupeCache.track(dedupeKey, promise);
@@ -899,7 +915,8 @@ class FetchEnh {
   setConfig(config: FetchEnhConfig): void {
     const knownKeys = new Set<string>([
       '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);
+    }
   }
 
   /**

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;
 }