diff --git a/.changeset/warm-donkeys-wait.md b/.changeset/warm-donkeys-wait.md new file mode 100644 index 0000000..ccfadb4 --- /dev/null +++ b/.changeset/warm-donkeys-wait.md @@ -0,0 +1,5 @@ +--- +"@api-wrappers/api-core": minor +--- + +Add an adaptive rate-limit plugin driven by standard and provider-specific response headers. diff --git a/src/__tests__/headerRateLimit.test.ts b/src/__tests__/headerRateLimit.test.ts new file mode 100644 index 0000000..5f07c43 --- /dev/null +++ b/src/__tests__/headerRateLimit.test.ts @@ -0,0 +1,68 @@ +import { describe, expect, it } from "bun:test"; +import type { RequestContext } from "../context/RequestContext"; +import type { ResponseContext } from "../context/ResponseContext"; +import { createHeaderRateLimitPlugin } from "../plugins/headerRateLimit/headerRateLimitPlugin"; + +const request: RequestContext = { + url: "https://api.test/graphql", + method: "POST", + headers: {}, + meta: {}, + retryCount: 0, + attempt: 0, +}; + +function response(headers: HeadersInit): ResponseContext { + return { + request, + response: new Response(null, { status: 200, headers }), + meta: {}, + }; +} + +describe("createHeaderRateLimitPlugin", () => { + it("waits until a server-reported reset after the budget is exhausted", async () => { + let currentTime = 1_000; + const waits: number[] = []; + const plugin = createHeaderRateLimitPlugin({ + now: () => currentTime, + wait: async (ms) => { + waits.push(ms); + currentTime += ms; + }, + }); + + await plugin.afterResponse?.( + response({ + "x-ratelimit-limit": "90", + "x-ratelimit-remaining": "0", + "x-ratelimit-reset": "3", + }), + ); + await plugin.beforeRequest?.(request); + + expect(waits).toEqual([2_000]); + expect(plugin.getState()).toEqual({ + limit: 90, + remaining: 0, + resetAt: 3_000, + blockedUntil: 3_000, + }); + }); + + it("honours Retry-After independently of remaining headers", async () => { + const waits: number[] = []; + const plugin = createHeaderRateLimitPlugin({ + now: () => 10_000, + wait: async (ms) => { + waits.push(ms); + }, + }); + + await plugin.afterResponse?.(response({ "retry-after": "5" })); + await plugin.beforeRequest?.(request); + + expect(waits).toEqual([5_000]); + expect(plugin.getState().blockedUntil).toBe(15_000); + }); +}); diff --git a/src/index.ts b/src/index.ts index b2b7fea..51b2a9b 100644 --- a/src/index.ts +++ b/src/index.ts @@ -81,7 +81,13 @@ export type { CachePluginOptions, CacheStore, } from "./plugins/cache/types"; - +export { createHeaderRateLimitPlugin } from "./plugins/headerRateLimit/headerRateLimitPlugin"; +export type { + HeaderRateLimitPlugin, + HeaderRateLimitPluginOptions, + HeaderRateLimitState, + RateLimitResetFormat, +} from "./plugins/headerRateLimit/types"; export { createLoggerPlugin } from "./plugins/logger/loggerPlugin"; export type { LoggerPluginOptions } from "./plugins/logger/types"; export { createRateLimitPlugin } from "./plugins/rateLimit/rateLimitPlugin"; diff --git a/src/plugins/headerRateLimit/headerRateLimitPlugin.ts b/src/plugins/headerRateLimit/headerRateLimitPlugin.ts new file mode 100644 index 0000000..7e08f05 --- /dev/null +++ b/src/plugins/headerRateLimit/headerRateLimitPlugin.ts @@ -0,0 +1,179 @@ +import type { ResponseContext } from "../../context/ResponseContext"; +import { createPassThroughError } from "../../plugin/passThroughError"; +import type { + HeaderRateLimitPlugin, + HeaderRateLimitPluginOptions, + HeaderRateLimitState, + RateLimitResetFormat, +} from "./types"; + +/** + * Adapts client-side request timing to rate-limit headers returned by an API. + * The plugin only pauses requests after the server reports an exhausted budget + * or supplies `Retry-After`; it does not assume a provider-specific limit. + */ +export function createHeaderRateLimitPlugin( + options: HeaderRateLimitPluginOptions = {}, +): HeaderRateLimitPlugin { + const limitHeader = normalizeHeader( + options.limitHeader ?? "x-ratelimit-limit", + ); + const remainingHeader = normalizeHeader( + options.remainingHeader ?? "x-ratelimit-remaining", + ); + const resetHeader = normalizeHeader( + options.resetHeader ?? "x-ratelimit-reset", + ); + const retryAfterHeader = normalizeHeader( + options.retryAfterHeader ?? "retry-after", + ); + const resetFormat = options.resetFormat ?? "unix-seconds"; + const bufferMs = options.bufferMs ?? 0; + const now = options.now ?? Date.now; + const wait = options.wait ?? waitWithSignal; + const state: HeaderRateLimitState = {}; + + if (bufferMs < 0) { + throw new Error("bufferMs must be greater than or equal to 0"); + } + + return { + name: "header-rate-limit", + priority: 0, + + async beforeRequest(ctx) { + const blockedUntil = state.blockedUntil ?? 0; + const waitMs = Math.max(0, blockedUntil - now()); + if (waitMs > 0) await wait(waitMs, ctx.signal); + return ctx; + }, + + afterResponse(ctx) { + updateState(ctx, { + limitHeader, + remainingHeader, + resetHeader, + retryAfterHeader, + resetFormat, + bufferMs, + now, + state, + }); + return ctx; + }, + + getState() { + return { ...state }; + }, + }; +} + +interface StateUpdateOptions { + limitHeader: string; + remainingHeader: string; + resetHeader: string; + retryAfterHeader: string; + resetFormat: RateLimitResetFormat; + bufferMs: number; + now: () => number; + state: HeaderRateLimitState; +} + +function updateState(ctx: ResponseContext, options: StateUpdateOptions): void { + const currentTime = options.now(); + const headers = ctx.response.headers; + const limit = parseFiniteNumber(headers.get(options.limitHeader)); + const remaining = parseFiniteNumber(headers.get(options.remainingHeader)); + const resetAt = parseResetAt( + headers.get(options.resetHeader), + options.resetFormat, + currentTime, + ); + const retryAt = parseRetryAfter( + headers.get(options.retryAfterHeader), + currentTime, + ); + + if (limit !== undefined) options.state.limit = limit; + if (remaining !== undefined) options.state.remaining = remaining; + if (resetAt !== undefined) options.state.resetAt = resetAt; + + let blockedUntil = options.state.blockedUntil ?? 0; + if (remaining !== undefined && remaining <= 0 && resetAt !== undefined) { + blockedUntil = Math.max(blockedUntil, resetAt + options.bufferMs); + } + if (retryAt !== undefined) { + blockedUntil = Math.max(blockedUntil, retryAt + options.bufferMs); + } + + if (blockedUntil > currentTime) { + options.state.blockedUntil = blockedUntil; + } else { + delete options.state.blockedUntil; + } +} + +function parseResetAt( + value: string | null, + format: RateLimitResetFormat, + now: number, +): number | undefined { + const parsed = parseFiniteNumber(value); + if (parsed === undefined) return undefined; + if (format === "unix-milliseconds") return Math.max(0, parsed); + if (format === "delay-seconds") return now + Math.max(0, parsed * 1_000); + return Math.max(0, parsed * 1_000); +} + +function parseRetryAfter( + value: string | null, + now: number, +): number | undefined { + if (!value) return undefined; + const seconds = Number(value); + if (Number.isFinite(seconds)) return now + Math.max(0, seconds * 1_000); + const date = Date.parse(value); + return Number.isNaN(date) ? undefined : Math.max(now, date); +} + +function parseFiniteNumber(value: string | null): number | undefined { + if (value === null || value.trim() === "") return undefined; + const parsed = Number(value); + return Number.isFinite(parsed) ? parsed : undefined; +} + +function normalizeHeader(value: string): string { + return value.toLowerCase(); +} + +function waitWithSignal(ms: number, signal?: AbortSignal): Promise { + return new Promise((resolve, reject) => { + if (signal?.aborted) { + reject(createPassThroughError(getAbortReason(signal))); + return; + } + + let onAbort: (() => void) | undefined; + const cleanup = () => { + if (onAbort) signal?.removeEventListener("abort", onAbort); + }; + const timer = setTimeout(() => { + cleanup(); + resolve(); + }, ms); + onAbort = () => { + clearTimeout(timer); + cleanup(); + reject(createPassThroughError(getAbortReason(signal))); + }; + signal?.addEventListener("abort", onAbort, { once: true }); + }); +} + +function getAbortReason(signal?: AbortSignal): unknown { + if (signal?.reason !== undefined) return signal.reason; + if (typeof DOMException !== "undefined") { + return new DOMException("The operation was aborted.", "AbortError"); + } + return new Error("The operation was aborted."); +} diff --git a/src/plugins/headerRateLimit/types.ts b/src/plugins/headerRateLimit/types.ts new file mode 100644 index 0000000..88c9698 --- /dev/null +++ b/src/plugins/headerRateLimit/types.ts @@ -0,0 +1,28 @@ +import type { ApiPlugin } from "../../plugin/types"; + +export type RateLimitResetFormat = + | "unix-seconds" + | "unix-milliseconds" + | "delay-seconds"; + +export interface HeaderRateLimitState { + limit?: number; + remaining?: number; + resetAt?: number; + blockedUntil?: number; +} + +export interface HeaderRateLimitPluginOptions { + limitHeader?: string; + remainingHeader?: string; + resetHeader?: string; + retryAfterHeader?: string; + resetFormat?: RateLimitResetFormat; + bufferMs?: number; + now?: () => number; + wait?: (ms: number, signal?: AbortSignal) => Promise; +} + +export interface HeaderRateLimitPlugin extends ApiPlugin { + getState(): Readonly; +}