pjandro/OmniRoute
1
0
Fork 0
mirror of https://github.com/diegosouzapw/OmniRoute.git synced 2026-08-06 23:32:12 +03:00
Code Issues Packages Projects Releases Wiki Activity

fix(resilience): enforce RPM with rolling leases (#9604)

Browse Source 
Validated in local merge-train (diegosouzapw batch)
This commit is contained in:
Diego Rodrigues de Sa e Souza
2026-08-06 10:41:04 -03:00
committed by GitHub
parent f2e36ad0ce
commit ece486dc38
11 changed files with 1089 additions and 289 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
changelog.d/fixes/9259-rolling-rpm-leases.md Normal file
View File

@@ -0,0 +1 @@
- **fix(resilience):** Enforce RPM limits with rolling leases and atomic global/provider/account admission ([#9259](https://github.com/diegosouzapw/OmniRoute/pull/9259)). The configured global RPM budget is shared across all enabled provider connections within one process; provider/account overrides add narrower scopes.

5
config/quality/eslint-suppressions.json
View File

@@ -249,11 +249,6 @@
"count": 1
}
},
"open-sse/services/rateLimitManager.ts": {
"no-restricted-syntax": {
"count": 1
}
},
"open-sse/services/taskAwareRouter.ts": {
"@typescript-eslint/no-explicit-any": {
"count": 4

5
config/quality/file-size-baseline.json
View File

@@ -1,5 +1,6 @@
{
"_rebaseline_2026_07_24_8470_hyperagent_sticky_thread": "PR #8470 (artickc, fix/hyperagent-tool-loop-thread-sticky) own growth: open-sse/executors/hyperagent.ts 936->1025 (wc -l; check-file-size.mjs counts via split(\"\\n\").length so the gate sees 937->1026, +89, crosses the 1000 cap). Fixes a real bug where a reverse-conversion proxy (text-Intent/JSON to Claude Code native tool_calls) rewrites assistant messages between agentic tool-loop turns, breaking HyperAgent's conversation-prefix fingerprint and cold-starting the thread mid tool-loop. Adds Anthropic tool_use/tool_result flattening to extractMessageText() plus a new rootUserFingerprint()/root-key lookup tier in resolveHyperAgentThreadBinding()/storeHyperAgentThreadAfterTurn() so the thread stays sticky across the tool loop. Cohesive additions inside the existing single-file executor; not extractable without splitting the executor mid-request-flow. Covered by tests/unit/executor-hyperagent.test.ts (19/19, +5 new cases for tool_result/tool_use flattening + root-key stickiness). Pre-merge review flagged a cross-conversation root-key collision risk (tracked in the PR's own mandatory pre-merge checklist, not yet addressed) — unrelated to this file-size ratchet, tracked separately by /fix-prs.",
"_rebaseline_2026_08_02_9259_rolling_rpm": "PR #9259 (issue #8733) own growth: open-sse/services/rateLimitManager.ts baseline 1060->1167 (+107; final source 1153). The existing withRateLimit chokepoint now composes process-local rolling RPM leases with Bottleneck admission, releases pre-dispatch leases on queue timeout/abort/connection disable, preserves caller abort reasons, and wires 429/header state into the extracted rollingRpmGate.ts. The remaining growth is irreducible lifecycle wiring at the dispatch boundary plus the real watchdog test hooks needed to verify queued-wedge recovery; moving it further would obscure lease ownership and Bottleneck cleanup. Covered by the focused rate-limit manager/sliding-window suite (33/33); distributed multi-instance coordination remains explicitly out of scope.",
"_rebaseline_2026_07_24_8470_hyperagent_sticky_thread": "PR #8470 (artickc, fix/hyperagent-tool-loop-thread-sticky) own growth: open-sse/executors/hyperagent.ts 936->1025 (wc -l; check-file-size.mjs counts via split(\"\\n\").length so the gate sees 937->1026, +89, crosses the 1000 cap). Fixes a real bug where a reverse-conversion proxy (text-Intent/JSON to Claude Code native tool_calls) rewrites assistant messages between agentic tool-loop turns, breaking HyperAgent’s conversation-prefix fingerprint and cold-starting the thread mid tool-loop. Adds Anthropic tool_use/tool_result flattening to extractMessageText() plus a new rootUserFingerprint()/root-key lookup tier in resolveHyperAgentThreadBinding()/storeHyperAgentThreadAfterTurn() so the thread stays sticky across the tool loop. Cohesive additions inside the existing single-file executor; not extractable without splitting the executor mid-request-flow. Covered by tests/unit/executor-hyperagent.test.ts (19/19, +5 new cases for tool_result/tool_use flattening + root-key stickiness). Pre-merge review flagged a cross-conversation root-key collision risk (tracked in the PR’s own mandatory pre-merge checklist, not yet addressed) — unrelated to this file-size ratchet, tracked separately by /fix-prs.",
"_rebaseline_2026_07_25_8494_capability_filter_fail_closed": "PR #8494 (fix/capability-filters-fail-closed, #8488) own growth: open-sse/services/combo.ts 3640->3693 (+53) adds a fail-closed guard after filterTargetsByRequestCompatibility() — when every eligible target is excluded by request-capability filtering (vision/tools/etc) instead of quota/health, the combo now returns an explicit `capability_mismatch` 400 (describeCapabilityFilterExhaustion, imported from combo/comboStructure.ts) rather than silently falling through to a generic no-targets error, plus a `compatFilterFailOpen` escape hatch (combo config OR settings) mirrored at both the main/auto and round-robin call sites for symmetry. combo/comboStructure.ts (previously under cap, un-frozen) grows 794->918 (+124) — new home for describeCapabilityFilterExhaustion + providerSupportsEmulatedToolCalling (#5240 emulated tool-calling exemption so fail-closed does not regress prompt-emulation-only combos like all-chatgpt-web). Irreducible orchestration wiring at the existing filter chokepoint (same precedent as #7301's universal-cooldown-retry generalization). Companion test tests/unit/combo-routing-engine.test.ts 3409->3449 (+40, fail-closed/fail-open coverage across both call sites) also rebaselined. Covered by tests/unit/8488-capability-filter-fail-closed.test.ts (new) + 95/95 passing across both files. Structural shrink of combo.ts tracked in #3501.",
"_rebaseline_2026_07_25_8499_ts7_result_union_predicates": "PR #8499 (backryun, chore/ts7-types-executor-scattered) own growth: muse-spark-web.ts 1396->1405 (+9, irreducible). Under this workspace's `strictNullChecks: false`, the boolean-literal discriminant on `GraphqlResult` (`{ ok: true } | { ok: false; error: string }`) narrows the positive `.ok===true` branch but leaves `!result.ok` at the full union under TS7, making `.error` unreachable to the checker at the two call sites (warmup, mode-switch). Fixed by adding a single `isGraphqlFailure()` type-predicate helper (doc comment + 3-line body) reused at both call sites instead of duplicating the predicate inline — not extractable to a shared module without splitting a single-file executor's local narrowing helper out of its own file. Covered by the existing muse-spark-web executor test suite (no behavior change, pure narrowing fix).",
"_rebaseline_2026_07_22_8131_windowshide_cloudflared_spawn": "PR #8167 (Dingding-leo, fix/windows-hide-child-process, #8131) own growth: src/lib/cloudflaredTunnel.ts 934->935 (+1, irreducible call-site wiring — the single `windowsHide: true` option added to the existing cloudflared spawn() options object so no transient conhost.exe/cmd console window flashes open on Windows). Covered by the pre-merge-fix regression test tests/unit/windows-hide-child-process-spawns-8131.test.ts (added for the two additional spawn() sites the PR missed: ServiceSupervisor.ts, versionManager/processManager.ts) plus the windowsHide assertion added to tests/unit/services/installers/runNpm-shell-5379.test.ts (installers/utils.ts buildNpmExecOptions).",
@@ -544,7 +545,7 @@
"open-sse/services/claudeCodeCompatible.ts": 1202,
"open-sse/services/combo.ts": 3648,
"open-sse/services/compression/strategySelector.ts": 1060,
"open-sse/services/rateLimitManager.ts": 1105,
"open-sse/services/rateLimitManager.ts": 1167,
"open-sse/translator/response/openai-responses.ts": 1204,
"open-sse/utils/cursorAgentProtobuf.ts": 1505,
"open-sse/utils/stream.ts": 2889,

6
open-sse/services/providerDefaultRateLimit.ts
View File

@@ -11,8 +11,10 @@
* API (#6846 Phase 1). Every other provider still gets zero behavior change; the
* whole path is a no-op unless an entry (or a resolved override, see below) exists.
*
* Wired as a pre-schedule gate in `withRateLimit` (rateLimitManager.ts). Bottleneck
* still applies on top — this only adds a floor for header-less providers.
* Composed into the rolling lease gate in `withRateLimit` (rateLimitManager.ts).
* The exported acquire helpers remain available for provider-specific callers
* and tests, while the main request path acquires global and provider scopes
* atomically.
*/
import { SlidingWindowLimiter, type RateLimitWindow } from "./slidingWindowLimiter.ts";

408
open-sse/services/rateLimitManager.ts
View File

@@ -13,7 +13,12 @@ import { parseRetryAfterFromBody } from "./accountFallback.ts";
import { getAntigravityQuotaFamily } from "./antigravityQuotaFamily.ts";
import { getProviderCategory } from "../config/providerRegistry.ts";
import { getCodexRateLimitKey } from "../executors/codex.ts";
import { awaitProviderDefaultSlot, setProviderQuotaOverrides } from "./providerDefaultRateLimit.ts";
import {
getProviderDefaultRateLimit,
setProviderQuotaOverrides,
} from "./providerDefaultRateLimit.ts";
import { keyContainsConnection, RollingRpmGate } from "./rollingRpmGate.ts";
import { toNumber } from "@/shared/utils/numeric";
import {
DEFAULT_RESILIENCE_SETTINGS,
resolveResilienceSettings,
@@ -39,25 +44,36 @@ interface LearnedLimitEntry {
interface LimiterUpdateSettings {
maxConcurrent?: number | null;
minTime: number;
reservoir?: number | null;
reservoirRefreshAmount?: number | null;
reservoirRefreshInterval?: number | null;
}
type JsonRecord = Record<string, unknown>;
type QueueTimeoutReason = "local-queue" | "upstream-cooldown";
function toRecord(value: unknown): JsonRecord {
return value && typeof value === "object" && !Array.isArray(value) ? (value as JsonRecord) : {};
}
function toNumber(value: unknown, fallback = 0): number {
const parsed =
typeof value === "number"
? value
: typeof value === "string" && value.trim().length > 0
? Number(value)
: Number.NaN;
return Number.isFinite(parsed) ? parsed : fallback;
function createQueueTimeoutError(
provider: string,
model: string | null,
maxWaitMs: number,
reason: QueueTimeoutReason = "local-queue",
cause?: unknown
) {
const target = model ? `${provider}/${model}` : provider;
const message =
reason === "upstream-cooldown"
? `Request dropped after waiting ${maxWaitMs}ms for an upstream rate-limit cooldown for ${target}. ` +
`The provider cooldown outlasted OmniRoute's local wait budget; this is not local queue saturation.`
: `Request dropped after exceeding the local rate-limit queue budget maxWaitMs (${maxWaitMs}ms) for ` +
`${target} — this is OmniRoute's request queue ` +
`(resilienceSettings.requestQueue.maxWaitMs), not an upstream timeout. Raise it in ` +
`Settings → Resilience if this is queue saturation rather than a slow provider.`;
const queueErr = new Error(message, cause === undefined ? undefined : { cause }) as Error & {
code?: string;
};
queueErr.code = "RATE_LIMIT_QUEUE_TIMEOUT";
return queueErr;
}
function isNodeTestRunnerChild(): boolean {
@@ -101,16 +117,15 @@ let initialized = false;
let currentRequestQueueSettings: RequestQueueSettings = DEFAULT_RESILIENCE_SETTINGS.requestQueue;
// Watchdog: detect Bottleneck limiters that are wedged (queue has work, but no
// jobs are dispatched). When the reservoir/refresh state desyncs from reality,
// this catches it and force-resets so traffic isn't stuck forever.
// jobs are dispatched). RPM admission happens before Bottleneck, so a queued
// Bottleneck job with no active work is a concurrency scheduler failure.
const lastDispatchAt = new Map<string, number>();
let nextJobTraceId = 1;
let watchdogInterval: ReturnType<typeof setInterval> | null = null;
const WATCHDOG_INTERVAL_MS = 30_000;
// Threshold has to exceed any *legitimate* gap between dispatches:
// - default reservoirRefreshInterval is 60s
// - adaptive minTime can climb to ~60s for 1-RPM providers (see updateFromHeaders)
// 120s gives a 2× margin against both, while still catching the actual wedge
// case we observed (queue stalled for 3+ minutes with no progress).
// Threshold has to exceed any legitimate gap caused by adaptive minTime while
// still catching the actual wedge case we observed (queue stalled for 3+
// minutes with no progress).
const WEDGE_THRESHOLD_MS = 120_000;
/**
@@ -128,19 +143,10 @@ function isAutoEnableActive(settings: RequestQueueSettings): boolean {
return settings.autoEnableApiKeyProviders;
}
// Sentinels for "no rate limit" / effectively infinite capacity. The reservoir
// value uses Number.MAX_SAFE_INTEGER so the bucket can never realistically be
// exhausted; maxConcurrent uses a smaller-but-still-vast ceiling since
// Bottleneck tracks concurrent jobs in memory and an unbounded number would
// risk internal counter overflow under sustained pressure.
const EFFECTIVELY_INFINITE = Number.MAX_SAFE_INTEGER;
// Bottleneck handles concurrency and pacing. RPM is enforced by the rolling
// lease limiter above rather than by a fixed-window reservoir.
const EFFECTIVELY_INFINITE_CONCURRENCY = 1000;
// Resolve an RPM override. 0 or missing means "infinite" (no rate cap).
function resolveRpm(override: number | undefined | null): number {
return typeof override === "number" && override > 0 ? override : EFFECTIVELY_INFINITE;
}
// Resolve a minTime override. 0 or missing means "no minimum gap".
function resolveMinTime(override: number | undefined | null): number {
return typeof override === "number" && override > 0 ? override : 0;
@@ -152,15 +158,9 @@ function resolveMaxConcurrent(override: number | undefined | null): number {
}
function buildLimiterDefaults() {
// 0 or missing values mean "infinite" / no rate limit applies. This treats
// the global request-queue settings the same way per-connection overrides
// are interpreted (see resolveRpm / resolveMinTime / resolveMaxConcurrent).
return {
maxConcurrent: resolveMaxConcurrent(currentRequestQueueSettings.concurrentRequests),
minTime: resolveMinTime(currentRequestQueueSettings.minTimeBetweenRequestsMs),
reservoir: resolveRpm(currentRequestQueueSettings.requestsPerMinute),
reservoirRefreshAmount: resolveRpm(currentRequestQueueSettings.requestsPerMinute),
reservoirRefreshInterval: 60 * 1000,
};
}
@@ -271,12 +271,18 @@ function reconcileEnabledConnections(
function watchdogTick() {
const now = Date.now();
rpmGate.cleanupExpired(now);
// Clean up idle limiters that haven't been used recently
for (const [key, limiter] of Array.from(limiters)) {
const lastUsed = limiterLastUsed.get(key) ?? 0;
if (now - lastUsed > INACTIVE_LIMITER_MS) {
const counts = limiter.counts();
if (counts.QUEUED === 0 && counts.RUNNING === 0 && counts.EXECUTING === 0) {
if (
counts.RECEIVED === 0 &&
counts.QUEUED === 0 &&
counts.RUNNING === 0 &&
counts.EXECUTING === 0
) {
limiters.delete(key);
lastDispatchAt.delete(key);
limiterLastUsed.delete(key);
@@ -289,7 +295,10 @@ function watchdogTick() {
}
for (const [key, limiter] of Array.from(limiters)) {
const counts = limiter.counts();
if (counts.QUEUED === 0) continue;
// RECEIVED-only work is still active and must not be evicted. Once a job
// is stably queued, Bottleneck reports it in QUEUED with RECEIVED=0; that
// is the state the wedge detector is designed to recover.
if (counts.RECEIVED > 0 || counts.QUEUED === 0) continue;
if (counts.RUNNING > 0 || counts.EXECUTING > 0) continue;
const lastDispatch = lastDispatchAt.get(key);
// No heartbeat yet → seed it and skip this tick. Prevents false wedge
@@ -302,7 +311,7 @@ function watchdogTick() {
if (stalledMs < WEDGE_THRESHOLD_MS) continue;
warnRateLimit(
`🚨 [RATE-LIMIT] WEDGED: ${key} queued=${counts.QUEUED} running=0 executing=0 stalled=${stalledMs}ms — force-resetting`
`🚨 [RATE-LIMIT] WEDGED: ${key} received=${counts.RECEIVED} queued=${counts.QUEUED} running=0 executing=0 stalled=${stalledMs}ms — force-resetting`
);
// Live incident (log id 1784465227489-a2cbc0): disconnect() releases the
// heartbeat timer but does NOT reject the QUEUED jobs already sitting on
@@ -353,15 +362,38 @@ export function stopRateLimitWatchdog(): void {
watchdogInterval = null;
}
export function __installLimiterForTests(
provider: string,
connectionId: string,
limiter: Bottleneck,
model = null
): void {
const key = getLimiterKey(provider, connectionId, model);
limiters.set(key, limiter);
lastDispatchAt.set(key, Date.now());
limiterLastUsed.set(key, Date.now());
}
export function __runRateLimitWatchdogForTests(): void {
watchdogTick();
}
export function __getLimiterForTests(provider: string, connectionId: string, model = null) {
return getLimiter(provider, connectionId, model);
}
export function __setLastDispatchAtForTests(
provider: string,
connectionId: string,
model: string | null,
timestamp: number
): void {
lastDispatchAt.set(getLimiterKey(provider, connectionId, model), timestamp);
}
function evictWedgeLimiter(key: string, limiter: Bottleneck): void {
if (limiters.get(key) !== limiter) return;
limiters.delete(key);
lastDispatchAt.delete(key);
limiterLastUsed.delete(key);
trackAsyncOperation(limiter.disconnect());
trackAsyncOperation(
limiter.stop({ dropWaitingJobs: true, dropErrorMessage: "rate-limit-watchdog-wedge-reset" })
);
evictLimiterAndDropQueued(key, limiter, "rate-limit-watchdog-wedge-reset");
}
/**
@@ -422,13 +454,10 @@ export async function initializeRateLimits() {
// budget + concurrency cap (nvidia today). No-op for every provider without
// an entry in either providerQuotaOverrides or PROVIDER_DEFAULT_RATE_LIMITS.
setProviderQuotaOverrides(resilience.providerQuotaOverrides);
const { explicitCount, autoCount } = reconcileEnabledConnections(
connections as unknown[],
currentRequestQueueSettings
);
await updateAllLimiterSettings();
// Load per-connection rate limit overrides
// Load per-connection rate limit overrides before reconciliation can create
// any limiter. The RPM gate reads these overrides at admission time, and
// Bottleneck still needs the non-RPM connection settings immediately.
connectionRateLimitOverrides.clear();
for (const conn of connections as Array<Record<string, unknown>>) {
const overrides = conn.rateLimitOverrides;
@@ -437,6 +466,12 @@ export async function initializeRateLimits() {
}
}
const { explicitCount, autoCount } = reconcileEnabledConnections(
connections as unknown[],
currentRequestQueueSettings
);
updateAllLimiterSettings();
if (explicitCount > 0 || autoCount > 0) {
logRateLimit(
`🛡️ [RATE-LIMIT] Loaded ${explicitCount} explicit + ${autoCount} auto-enabled protection(s)`
@@ -474,22 +509,14 @@ export function enableRateLimitProtection(connectionId) {
*/
export function disableRateLimitProtection(connectionId) {
enabledConnections.delete(connectionId);
// Evict limiters for this connection from the cache. Do NOT call limiter.stop() —
// it permanently rejects future .schedule() calls with "This limiter has been stopped",
// and in-flight requests holding a reference to the old instance would fail with 502.
// Call disconnect() (not stop()) to release Bottleneck's internal heartbeat timer
// without permanently poisoning the instance for any remaining in-flight jobs.
// Eviction-only would leak the heartbeat timer until GC; disconnect() releases it
// synchronously so the runtime memory footprint stays flat under heavy connection churn.
// .stop() is reserved exclusively for SIGTERM/SIGINT shutdown (see shutdownLimiters).
// Drop queued jobs before evicting the limiter. Otherwise disconnect() leaves
// callers waiting on an instance that is no longer reachable from the cache.
for (const [key, limiter] of Array.from(limiters)) {
if (key.includes(connectionId)) {
limiters.delete(key);
lastDispatchAt.delete(key);
limiterLastUsed.delete(key);
trackAsyncOperation(limiter.disconnect());
if (keyContainsConnection(key, connectionId)) {
evictLimiterAndDropQueued(key, limiter, "rate-limit-connection-disabled");
}
}
rpmGate.clearConnection(connectionId);
}
/**
@@ -517,13 +544,11 @@ export function refreshConnectionRateLimits(connectionId, overrides) {
}
// Evict limiters referencing this connection so they get recreated on next use
for (const [key, limiter] of Array.from(limiters)) {
if (key.includes(connectionId)) {
limiters.delete(key);
lastDispatchAt.delete(key);
limiterLastUsed.delete(key);
trackAsyncOperation(limiter.disconnect());
if (keyContainsConnection(key, connectionId)) {
evictLimiterAndDropQueued(key, limiter, "rate-limit-settings-refresh");
}
}
rpmGate.clearConnection(connectionId);
}
/**
@@ -546,6 +571,15 @@ function getLimiterKey(provider, connectionId, model = null) {
return `${provider}:${connectionId}`;
}
const rpmGate = new RollingRpmGate({
getGlobalRpm: () => currentRequestQueueSettings.requestsPerMinute,
getProviderWindow: getProviderDefaultRateLimit,
getConnectionRpm: (connectionId) => connectionRateLimitOverrides.get(connectionId)?.rpm,
getLimiterKey,
createQueueTimeoutError: (provider, model, maxWaitMs, reason) =>
createQueueTimeoutError(provider, model, maxWaitMs, reason),
});
function getLimiter(provider, connectionId, model = null) {
const key = getLimiterKey(provider, connectionId, model);
@@ -554,23 +588,14 @@ function getLimiter(provider, connectionId, model = null) {
const overrides = connectionRateLimitOverrides.get(connectionId);
if (overrides) {
// 0 (or missing) means "no override — fall through to buildLimiterDefaults()".
// Without this guard, an rpm of 0 sets reservoir=0, which Bottleneck treats
// as "depleted" and blocks ALL requests indefinitely. Treating 0 as "use
// default" lets users effectively disable per-connection limits without
// globally raising the system default.
if (typeof overrides.maxConcurrent === "number" && overrides.maxConcurrent > 0) {
defaults.maxConcurrent = overrides.maxConcurrent;
}
if (typeof overrides.minTime === "number" && overrides.minTime > 0) {
defaults.minTime = overrides.minTime;
}
if (typeof overrides.rpm === "number" && overrides.rpm > 0) {
defaults.reservoir = overrides.rpm;
defaults.reservoirRefreshAmount = overrides.rpm;
defaults.reservoirRefreshInterval = 60 * 1000;
}
// TODO: TPM/TPD integration — requires a token-bucket vs request-bucket
// separation (Bottleneck's reservoir is request-count, not token-count).
// separation. RPM is handled by the rolling lease gate below.
// When added, treat 0/missing the same way: fall through to system default.
}
const limiter = new Bottleneck({
@@ -593,6 +618,15 @@ function getLimiter(provider, connectionId, model = null) {
return limiters.get(key);
}
function evictLimiterAndDropQueued(key: string, limiter: Bottleneck, reason: string): void {
if (limiters.get(key) === limiter) {
limiters.delete(key);
lastDispatchAt.delete(key);
limiterLastUsed.delete(key);
}
trackAsyncOperation(limiter.stop({ dropWaitingJobs: true, dropErrorMessage: reason }));
}
/**
* Acquire a rate limit slot before making a request.
* If rate limiting is disabled for this connection, returns immediately.
@@ -604,32 +638,7 @@ function getLimiter(provider, connectionId, model = null) {
* @param {AbortSignal} signal - Optional abort signal to cancel waiting
* @returns {Promise<unknown>} Result of fn()
*/
async function getQueueHealthSnapshot(key: string, limiter: Bottleneck) {
const counts = limiter.counts();
let reservoirRemaining: number | null = null;
try {
reservoirRemaining = await limiter.currentReservoir();
} catch {
// Snapshot logging must never affect request handling.
}
const lastDispatch = lastDispatchAt.get(key);
return {
queued: counts.QUEUED,
running: counts.RUNNING,
executing: counts.EXECUTING,
reservoirRemaining,
lastDispatchAgeMs: lastDispatch ? Date.now() - lastDispatch : null,
};
}
export async function withRateLimit(
provider,
connectionId,
model,
fn,
signal = null,
retryAfterWedge = true
) {
export async function withRateLimit(provider, connectionId, model, fn, signal = null) {
if (!enabledConnections.has(connectionId)) {
return fn();
}
@@ -642,18 +651,20 @@ export async function withRateLimit(
throw err;
}
// Proactive sliding-window fallback for header-less providers with a declared cap
// (Fase 8.2). No-op unless PROVIDER_DEFAULT_RATE_LIMITS has an entry for `provider`.
await awaitProviderDefaultSlot(
const maxWaitMs = currentRequestQueueSettings.maxWaitMs;
const queueStartedAt = Date.now();
const rpmLease = await rpmGate.acquire(
provider,
connectionId,
model,
signal,
currentRequestQueueSettings.maxWaitMs
maxWaitMs,
queueStartedAt
);
const limiter = getLimiter(provider, connectionId, model);
const maxWaitMs = currentRequestQueueSettings.maxWaitMs;
const scheduleOpts = maxWaitMs && maxWaitMs > 0 ? { expiration: maxWaitMs } : {};
const key = getLimiterKey(provider, connectionId, model);
const jobId = `${key}:job-${nextJobTraceId++}`;
const scheduleOpts = { id: jobId };
// Issue #6593: opt-in admission cap — fast-reject before Bottleneck's
// schedule() (and before any downstream compression/prompt work runs) when
@@ -664,30 +675,77 @@ export async function withRateLimit(
model ? `${provider}/${model}` : provider
);
if (admissionErr) {
rpmLease?.release();
logRateLimit(
`🚧 [RATE-LIMIT] ${getLimiterKey(provider, connectionId, model)} — queue full, rejecting fast (maxQueueDepth=${currentRequestQueueSettings.maxQueueDepth})`
);
throw admissionErr;
}
let dispatched = false;
let queueExpired = false;
let dispatchCancelled = false;
let queueTimer: ReturnType<typeof setTimeout> | undefined;
const remainingWaitMs =
maxWaitMs > 0 ? Math.max(1, maxWaitMs - (Date.now() - queueStartedAt)) : 0;
const queueTimeoutPromise =
remainingWaitMs > 0
? new Promise<never>((_, reject) => {
queueTimer = setTimeout(() => {
if (dispatched) return;
queueExpired = true;
logRateLimit(
`⏰ [RATE-LIMIT] ${key} — job exceeded ${Math.ceil(maxWaitMs / 1000)}s queue wait budget, dropping`
);
reject(new Error("rate-limit-queue-timeout"));
}, remainingWaitMs);
})
: null;
const scheduled = limiter.schedule(scheduleOpts, async () => {
if (queueExpired) {
throw createQueueTimeoutError(provider, model, maxWaitMs);
}
if (dispatchCancelled) {
const error = new Error("The operation was aborted before limiter dispatch");
error.name = "AbortError";
throw error;
}
if (signal?.aborted) {
const error = new Error("The operation was aborted before limiter dispatch");
error.name = "AbortError";
throw error;
}
dispatched = true;
if (queueTimer) clearTimeout(queueTimer);
return fn();
});
try {
if (signal) {
let abortListener: (() => void) | undefined;
const abortPromise = new Promise<never>((_, reject) => {
const onAbort = () => {
const reason = signal.reason;
// Preserve native Error reasons (including AbortController's
// read-only DOMException) instead of mutating or wrapping them.
// Reject before evicting the queued job so the caller observes its
// abort reason instead of Bottleneck's internal drop error.
if (reason instanceof Error) {
reject(reason);
return;
} else {
const err = new Error(
typeof reason === "string" ? reason : "The operation was aborted"
);
err.name = "AbortError";
if (reason !== undefined) {
(err as Error & { cause?: unknown }).cause = reason;
}
reject(err);
}
const err = new Error(typeof reason === "string" ? reason : "The operation was aborted");
err.name = "AbortError";
if (reason !== undefined) {
(err as Error & { cause?: unknown }).cause = reason;
if (!dispatched) {
dispatchCancelled = true;
if (queueTimer) clearTimeout(queueTimer);
// Leave the cancelled job in Bottleneck so queued peers are not dropped.
// Its scheduled callback will consume one queue turn and exit before fn().
}
reject(err);
};
if (signal.aborted) {
onAbort();
@@ -698,51 +756,30 @@ export async function withRateLimit(
});
try {
return await Promise.race([limiter.schedule(scheduleOpts, fn), abortPromise]);
const races: Promise<unknown>[] = [scheduled, abortPromise];
if (queueTimeoutPromise) races.push(queueTimeoutPromise);
return await Promise.race(races);
} finally {
if (abortListener) {
signal.removeEventListener("abort", abortListener);
}
}
} else {
return await limiter.schedule(scheduleOpts, fn);
return await (queueTimeoutPromise
? Promise.race([scheduled, queueTimeoutPromise])
: scheduled);
}
} catch (err) {
// Bottleneck's raw `This job timed out after <maxWaitMs> ms.` is
// indistinguishable from an upstream gateway timeout, so it leaks into 502
// bodies / call-log `last_error` and gets misdiagnosed as a provider outage
// (#4165). Rewrite it into a clear, OmniRoute-owned error (knob named,
// upstream disclaimed, original kept as `cause`, `code` for classification).
// If the limiter is idle with capacity after the expiry, the scheduler is wedged.
// Reset it and retry this never-dispatched function once on a fresh limiter.
if (err?.message?.includes("This job timed out")) {
const key = getLimiterKey(provider, connectionId, model);
const queueState = await getQueueHealthSnapshot(key, limiter);
logRateLimit(
`⏰ [RATE-LIMIT] ${key} — job expired after ${Math.ceil((maxWaitMs || 0) / 1000)}s in queue, dropping`
);
const limiterIsWedged =
retryAfterWedge &&
queueState.running === 0 &&
queueState.executing === 0 &&
typeof queueState.reservoirRemaining === "number" &&
queueState.reservoirRemaining > 0 &&
typeof queueState.lastDispatchAgeMs === "number" &&
queueState.lastDispatchAgeMs >= Math.max(1, maxWaitMs || 0);
if (limiterIsWedged) {
logRateLimit(`🔄 [RATE-LIMIT] ${key} — recovering idle limiter after queue expiry`);
evictWedgeLimiter(key, limiter);
return withRateLimit(provider, connectionId, model, fn, signal, false);
}
const queueErr = new Error(
`Request dropped after exceeding the local rate-limit queue budget maxWaitMs (${maxWaitMs}ms) for ` +
`${model ? `${provider}/${model}` : provider} — this is OmniRoute's request queue ` +
`(resilienceSettings.requestQueue.maxWaitMs), not an upstream timeout. Raise it in ` +
`Settings → Resilience if this is queue saturation rather than a slow provider.`,
if (queueTimer) clearTimeout(queueTimer);
if (!dispatched) rpmLease?.release();
if (err?.message === "rate-limit-upstream-429") {
const rateLimitErr = new Error(
`Request dropped while the ${provider} connection was under an upstream rate-limit cooldown`,
{ cause: err }
) as Error & { code?: string };
queueErr.code = "RATE_LIMIT_QUEUE_TIMEOUT";
throw queueErr;
) as Error & { code?: string; status?: number };
rateLimitErr.code = "RATE_LIMIT_UPSTREAM_429";
rateLimitErr.status = 429;
throw rateLimitErr;
}
// The watchdog's stop({ dropWaitingJobs: true }) wedge-recovery (above) rejects
// queued jobs with this exact message. Rewrite it the same way as the timeout
@@ -759,6 +796,9 @@ export async function withRateLimit(
wedgeErr.code = "RATE_LIMIT_QUEUE_WEDGED";
throw wedgeErr;
}
if (err?.message === "rate-limit-queue-timeout") {
throw createQueueTimeoutError(provider, model, maxWaitMs);
}
throw err;
}
}
@@ -802,20 +842,12 @@ export function updateFromHeaders(provider, connectionId, headers, status, model
`🚫 [RATE-LIMIT] ${provider}:${connectionId.slice(0, 8)} — 429 received, pausing for ${Math.ceil(retryAfterMs / 1000)}s, dropping ${counts.QUEUED} queued request(s)`
);
// Evict from the cache so follow-up learning from the same error body
// can materialize a fresh limiter immediately. Do NOT call limiter.stop() —
// it permanently rejects future .schedule() calls with "This limiter has been stopped".
// In-flight requests holding a reference to the evicted instance will fail (they
// were already going to fail — the 429 means the API rejected them), but future
// requests will get a fresh Bottleneck instance via getLimiter().
// Call disconnect() (not stop()) to release Bottleneck's internal heartbeat timer
// without permanently poisoning the instance for any remaining in-flight jobs.
// Without disconnect() here, every 429 leaks a heartbeat timer until GC reclaims
// the abandoned Bottleneck; under sustained quota pressure that is a real leak.
limiters.delete(limiterKey);
lastDispatchAt.delete(limiterKey);
limiterLastUsed.delete(limiterKey);
trackAsyncOperation(limiter.disconnect());
rpmGate.block(provider, connectionId, model, retryAfterMs);
// Evict from the cache before stopping so follow-up requests get a fresh
// instance. Stopping the unreachable instance rejects its queued jobs and
// releases its heartbeat without poisoning the replacement limiter.
evictLimiterAndDropQueued(limiterKey, limiter, "rate-limit-upstream-429");
return;
}
@@ -830,28 +862,31 @@ export function updateFromHeaders(provider, connectionId, headers, status, model
// Normal response — update limiter from headers
if (!isNaN(limit) && limit > 0) {
const resetMs = parseResetTime(resetStr) || 60000;
// Calculate optimal minTime from RPM limit
const minTime = Math.max(0, Math.floor(60000 / limit) - 10); // Small buffer
const updates: LimiterUpdateSettings = { minTime };
const resetMs = parseResetTime(resetStr) || 60000;
// If remaining is low (< 10% of limit), set reservoir to throttle immediately
// Keep adaptive pacing from response headers, but do not mutate an RPM
// reservoir. RPM admission is enforced by the rolling lease gate.
if (!isNaN(remaining)) {
if (remaining < limit * 0.1) {
updates.reservoir = remaining;
updates.reservoirRefreshAmount = limit;
updates.reservoirRefreshInterval = resetMs;
rpmGate.learnHeaderWindow(
provider,
connectionId,
model,
remaining,
resetMs,
Date.now() + resetMs
);
logRateLimit(
`⚠️ [RATE-LIMIT] ${provider}:${connectionId.slice(0, 8)} — ${remaining}/${limit} remaining, throttling`
);
} else if (remaining > limit * 0.5) {
// Plenty of headroom — relax the limiter
updates.minTime = 0;
updates.reservoir = null;
updates.reservoirRefreshAmount = null;
updates.reservoirRefreshInterval = null;
rpmGate.clearLearnedHeaderWindow(provider, connectionId, model);
}
}
@@ -984,6 +1019,8 @@ export async function __resetRateLimitManagerForTests() {
}
limiters.clear();
enabledConnections.clear();
connectionRateLimitOverrides.clear();
rpmGate.reset();
initialized = false;
lastDispatchAt.clear();
limiterLastUsed.clear();
@@ -1088,17 +1125,10 @@ export function updateFromResponseBody(provider, connectionId, responseBody, sta
const { retryAfterMs, reason } = parseRetryAfterFromBody(responseBody);
if (retryAfterMs && retryAfterMs > 0) {
const limiter = getLimiter(provider, connectionId, model);
getLimiter(provider, connectionId, model);
logRateLimit(
`🚫 [RATE-LIMIT] ${provider}:${connectionId.slice(0, 8)} — body-parsed retry: ${Math.ceil(retryAfterMs / 1000)}s (${reason})`
);
trackAsyncOperation(
applyLimiterSettings(limiter, {
reservoir: 0,
reservoirRefreshAmount: 60,
reservoirRefreshInterval: retryAfterMs,
})
);
rpmGate.block(provider, connectionId, model, retryAfterMs);
}
}

236
open-sse/services/rollingRpmGate.ts Normal file
View File

@@ -0,0 +1,236 @@
import {
SlidingWindowLimiter,
type RateLimitScope,
type RateLimitWindow,
type SlidingWindowLease,
} from "./slidingWindowLimiter.ts";
type GetLimiterKey = (provider: string, connectionId: string, model?: string | null) => string;
type QueueTimeoutReason = "local-queue" | "upstream-cooldown";
type QueueTimeoutErrorFactory = (
provider: string,
model: string | null,
maxWaitMs: number,
reason?: QueueTimeoutReason
) => Error;
export interface RollingRpmGateOptions {
getGlobalRpm: () => number | null | undefined;
getProviderWindow: (provider: string) => RateLimitWindow | undefined;
getConnectionRpm: (connectionId: string) => number | null | undefined;
getLimiterKey: GetLimiterKey;
createQueueTimeoutError: QueueTimeoutErrorFactory;
}
interface LearnedHeaderWindow {
window: RateLimitWindow;
expiresAt: number;
}
function createAbortError(signal: AbortSignal): Error {
const reason = signal.reason;
if (reason instanceof Error) return reason;
const error = new Error(typeof reason === "string" ? reason : "The operation was aborted");
error.name = "AbortError";
if (reason !== undefined) (error as Error & { cause?: unknown }).cause = reason;
return error;
}
function sleepOrAbort(ms: number, signal: AbortSignal | null): Promise<void> {
return new Promise((resolve, reject) => {
const timer = setTimeout(() => {
signal?.removeEventListener("abort", onAbort);
resolve();
}, ms);
const onAbort = () => {
clearTimeout(timer);
signal?.removeEventListener("abort", onAbort);
reject(createAbortError(signal as AbortSignal));
};
if (signal) {
if (signal.aborted) onAbort();
else signal.addEventListener("abort", onAbort, { once: true });
}
});
}
/**
* Process-local trailing-window RPM admission. Distributed deployments need a
* shared coordination store before this scope can be treated as cluster-wide.
*/
export function keyContainsConnection(key: string, connectionId: string): boolean {
const marker = `:${connectionId}`;
return key.endsWith(marker) || key.includes(`${marker}:`);
}
export class RollingRpmGate {
private readonly limiter = new SlidingWindowLimiter();
private readonly blockedUntil = new Map<string, number>();
private readonly learnedHeaderWindows = new Map<string, LearnedHeaderWindow>();
private readonly windowMs = 60_000;
constructor(private readonly options: RollingRpmGateOptions) {}
async acquire(
provider: string,
connectionId: string,
model: string | null,
signal: AbortSignal | null,
maxWaitMs: number,
startedAt: number
): Promise<SlidingWindowLease | null> {
const blockKey = this.options.getLimiterKey(provider, connectionId, model);
let scopes = this.getScopes(provider, connectionId, model);
if (scopes.length === 0 && (this.blockedUntil.get(blockKey) ?? 0) <= Date.now()) {
return null;
}
for (;;) {
if (signal?.aborted) throw createAbortError(signal);
scopes = this.getScopes(provider, connectionId, model);
if (scopes.length === 0 && (this.blockedUntil.get(blockKey) ?? 0) <= Date.now()) {
return null;
}
const now = Date.now();
const blockedUntil = this.blockedUntil.get(blockKey) ?? 0;
if (blockedUntil <= now) this.blockedUntil.delete(blockKey);
let scopeBlockedUntil = 0;
for (const scope of scopes) {
const scopeBlocked = this.blockedUntil.get(scope.key) ?? 0;
if (scopeBlocked > now) scopeBlockedUntil = Math.max(scopeBlockedUntil, scopeBlocked);
else if (scopeBlocked > 0) this.blockedUntil.delete(scope.key);
}
const forcedWaitMs = Math.max(0, blockedUntil - now, scopeBlockedUntil - now);
if (forcedWaitMs === 0) {
const result = this.limiter.tryAcquireMany(scopes);
if (result.allowed) return result.lease ?? null;
const retryAfterMs = Math.max(1, result.retryAfterMs);
const remainingMs = maxWaitMs > 0 ? maxWaitMs - (now - startedAt) : retryAfterMs;
if (maxWaitMs > 0 && remainingMs <= 0) {
throw this.options.createQueueTimeoutError(provider, model, maxWaitMs);
}
await sleepOrAbort(
Math.min(retryAfterMs, maxWaitMs > 0 ? remainingMs : retryAfterMs),
signal
);
continue;
}
const remainingMs = maxWaitMs > 0 ? maxWaitMs - (now - startedAt) : forcedWaitMs;
if (maxWaitMs > 0 && remainingMs <= 0) {
throw this.options.createQueueTimeoutError(provider, model, maxWaitMs, "upstream-cooldown");
}
await sleepOrAbort(
Math.min(forcedWaitMs, maxWaitMs > 0 ? remainingMs : forcedWaitMs),
signal
);
}
}
block(provider: string, connectionId: string, model: string | null, retryAfterMs: number): void {
if (retryAfterMs > 0) {
const key = this.options.getLimiterKey(provider, connectionId, model);
this.blockedUntil.set(key, Date.now() + retryAfterMs);
}
}
learnHeaderWindow(
provider: string,
connectionId: string,
model: string | null,
requests: number,
windowMs: number,
expiresAt: number
): void {
const key = `header:${this.options.getLimiterKey(provider, connectionId, model)}`;
if (requests <= 0) {
this.learnedHeaderWindows.delete(key);
this.blockedUntil.set(key, expiresAt);
return;
}
this.blockedUntil.delete(key);
this.learnedHeaderWindows.set(key, {
window: { requests, windowMs },
expiresAt,
});
}
clearLearnedHeaderWindow(provider: string, connectionId: string, model: string | null): void {
const key = `header:${this.options.getLimiterKey(provider, connectionId, model)}`;
this.learnedHeaderWindows.delete(key);
this.blockedUntil.delete(key);
}
clearConnection(connectionId: string): void {
for (const key of this.blockedUntil.keys()) {
if (keyContainsConnection(key, connectionId)) this.blockedUntil.delete(key);
}
for (const key of this.learnedHeaderWindows.keys()) {
if (keyContainsConnection(key, connectionId)) this.learnedHeaderWindows.delete(key);
}
}
reset(): void {
this.limiter.reset();
this.blockedUntil.clear();
this.learnedHeaderWindows.clear();
}
cleanupExpired(now = Date.now()): void {
for (const [key, expiresAt] of this.blockedUntil) {
if (expiresAt <= now) this.blockedUntil.delete(key);
}
for (const [key, window] of this.learnedHeaderWindows) {
if (window.expiresAt <= now) this.learnedHeaderWindows.delete(key);
}
}
private getScopes(
provider: string,
connectionId: string,
model: string | null
): RateLimitScope[] {
const scopes: RateLimitScope[] = [];
const globalRpm = this.options.getGlobalRpm();
if (typeof globalRpm === "number" && globalRpm > 0) {
scopes.push({
key: "global",
window: { requests: globalRpm, windowMs: this.windowMs },
});
}
const providerWindow = this.options.getProviderWindow(provider);
if (providerWindow) scopes.push({ key: `provider:${provider}`, window: providerWindow });
const connectionRpm = this.options.getConnectionRpm(connectionId);
if (typeof connectionRpm === "number" && connectionRpm > 0) {
scopes.push({
key: `provider-account:${provider}:${connectionId}`,
window: { requests: connectionRpm, windowMs: this.windowMs },
});
}
const headerKey = `header:${this.options.getLimiterKey(provider, connectionId, model)}`;
const headerBlockedUntil = this.blockedUntil.get(headerKey) ?? 0;
const headerRemainingMs = headerBlockedUntil - Date.now();
if (headerRemainingMs > 0) {
scopes.push({
key: headerKey,
window: { requests: 1, windowMs: headerRemainingMs },
});
return scopes;
}
if (headerBlockedUntil > 0) this.blockedUntil.delete(headerKey);
const headerWindow = this.learnedHeaderWindows.get(headerKey);
if (headerWindow) {
if (headerWindow.expiresAt > Date.now()) {
scopes.push({ key: headerKey, window: headerWindow.window });
} else {
this.learnedHeaderWindows.delete(headerKey);
}
}
return scopes;
}
}

94
open-sse/services/slidingWindowLimiter.ts
View File

@@ -27,13 +27,33 @@ export interface AcquireResult {
retryAfterMs: number;
}
export interface RateLimitScope {
key: string;
window: RateLimitWindow;
}
export interface SlidingWindowLease {
/** Release a lease that was acquired but never dispatched upstream. */
release(): void;
}
export interface MultiAcquireResult extends AcquireResult {
lease?: SlidingWindowLease;
}
interface Hit {
id: number;
timestamp: number;
}
// Hard ceiling on distinct keys tracked, so a pathological key space (e.g. a
// per-request id leaking into the key) can never grow the map without bound.
const MAX_KEYS = 5000;
export class SlidingWindowLimiter {
private readonly hits = new Map<string, number[]>();
private readonly hits = new Map<string, Hit[]>();
private readonly now: () => number;
private nextHitId = 1;
constructor(opts: { now?: () => number } = {}) {
this.now = opts.now ?? Date.now;
@@ -45,26 +65,60 @@ export class SlidingWindowLimiter {
* (without recording) when the trailing window is saturated.
*/
tryAcquire(key: string, window: RateLimitWindow): AcquireResult {
const { requests, windowMs } = window;
// A non-positive cap or window means "no limit configured" → always allow.
if (!(requests > 0) || !(windowMs > 0)) return { allowed: true, retryAfterMs: 0 };
const result = this.tryAcquireMany([{ key, window }]);
return { allowed: result.allowed, retryAfterMs: result.retryAfterMs };
}
/**
* Acquire all supplied scopes atomically. No scope is recorded unless every
* configured scope has capacity, preventing a global lease from being held
* while a narrower provider/account lease is unavailable.
*/
tryAcquireMany(scopes: readonly RateLimitScope[]): MultiAcquireResult {
const activeScopes = scopes.filter(({ window }) => window.requests > 0 && window.windowMs > 0);
if (activeScopes.length === 0) return { allowed: true, retryAfterMs: 0 };
const now = this.now();
const cutoff = now - windowMs;
const previous = this.hits.get(key);
// Drop timestamps that have aged out of the trailing window.
const live = previous ? previous.filter((ts) => ts > cutoff) : [];
const prepared = activeScopes.map((scope) => {
const cutoff = now - scope.window.windowMs;
const previous = this.hits.get(scope.key);
const live = previous ? previous.filter((hit) => hit.timestamp > cutoff) : [];
const retryAfterMs =
live.length >= scope.window.requests
? Math.max(0, live[0].timestamp + scope.window.windowMs - now)
: 0;
return { scope, live, retryAfterMs };
});
if (live.length >= requests) {
// The oldest in-window hit is the first to expire and free a slot.
const retryAfterMs = Math.max(0, live[0] + windowMs - now);
this.hits.set(key, live); // persist the pruned list; do NOT record a blocked attempt
return { allowed: false, retryAfterMs };
}
const retryAfterMs = prepared.reduce((max, entry) => Math.max(max, entry.retryAfterMs), 0);
for (const entry of prepared) this.set(entry.scope.key, entry.live);
if (retryAfterMs > 0) return { allowed: false, retryAfterMs };
live.push(now);
this.set(key, live);
return { allowed: true, retryAfterMs: 0 };
const entries = prepared.map((entry) => {
const hit = { id: this.nextHitId++, timestamp: now };
entry.live.push(hit);
this.set(entry.scope.key, entry.live);
return { key: entry.scope.key, id: hit.id };
});
let released = false;
return {
allowed: true,
retryAfterMs: 0,
lease: {
release: () => {
if (released) return;
released = true;
for (const entry of entries) {
const live = this.hits.get(entry.key);
if (!live) continue;
const remaining = live.filter((hit) => hit.id !== entry.id);
if (remaining.length > 0) this.hits.set(entry.key, remaining);
else this.hits.delete(entry.key);
}
},
},
};
}
/** Clear history for one key, or all keys when called with no argument. */
@@ -73,7 +127,11 @@ export class SlidingWindowLimiter {
else this.hits.delete(key);
}
private set(key: string, live: number[]): void {
private set(key: string, live: Hit[]): void {
if (live.length === 0) {
this.hits.delete(key);
return;
}
if (!this.hits.has(key) && this.hits.size >= MAX_KEYS) {
// Evict the least-recently-inserted key (Map preserves insertion order).
const oldest = this.hits.keys().next().value;

7
tests/unit/chat-rate-limit-body-lock.test.ts
View File

@@ -32,8 +32,8 @@ test.after(async () => {
test("handleChat applies body-derived retry-after to the runtime limiter", async () => {
const connection = await seedConnection("openai", { apiKey: "sk-openai-body-retry" });
globalThis.fetch = async () =>
new Response(
globalThis.fetch = async () => {
return new Response(
JSON.stringify({
error: {
message: "Rate limit exceeded. Please retry after 20s.",
@@ -44,6 +44,7 @@ test("handleChat applies body-derived retry-after to the runtime limiter", async
headers: { "Content-Type": "application/json" },
}
);
};
const response = await handleChat(
buildRequest({
@@ -65,7 +66,7 @@ test("handleChat applies body-derived retry-after to the runtime limiter", async
"gpt-4.1"
);
assert.ok(limiterState, "expected limiter state to exist for the active connection");
assert.equal(limiterState.reservoir, 0, "body-derived retry-after should drain the limiter");
assert.equal(limiterState.reservoir, null, "RPM is enforced by the rolling lease gate");
});
test("handleChat tolerates non-JSON rate-limit bodies without breaking fallback flow", async () => {

446
tests/unit/rate-limit-manager.test.ts
View File

@@ -1,5 +1,6 @@
import test from "node:test";
import assert from "node:assert/strict";
import Bottleneck from "bottleneck";
import fs from "node:fs";
import os from "node:os";
import path from "node:path";
@@ -12,7 +13,6 @@ const providersDb = await import("../../src/lib/db/providers.ts");
const resilienceSettings = await import("../../src/lib/resilience/settings.ts");
const rateLimitManager = await import("../../open-sse/services/rateLimitManager.ts");
const accountFallback = await import("../../open-sse/services/accountFallback.ts");
const Bottleneck = (await import("bottleneck")).default;
function wait(ms) {
return new Promise((resolve) => setTimeout(resolve, ms));
@@ -60,43 +60,296 @@ test("rate limit manager bypasses disabled connections and exposes inactive stat
assert.deepEqual(rateLimitManager.getAllRateLimitStatus(), {});
});
test("idle-capacity queue expiry resets the limiter and retries once", async () => {
test("queue expiry does not invoke the provider after a late dispatch", async () => {
await rateLimitManager.applyRequestQueueSettings({
...resilienceSettings.DEFAULT_RESILIENCE_SETTINGS.requestQueue,
autoEnableApiKeyProviders: false,
maxWaitMs: 20,
maxWaitMs: 100,
requestsPerMinute: 0,
concurrentRequests: 1,
minTimeBetweenRequestsMs: 0,
maxQueueDepth: 0,
});
const originalSchedule = Bottleneck.prototype.schedule;
let attempts = 0;
Bottleneck.prototype.schedule = function (...args) {
attempts++;
if (attempts === 1) {
return new Promise((_, reject) => {
setTimeout(() => reject(new Error("This job timed out after 20 ms.")), 30);
});
rateLimitManager.enableRateLimitProtection("queue-expiry-conn");
let resolveFirstStarted: () => void = () => undefined;
const firstStarted = new Promise<void>((resolve) => {
resolveFirstStarted = resolve;
});
const first = rateLimitManager.withRateLimit(
"openai",
"queue-expiry-conn",
"gpt-4o",
async () => {
resolveFirstStarted();
await wait(300);
return "first";
}
return originalSchedule.apply(this, args);
};
);
await firstStarted;
try {
rateLimitManager.enableRateLimitProtection("idle-capacity-conn");
const result = await rateLimitManager.withRateLimit(
"openai",
"idle-capacity-conn",
"gpt-4o",
async () => "recovered"
);
let secondCalls = 0;
await assert.rejects(
rateLimitManager.withRateLimit("openai", "queue-expiry-conn", "gpt-4o", async () => {
secondCalls++;
return "late";
}),
(error: { code?: string }) => error.code === "RATE_LIMIT_QUEUE_TIMEOUT"
);
assert.equal(result, "recovered");
assert.equal(attempts, 2, "the expired job should be retried once on a fresh limiter");
} finally {
Bottleneck.prototype.schedule = originalSchedule;
await first;
await wait(50);
assert.equal(secondCalls, 0, "a queue-expired job must not invoke the provider later");
});
test("queue expiry does not drop other queued jobs", async () => {
await rateLimitManager.applyRequestQueueSettings({
...resilienceSettings.DEFAULT_RESILIENCE_SETTINGS.requestQueue,
autoEnableApiKeyProviders: false,
maxWaitMs: 500,
requestsPerMinute: 0,
concurrentRequests: 1,
minTimeBetweenRequestsMs: 0,
maxQueueDepth: 0,
});
rateLimitManager.enableRateLimitProtection("queue-peer-conn");
let resolveFirstExecuting: () => void = () => undefined;
const firstExecuting = new Promise<void>((resolve) => {
resolveFirstExecuting = resolve;
});
let releaseFirst: () => void = () => undefined;
const first = rateLimitManager.withRateLimit("openai", "queue-peer-conn", null, async () => {
resolveFirstExecuting();
await new Promise<void>((resolve) => {
releaseFirst = resolve;
});
return "first";
});
await firstExecuting;
const second = rateLimitManager.withRateLimit(
"openai",
"queue-peer-conn",
null,
async () => "expired"
);
await wait(400);
let thirdCalls = 0;
const third = rateLimitManager.withRateLimit("openai", "queue-peer-conn", null, async () => {
thirdCalls++;
return "third";
});
await assert.rejects(
second,
(error: { code?: string }) => error.code === "RATE_LIMIT_QUEUE_TIMEOUT"
);
releaseFirst();
await Promise.all([first, third]);
assert.equal(thirdCalls, 1, "a peer queued job must survive another job's expiry");
});
test("global RPM lease is shared across enabled provider connections", async () => {
await rateLimitManager.applyRequestQueueSettings({
...resilienceSettings.DEFAULT_RESILIENCE_SETTINGS.requestQueue,
autoEnableApiKeyProviders: false,
maxWaitMs: 1000,
requestsPerMinute: 2,
concurrentRequests: 10,
minTimeBetweenRequestsMs: 0,
maxQueueDepth: 0,
});
rateLimitManager.enableRateLimitProtection("global-rpm-a");
rateLimitManager.enableRateLimitProtection("global-rpm-b");
let calls = 0;
await rateLimitManager.withRateLimit("openai", "global-rpm-a", null, async () => {
calls++;
});
await rateLimitManager.withRateLimit("anthropic", "global-rpm-b", null, async () => {
calls++;
});
await assert.rejects(
rateLimitManager.withRateLimit("openai", "global-rpm-a", null, async () => {
calls++;
}),
(error: { code?: string }) => error.code === "RATE_LIMIT_QUEUE_TIMEOUT"
);
assert.equal(calls, 2, "the global lease blocks the third dispatch across providers");
});
test("provider/account RPM lease failure does not consume the global lease", async () => {
await rateLimitManager.applyRequestQueueSettings({
...resilienceSettings.DEFAULT_RESILIENCE_SETTINGS.requestQueue,
autoEnableApiKeyProviders: false,
maxWaitMs: 1000,
requestsPerMinute: 2,
concurrentRequests: 10,
minTimeBetweenRequestsMs: 0,
maxQueueDepth: 0,
});
rateLimitManager.enableRateLimitProtection("provider-rpm-a");
rateLimitManager.enableRateLimitProtection("provider-rpm-b");
rateLimitManager.refreshConnectionRateLimits("provider-rpm-a", { rpm: 1 });
let calls = 0;
await rateLimitManager.withRateLimit("openai", "provider-rpm-a", null, async () => {
calls++;
});
await assert.rejects(
rateLimitManager.withRateLimit("openai", "provider-rpm-a", null, async () => {
calls++;
}),
(error: { code?: string }) => error.code === "RATE_LIMIT_QUEUE_TIMEOUT"
);
await rateLimitManager.withRateLimit("anthropic", "provider-rpm-b", null, async () => {
calls++;
});
assert.equal(calls, 2, "the failed provider lease did not consume the second global lease");
});
test("aborted queued work releases its pre-dispatch RPM lease", async () => {
await rateLimitManager.applyRequestQueueSettings({
...resilienceSettings.DEFAULT_RESILIENCE_SETTINGS.requestQueue,
autoEnableApiKeyProviders: false,
maxWaitMs: 1000,
requestsPerMinute: 2,
concurrentRequests: 1,
minTimeBetweenRequestsMs: 0,
maxQueueDepth: 0,
});
rateLimitManager.enableRateLimitProtection("abort-lease-conn");
rateLimitManager.enableRateLimitProtection("abort-lease-other");
let resolveFirstExecuting: () => void = () => undefined;
const firstExecuting = new Promise<void>((resolve) => {
resolveFirstExecuting = resolve;
});
let releaseFirst: () => void = () => undefined;
const firstStarted = new Promise<void>((resolve) => {
releaseFirst = resolve;
});
const first = rateLimitManager.withRateLimit("openai", "abort-lease-conn", null, async () => {
resolveFirstExecuting();
await firstStarted;
return "first";
});
await firstExecuting;
const controller = new AbortController();
let abortedCalls = 0;
const queued = rateLimitManager.withRateLimit(
"openai",
"abort-lease-conn",
null,
async () => {
abortedCalls++;
return "should-not-dispatch";
},
controller.signal
);
await wait(20);
controller.abort();
await assert.rejects(queued, (error: { name?: string }) => error.name === "AbortError");
let thirdCalls = 0;
await rateLimitManager.withRateLimit("anthropic", "abort-lease-other", null, async () => {
thirdCalls++;
});
releaseFirst();
await first;
assert.equal(abortedCalls, 0, "aborted queued work must not invoke the provider");
assert.equal(thirdCalls, 1, "aborted work must return its unused global lease");
});
test("aborting one queued request does not drop queued peers", async () => {
await rateLimitManager.applyRequestQueueSettings({
...resilienceSettings.DEFAULT_RESILIENCE_SETTINGS.requestQueue,
autoEnableApiKeyProviders: false,
maxWaitMs: 1000,
requestsPerMinute: 0,
concurrentRequests: 1,
minTimeBetweenRequestsMs: 0,
maxQueueDepth: 0,
});
const connectionId = "abort-peer-conn";
rateLimitManager.enableRateLimitProtection(connectionId);
let resolveFirstExecuting: () => void = () => undefined;
const firstExecuting = new Promise<void>((resolve) => {
resolveFirstExecuting = resolve;
});
let releaseFirst: () => void = () => undefined;
const first = rateLimitManager.withRateLimit("test-provider", connectionId, null, async () => {
resolveFirstExecuting();
await new Promise<void>((resolve) => {
releaseFirst = resolve;
});
});
await firstExecuting;
const limiter = rateLimitManager.__getLimiterForTests("test-provider", connectionId);
const controller = new AbortController();
let abortedCalls = 0;
const aborted = rateLimitManager.withRateLimit(
"test-provider",
connectionId,
null,
async () => {
abortedCalls++;
},
controller.signal
);
let peerCalls = 0;
const peer = rateLimitManager.withRateLimit("test-provider", connectionId, null, async () => {
peerCalls++;
});
for (let attempt = 0; attempt < 200 && limiter.counts().QUEUED < 2; attempt++) {
await wait(5);
}
assert.ok(limiter.counts().QUEUED >= 2, "both queued requests must be present before abort");
controller.abort();
await assert.rejects(aborted, (error: { name?: string }) => error.name === "AbortError");
releaseFirst();
await Promise.all([first, peer]);
assert.equal(abortedCalls, 0, "aborted queued work must not invoke the provider");
assert.equal(peerCalls, 1);
});
test("dispatched provider failures retain their RPM lease", async () => {
await rateLimitManager.applyRequestQueueSettings({
...resilienceSettings.DEFAULT_RESILIENCE_SETTINGS.requestQueue,
autoEnableApiKeyProviders: false,
maxWaitMs: 500,
requestsPerMinute: 1,
concurrentRequests: 10,
minTimeBetweenRequestsMs: 0,
maxQueueDepth: 0,
});
rateLimitManager.enableRateLimitProtection("failed-dispatch-a");
rateLimitManager.enableRateLimitProtection("failed-dispatch-b");
await assert.rejects(
rateLimitManager.withRateLimit("openai", "failed-dispatch-a", null, async () => {
throw new Error("upstream failure");
}),
/upstream failure/
);
let secondCalls = 0;
await assert.rejects(
rateLimitManager.withRateLimit("anthropic", "failed-dispatch-b", null, async () => {
secondCalls++;
}),
(error: { code?: string }) => error.code === "RATE_LIMIT_QUEUE_TIMEOUT"
);
assert.equal(secondCalls, 0, "a dispatched failure still counts against the RPM window");
});
test("withRateLimit forwards AbortController DOMException without mutating it", async () => {
@@ -209,6 +462,130 @@ test("rate limit manager handles 429 limiter teardown and disable cleanup", asyn
assert.equal(rateLimitManager.getRateLimitStatus("gemini", "conn-disable").active, false);
});
test("rate limit manager blocks admission after an upstream 429 retry hint", async () => {
await rateLimitManager.applyRequestQueueSettings({
concurrentRequests: 1,
requestsPerMinute: 0,
maxWaitMs: 100,
});
rateLimitManager.enableRateLimitProtection("conn-429-block");
rateLimitManager.updateFromHeaders(
"openai",
"conn-429-block",
{ "retry-after": "1s" },
429,
"gpt-4o"
);
let providerCalls = 0;
await assert.rejects(
rateLimitManager.withRateLimit("openai", "conn-429-block", "gpt-4o", async () => {
providerCalls++;
}),
(error: unknown) => {
const code = error && typeof error === "object" && "code" in error ? error.code : undefined;
assert.equal(code, "RATE_LIMIT_QUEUE_TIMEOUT");
assert.match(String((error as Error).message), /upstream rate-limit cooldown/);
return true;
}
);
assert.equal(providerCalls, 0);
});
test("rate limit manager blocks a zero-remaining header window until reset", async () => {
await rateLimitManager.applyRequestQueueSettings({
concurrentRequests: 1,
requestsPerMinute: 0,
maxWaitMs: 100,
});
rateLimitManager.enableRateLimitProtection("conn-zero-remaining");
rateLimitManager.updateFromHeaders(
"openai",
"conn-zero-remaining",
{
"x-ratelimit-limit-requests": "10",
"x-ratelimit-remaining-requests": "0",
"x-ratelimit-reset-requests": "1s",
},
200
);
let providerCalls = 0;
await assert.rejects(
rateLimitManager.withRateLimit("openai", "conn-zero-remaining", null, async () => {
providerCalls++;
}),
(error: unknown) => {
const code = error && typeof error === "object" && "code" in error ? error.code : undefined;
assert.equal(code, "RATE_LIMIT_QUEUE_TIMEOUT");
assert.match(String((error as Error).message), /upstream rate-limit cooldown/);
return true;
}
);
assert.equal(providerCalls, 0);
});
test("rate limit manager keeps learned header windows model-scoped where limiters are model-scoped", async () => {
await rateLimitManager.applyRequestQueueSettings({
concurrentRequests: 1,
requestsPerMinute: 0,
maxWaitMs: 100,
});
rateLimitManager.enableRateLimitProtection("conn-model-header");
rateLimitManager.updateFromHeaders(
"github",
"conn-model-header",
{
"x-ratelimit-limit-requests": "10",
"x-ratelimit-remaining-requests": "0",
"x-ratelimit-reset-requests": "1s",
},
200,
"model-a"
);
let providerCalls = 0;
await rateLimitManager.withRateLimit("github", "conn-model-header", "model-b", async () => {
providerCalls++;
});
assert.equal(providerCalls, 1);
});
test("rate limit watchdog resets a queued limiter with received work", async () => {
await rateLimitManager.applyRequestQueueSettings({
concurrentRequests: 1,
requestsPerMinute: 0,
maxWaitMs: 5_000,
});
rateLimitManager.enableRateLimitProtection("conn-wedge");
const limiter = new Bottleneck({ reservoir: 0, id: "test-provider:conn-wedge" });
rateLimitManager.__installLimiterForTests("test-provider", "conn-wedge", limiter);
let providerCalls = 0;
const pending = rateLimitManager.withRateLimit("test-provider", "conn-wedge", null, async () => {
providerCalls++;
});
await wait(100);
const counts = limiter.counts();
assert.equal(counts.RECEIVED, 0);
assert.ok(counts.QUEUED > 0);
rateLimitManager.__setLastDispatchAtForTests(
"test-provider",
"conn-wedge",
null,
Date.now() - 120_001
);
rateLimitManager.__runRateLimitWatchdogForTests();
await assert.rejects(pending, (error: unknown) => {
const code = error && typeof error === "object" && "code" in error ? error.code : undefined;
assert.equal(code, "RATE_LIMIT_QUEUE_WEDGED");
return true;
});
assert.equal(providerCalls, 0);
});
test("rate limit manager uses model-scoped limiter keys for GitHub Copilot (#1624)", async () => {
rateLimitManager.enableRateLimitProtection("conn-github");
rateLimitManager.updateFromHeaders(
@@ -240,6 +617,11 @@ test("rate limit manager uses model-scoped limiter keys for GitHub Copilot (#162
});
test("rate limit manager parses retry hints from response bodies and locks models", async () => {
await rateLimitManager.applyRequestQueueSettings({
...resilienceSettings.DEFAULT_RESILIENCE_SETTINGS.requestQueue,
maxWaitMs: 100,
requestsPerMinute: 0,
});
rateLimitManager.enableRateLimitProtection("conn-body");
rateLimitManager.updateFromResponseBody(
"openai",
@@ -263,6 +645,20 @@ test("rate limit manager parses retry hints from response bodies and locks model
assert.equal(limiterState?.key, "openai:conn-body");
assert.equal(rateLimitManager.getRateLimitStatus("openai", "conn-body").active, true);
let providerCalls = 0;
await assert.rejects(
rateLimitManager.withRateLimit("openai", "conn-body", "gpt-4o", async () => {
providerCalls++;
}),
(error: unknown) => {
const code = error && typeof error === "object" && "code" in error ? error.code : undefined;
assert.equal(code, "RATE_LIMIT_QUEUE_TIMEOUT");
assert.match(String((error as Error).message), /upstream rate-limit cooldown/);
return true;
}
);
assert.equal(providerCalls, 0);
rateLimitManager.updateFromResponseBody(
"openai",
"conn-body",

97
tests/unit/rate-limit-queue-timeout-message-4165.test.ts
View File

@@ -1,19 +1,13 @@
/**
* #4165 — surface a clear error when the request-queue (Bottleneck) drops a job.
*
* OmniRoute schedules every rate-limited request through Bottleneck with
* `{ expiration: requestQueue.maxWaitMs }` (open-sse/services/rateLimitManager.ts).
* When a job exceeds that budget Bottleneck throws the raw message
* `"This job timed out after <N> ms."` — which is indistinguishable from an
* upstream gateway timeout. In #4165 an operator spent ~3h misdiagnosing local
* queue saturation as a provider outage because the 502 body / call-log
* `last_error` carried that upstream-looking string across many providers.
* Queue waiting is bounded by a separate timer. Bottleneck's job expiration is
* intentionally not used because it measures the entire scheduled lifetime and
* would kill an already-dispatched provider call that is making progress.
*
* The fix rewrites that specific Bottleneck error into a clear, OmniRoute-owned
* message that names the knob (`resilienceSettings.requestQueue.maxWaitMs`) and
* explicitly says it is NOT an upstream timeout, while preserving the original
* error as `.cause` and tagging `.code = "RATE_LIMIT_QUEUE_TIMEOUT"` so callers
* can classify it. Behavior is unchanged: the job is still dropped.
* The queue-only timer still rewrites pre-dispatch expiry into a clear,
* OmniRoute-owned error that names the knob (`resilienceSettings.requestQueue.maxWaitMs`)
* and explicitly says it is NOT an upstream timeout.
*/
import test from "node:test";
import assert from "node:assert/strict";
@@ -41,8 +35,7 @@ test.after(() => {
fs.rmSync(TEST_DATA_DIR, { recursive: true, force: true });
});
// Drive a real Bottleneck `expiration` failure: a tiny maxWaitMs and a job that
// runs longer than it.
// A dispatched provider call may run longer than maxWaitMs without being killed.
async function triggerQueueTimeout() {
await rateLimitManager.applyRequestQueueSettings({
...resilienceSettings.DEFAULT_RESILIENCE_SETTINGS.requestQueue,
@@ -60,36 +53,60 @@ async function triggerQueueTimeout() {
});
}
test("#4165 queue-timeout surfaces a clear OmniRoute error, not the raw upstream-looking string", async () => {
let caught: (Error & { code?: string; cause?: { message?: string } }) | undefined;
async function triggerQueuedTimeout() {
await rateLimitManager.applyRequestQueueSettings({
...resilienceSettings.DEFAULT_RESILIENCE_SETTINGS.requestQueue,
autoEnableApiKeyProviders: false,
concurrentRequests: 1,
requestsPerMinute: 0,
minTimeBetweenRequestsMs: 0,
maxWaitMs: 40,
});
const connectionId = "conn-queued-timeout";
rateLimitManager.enableRateLimitProtection(connectionId);
let resolveFirstExecuting: () => void = () => undefined;
const firstExecuting = new Promise<void>((resolve) => {
resolveFirstExecuting = resolve;
});
let releaseFirst: () => void = () => undefined;
const first = rateLimitManager.withRateLimit("test-provider", connectionId, null, async () => {
resolveFirstExecuting();
await new Promise<void>((resolve) => {
releaseFirst = resolve;
});
});
await firstExecuting;
let caught: unknown;
try {
await triggerQueueTimeout();
assert.fail("expected the queued job to be dropped");
} catch (err) {
caught = err as Error & { code?: string; cause?: { message?: string } };
await rateLimitManager.withRateLimit(
"test-provider",
connectionId,
null,
async () => "should-not-dispatch"
);
assert.fail("expected the queued job to expire");
} catch (error) {
caught = error;
} finally {
releaseFirst();
await first;
}
assert.ok(caught, "an error should have been thrown");
return caught;
}
// Tagged so combo / callers can classify it as a local queue drop.
assert.equal(caught.code, "RATE_LIMIT_QUEUE_TIMEOUT", "error must carry the queue-timeout code");
test("#4165 a dispatched provider call is not killed by the queue budget", async () => {
const result = await triggerQueueTimeout();
assert.equal(result, "should-not-reach");
});
// The surfaced message must read as a local queue limit, naming the knob,
// and must NOT masquerade as an upstream "This job timed out" gateway error.
assert.match(caught.message, /maxWaitMs/, "message should name the maxWaitMs knob");
assert.match(
caught.message,
/not an upstream/i,
"message should explicitly disclaim an upstream timeout"
);
assert.doesNotMatch(
caught.message,
/This job timed out/,
"raw Bottleneck/upstream-looking string must not leak into the surfaced message"
);
// The original Bottleneck error is preserved for debugging.
assert.ok(caught.cause, "original error should be preserved as cause");
assert.match(String(caught.cause?.message ?? ""), /This job timed out/);
test("#4165 queue expiry surfaces a clear local error", async () => {
const caught = (await triggerQueuedTimeout()) as Error & { code?: string };
assert.equal(caught.code, "RATE_LIMIT_QUEUE_TIMEOUT");
assert.match(caught.message, /maxWaitMs/);
assert.match(caught.message, /not an upstream/i);
assert.doesNotMatch(caught.message, /This job timed out/);
});
test("#4165 a job that completes within maxWaitMs is unaffected", async () => {

73
tests/unit/sliding-window-limiter.test.ts
View File

@@ -25,7 +25,10 @@ test("allows up to N requests in the window and blocks the (N+1)-th", () => {
const blocked = limiter.tryAcquire("k", win);
assert.equal(blocked.allowed, false, "the 4th request in a 3/1000ms window is blocked");
assert.ok(blocked.retryAfterMs > 0 && blocked.retryAfterMs <= 1000, "retryAfterMs points at the oldest hit expiry");
assert.ok(
blocked.retryAfterMs > 0 && blocked.retryAfterMs <= 1000,
"retryAfterMs points at the oldest hit expiry"
);
});
test("the window slides: a slot frees once the oldest hit ages out", () => {
@@ -39,7 +42,11 @@ test("the window slides: a slot frees once the oldest hit ages out", () => {
assert.equal(limiter.tryAcquire("k", win).allowed, false, "2/1000ms is saturated at t=400");
clock.advance(601); // t=1001 — the t=0 hit (>1000ms old) ages out
assert.equal(limiter.tryAcquire("k", win).allowed, true, "a slot frees once the oldest hit leaves the window");
assert.equal(
limiter.tryAcquire("k", win).allowed,
true,
"a slot frees once the oldest hit leaves the window"
);
});
test("retryAfterMs reflects when the oldest in-window hit expires", () => {
@@ -60,7 +67,11 @@ test("keys are isolated from one another", () => {
const win = { requests: 1, windowMs: 1000 };
assert.equal(limiter.tryAcquire("a", win).allowed, true);
assert.equal(limiter.tryAcquire("b", win).allowed, true, "key b is unaffected by key a being saturated");
assert.equal(
limiter.tryAcquire("b", win).allowed,
true,
"key b is unaffected by key a being saturated"
);
assert.equal(limiter.tryAcquire("a", win).allowed, false);
});
@@ -78,7 +89,11 @@ test("reset clears a single key's history", () => {
assert.equal(limiter.tryAcquire("k", win).allowed, true);
assert.equal(limiter.tryAcquire("k", win).allowed, false);
limiter.reset("k");
assert.equal(limiter.tryAcquire("k", win).allowed, true, "history cleared → slot available again");
assert.equal(
limiter.tryAcquire("k", win).allowed,
true,
"history cleared → slot available again"
);
});
test("blocked attempts do not consume a slot (no double counting)", () => {
@@ -92,5 +107,53 @@ test("blocked attempts do not consume a slot (no double counting)", () => {
limiter.tryAcquire("k", win);
limiter.tryAcquire("k", win);
clock.advance(1001);
assert.equal(limiter.tryAcquire("k", win).allowed, true, "only the single successful hit aged out");
assert.equal(
limiter.tryAcquire("k", win).allowed,
true,
"only the single successful hit aged out"
);
});
test("multi-scope acquisition is atomic", () => {
const clock = fakeClock();
const limiter = new SlidingWindowLimiter({ now: clock.now });
const global = { key: "global", window: { requests: 2, windowMs: 1000 } };
const provider = { key: "provider:openference", window: { requests: 1, windowMs: 1000 } };
const first = limiter.tryAcquireMany([global, provider]);
assert.equal(first.allowed, true);
const blocked = limiter.tryAcquireMany([global, provider]);
assert.equal(blocked.allowed, false, "the provider scope blocks the second request");
clock.advance(1001);
const second = limiter.tryAcquireMany([global, provider]);
assert.equal(second.allowed, true);
});
test("releasing an un-dispatched multi-scope lease returns every scope", () => {
const limiter = new SlidingWindowLimiter();
const scopes = [
{ key: "global", window: { requests: 1, windowMs: 1000 } },
{ key: "provider", window: { requests: 1, windowMs: 1000 } },
];
const result = limiter.tryAcquireMany(scopes);
assert.equal(result.allowed, true);
result.lease?.release();
assert.equal(limiter.tryAcquireMany(scopes).allowed, true);
});
test("rolling leases do not reset as a burst at a fixed boundary", () => {
const clock = fakeClock();
const limiter = new SlidingWindowLimiter({ now: clock.now });
const window = { key: "global", window: { requests: 2, windowMs: 1000 } };
assert.equal(limiter.tryAcquireMany([window]).allowed, true); // t=0
clock.advance(900);
assert.equal(limiter.tryAcquireMany([window]).allowed, true); // t=900
assert.equal(limiter.tryAcquireMany([window]).allowed, false);
clock.advance(100);
assert.equal(limiter.tryAcquireMany([window]).allowed, true, "only the t=0 lease returned");
assert.equal(limiter.tryAcquireMany([window]).allowed, false, "the t=900 lease remains active");
});