diff --git a/changelog.d/fixes/9041-rate-limit-idle-capacity-wedge.md b/changelog.d/fixes/9041-rate-limit-idle-capacity-wedge.md new file mode 100644 index 0000000000..cfb180cabd --- /dev/null +++ b/changelog.d/fixes/9041-rate-limit-idle-capacity-wedge.md @@ -0,0 +1 @@ +- **fix(resilience):** Detect and reset idle-capacity rate-limit queue wedges on an eligible watchdog scan so routing can fall back promptly ([#9041](https://github.com/diegosouzapw/OmniRoute/pull/9041)) diff --git a/docs/architecture/RESILIENCE_GUIDE.md b/docs/architecture/RESILIENCE_GUIDE.md index 8e68322e96..e2f9d3dfd2 100644 --- a/docs/architecture/RESILIENCE_GUIDE.md +++ b/docs/architecture/RESILIENCE_GUIDE.md @@ -224,12 +224,16 @@ rate limit. Bounded by `comboCooldownWait` (`enabled`, `maxWaitMs`, `maxAttempts **Scope**: the local per-provider+connection rate-limit queue (`open-sse/services/rateLimitManager.ts`, backed by Bottleneck), one layer below the three mechanisms above. -**`maxWaitMs` default lowered 120s → 15s.** `resilienceSettings.requestQueue.maxWaitMs` -bounds how long a request may wait in the local queue before it is dropped -(`code: "RATE_LIMIT_QUEUE_TIMEOUT"`, #4165). The factory default fell from 120000ms to -15000ms so a saturated queue fails fast instead of holding a caller for two -minutes; override via `RATE_LIMIT_MAX_WAIT_MS` (env) or the dashboard -(**Settings → Resilience**, 1–30000ms UI ceiling). +**`maxWaitMs` is a legacy persisted name for execution expiration.** +`resilienceSettings.requestQueue.maxWaitMs` is passed to Bottleneck as a job +`expiration`, whose timer starts only after dispatch. It therefore bounds +limiter-managed execution, not time spent in the local queue. Expiration is +surfaced as trusted local `code: "RATE_LIMIT_EXECUTION_TIMEOUT"` (HTTP 504); +the former queue-timeout code name is accepted only for trusted internal +backward compatibility. The default is 15000ms; override via +`RATE_LIMIT_MAX_WAIT_MS` (env) or the dashboard (**Settings → Resilience**, +1–30000ms UI ceiling). Queue residence has no time deadline; use +`maxQueueDepth` below to bound queued callers. **`maxQueueDepth` — opt-in admission cap (new).** `resilienceSettings.requestQueue.maxQueueDepth` bounds how many requests may sit queued (not yet dispatched) for one @@ -252,7 +256,7 @@ it is unit-testable without a real Bottleneck limiter. > around the `resolveCompressionSettings`/`selectCompressionStrategy` block), > not HTTP response compression on synthesized 429 bodies — there is no > matching code path for a literal bypass flag. That prompt-compression step -> also currently runs *before* `withRateLimit()` in the request pipeline, so +> also currently runs _before_ `withRateLimit()` in the request pipeline, so > reordering to skip it on a queue-full rejection is a separate, larger > change than this issue's scope; it was intentionally **not** implemented > here and is left as a follow-up if the CPU-saving win is worth the diff --git a/docs/security/AGENTROUTER_WAF.md b/docs/security/AGENTROUTER_WAF.md index 8aaaa941c6..0311da0a07 100644 --- a/docs/security/AGENTROUTER_WAF.md +++ b/docs/security/AGENTROUTER_WAF.md @@ -1,5 +1,5 @@ --- -title: "AgentRouter WAF" +title: "agentrouter.org WAF (Web Application Firewall)" version: 3.8.50 lastUpdated: 2026-08-03 --- @@ -94,4 +94,4 @@ The current filter is overly aggressive — it blocks "Lorem ipsum" in `tool_result` blocks even though the operator clearly did not intend to inject a prompt. Operators who want this fixed at the source should contact `agentrouter.org` to report the false positives. The blocklist -above is the empirical result of probing the upstream as of 2026-08-03. \ No newline at end of file +above is the empirical result of probing the upstream as of 2026-08-03. diff --git a/open-sse/handlers/chatCore.ts b/open-sse/handlers/chatCore.ts index 99be5cda82..fe9da81aeb 100644 --- a/open-sse/handlers/chatCore.ts +++ b/open-sse/handlers/chatCore.ts @@ -302,6 +302,7 @@ import { updateFromResponseBody, initializeRateLimits, } from "../services/rateLimitManager.ts"; +import * as localLimiterErrors from "../services/rateLimitManager/errors.ts"; import { acquire as acquireAccountSemaphore, markBlocked as markAccountSemaphoreBlocked, @@ -3362,27 +3363,22 @@ export async function handleChatCore({ errorCode: error.code, }; } - // abort(reason) can reject the upstream fetch with a raw string reason - // (e.g. "request_signal_aborted") that has no `name`/`status`; classify - // via isLocalStreamLifecycleError so those map to 499 instead of falling - // through to the 502 provider-failure default. + // abort(reason) can reject with a raw string lacking `name`/`status`; classify + // it through isLocalStreamLifecycleError so it maps to 499 rather than the + // 502 provider-failure default. const isRequestAborted = isLocalStreamLifecycleError(error); - // #8376: an unreachable upstream proxy (ECONNREFUSED/ECONNRESET/...) is tagged by - // proxyFetch.ts (tagProxyUnreachable) with `.errorCode = "proxy_unreachable"` before - // it reaches this catch. Classify it explicitly to 502 instead of falling through - // the generic `error.status` branch (a raw connect-refused error has no `.status` at - // all, so it used to collapse into an ordinary 502/504 the provider-breaker predicate - // can't tell apart from a per-model 5xx). + // #8376: proxyFetch tags unreachable transport failures so they remain + // distinguishable from ordinary provider 5xx responses. const isProxyUnreachableFailure = !isRequestAborted && (error as { errorCode?: unknown })?.errorCode === "proxy_unreachable"; const errorCode = getUpstreamErrorIdentifier(error); - const isLocalQueueTimeout = errorCode === "RATE_LIMIT_QUEUE_TIMEOUT"; + const localRateLimitFailure = localLimiterErrors.getClientSafeLocalRateLimitError(error); const failureStatus = isRequestAborted ? 499 : isProxyUnreachableFailure ? HTTP_STATUS.BAD_GATEWAY - : isLocalQueueTimeout - ? HTTP_STATUS.SERVICE_UNAVAILABLE + : localRateLimitFailure + ? localRateLimitFailure.status : error.name === "TimeoutError" || error.name === "BodyTimeoutError" ? HTTP_STATUS.GATEWAY_TIMEOUT : error.status && typeof error.status === "number" @@ -3390,8 +3386,9 @@ export async function handleChatCore({ : HTTP_STATUS.BAD_GATEWAY; const failureMessage = isRequestAborted ? "Request aborted" - : formatProviderError(error, provider, model, failureStatus); - const upstreamErrorCode = isProxyUnreachableFailure ? "proxy_unreachable" : errorCode; + : formatProviderError(localRateLimitFailure ?? error, provider, model, failureStatus); + const upstreamErrorCode = + localRateLimitFailure?.code ?? (isProxyUnreachableFailure ? "proxy_unreachable" : errorCode); // Tag our own deadline timeouts (fetch-start TimeoutError / body BodyTimeoutError, // both surfaced as a 504) as "upstream_timeout" so the cooldown layer can tell a // slow-but-not-failed request apart from a real provider 5xx. (Antigravity already @@ -3441,19 +3438,22 @@ export async function handleChatCore({ upstreamErrorCode, upstreamErrorType ); + localLimiterErrors.markTrustedLocalRateLimitResponse(result.response, error); return { ...result, errorType: upstreamErrorType, errorCode: upstreamErrorCode, }; } - return createErrorResult( + const result = createErrorResult( failureStatus, failureMessage, null, upstreamErrorCode, upstreamErrorType ); + localLimiterErrors.markTrustedLocalRateLimitResponse(result.response, error); + return result; } let upstreamErrorParsed = false; let parsedStatusCode = providerResponse.status; diff --git a/open-sse/services/combo.ts b/open-sse/services/combo.ts index 7db96ae231..7b73108df5 100644 --- a/open-sse/services/combo.ts +++ b/open-sse/services/combo.ts @@ -1594,7 +1594,7 @@ export async function handleComboChat({ : undefined, } : undefined; - const scopedFailure = isScopedFailure(result.status, errorText, structuredError); + const scopedFailure = isScopedFailure(result, errorText, structuredError); // #8375: input-bound request-scoped failures (context_length_exceeded) are // deterministic for the same input — retrying on other accounts of the same @@ -1675,6 +1675,7 @@ export async function handleComboChat({ rawModel, isTokenLimitBreach, allAccountsRateLimited: false, + requestScopedFailure: scopedFailure, sets: { exhaustedProviders, exhaustedConnections, transientRateLimitedProviders }, log, tag: "COMBO", @@ -1767,6 +1768,7 @@ export async function handleComboChat({ const isTransient = !isStreamReadinessFailure && !isTokenLimitBreach && + !scopedFailure && [408, 429, 500, 502, 503, 504].includes(result.status); if (retry < maxRetries && isTransient && !providerExhausted) { if ( @@ -2841,7 +2843,7 @@ async function handleRoundRobinCombo({ : undefined, } : undefined; - const scopedFailure = isScopedFailure(result.status, errorText, structuredError); + const scopedFailure = isScopedFailure(result, errorText, structuredError); const fallbackResult = checkFallbackError( result.status, errorText, @@ -2880,6 +2882,7 @@ async function handleRoundRobinCombo({ rawModel: parseModel(modelStr).model || modelStr, isTokenLimitBreach, allAccountsRateLimited: isAllAccountsRateLimited, + requestScopedFailure: scopedFailure, sets: { exhaustedProviders, exhaustedConnections, transientRateLimitedProviders }, log, tag: "COMBO-RR", @@ -2913,6 +2916,7 @@ async function handleRoundRobinCombo({ const isTransient = !isStreamReadinessFailure && !isTokenLimitBreach && + !scopedFailure && [408, 429, 500, 502, 503, 504].includes(result.status); if (retry < maxRetries && isTransient && !providerExhausted) { continue; diff --git a/open-sse/services/combo/comboPredicates.ts b/open-sse/services/combo/comboPredicates.ts index cc29f710fa..93f1568c96 100644 --- a/open-sse/services/combo/comboPredicates.ts +++ b/open-sse/services/combo/comboPredicates.ts @@ -12,6 +12,7 @@ import { isSelfInflictedUpstreamTimeout } from "../../handlers/chatCore/cooldown import { isLocalStreamLifecycleError } from "@/shared/utils/circuitBreaker"; import { CONTEXT_OVERFLOW_PATTERNS, MODEL_ACCESS_DENIED_PATTERNS } from "../accountFallback.ts"; import { isResourceNotFoundResponse } from "../errorClassifier.ts"; +import { getTrustedLocalRateLimitResponse } from "../rateLimitManager/errors.ts"; import type { ResolvedComboTarget } from "./types.ts"; // Status codes that should mark round-robin target semaphores as cooling down. @@ -190,13 +191,13 @@ export function shouldRecordProviderBreakerFailure(args: { ); } -const REQUEST_SCOPED_UPSTREAM_ERROR_CODES = new Set([ - "context_length_exceeded", - "upstream_empty_response", - "upstream_response_failed", +const REQUEST_SCOPED_UPSTREAM_ERROR_CODES: Record = { + context_length_exceeded: true, + upstream_empty_response: true, + upstream_response_failed: true, // Local combo per-target timer (targetTimeoutRunner) — not a connection health signal. - "combo_target_timeout", -]); + combo_target_timeout: true, +}; /** Request/model-specific failures must not poison provider-wide resilience state. */ export function isRequestScopedUpstreamFailure(error?: { @@ -205,18 +206,19 @@ export function isRequestScopedUpstreamFailure(error?: { }): boolean { const code = typeof error?.code === "string" ? error.code.toLowerCase() : ""; const type = typeof error?.type === "string" ? error.type.toLowerCase() : ""; - return REQUEST_SCOPED_UPSTREAM_ERROR_CODES.has(code) || type === "context_length_exceeded"; + return REQUEST_SCOPED_UPSTREAM_ERROR_CODES[code] === true || type === "context_length_exceeded"; } /** Request-scoped classification that also has access to the HTTP body. */ export function isComboRequestScopedFailure( - status: number, + response: Response, errorText: string, error?: { code?: string | null; type?: string | null } ): boolean { return ( + getTrustedLocalRateLimitResponse(response) !== null || isRequestScopedUpstreamFailure(error) || - (status === 404 && isResourceNotFoundResponse(errorText)) + (response.status === 404 && isResourceNotFoundResponse(errorText)) ); } @@ -255,6 +257,7 @@ export function isInputBoundRequestFailure(error?: { export function shouldSkipConnDisable( result: { status: number; + response?: Response; errorCode?: string | null; errorType?: string | null; error?: unknown; @@ -270,6 +273,7 @@ export function shouldSkipConnDisable( // Client abort surfaced as a bare error (no statusCode → defaults to 502): // a local lifecycle event, not a provider failure (#4602 policy). isLocalStreamLifecycleError(result.error) || + (result.response ? getTrustedLocalRateLimitResponse(result.response) !== null : false) || result.errorCode === "plugin_block" || result.errorType === "plugin_block" || (is401 && hasExtraKeys) || diff --git a/open-sse/services/combo/targetExhaustion.ts b/open-sse/services/combo/targetExhaustion.ts index 02989b68f2..c4a880e79a 100644 --- a/open-sse/services/combo/targetExhaustion.ts +++ b/open-sse/services/combo/targetExhaustion.ts @@ -65,6 +65,7 @@ export type ApplyComboTargetExhaustionOptions = { rawModel: string; isTokenLimitBreach: boolean; allAccountsRateLimited: boolean; + requestScopedFailure: boolean; sets: ComboExhaustionSets; log: ComboLogger; tag: string; @@ -154,12 +155,25 @@ function isProviderQuotaExhausted( provider: string | null | undefined, opts: Pick< ApplyComboTargetExhaustionOptions, - "rawModel" | "fallbackResult" | "structuredError" | "errorText" | "allAccountsRateLimited" + | "rawModel" + | "fallbackResult" + | "structuredError" + | "errorText" + | "allAccountsRateLimited" + | "requestScopedFailure" > ): boolean { - const { rawModel, fallbackResult, structuredError, errorText, allAccountsRateLimited } = opts; + const { + rawModel, + fallbackResult, + structuredError, + errorText, + allAccountsRateLimited, + requestScopedFailure, + } = opts; return ( Boolean(provider && provider !== "unknown") && + !(requestScopedFailure || isRequestScopedUpstreamFailure(structuredError)) && !hasPerModelQuota(provider as string, rawModel) && (isProviderExhaustedReason(fallbackResult) || classifyErrorText(structuredError?.code || errorText) === RateLimitReason.QUOTA_EXHAUSTED || @@ -189,7 +203,17 @@ function markTransientOrConnectionLevel( target: ResolvedComboTarget, opts: ApplyComboTargetExhaustionOptions ): void { - const { result, errorText, rawModel, isTokenLimitBreach, sets, log, tag, structuredError } = opts; + const { + result, + errorText, + rawModel, + isTokenLimitBreach, + requestScopedFailure, + sets, + log, + tag, + structuredError, + } = opts; const provider = target.provider; if (result.status === 429 && !isTokenLimitBreach && provider && provider !== "unknown") { sets.transientRateLimitedProviders.add(provider); @@ -201,6 +225,7 @@ function markTransientOrConnectionLevel( log, tag, rawModel, + requestScopedFailure, structuredError, }); } @@ -244,16 +269,25 @@ function markConnectionLevelExhaustion( target: ResolvedComboTarget, opts: Pick< ApplyComboTargetExhaustionOptions, - "result" | "errorText" | "sets" | "log" | "tag" | "rawModel" | "structuredError" + | "result" + | "errorText" + | "sets" + | "log" + | "tag" + | "rawModel" + | "requestScopedFailure" + | "structuredError" > ): void { - const { result, errorText, sets, log, tag, rawModel, structuredError } = opts; + const { result, errorText, sets, log, tag, rawModel, requestScopedFailure, structuredError } = + opts; const provider = target.provider; if ( !provider || provider === "unknown" || !CONNECTION_LEVEL_ERROR_STATUSES.includes(result.status) || isProviderCircuitOpenResult(result, errorText) || + requestScopedFailure || isRequestScopedUpstreamFailure(structuredError) || // #5085: empty-content 502 is a healthy connection returning no body — model-level, not // connection-level. Don't exhaust the provider; let the remaining legs (incl. same-provider) diff --git a/open-sse/services/rateLimitManager.ts b/open-sse/services/rateLimitManager.ts index cb4350b2f8..fbc7d0d2fc 100644 --- a/open-sse/services/rateLimitManager.ts +++ b/open-sse/services/rateLimitManager.ts @@ -13,12 +13,7 @@ import { parseRetryAfterFromBody } from "./accountFallback.ts"; import { getAntigravityQuotaFamily } from "./antigravityQuotaFamily.ts"; import { getProviderCategory } from "../config/providerRegistry.ts"; import { getCodexRateLimitKey } from "../executors/codex.ts"; -import { - getProviderDefaultRateLimit, - setProviderQuotaOverrides, -} from "./providerDefaultRateLimit.ts"; -import { keyContainsConnection, RollingRpmGate } from "./rollingRpmGate.ts"; -import { toNumber } from "@/shared/utils/numeric"; +import { awaitProviderDefaultSlot, setProviderQuotaOverrides } from "./providerDefaultRateLimit.ts"; import { DEFAULT_RESILIENCE_SETTINGS, resolveResilienceSettings, @@ -31,6 +26,13 @@ import { toPlainHeaders, } from "./rateLimitManager/headers"; import { checkQueueAdmission } from "./rateLimitManager/admission"; +import { + markLocalRateLimitError, + RATE_LIMIT_EXECUTION_TIMEOUT_CODE, + RATE_LIMIT_QUEUE_WEDGED_CODE, +} from "./rateLimitManager/errors"; +import { LimiterWedgeWatchdog, WATCHDOG_INTERVAL_MS } from "./rateLimitManager/wedgeWatchdog"; +import { toNumber } from "@/shared/utils/numeric"; interface LearnedLimitEntry { provider: string; @@ -44,38 +46,17 @@ interface LearnedLimitEntry { interface LimiterUpdateSettings { maxConcurrent?: number | null; minTime: number; + reservoir?: number | null; + reservoirRefreshAmount?: number | null; + reservoirRefreshInterval?: number | null; } type JsonRecord = Record; -type QueueTimeoutReason = "local-queue" | "upstream-cooldown"; function toRecord(value: unknown): JsonRecord { return value && typeof value === "object" && !Array.isArray(value) ? (value as JsonRecord) : {}; } -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 { return typeof process.env.NODE_TEST_CONTEXT === "string"; } @@ -105,7 +86,6 @@ const connectionRateLimitOverrides = new Map>(); // Store learned limits for persistence (debounced) const learnedLimits: Record = {}; const MAX_LEARNED_LIMITS = 200; -const INACTIVE_LIMITER_MS = 10 * 60 * 1000; const limiterLastUsed = new Map(); let persistTimer: ReturnType | null = null; const pendingAsyncOperations = new Set>(); @@ -116,17 +96,24 @@ 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). RPM admission happens before Bottleneck, so a queued -// Bottleneck job with no active work is a concurrency scheduler failure. -const lastDispatchAt = new Map(); -let nextJobTraceId = 1; +const limiterEffectiveSettings = new WeakMap(); +const preservedReplacementSettings = new Map(); +const limiterWatchdog = new LimiterWedgeWatchdog({ + limiters, + limiterLastUsed, + limiterEffectiveSettings, + preservedReplacementSettings, + trackBackground: (promise) => { + trackAsyncOperation(promise); + }, + log: logRateLimit, + warn: warnRateLimit, +}); let watchdogInterval: ReturnType | null = null; -const WATCHDOG_INTERVAL_MS = 30_000; -// 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; + +type LimiterFactory = (options: Bottleneck.ConstructorOptions) => Bottleneck; +const defaultLimiterFactory: LimiterFactory = (options) => new Bottleneck(options); +let limiterFactory: LimiterFactory = defaultLimiterFactory; /** * Env-var override for the auto-enable safety net. Highest priority — wins @@ -143,10 +130,19 @@ function isAutoEnableActive(settings: RequestQueueSettings): boolean { return settings.autoEnableApiKeyProviders; } -// Bottleneck handles concurrency and pacing. RPM is enforced by the rolling -// lease limiter above rather than by a fixed-window reservoir. +// 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; 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; @@ -158,62 +154,38 @@ 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, }; } -/** - * Apply new settings to a Bottleneck limiter and re-arm its reservoir-refresh - * heartbeat. - * - * Bottleneck 2.19.5 (frozen upstream dependency, no release since 2019) has a - * bug in `LocalDatastore#_startHeartbeat()` - * (node_modules/bottleneck/lib/LocalDatastore.js:29,56): the guard - * `if (this.heartbeat == null && ...)` only (re)creates the periodic - * reservoir-refresh interval the FIRST time it runs. Every later call — - * including the one `updateSettings()` itself triggers internally — falls - * into the `else` branch and does `clearInterval(this.heartbeat)` WITHOUT - * resetting `this.heartbeat` back to `null`. Because the stale reference is - * left in place, every future `_startHeartbeat()` call keeps taking the same - * dead `else` branch: the periodic reservoir refresh is gone forever after - * the FIRST manual `updateSettings()` call on a limiter — every limiter here - * starts with a live heartbeat (buildLimiterDefaults() always sets - * reservoirRefreshInterval/reservoirRefreshAmount), so that "first call" is - * whichever of the 5 updateSettings() call sites in this file runs first. - * - * Work around it here instead of patching node_modules: null out the stale - * reference ourselves and re-invoke `_startHeartbeat()` so it takes the - * "start a fresh interval" branch again. Every `limiter.updateSettings(...)` - * call in this file MUST go through this helper, never Bottleneck's method - * directly. - */ -async function applyLimiterSettings( +function updateLimiterSettings( limiter: Bottleneck, updates: Bottleneck.ConstructorOptions -): Promise { - await limiter.updateSettings(updates); - const store = ( - limiter as unknown as { - _store?: { - heartbeat?: ReturnType | null; - _startHeartbeat?: () => void; - }; - } - )._store; - if (store && typeof store._startHeartbeat === "function") { - if (store.heartbeat != null) clearInterval(store.heartbeat); - store.heartbeat = null; - store._startHeartbeat(); +): Bottleneck { + const effective = limiterEffectiveSettings.get(limiter) ?? {}; + limiterEffectiveSettings.set(limiter, { ...effective, ...updates }); + return limiter.updateSettings(updates); +} + +function updateAllLimiterSettings() { + const defaults = buildLimiterDefaults(); + for (const limiter of limiters.values()) { + updateLimiterSettings(limiter, defaults); } } -async function updateAllLimiterSettings() { - const defaults = buildLimiterDefaults(); - await Promise.all( - Array.from(limiters.values(), (limiter) => applyLimiterSettings(limiter, defaults)) - ); +function clearPreservedReplacementSettings(connectionId: string): void { + for (const key of preservedReplacementSettings.keys()) { + if (key.includes(connectionId)) preservedReplacementSettings.delete(key); + } } function reconcileEnabledConnections( @@ -246,9 +218,8 @@ function reconcileEnabledConnections( nextEnabledConnections.add(connectionId); autoCount++; - // Route through getLimiter so the `queued`/`executing` listeners and - // lastDispatchAt heartbeat are wired up — otherwise the watchdog sees - // `stalledMs = now - 0` and falsely flags healthy idle limiters as wedged. + // Route through getLimiter so the queue-progress listeners are wired up. + // Otherwise a limiter created here could not be evaluated safely by the watchdog. getLimiter(provider, connectionId); } } @@ -269,82 +240,16 @@ 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.RECEIVED === 0 && - counts.QUEUED === 0 && - counts.RUNNING === 0 && - counts.EXECUTING === 0 - ) { - limiters.delete(key); - lastDispatchAt.delete(key); - limiterLastUsed.delete(key); - logRateLimit( - `🧹 [RATE-LIMIT] Evicting idle limiter: ${key} (inactive for ${Math.round((now - lastUsed) / 1000)}s)` - ); - trackAsyncOperation(limiter.disconnect()); - } - } - } - for (const [key, limiter] of Array.from(limiters)) { - const counts = limiter.counts(); - // 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 - // detection on a brand-new limiter or one created outside getLimiter. - if (lastDispatch === undefined) { - lastDispatchAt.set(key, now); - continue; - } - const stalledMs = now - lastDispatch; - if (stalledMs < WEDGE_THRESHOLD_MS) continue; - - warnRateLimit( - `🚨 [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 - // this instance — withRateLimit's `limiter.schedule()` for those callers - // then just hangs forever (nothing will ever dequeue them; getLimiter() - // only hands out a FRESH instance to future callers), leaving the - // dispatch orphaned until the outer ~300s per-target timeout eventually - // aborts it. Real clients routinely give up (and retry) well before that - // — this specific incident's client aborted at ~60s having never reached - // the provider at all (queued=2 running=0 executing=0 the entire time). - // - // stop({ dropWaitingJobs: true }) rejects exactly the RECEIVED/QUEUED/ - // RUNNING jobs on THIS instance immediately (Bottleneck's own contract — - // see node_modules/bottleneck/bottleneck.d.ts StopOptions) so those - // withRateLimit() callers reject right away instead of hanging, letting - // combo's fallback/cooldown-wait engage within seconds instead of minutes. - // This is safe against the previously-documented "spurious 502 bursts" - // concern: the wedge condition checked above already requires - // RUNNING === 0 && EXECUTING === 0, so no job that's actually progressing - // can be caught by this — only ones already confirmed stuck. The instance - // is deleted from `limiters` synchronously (above) before this call, so - // no future getLimiter() call can ever hand out this now-stopped instance - // — the "permanently rejects future .schedule()" behavior stop() has is - // therefore moot; nothing will call .schedule() on it again. - evictWedgeLimiter(key, limiter); - } -} - let shutdownHandlersRegistered = false; export function startRateLimitWatchdog(): void { if (watchdogInterval) return; - watchdogInterval = setInterval(watchdogTick, WATCHDOG_INTERVAL_MS); + watchdogInterval = setInterval(() => { + const run = trackAsyncOperation(limiterWatchdog.run()); + void run.then(undefined, (error) => { + errorRateLimit("[RATE-LIMIT] Watchdog scan failed:", error); + }); + }, WATCHDOG_INTERVAL_MS); watchdogInterval.unref?.(); // Register SIGTERM/SIGINT shutdown handlers once, lazily, on first watchdog start. // Registering here (rather than at module load) avoids interfering with test runner @@ -362,54 +267,18 @@ 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; - evictLimiterAndDropQueued(key, limiter, "rate-limit-watchdog-wedge-reset"); -} - /** * Gracefully stop all limiters for process shutdown. - * ONLY call this from SIGTERM/SIGINT handlers — not during runtime resets. - * Calling .stop() during runtime (e.g. on 429 or connection disable) permanently - * rejects future .schedule() calls, causing 502 bursts. This function is the - * sole legitimate use of limiter.stop() in this module. + * Runtime wedge recovery also uses stop(), but only after synchronously + * removing that limiter from the cache so it can never accept new work. */ function shutdownLimiters(): void { for (const limiter of limiters.values()) { limiter.stop({ dropWaitingJobs: false }); } limiters.clear(); - lastDispatchAt.clear(); limiterLastUsed.clear(); + preservedReplacementSettings.clear(); } // Only register shutdown handlers when there are active limiters to shut down. @@ -454,10 +323,13 @@ 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 + ); + updateAllLimiterSettings(); - // 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. + // Load per-connection rate limit overrides connectionRateLimitOverrides.clear(); for (const conn of connections as Array>) { const overrides = conn.rateLimitOverrides; @@ -466,12 +338,6 @@ 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)` @@ -491,16 +357,21 @@ export async function initializeRateLimits() { export async function applyRequestQueueSettings(nextSettings: RequestQueueSettings) { currentRequestQueueSettings = { ...nextSettings }; + // Global policy changes invalidate snapshots from the previous generation. + preservedReplacementSettings.clear(); const { getCachedProviderConnections } = await import("@/lib/localDb"); const connections = await getCachedProviderConnections(); + // Also discard any snapshot created while the asynchronous DB read yielded. + preservedReplacementSettings.clear(); reconcileEnabledConnections(connections as unknown[], currentRequestQueueSettings); - await updateAllLimiterSettings(); + updateAllLimiterSettings(); } /** * Get or create a limiter for a given provider+connection combination */ export function enableRateLimitProtection(connectionId) { + if (!enabledConnections.has(connectionId)) clearPreservedReplacementSettings(connectionId); enabledConnections.add(connectionId); } @@ -509,14 +380,19 @@ export function enableRateLimitProtection(connectionId) { */ export function disableRateLimitProtection(connectionId) { enabledConnections.delete(connectionId); - // Drop queued jobs before evicting the limiter. Otherwise disconnect() leaves - // callers waiting on an instance that is no longer reachable from the cache. + clearPreservedReplacementSettings(connectionId); + // Ordinary administrative eviction uses disconnect(), not stop(), so + // in-flight requests can finish. Wedge recovery is the deliberate exception: + // it removes the limiter from the cache first, then stops it to settle jobs + // that were already proven stranded. for (const [key, limiter] of Array.from(limiters)) { - if (keyContainsConnection(key, connectionId)) { - evictLimiterAndDropQueued(key, limiter, "rate-limit-connection-disabled"); + if (key.includes(connectionId)) { + limiters.delete(key); + limiterWatchdog.forget(limiter); + limiterLastUsed.delete(key); + trackAsyncOperation(limiter.disconnect()); } } - rpmGate.clearConnection(connectionId); } /** @@ -542,13 +418,16 @@ export function refreshConnectionRateLimits(connectionId, overrides) { } else { connectionRateLimitOverrides.set(connectionId, overrides); } + clearPreservedReplacementSettings(connectionId); // Evict limiters referencing this connection so they get recreated on next use for (const [key, limiter] of Array.from(limiters)) { - if (keyContainsConnection(key, connectionId)) { - evictLimiterAndDropQueued(key, limiter, "rate-limit-settings-refresh"); + if (key.includes(connectionId)) { + limiters.delete(key); + limiterWatchdog.forget(limiter); + limiterLastUsed.delete(key); + trackAsyncOperation(limiter.disconnect()); } } - rpmGate.clearConnection(connectionId); } /** @@ -571,46 +450,51 @@ 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); if (!limiters.has(key)) { - const defaults = buildLimiterDefaults(); - const overrides = connectionRateLimitOverrides.get(connectionId); - if (overrides) { - // 0 (or missing) means "no override — fall through to buildLimiterDefaults()". - if (typeof overrides.maxConcurrent === "number" && overrides.maxConcurrent > 0) { - defaults.maxConcurrent = overrides.maxConcurrent; + const preserved = preservedReplacementSettings.get(key); + let options: Bottleneck.ConstructorOptions; + if (preserved) { + preservedReplacementSettings.delete(key); + options = { ...preserved, id: key }; + } else { + const defaults = buildLimiterDefaults(); + 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. + 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 separate token and request buckets. } - if (typeof overrides.minTime === "number" && overrides.minTime > 0) { - defaults.minTime = overrides.minTime; - } - // TODO: TPM/TPD integration — requires a token-bucket vs request-bucket - // separation. RPM is handled by the rolling lease gate below. - // When added, treat 0/missing the same way: fall through to system default. + options = { ...defaults, id: key }; } - const limiter = new Bottleneck({ - ...defaults, - id: key, - }); - // Heartbeat: timestamp every dispatch so the watchdog can tell a healthy - // queue (just dispatched a job) from a wedged one (queue has work but - // nothing has been dispatched in a while). - limiter.on("executing", () => { - lastDispatchAt.set(key, Date.now()); + const limiter = limiterFactory(options); + limiterEffectiveSettings.set(limiter, { ...options }); + limiter.on("queued", () => { + limiterWatchdog.noteQueued(key, limiter); }); + const markQueueProgress = () => { + limiterWatchdog.noteProgress(key, limiter); + }; + limiter.on("executing", markQueueProgress); + // A long-running job can leave older work queued. Start the idle grace + // from its completion, not from when that waiting work first arrived. + limiter.on("done", markQueueProgress); limiters.set(key, limiter); - lastDispatchAt.set(key, Date.now()); limiterLastUsed.set(key, Date.now()); } @@ -618,15 +502,6 @@ 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. @@ -651,20 +526,22 @@ export async function withRateLimit(provider, connectionId, model, fn, signal = throw err; } - const maxWaitMs = currentRequestQueueSettings.maxWaitMs; - const queueStartedAt = Date.now(); - const rpmLease = await rpmGate.acquire( + // 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( provider, connectionId, - model, signal, - maxWaitMs, - queueStartedAt + currentRequestQueueSettings.maxWaitMs ); + const limiter = getLimiter(provider, connectionId, model); - const key = getLimiterKey(provider, connectionId, model); - const jobId = `${key}:job-${nextJobTraceId++}`; - const scheduleOpts = { id: jobId }; + // Bottleneck's `expiration` starts only after a job leaves QUEUED. The + // legacy maxWaitMs setting therefore bounds limiter-managed execution; it + // is not a queue-wait deadline. + const executionExpirationMs = currentRequestQueueSettings.maxWaitMs; + const scheduleOpts = + executionExpirationMs && executionExpirationMs > 0 ? { expiration: executionExpirationMs } : {}; // Issue #6593: opt-in admission cap — fast-reject before Bottleneck's // schedule() (and before any downstream compression/prompt work runs) when @@ -675,129 +552,96 @@ export async function withRateLimit(provider, connectionId, model, fn, signal = 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 | undefined; - const remainingWaitMs = - maxWaitMs > 0 ? Math.max(1, maxWaitMs - (Date.now() - queueStartedAt)) : 0; - const queueTimeoutPromise = - remainingWaitMs > 0 - ? new Promise((_, 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((_, reject) => { - const onAbort = () => { - const reason = signal.reason; - // 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); - } 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); - } - 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(). - } - }; - if (signal.aborted) { - onAbort(); + const { promise: abortPromise, reject: rejectAbort } = Promise.withResolvers(); + const onAbort = () => { + const reason = signal.reason; + // Preserve native Error reasons (including AbortController's + // read-only DOMException) instead of mutating or wrapping them. + if (reason instanceof Error) { + rejectAbort(reason); return; } + 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; + } + rejectAbort(err); + }; + if (signal.aborted) { + onAbort(); + } else { abortListener = onAbort; signal.addEventListener("abort", abortListener, { once: true }); - }); + } try { - const races: Promise[] = [scheduled, abortPromise]; - if (queueTimeoutPromise) races.push(queueTimeoutPromise); - return await Promise.race(races); + return await Promise.race([limiter.schedule(scheduleOpts, fn), abortPromise]); } finally { if (abortListener) { signal.removeEventListener("abort", abortListener); } } } else { - return await (queueTimeoutPromise - ? Promise.race([scheduled, queueTimeoutPromise]) - : scheduled); + return await limiter.schedule(scheduleOpts, fn); } } catch (err) { - 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; status?: number }; - rateLimitErr.code = "RATE_LIMIT_UPSTREAM_429"; - rateLimitErr.status = 429; - throw rateLimitErr; + // Only Bottleneck-owned failures are rewritten. Application code can throw + // the same text and must retain its original identity and semantics. + if ( + err instanceof Bottleneck.BottleneckError && + /^This job timed out after \d+ ms\.$/.test(err.message) + ) { + const key = getLimiterKey(provider, connectionId, model); + logRateLimit( + `⏰ [RATE-LIMIT] ${key} — limiter-managed execution expired after ${Math.ceil((executionExpirationMs || 0) / 1000)}s` + ); + throw markLocalRateLimitError( + new Error( + `Request exceeded OmniRoute's local rate-limit execution expiration ` + + `(legacy resilienceSettings.requestQueue.maxWaitMs=${executionExpirationMs}ms) for ` + + `${model ? `${provider}/${model}` : provider}. Bottleneck applies this deadline only ` + + `after dispatch; it does not bound queue wait and is not an upstream-generated timeout.`, + { cause: err } + ), + RATE_LIMIT_EXECUTION_TIMEOUT_CODE + ); } - // The watchdog's stop({ dropWaitingJobs: true }) wedge-recovery (above) rejects - // queued jobs with this exact message. Rewrite it the same way as the timeout - // case — a clear, OmniRoute-owned, classifiable error — so combo's transient-error - // handling (which already treats a 502 as retryable) falls back to the next target - // immediately instead of surfacing Bottleneck's internal wording. - if (err?.message === "rate-limit-watchdog-wedge-reset") { + + if ( + err instanceof Bottleneck.BottleneckError && + err.message === "rate-limit-watchdog-wedge-reset" + ) { + const cleanup = limiterWatchdog.getEviction(limiter); + if (!cleanup) throw err; + + let cleanupError: unknown; + try { + await cleanup; + } catch (error) { + cleanupError = error; + errorRateLimit("[RATE-LIMIT] Wedge cleanup failed:", error); + } + + const key = getLimiterKey(provider, connectionId, model); + logRateLimit(`↪️ [RATE-LIMIT] ${key} — surfacing local wedge; caller will not be replayed`); const wedgeErr = new Error( `Request dropped: the local rate-limit queue for ${model ? `${provider}/${model}` : provider} ` + - `was detected as wedged (stalled with nothing executing) and force-reset. This is OmniRoute's ` + - `own queue recovering, not an upstream error.`, + `was detected as wedged (stalled with nothing executing) and force-reset. OmniRoute does ` + + `not replay dropped work automatically; combo routing may fall back to another target.`, { cause: err } - ) as Error & { code?: string }; - wedgeErr.code = "RATE_LIMIT_QUEUE_WEDGED"; - throw wedgeErr; - } - if (err?.message === "rate-limit-queue-timeout") { - throw createQueueTimeoutError(provider, model, maxWaitMs); + ) as Error & { cleanupError?: unknown }; + if (cleanupError !== undefined) wedgeErr.cleanupError = cleanupError; + throw markLocalRateLimitError(wedgeErr, RATE_LIMIT_QUEUE_WEDGED_CODE); } throw err; } @@ -842,12 +686,21 @@ 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)` ); - 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"); + // 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); + limiterWatchdog.forget(limiter); + limiterLastUsed.delete(limiterKey); + preservedReplacementSettings.delete(limiterKey); + trackAsyncOperation(limiter.disconnect()); return; } @@ -856,41 +709,40 @@ export function updateFromHeaders(provider, connectionId, headers, status, model logRateLimit( `⚠️ [RATE-LIMIT] ${provider}:${connectionId.slice(0, 8)} — near capacity, slowing down` ); - trackAsyncOperation(applyLimiterSettings(limiter, { minTime: 200 })); + updateLimiterSettings(limiter, { + minTime: 200, // Add 200ms between requests + }); return; } // 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; - // Keep adaptive pacing from response headers, but do not mutate an RPM - // reservoir. RPM admission is enforced by the rolling lease gate. + // If remaining is low (< 10% of limit), set reservoir to throttle immediately if (!isNaN(remaining)) { if (remaining < limit * 0.1) { - rpmGate.learnHeaderWindow( - provider, - connectionId, - model, - remaining, - resetMs, - Date.now() + resetMs - ); + updates.reservoir = remaining; + updates.reservoirRefreshAmount = limit; + updates.reservoirRefreshInterval = 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; - rpmGate.clearLearnedHeaderWindow(provider, connectionId, model); + updates.reservoir = null; + updates.reservoirRefreshAmount = null; + updates.reservoirRefreshInterval = null; } } - trackAsyncOperation(applyLimiterSettings(limiter, updates)); + updateLimiterSettings(limiter, updates); // Persist learned limits (debounced) recordLearnedLimit( @@ -1003,6 +855,14 @@ export async function __flushLearnedLimitsForTests() { } } +export function __setLimiterFactoryForTests(factory: LimiterFactory): void { + limiterFactory = factory; +} + +export async function __runLimiterWatchdogForTests(now = Date.now()): Promise { + await limiterWatchdog.run(now); +} + export async function __resetRateLimitManagerForTests() { if (persistTimer) { clearTimeout(persistTimer); @@ -1019,11 +879,11 @@ export async function __resetRateLimitManagerForTests() { } limiters.clear(); enabledConnections.clear(); - connectionRateLimitOverrides.clear(); - rpmGate.reset(); initialized = false; - lastDispatchAt.clear(); limiterLastUsed.clear(); + preservedReplacementSettings.clear(); + limiterFactory = defaultLimiterFactory; + limiterWatchdog.reset(); shutdownHandlersRegistered = false; for (const key of Object.keys(learnedLimits)) { @@ -1094,7 +954,7 @@ async function loadPersistedLimits() { const limiter = limiters.get(key); if (limiter && limit > 0) { const inferredMinTime = minTime || Math.max(0, Math.floor(60000 / limit) - 10); - await applyLimiterSettings(limiter, { minTime: inferredMinTime }); + updateLimiterSettings(limiter, { minTime: inferredMinTime }); count++; } } @@ -1125,10 +985,15 @@ export function updateFromResponseBody(provider, connectionId, responseBody, sta const { retryAfterMs, reason } = parseRetryAfterFromBody(responseBody); if (retryAfterMs && retryAfterMs > 0) { - getLimiter(provider, connectionId, model); + const limiter = getLimiter(provider, connectionId, model); logRateLimit( `🚫 [RATE-LIMIT] ${provider}:${connectionId.slice(0, 8)} — body-parsed retry: ${Math.ceil(retryAfterMs / 1000)}s (${reason})` ); - rpmGate.block(provider, connectionId, model, retryAfterMs); + + updateLimiterSettings(limiter, { + reservoir: 0, + reservoirRefreshAmount: 60, + reservoirRefreshInterval: retryAfterMs, + }); } } diff --git a/open-sse/services/rateLimitManager/admission.ts b/open-sse/services/rateLimitManager/admission.ts index d9ace7bf13..af3192e0d5 100644 --- a/open-sse/services/rateLimitManager/admission.ts +++ b/open-sse/services/rateLimitManager/admission.ts @@ -13,8 +13,10 @@ * @module services/rateLimitManager/admission */ +import { markLocalRateLimitError, RATE_LIMIT_QUEUE_FULL_CODE } from "./errors"; + export interface QueueFullError extends Error { - code: "RATE_LIMIT_QUEUE_FULL"; + code: typeof RATE_LIMIT_QUEUE_FULL_CODE; status: 429; } @@ -36,13 +38,8 @@ export function checkQueueAdmission( `queued request(s), at or above the configured admission cap maxQueueDepth (${maxQueueDepth}) ` + `— this is OmniRoute's request queue (resilienceSettings.requestQueue.maxQueueDepth), not an ` + `upstream rejection. Raise it in Settings → Resilience if this is expected burst traffic.` - ) as Error & { code?: string; status?: number }; - err.code = "RATE_LIMIT_QUEUE_FULL"; - // chatCore's generic catch-all fallback (open-sse/handlers/chatCore.ts) maps a - // status-less error to HTTP 502 — which also risks tripping the whole-provider - // circuit breaker (PROVIDER_BREAKER_FAILURE_STATUSES includes 502) for what is a - // purely local, in-process admission decision. Tag 429 explicitly so it is read - // via `error.status` before that fallback kicks in. - err.status = 429; - return err as QueueFullError; + ); + // The public code/status remain useful to callers, while the WeakMap brand + // is the provenance signal used by health and routing decisions. + return markLocalRateLimitError(err, RATE_LIMIT_QUEUE_FULL_CODE) as QueueFullError; } diff --git a/open-sse/services/rateLimitManager/errors.ts b/open-sse/services/rateLimitManager/errors.ts new file mode 100644 index 0000000000..167f92c834 --- /dev/null +++ b/open-sse/services/rateLimitManager/errors.ts @@ -0,0 +1,94 @@ +export const RATE_LIMIT_EXECUTION_TIMEOUT_CODE = "RATE_LIMIT_EXECUTION_TIMEOUT"; +export const RATE_LIMIT_QUEUE_FULL_CODE = "RATE_LIMIT_QUEUE_FULL"; +export const RATE_LIMIT_QUEUE_WEDGED_CODE = "RATE_LIMIT_QUEUE_WEDGED"; +export const LEGACY_RATE_LIMIT_QUEUE_TIMEOUT_CODE = "RATE_LIMIT_QUEUE_TIMEOUT"; + +export type LocalRateLimitErrorCode = + | typeof RATE_LIMIT_EXECUTION_TIMEOUT_CODE + | typeof RATE_LIMIT_QUEUE_FULL_CODE + | typeof RATE_LIMIT_QUEUE_WEDGED_CODE; + +export type TrustedLocalRateLimitErrorCode = + LocalRateLimitErrorCode | typeof LEGACY_RATE_LIMIT_QUEUE_TIMEOUT_CODE; + +export interface TrustedLocalRateLimitFailure { + code: TrustedLocalRateLimitErrorCode; + status: 429 | 503 | 504; +} + +const localRateLimitErrors = new WeakMap(); +const localRateLimitResponses = new WeakMap(); + +function getStatusForCode(code: TrustedLocalRateLimitErrorCode): 429 | 503 | 504 { + switch (code) { + case RATE_LIMIT_QUEUE_FULL_CODE: + return 429; + case RATE_LIMIT_EXECUTION_TIMEOUT_CODE: + return 504; + case RATE_LIMIT_QUEUE_WEDGED_CODE: + case LEGACY_RATE_LIMIT_QUEUE_TIMEOUT_CODE: + return 503; + } +} + +/** + * Brand an error created by OmniRoute's local limiter. The WeakMap identity, + * not the public code string, is the trusted provenance signal. + */ +export function markLocalRateLimitError( + error: T, + code: TrustedLocalRateLimitErrorCode +): T & { code: TrustedLocalRateLimitErrorCode; status: 429 | 503 | 504 } { + const failure = Object.freeze({ code, status: getStatusForCode(code) }); + localRateLimitErrors.set(error, failure); + const branded = error as T & { + code: TrustedLocalRateLimitErrorCode; + status: 429 | 503 | 504; + }; + branded.code = failure.code; + branded.status = failure.status; + return branded; +} + +export function getTrustedLocalRateLimitError(error: unknown): TrustedLocalRateLimitFailure | null { + if (!error || (typeof error !== "object" && typeof error !== "function")) return null; + return localRateLimitErrors.get(error as object) ?? null; +} + +/** + * Return the public fields for a trusted local failure without its low-level + * Bottleneck cause, which must remain server-side diagnostic context. + */ +export function getClientSafeLocalRateLimitError( + error: unknown +): (TrustedLocalRateLimitFailure & { message: string }) | null { + const failure = getTrustedLocalRateLimitError(error); + if (!failure) return null; + return { + ...failure, + message: error instanceof Error ? error.message : "Local rate-limit failure", + }; +} + +/** + * Transfer trusted local provenance from a branded error to its generated + * internal Response. Provider-controlled bodies and headers cannot set this. + */ +export function markTrustedLocalRateLimitResponse(response: Response, error: unknown): Response { + const failure = getTrustedLocalRateLimitError(error); + if (failure) localRateLimitResponses.set(response, failure); + return response; +} + +export function getTrustedLocalRateLimitResponse( + response: Response +): TrustedLocalRateLimitFailure | null { + return localRateLimitResponses.get(response) ?? null; +} + +/** Preserve trusted provenance when an internal response wrapper must allocate. */ +export function inheritTrustedLocalRateLimitResponse(source: Response, target: Response): Response { + const failure = localRateLimitResponses.get(source); + if (failure) localRateLimitResponses.set(target, failure); + return target; +} diff --git a/open-sse/services/rateLimitManager/wedgeWatchdog.ts b/open-sse/services/rateLimitManager/wedgeWatchdog.ts new file mode 100644 index 0000000000..624a413880 --- /dev/null +++ b/open-sse/services/rateLimitManager/wedgeWatchdog.ts @@ -0,0 +1,210 @@ +import Bottleneck from "bottleneck"; + +export const WATCHDOG_INTERVAL_MS = 30_000; + +const INACTIVE_LIMITER_MS = 10 * 60 * 1000; +const IDLE_CAPACITY_WEDGE_GRACE_MS = 10_000; + +interface IdleCapacitySnapshot { + lastProgress: number; + reservoir: number | null; +} + +interface LimiterWedgeWatchdogDependencies { + limiters: Map; + limiterLastUsed: Map; + limiterEffectiveSettings: WeakMap; + preservedReplacementSettings: Map; + trackBackground: (promise: Promise) => void; + log: (...args: unknown[]) => void; + warn: (...args: unknown[]) => void; +} + +/** + * Detects a Bottleneck queue that has remained idle despite immediately usable + * capacity. State is keyed by limiter identity so late events from an evicted + * instance cannot mutate the replacement's progress record. + */ +export class LimiterWedgeWatchdog { + private queueProgressAt = new WeakMap(); + private evictions = new WeakMap>(); + private currentRun: Promise | null = null; + + constructor(private readonly dependencies: LimiterWedgeWatchdogDependencies) {} + + noteQueued(key: string, limiter: Bottleneck): void { + if (this.dependencies.limiters.get(key) !== limiter) return; + if (!this.queueProgressAt.has(limiter)) this.queueProgressAt.set(limiter, Date.now()); + } + + noteProgress(key: string, limiter: Bottleneck): void { + if (this.dependencies.limiters.get(key) !== limiter) return; + if (limiter.counts().QUEUED > 0) { + this.queueProgressAt.set(limiter, Date.now()); + } else { + this.queueProgressAt.delete(limiter); + } + } + + forget(limiter: Bottleneck): void { + this.queueProgressAt.delete(limiter); + } + + getEviction(limiter: Bottleneck): Promise | undefined { + return this.evictions.get(limiter); + } + + run(now = Date.now()): Promise { + if (this.currentRun) return this.currentRun; + const run = this.tick(now); + this.currentRun = run; + void run.then( + () => { + if (this.currentRun === run) this.currentRun = null; + }, + () => { + if (this.currentRun === run) this.currentRun = null; + } + ); + return run; + } + + reset(): void { + this.queueProgressAt = new WeakMap(); + this.evictions = new WeakMap(); + this.currentRun = null; + } + + private async tick(now: number): Promise { + const { limiters, limiterLastUsed, log, trackBackground, warn } = this.dependencies; + + for (const [key, limiter] of Array.from(limiters)) { + const lastUsed = limiterLastUsed.get(key) ?? 0; + if (now - lastUsed <= INACTIVE_LIMITER_MS) continue; + + const counts = limiter.counts(); + if (counts.QUEUED > 0 || counts.RUNNING > 0 || counts.EXECUTING > 0) continue; + + limiters.delete(key); + this.queueProgressAt.delete(limiter); + limiterLastUsed.delete(key); + log( + `[RATE-LIMIT] Evicting idle limiter: ${key} ` + + `(inactive for ${Math.round((now - lastUsed) / 1000)}s)` + ); + trackBackground(limiter.disconnect()); + } + + for (const [key, limiter] of Array.from(limiters)) { + const snapshot = await this.getStableIdleCapacity(key, limiter, now); + if (!snapshot) continue; + + const counts = limiter.counts(); + const cleanup = this.evict(key, limiter, snapshot); + if (!cleanup) continue; + + warn( + `[RATE-LIMIT] WEDGED: ${key} queued=${counts.QUEUED} running=0 executing=0 ` + + `stalled=${now - snapshot.lastProgress}ms with idle capacity — force-resetting` + ); + await cleanup; + } + } + + private async getStableIdleCapacity( + key: string, + limiter: Bottleneck, + now: number + ): Promise { + const before = limiter.counts(); + if (before.QUEUED === 0) { + this.queueProgressAt.delete(limiter); + return null; + } + if (before.RUNNING > 0 || before.EXECUTING > 0) return null; + + const lastProgress = this.queueProgressAt.get(limiter); + if (lastProgress === undefined) { + this.queueProgressAt.set(limiter, now); + return null; + } + if (now - lastProgress < IDLE_CAPACITY_WEDGE_GRACE_MS) return null; + + let canRunNow: boolean; + let reservoir: number | null; + try { + // Every job this manager submits has Bottleneck's default weight of 1. + // check(1) is an eligibility query for exactly that shape, not a generic + // query about an arbitrary weighted queue head. + canRunNow = await limiter.check(1); + if (!canRunNow) return null; + reservoir = await limiter.currentReservoir(); + } catch { + return null; + } + if (this.dependencies.limiters.get(key) !== limiter) return null; + + const after = limiter.counts(); + if ( + after.QUEUED === 0 || + after.RUNNING > 0 || + after.EXECUTING > 0 || + this.queueProgressAt.get(limiter) !== lastProgress + ) { + return null; + } + return { lastProgress, reservoir }; + } + + private evict( + key: string, + limiter: Bottleneck, + snapshot: IdleCapacitySnapshot + ): Promise | null { + const { limiterEffectiveSettings, limiterLastUsed, limiters, preservedReplacementSettings } = + this.dependencies; + if (limiters.get(key) !== limiter) return null; + + const counts = limiter.counts(); + if ( + counts.QUEUED === 0 || + counts.RUNNING > 0 || + counts.EXECUTING > 0 || + this.queueProgressAt.get(limiter) !== snapshot.lastProgress + ) { + return null; + } + + const effectiveSettings = limiterEffectiveSettings.get(limiter) ?? {}; + preservedReplacementSettings.set(key, { + ...effectiveSettings, + id: key, + // Carry consumed capacity forward. Restarting the refresh interval from + // replacement creation is conservative and cannot grant an early burst. + reservoir: snapshot.reservoir, + }); + limiters.delete(key); + this.queueProgressAt.delete(limiter); + limiterLastUsed.delete(key); + + // Register this Promise before stop() runs. Every dropped caller awaits the + // same cleanup and is surfaced exactly once; none is replayed automatically. + const stopped = Promise.resolve().then(() => + limiter.stop({ + dropWaitingJobs: true, + dropErrorMessage: "rate-limit-watchdog-wedge-reset", + }) + ); + const cleanup = stopped + .then( + () => limiter.disconnect(), + async (stopError: unknown) => { + await limiter.disconnect(); + throw stopError; + } + ) + .then(() => true); + this.evictions.set(limiter, cleanup); + return cleanup; + } +} diff --git a/src/lib/api/modelTestRunner.ts b/src/lib/api/modelTestRunner.ts index e3041e7c1f..bb2aeb8dba 100644 --- a/src/lib/api/modelTestRunner.ts +++ b/src/lib/api/modelTestRunner.ts @@ -17,6 +17,7 @@ import { isDailyQuotaExhausted, } from "@omniroute/open-sse/services/accountFallback"; import { looksLikeQuotaExhausted } from "@/shared/utils/classify429"; +import { getTrustedLocalRateLimitError } from "@omniroute/open-sse/services/rateLimitManager/errors"; const INTERNAL_ORIGIN = "http://omniroute.internal"; export const DEFAULT_MODEL_TEST_TIMEOUT_MS = 30_000; @@ -484,12 +485,14 @@ export async function runSingleModelTest( rateLimited: true, }; } + const localRateLimitFailure = getTrustedLocalRateLimitError(error); return { modelId: fullModelStr, - status: "error", + status: localRateLimitFailure?.status === 429 ? "rate_limited" : "error", latencyMs, - httpStatus: 500, + httpStatus: localRateLimitFailure?.status ?? 500, error: getErrorMessage(error), + ...(localRateLimitFailure?.status === 429 ? { rateLimited: true } : {}), }; } let latencyMs = Date.now() - startTime; diff --git a/src/lib/resilience/settings/types.ts b/src/lib/resilience/settings/types.ts index 8e2ca28723..0bc16d48b2 100644 --- a/src/lib/resilience/settings/types.ts +++ b/src/lib/resilience/settings/types.ts @@ -16,6 +16,10 @@ export interface RequestQueueSettings { requestsPerMinute: number; minTimeBetweenRequestsMs: number; concurrentRequests: number; + /** + * Legacy persisted key used as Bottleneck's post-dispatch execution + * expiration. It does not bound time spent in Bottleneck's QUEUED state. + */ maxWaitMs: number; /** * Issue #6593: opt-in admission cap on the local rate-limit queue. When the diff --git a/src/sse/handlers/chatHelpers.ts b/src/sse/handlers/chatHelpers.ts index 835c0d58e6..ddfdaee039 100644 --- a/src/sse/handlers/chatHelpers.ts +++ b/src/sse/handlers/chatHelpers.ts @@ -17,6 +17,7 @@ import { providerCircuitOpenResponse, unavailableResponse, } from "@omniroute/open-sse/utils/error.ts"; +import { inheritTrustedLocalRateLimitResponse } from "@omniroute/open-sse/services/rateLimitManager/errors.ts"; import { HTTP_STATUS } from "@omniroute/open-sse/config/constants.ts"; import { runWithProxyContext, @@ -898,7 +899,7 @@ export function withSessionHeader(response: Response, sessionId: string | null): headers: response.headers, }); cloned.headers.set("X-OmniRoute-Session-Id", sessionId); - return cloned; + return inheritTrustedLocalRateLimitResponse(response, cloned); } } @@ -915,7 +916,7 @@ export function withCorrelationId(response: Response, correlationId: string | nu headers: response.headers, }); cloned.headers.set("X-Correlation-Id", correlationId); - return cloned; + return inheritTrustedLocalRateLimitResponse(response, cloned); } } @@ -960,6 +961,6 @@ export function withSelectedConnectionHeader( headers: response.headers, }); cloned.headers.set("X-OmniRoute-Selected-Connection-Id", connectionId); - return cloned; + return inheritTrustedLocalRateLimitResponse(response, cloned); } } diff --git a/src/sse/handlers/chatPredicates.ts b/src/sse/handlers/chatPredicates.ts index 2f8c8e5bd4..cbff12d506 100644 --- a/src/sse/handlers/chatPredicates.ts +++ b/src/sse/handlers/chatPredicates.ts @@ -1,5 +1,6 @@ import { isLocalStreamLifecycleError } from "../../shared/utils/circuitBreaker"; import { isRequestScopedUpstreamFailure } from "./comboFailureLogging"; +import { getTrustedLocalRateLimitResponse } from "@omniroute/open-sse/services/rateLimitManager/errors"; export const PROVIDER_BREAKER_FAILURE_STATUSES = new Set([408, 500, 502, 503, 504]); @@ -12,7 +13,13 @@ export function isProviderBreakerFailureStatus(status: number): boolean { // otherwise a client abort (502 default, error='request_signal_aborted') trips the // provider-wide breaker. Pure predicate, unit-testable without the full request path. export function shouldTripProviderBreakerForResult( - result: { status: number; errorCode?: string | null; errorType?: string | null; error?: unknown }, + result: { + status: number; + response?: Response; + errorCode?: string | null; + errorType?: string | null; + error?: unknown; + }, isCombo: boolean, forceLiveComboTest: boolean ): boolean { @@ -20,6 +27,7 @@ export function shouldTripProviderBreakerForResult( !forceLiveComboTest && !isCombo && !isRequestScopedUpstreamFailure({ code: result.errorCode, type: result.errorType }) && + !(result.response && getTrustedLocalRateLimitResponse(result.response)) && !isLocalStreamLifecycleError(result.error) && PROVIDER_BREAKER_FAILURE_STATUSES.has(Number(result.status)) ); diff --git a/stryker.conf.json b/stryker.conf.json index 7a0cc3dfcc..b512eedb28 100644 --- a/stryker.conf.json +++ b/stryker.conf.json @@ -280,6 +280,8 @@ "tests/unit/quota-streaming-consumption-usd.test.ts", "tests/unit/qwen-web-content-array-serialization.test.ts", "tests/unit/rate-limit-enhanced.test.ts", + "tests/unit/rate-limit-execution-timeout-message-4165.test.ts", + "tests/unit/rate-limit-local-error-classification.test.ts", "tests/unit/rate-limit-manager.test.ts", "tests/unit/rate-limit-queue-timeout-lockout.test.ts", "tests/unit/repro-7503-no-choices.test.ts", diff --git a/tests/unit/model-test-runner.test.ts b/tests/unit/model-test-runner.test.ts index 4ca0bc9472..bca51db207 100644 --- a/tests/unit/model-test-runner.test.ts +++ b/tests/unit/model-test-runner.test.ts @@ -10,6 +10,13 @@ import { resolveModelTestTimeoutMs, classifyTestErrorQuota, } from "@/lib/api/modelTestRunner.ts"; +import Bottleneck from "bottleneck"; +import * as rateLimitManager from "@omniroute/open-sse/services/rateLimitManager.ts"; +import { + markLocalRateLimitError, + RATE_LIMIT_EXECUTION_TIMEOUT_CODE, + RATE_LIMIT_QUEUE_WEDGED_CODE, +} from "@omniroute/open-sse/services/rateLimitManager/errors.ts"; // --------------------------------------------------------------------------- // parseRetryAfterHeader — Retry-After is either delta-seconds or an HTTP-date. @@ -375,3 +382,46 @@ test("classifyTestErrorQuota: daily-quota wins over credits-exhausted (isTransie assert.equal(result.isQuota, true); assert.equal(result.isTransient, true); }); +test("runSingleModelTest preserves trusted local limiter HTTP statuses", async () => { + const connection = await createProviderConnection({ + provider: "openai", + authType: "apikey", + name: "model-test-local-limiter-errors", + apiKey: "sk-model-test-local-limiter-errors", + isActive: true, + testStatus: "active", + }); + + try { + for (const [code, status] of [ + [RATE_LIMIT_QUEUE_WEDGED_CODE, 503], + [RATE_LIMIT_EXECUTION_TIMEOUT_CODE, 504], + ] as const) { + await rateLimitManager.__resetRateLimitManagerForTests(); + rateLimitManager.enableRateLimitProtection(connection.id); + rateLimitManager.__setLimiterFactoryForTests((options) => { + const limiter = new Bottleneck(options); + Object.defineProperty(limiter, "schedule", { + configurable: true, + value: async () => { + throw markLocalRateLimitError(new Error(`trusted ${code}`), code); + }, + }); + return limiter; + }); + + const result = await runSingleModelTest({ + providerId: "openai", + modelId: "gpt-4o", + connectionId: connection.id, + timeoutMs: 5_000, + }); + + assert.equal(result.status, "error"); + assert.equal(result.httpStatus, status); + assert.equal(result.error, `Error: trusted ${code}`); + } + } finally { + await rateLimitManager.__resetRateLimitManagerForTests(); + } +}); diff --git a/tests/unit/rate-limit-execution-timeout-message-4165.test.ts b/tests/unit/rate-limit-execution-timeout-message-4165.test.ts new file mode 100644 index 0000000000..3cae26ebbd --- /dev/null +++ b/tests/unit/rate-limit-execution-timeout-message-4165.test.ts @@ -0,0 +1,129 @@ +/** + * #4165 — classify Bottleneck's execution expiration accurately. + * + * OmniRoute passes the legacy `requestQueue.maxWaitMs` value to Bottleneck as + * the job `expiration`. Bottleneck starts that timer only after a job leaves + * QUEUED, so it bounds limiter-managed execution and does not bound queue wait. + * + * The raw Bottleneck message (`This job timed out after ms.`) still needs an + * OmniRoute-owned code and message so it cannot masquerade as an upstream- + * generated timeout. The original error remains available as `.cause`. + */ +import test from "node:test"; +import assert from "node:assert/strict"; +import fs from "node:fs"; +import os from "node:os"; +import path from "node:path"; + +const TEST_DATA_DIR = fs.mkdtempSync(path.join(os.tmpdir(), "omniroute-rl-execution-timeout-")); +process.env.DATA_DIR = TEST_DATA_DIR; + +// Dynamic imports are required because DATA_DIR must be set before DB modules evaluate. +const core = await import("../../src/lib/db/core.ts"); +const resilienceSettings = await import("../../src/lib/resilience/settings.ts"); +const rateLimitManager = await import("../../open-sse/services/rateLimitManager.ts"); +const { getClientSafeLocalRateLimitError, getTrustedLocalRateLimitError } = + await import("../../open-sse/services/rateLimitManager/errors.ts"); +const { formatProviderError } = await import("../../open-sse/utils/error.ts"); + +// This contract test deliberately drives Bottleneck's real expiration timer. +function wait(ms: number) { + const { promise, resolve } = Promise.withResolvers(); + setTimeout(resolve, ms); + return promise; +} + +test.afterEach(async () => { + await rateLimitManager.__resetRateLimitManagerForTests(); +}); + +test.after(() => { + core.resetDbInstance(); + fs.rmSync(TEST_DATA_DIR, { recursive: true, force: true }); +}); + +// Drive a real Bottleneck execution expiration with a function that outlives it. +async function triggerExecutionExpiration() { + await rateLimitManager.applyRequestQueueSettings({ + ...resilienceSettings.DEFAULT_RESILIENCE_SETTINGS.requestQueue, + autoEnableApiKeyProviders: false, + concurrentRequests: 1, + requestsPerMinute: 100000, + minTimeBetweenRequestsMs: 0, + maxWaitMs: 40, + }); + rateLimitManager.enableRateLimitProtection("conn-execution-timeout"); + + return rateLimitManager.withRateLimit("openai", "conn-execution-timeout", "gpt-4o", async () => { + await wait(400); // > maxWaitMs (40ms) → Bottleneck fails the job + return "should-not-reach"; + }); +} + +test("#4165 execution expiration is local and accurately named", async () => { + let caught: (Error & { code?: string; cause?: { message?: string } }) | undefined; + try { + await triggerExecutionExpiration(); + assert.fail("expected the limiter-managed execution to expire"); + } catch (err) { + caught = err as Error & { code?: string; cause?: { message?: string } }; + } + assert.ok(caught, "an error should have been thrown"); + + assert.equal( + caught.code, + "RATE_LIMIT_EXECUTION_TIMEOUT", + "error must carry the local execution-expiration code" + ); + + assert.match(caught.message, /execution expiration/i); + assert.match(caught.message, /does not bound queue wait/i); + assert.match( + caught.message, + /not an upstream-generated timeout/i, + "message should explicitly disclaim an upstream-generated 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/); + + assert.deepEqual(getTrustedLocalRateLimitError(caught), { + code: "RATE_LIMIT_EXECUTION_TIMEOUT", + status: 504, + }); + const safeError = getClientSafeLocalRateLimitError(caught); + assert.ok(safeError); + const clientMessage = formatProviderError(safeError, "openai", "gpt-4o", 504); + assert.match(clientMessage, /execution expiration/i); + assert.doesNotMatch( + clientMessage, + /This job timed out/, + "client and call-log formatting must not append the retained Bottleneck cause" + ); +}); + +test("#4165 a job that completes within the execution expiration is unaffected", async () => { + await rateLimitManager.applyRequestQueueSettings({ + ...resilienceSettings.DEFAULT_RESILIENCE_SETTINGS.requestQueue, + autoEnableApiKeyProviders: false, + concurrentRequests: 1, + requestsPerMinute: 100000, + minTimeBetweenRequestsMs: 0, + maxWaitMs: 5000, + }); + rateLimitManager.enableRateLimitProtection("conn-fast"); + + const result = await rateLimitManager.withRateLimit( + "openai", + "conn-fast", + "gpt-4o", + async () => "ok" + ); + assert.equal(result, "ok"); +}); diff --git a/tests/unit/rate-limit-local-error-classification.test.ts b/tests/unit/rate-limit-local-error-classification.test.ts new file mode 100644 index 0000000000..aad8ebb57d --- /dev/null +++ b/tests/unit/rate-limit-local-error-classification.test.ts @@ -0,0 +1,355 @@ +import test from "node:test"; +import assert from "node:assert/strict"; +import fs from "node:fs"; +import os from "node:os"; +import path from "node:path"; + +const TEST_DATA_DIR = fs.mkdtempSync(path.join(os.tmpdir(), "omniroute-rl-local-errors-")); +process.env.DATA_DIR = TEST_DATA_DIR; +process.env.API_KEY_SECRET = "test-rate-limit-local-error-secret"; + +// Dynamic imports are required because DATA_DIR must be set before DB modules evaluate. +const core = await import("../../src/lib/db/core.ts"); +const providersDb = await import("../../src/lib/db/providers.ts"); +const { handleComboChat } = await import("../../open-sse/services/combo.ts"); +const { + isComboRequestScopedFailure, + isRequestScopedUpstreamFailure, + shouldRecordProviderBreakerFailure, + shouldSkipConnDisable, +} = await import("../../open-sse/services/combo/comboPredicates.ts"); +const { + LEGACY_RATE_LIMIT_QUEUE_TIMEOUT_CODE, + RATE_LIMIT_EXECUTION_TIMEOUT_CODE, + RATE_LIMIT_QUEUE_WEDGED_CODE, + getTrustedLocalRateLimitError, + getTrustedLocalRateLimitResponse, + inheritTrustedLocalRateLimitResponse, + markLocalRateLimitError, + markTrustedLocalRateLimitResponse, +} = await import("../../open-sse/services/rateLimitManager/errors.ts"); +const accountFallback = await import("../../open-sse/services/accountFallback.ts"); +const providerCooldown = await import("../../open-sse/services/providerCooldownTracker.ts"); +const rateLimitSemaphore = await import("../../open-sse/services/rateLimitSemaphore.ts"); +const { createStreamingErrorResult } = + await import("../../open-sse/handlers/chatCore/streamErrorResult.ts"); +const { shouldTripProviderBreakerForResult } = + await import("../../src/sse/handlers/chatPredicates.ts"); + +const LOCAL_ERROR_MESSAGE = "OmniRoute repaired a local limiter queue"; + +function createLocalLimiterSseResponse(connectionId: string, code = RATE_LIMIT_QUEUE_WEDGED_CODE) { + const error = markLocalRateLimitError(new Error(LOCAL_ERROR_MESSAGE), code); + const { response } = createStreamingErrorResult( + getTrustedLocalRateLimitError(error)?.status ?? 503, + LOCAL_ERROR_MESSAGE, + code, + "rate_limit_queue_wedged" + ); + response.headers.set("X-OmniRoute-Selected-Connection-Id", connectionId); + return markTrustedLocalRateLimitResponse(response, error); +} + +function createUpstreamCollisionResponse(connectionId: string) { + return new Response( + JSON.stringify({ + error: { + message: "Provider emitted a colliding code", + code: RATE_LIMIT_QUEUE_WEDGED_CODE, + type: "rate_limit_queue_wedged", + }, + }), + { + status: 503, + headers: { + "content-type": "application/json", + "X-OmniRoute-Selected-Connection-Id": connectionId, + }, + } + ); +} + +function createSuccessResponse(connectionId: string) { + return new Response(JSON.stringify({ choices: [{ message: { content: "fallback ok" } }] }), { + status: 200, + headers: { + "content-type": "application/json", + "X-OmniRoute-Selected-Connection-Id": connectionId, + }, + }); +} + +const log = { info() {}, warn() {}, error() {}, debug() {} }; +const settings = { + modelLockout: { + enabled: true, + errorCodes: [503], + baseCooldownMs: 3_000, + maxCooldownMs: 5_000, + maxBackoffSteps: 10, + useExponentialBackoff: true, + }, +}; + +test.afterEach(() => { + accountFallback.clearAllModelLockouts(); + accountFallback.clearProviderFailure("openai"); + providerCooldown.clearCooldownState(); + rateLimitSemaphore.resetAll(); + core.resetDbInstance(); + fs.rmSync(TEST_DATA_DIR, { recursive: true, force: true }); + fs.mkdirSync(TEST_DATA_DIR, { recursive: true }); +}); + +test.after(() => { + accountFallback.clearAllModelLockouts(); + accountFallback.clearProviderFailure("openai"); + providerCooldown.clearCooldownState(); + rateLimitSemaphore.resetAll(); + core.resetDbInstance(); + fs.rmSync(TEST_DATA_DIR, { recursive: true, force: true }); +}); + +test("trusted provenance, not a public code string, classifies local limiter failures", () => { + const executionError = markLocalRateLimitError( + new Error("local execution expiration"), + RATE_LIMIT_EXECUTION_TIMEOUT_CODE + ); + const localResponse = markTrustedLocalRateLimitResponse( + new Response("local", { status: 504 }), + executionError + ); + const collisionResponse = createUpstreamCollisionResponse("collision-conn"); + + assert.deepEqual(getTrustedLocalRateLimitError(executionError), { + code: RATE_LIMIT_EXECUTION_TIMEOUT_CODE, + status: 504, + }); + assert.equal(getTrustedLocalRateLimitResponse(localResponse)?.status, 504); + const wrappedResponse = inheritTrustedLocalRateLimitResponse( + localResponse, + new Response("wrapped local", { status: 504 }) + ); + assert.equal(getTrustedLocalRateLimitResponse(wrappedResponse)?.status, 504); + assert.equal( + isRequestScopedUpstreamFailure({ code: RATE_LIMIT_EXECUTION_TIMEOUT_CODE }), + false, + "an upstream-controlled code string must not establish local provenance" + ); + assert.equal( + isComboRequestScopedFailure(localResponse, "local execution expiration", { + code: RATE_LIMIT_EXECUTION_TIMEOUT_CODE, + }), + true + ); + assert.equal( + isComboRequestScopedFailure(collisionResponse, "provider collision", { + code: RATE_LIMIT_QUEUE_WEDGED_CODE, + }), + false + ); + assert.equal( + shouldTripProviderBreakerForResult( + { + status: 504, + response: localResponse, + errorCode: RATE_LIMIT_EXECUTION_TIMEOUT_CODE, + }, + false, + false + ), + false + ); + assert.equal( + shouldTripProviderBreakerForResult( + { + status: 503, + response: collisionResponse, + errorCode: RATE_LIMIT_QUEUE_WEDGED_CODE, + }, + false, + false + ), + true, + "an untrusted upstream collision must remain a provider-health failure" + ); + assert.equal( + shouldSkipConnDisable( + { + status: 504, + response: localResponse, + errorCode: RATE_LIMIT_EXECUTION_TIMEOUT_CODE, + }, + false, + false, + "openai" + ), + true + ); + assert.equal( + shouldSkipConnDisable( + { + status: 503, + response: collisionResponse, + errorCode: RATE_LIMIT_QUEUE_WEDGED_CODE, + }, + false, + false, + "openai" + ), + false + ); + assert.equal( + shouldRecordProviderBreakerFailure({ + isStreamReadinessFailure: false, + status: 504, + sameProviderNext: false, + skipProviderBreaker: false, + requestScopedFailure: true, + error: executionError, + isProxyUnreachable: false, + }), + false + ); +}); + +test("legacy queue-timeout compatibility also requires trusted local provenance", () => { + const untrusted = new Response("legacy collision", { status: 503 }); + const legacyError = markLocalRateLimitError( + new Error("legacy local timeout"), + LEGACY_RATE_LIMIT_QUEUE_TIMEOUT_CODE + ); + const trusted = markTrustedLocalRateLimitResponse( + new Response("legacy local timeout", { status: 503 }), + legacyError + ); + + assert.equal( + isComboRequestScopedFailure(untrusted, "legacy collision", { + code: LEGACY_RATE_LIMIT_QUEUE_TIMEOUT_CODE, + }), + false + ); + assert.equal( + isComboRequestScopedFailure(trusted, "legacy local timeout", { + code: LEGACY_RATE_LIMIT_QUEUE_TIMEOUT_CODE, + }), + true + ); +}); + +for (const strategy of ["priority", "round-robin"] as const) { + test(`${strategy} fallback preserves all health state for a trusted local SSE failure`, async () => { + const connection = await providersDb.createProviderConnection({ + provider: "openai", + authType: "apikey", + name: `local-wedge-${strategy}`, + apiKey: `sk-local-wedge-${strategy}`, + isActive: true, + testStatus: "active", + rateLimitedUntil: null, + backoffLevel: 0, + providerSpecificData: {}, + }); + const models = [ + { + kind: "model", + model: "openai/gpt-local-first", + connectionId: connection.id, + }, + { + kind: "model", + model: "openai/gpt-local-second", + connectionId: connection.id, + }, + ]; + const calls: string[] = []; + + const result = await handleComboChat({ + body: {}, + combo: { + name: `local-wedge-${strategy}-combo`, + strategy, + models, + config: { + maxRetries: 1, + retryDelayMs: 0, + fallbackDelayMs: 0, + maxConcurrency: 1, + }, + }, + handleSingleModel: async (_body, modelStr) => { + calls.push(modelStr); + return calls.length === 1 + ? createLocalLimiterSseResponse(connection.id) + : createSuccessResponse(connection.id); + }, + isModelAvailable: async () => true, + log, + settings, + allCombos: null, + }); + + assert.equal(result.status, 200, `attempted targets: ${calls.join(", ")}`); + assert.deepEqual(calls, ["openai/gpt-local-first", "openai/gpt-local-second"]); + assert.equal(accountFallback.isModelLocked("openai", connection.id, "gpt-local-first"), false); + assert.equal( + accountFallback.getProviderBreakerState("openai")?.failureCount ?? 0, + 0, + "local failure must not increment the provider breaker" + ); + assert.equal( + providerCooldown.isProviderInCooldown("openai", connection.id), + false, + "local failure must not enter provider cooldown" + ); + const semaphoreStates = Object.values(rateLimitSemaphore.getStats()); + assert.equal( + semaphoreStates.some((state) => state.rateLimitedUntil !== null), + false, + "local failure must not cool a round-robin semaphore" + ); + const storedConnection = await providersDb.getProviderConnectionById(connection.id); + assert.equal(storedConnection?.testStatus, "active"); + assert.equal(storedConnection?.rateLimitedUntil ?? null, null); + }); +} + +test("an untrusted upstream code collision retains ordinary health penalties", async () => { + const connection = await providersDb.createProviderConnection({ + provider: "openai", + authType: "apikey", + name: "upstream-local-code-collision", + apiKey: "sk-upstream-local-code-collision", + isActive: true, + testStatus: "active", + providerSpecificData: {}, + }); + + const result = await handleComboChat({ + body: {}, + combo: { + name: "upstream-local-code-collision-combo", + strategy: "priority", + models: [ + { + kind: "model", + model: "openai/gpt-collision", + connectionId: connection.id, + }, + ], + config: { maxRetries: 1, retryDelayMs: 0, fallbackDelayMs: 0 }, + }, + handleSingleModel: async () => createUpstreamCollisionResponse(connection.id), + isModelAvailable: async () => true, + log, + settings, + allCombos: null, + }); + + assert.equal(result.status, 503); + assert.ok( + (accountFallback.getProviderBreakerState("openai")?.failureCount ?? 0) >= 1, + "the upstream 503 must remain eligible for provider-breaker accounting" + ); +}); diff --git a/tests/unit/rate-limit-manager.test.ts b/tests/unit/rate-limit-manager.test.ts index 1d2ec645e1..b1d411e62a 100644 --- a/tests/unit/rate-limit-manager.test.ts +++ b/tests/unit/rate-limit-manager.test.ts @@ -1,6 +1,5 @@ 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"; @@ -8,19 +7,86 @@ import path from "node:path"; const TEST_DATA_DIR = fs.mkdtempSync(path.join(os.tmpdir(), "omniroute-rate-limit-manager-")); process.env.DATA_DIR = TEST_DATA_DIR; +// Dynamic imports are required because DATA_DIR must be set before DB modules evaluate. const core = await import("../../src/lib/db/core.ts"); 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 rateLimitErrors = await import("../../open-sse/services/rateLimitManager/errors.ts"); const accountFallback = await import("../../open-sse/services/accountFallback.ts"); +const Bottleneck = (await import("bottleneck")).default; +// These integration-style tests exercise real Bottleneck timer/event behavior. function wait(ms) { - return new Promise((resolve) => setTimeout(resolve, ms)); + const { promise, resolve } = Promise.withResolvers(); + setTimeout(resolve, ms); + return promise; +} + +// A real deadline is intentional: these tests drive real Bottleneck queues, and +// a broken cleanup path otherwise leaves Node's test process pending forever. +async function settleWithin( + promise: Promise, + message: string, + timeoutMs = 2_000 +): Promise { + let timeout: NodeJS.Timeout; + const deadline = new Promise((_, reject) => { + timeout = setTimeout(() => reject(new Error(message)), timeoutMs); + }); + try { + return await Promise.race([promise, deadline]); + } finally { + clearTimeout(timeout); + } +} + +type TestBottleneck = InstanceType & { + _drainAll: (...args: unknown[]) => Promise; +}; + +/** + * Fault injection for the observed Bottleneck failure mode: jobs enter the real + * Bottleneck queue, but its internal drain loop stops making progress. Keep the + * private mutation in this one helper so the tests otherwise exercise public + * manager and Bottleneck behavior. + */ +function injectDrainWedge(limiter: InstanceType): TestBottleneck { + const wedged = limiter as TestBottleneck; + wedged._drainAll = () => Promise.resolve(null); + return wedged; +} + +async function waitForCondition( + condition: () => boolean | Promise, + message: string +): Promise { + const deadline = Date.now() + 1_000; + while (!(await condition())) { + if (Date.now() >= deadline) throw new Error(message); + await wait(5); + } +} + +async function expectWedgeError(promise: Promise): Promise { + await assert.rejects( + settleWithin(promise, "stranded limiter caller did not reject after wedge recovery"), + (error: Error & { code?: string }) => { + assert.equal(error.code, "RATE_LIMIT_QUEUE_WEDGED"); + assert.deepEqual(rateLimitErrors.getTrustedLocalRateLimitError(error), { + code: "RATE_LIMIT_QUEUE_WEDGED", + status: 503, + }); + return true; + } + ); } async function flushBackgroundWork() { await wait(50); - await new Promise((resolve) => setImmediate(resolve)); + const { promise, resolve } = Promise.withResolvers(); + setImmediate(resolve); + await promise; } async function resetStorage() { @@ -60,296 +126,561 @@ test("rate limit manager bypasses disabled connections and exposes inactive stat assert.deepEqual(rateLimitManager.getAllRateLimitStatus(), {}); }); -test("queue expiry does not invoke the provider after a late dispatch", async () => { +test("idle-capacity watchdog honors grace, cleans up in order, and rejects the stranded caller", async () => { await rateLimitManager.applyRequestQueueSettings({ ...resilienceSettings.DEFAULT_RESILIENCE_SETTINGS.requestQueue, autoEnableApiKeyProviders: false, - maxWaitMs: 100, + maxWaitMs: 240_000, requestsPerMinute: 0, concurrentRequests: 1, minTimeBetweenRequestsMs: 0, maxQueueDepth: 0, }); - rateLimitManager.enableRateLimitProtection("queue-expiry-conn"); - let resolveFirstStarted: () => void = () => undefined; - const firstStarted = new Promise((resolve) => { - resolveFirstStarted = resolve; + const cleanupEvents: string[] = []; + let limitersCreated = 0; + rateLimitManager.__setLimiterFactoryForTests((options) => { + const limiter = new Bottleneck(options); + limitersCreated++; + if (limitersCreated === 1) { + injectDrainWedge(limiter); + const originalStop = limiter.stop.bind(limiter); + const originalDisconnect = limiter.disconnect.bind(limiter); + limiter.stop = async (stopOptions) => { + cleanupEvents.push("stop:start"); + await originalStop(stopOptions); + cleanupEvents.push("stop:done"); + }; + limiter.disconnect = async (flush) => { + cleanupEvents.push("disconnect"); + await originalDisconnect(flush); + }; + } + return limiter; }); - const first = rateLimitManager.withRateLimit( + + rateLimitManager.enableRateLimitProtection("idle-capacity-conn"); + let executions = 0; + const pending = rateLimitManager.withRateLimit( "openai", - "queue-expiry-conn", + "idle-capacity-conn", "gpt-4o", async () => { - resolveFirstStarted(); - await wait(300); - return "first"; + executions++; + return "must-not-run"; } ); - await firstStarted; - 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" + await waitForCondition( + () => rateLimitManager.getRateLimitStatus("openai", "idle-capacity-conn").queued === 1, + "the injected drain failure never established a real queued job" + ); + const queuedObservedAt = Date.now(); + + await settleWithin( + rateLimitManager.__runLimiterWatchdogForTests(queuedObservedAt + 9_000), + "watchdog grace-period scan did not finish" + ); + assert.equal( + rateLimitManager.getRateLimitStatus("openai", "idle-capacity-conn").queued, + 1, + "the queue must survive before the 10s stability grace" ); - await first; - await wait(50); - assert.equal(secondCalls, 0, "a queue-expired job must not invoke the provider later"); + await settleWithin( + rateLimitManager.__runLimiterWatchdogForTests(queuedObservedAt + 11_000), + "watchdog wedge cleanup did not finish" + ); + await expectWedgeError(pending); + + assert.equal(executions, 0, "watchdog recovery must never replay application work"); + assert.equal(limitersCreated, 1, "dropped callers must not create a replacement limiter"); + assert.deepEqual(cleanupEvents, ["stop:start", "stop:done", "disconnect"]); }); -test("queue expiry does not drop other queued jobs", async () => { +test("wedge eviction rejects every queued caller and preserves learned state for future traffic", async () => { await rateLimitManager.applyRequestQueueSettings({ ...resilienceSettings.DEFAULT_RESILIENCE_SETTINGS.requestQueue, autoEnableApiKeyProviders: false, - maxWaitMs: 500, - requestsPerMinute: 0, - concurrentRequests: 1, + maxWaitMs: 240_000, + requestsPerMinute: 60, + concurrentRequests: 6, minTimeBetweenRequestsMs: 0, maxQueueDepth: 0, }); - rateLimitManager.enableRateLimitProtection("queue-peer-conn"); - let resolveFirstExecuting: () => void = () => undefined; - const firstExecuting = new Promise((resolve) => { - resolveFirstExecuting = resolve; - }); - let releaseFirst: () => void = () => undefined; - const first = rateLimitManager.withRateLimit("openai", "queue-peer-conn", null, async () => { - resolveFirstExecuting(); - await new Promise((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, + const createdOptions: Bottleneck.ConstructorOptions[] = []; + const createdLimiters: InstanceType[] = []; + rateLimitManager.__setLimiterFactoryForTests((options) => { + createdOptions.push({ ...options }); + const limiter = new Bottleneck(options); + createdLimiters.push(limiter); + if (createdLimiters.length === 1) injectDrainWedge(limiter); + return limiter; }); - 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((resolve) => { - resolveFirstExecuting = resolve; - }); - let releaseFirst: () => void = () => undefined; - const firstStarted = new Promise((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"; + const connectionId = "learned-state-conn"; rateLimitManager.enableRateLimitProtection(connectionId); - let resolveFirstExecuting: () => void = () => undefined; - const firstExecuting = new Promise((resolve) => { - resolveFirstExecuting = resolve; - }); - let releaseFirst: () => void = () => undefined; - const first = rateLimitManager.withRateLimit("test-provider", connectionId, null, async () => { - resolveFirstExecuting(); - await new Promise((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", + rateLimitManager.updateFromHeaders( + "openai", connectionId, - null, - async () => { - abortedCalls++; + { + "x-ratelimit-limit-requests": "100", + "x-ratelimit-remaining-requests": "1", + "x-ratelimit-reset-requests": "60s", }, - controller.signal + 200, + "gpt-4o" + ); + await waitForCondition( + async () => + (await rateLimitManager.__getLimiterStateForTests("openai", connectionId, "gpt-4o")) + ?.reservoir === 1, + "the learned reservoir was not applied" ); - 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); + let executions = 0; + const stranded = Array.from({ length: 3 }, () => + rateLimitManager.withRateLimit("openai", connectionId, "gpt-4o", async () => { + executions++; + return "must-not-run"; + }) + ); + await waitForCondition( + () => rateLimitManager.getRateLimitStatus("openai", connectionId).queued === 3, + "all callers did not enter the wedged queue" + ); + + await settleWithin( + rateLimitManager.__runLimiterWatchdogForTests(Date.now() + 11_000), + "multi-caller wedge cleanup did not finish" + ); + const settled = await settleWithin( + Promise.allSettled(stranded), + "not every stranded limiter caller settled" + ); + assert.equal(executions, 0); + assert.equal(createdLimiters.length, 1, "wedge recovery must not retry any dropped caller"); + for (const result of settled) { + assert.equal(result.status, "rejected"); + assert.equal((result as PromiseRejectedResult).reason.code, "RATE_LIMIT_QUEUE_WEDGED"); + } + + assert.equal( + await rateLimitManager.withRateLimit("openai", connectionId, "gpt-4o", async () => "future"), + "future" + ); + assert.equal(createdLimiters.length, 2, "future traffic should create one replacement limiter"); + assert.equal(createdOptions[1].reservoir, 1, "replacement must retain the remaining reservoir"); + assert.equal(createdOptions[1].minTime, 590, "replacement must retain learned request spacing"); + + const queuedAfterPreservedPermit = rateLimitManager.withRateLimit( + "openai", + connectionId, + "gpt-4o", + async () => "after-refill" + ); + await waitForCondition( + () => rateLimitManager.getRateLimitStatus("openai", connectionId).queued === 1, + "the preserved reservoir should allow only one request" + ); + await createdLimiters[1].incrementReservoir(1); + assert.equal(await queuedAfterPreservedPermit, "after-refill"); }); -test("dispatched provider failures retain their RPM lease", async () => { +test("global settings changed after eviction replace stale pending configuration", async () => { + const connection = await providersDb.createProviderConnection({ + provider: "openai", + authType: "apikey", + name: "wedge-global-settings", + apiKey: "sk-wedge-global-settings", + isActive: true, + rateLimitProtection: true, + }); await rateLimitManager.applyRequestQueueSettings({ ...resilienceSettings.DEFAULT_RESILIENCE_SETTINGS.requestQueue, autoEnableApiKeyProviders: false, - maxWaitMs: 500, + maxWaitMs: 240_000, + requestsPerMinute: 60, + concurrentRequests: 6, + minTimeBetweenRequestsMs: 0, + }); + + const createdOptions: Bottleneck.ConstructorOptions[] = []; + rateLimitManager.__setLimiterFactoryForTests((options) => { + createdOptions.push({ ...options }); + const limiter = new Bottleneck(options); + if (createdOptions.length === 1) injectDrainWedge(limiter); + return limiter; + }); + + const pending = rateLimitManager.withRateLimit( + "openai", + connection.id, + "gpt-4o", + async () => "must-not-run" + ); + await waitForCondition( + () => rateLimitManager.getRateLimitStatus("openai", connection.id).queued === 1, + "global-settings caller did not enter the wedged queue" + ); + await settleWithin( + rateLimitManager.__runLimiterWatchdogForTests(Date.now() + 11_000), + "global-settings wedge cleanup did not finish" + ); + await expectWedgeError(pending); + + await rateLimitManager.applyRequestQueueSettings({ + ...resilienceSettings.DEFAULT_RESILIENCE_SETTINGS.requestQueue, + autoEnableApiKeyProviders: false, + maxWaitMs: 240_000, + requestsPerMinute: 8, + concurrentRequests: 3, + minTimeBetweenRequestsMs: 31, + }); + assert.equal( + await rateLimitManager.withRateLimit( + "openai", + connection.id, + "gpt-4o", + async () => "new-policy" + ), + "new-policy" + ); + assert.equal(createdOptions[1].reservoir, 8); + assert.equal(createdOptions[1].maxConcurrent, 3); + assert.equal(createdOptions[1].minTime, 31); +}); + +test("connection overrides changed after eviction replace stale pending configuration", async () => { + await rateLimitManager.applyRequestQueueSettings({ + ...resilienceSettings.DEFAULT_RESILIENCE_SETTINGS.requestQueue, + autoEnableApiKeyProviders: false, + maxWaitMs: 240_000, + requestsPerMinute: 60, + concurrentRequests: 6, + minTimeBetweenRequestsMs: 0, + }); + const createdOptions: Bottleneck.ConstructorOptions[] = []; + rateLimitManager.__setLimiterFactoryForTests((options) => { + createdOptions.push({ ...options }); + const limiter = new Bottleneck(options); + if (createdOptions.length === 1) injectDrainWedge(limiter); + return limiter; + }); + + const connectionId = "wedge-override-conn"; + rateLimitManager.enableRateLimitProtection(connectionId); + const pending = rateLimitManager.withRateLimit( + "openai", + connectionId, + "gpt-4o", + async () => "must-not-run" + ); + await waitForCondition( + () => rateLimitManager.getRateLimitStatus("openai", connectionId).queued === 1, + "override caller did not enter the wedged queue" + ); + await settleWithin( + rateLimitManager.__runLimiterWatchdogForTests(Date.now() + 11_000), + "override wedge cleanup did not finish" + ); + await expectWedgeError(pending); + + rateLimitManager.refreshConnectionRateLimits(connectionId, { + rpm: 7, + maxConcurrent: 2, + minTime: 25, + }); + assert.equal( + await rateLimitManager.withRateLimit( + "openai", + connectionId, + "gpt-4o", + async () => "new-override" + ), + "new-override" + ); + assert.equal(createdOptions[1].reservoir, 7); + assert.equal(createdOptions[1].maxConcurrent, 2); + assert.equal(createdOptions[1].minTime, 25); +}); + +test("disable and re-enable discard learned state preserved by an earlier wedge", async () => { + await rateLimitManager.applyRequestQueueSettings({ + ...resilienceSettings.DEFAULT_RESILIENCE_SETTINGS.requestQueue, + autoEnableApiKeyProviders: false, + maxWaitMs: 240_000, + requestsPerMinute: 60, + concurrentRequests: 6, + minTimeBetweenRequestsMs: 0, + }); + const createdOptions: Bottleneck.ConstructorOptions[] = []; + rateLimitManager.__setLimiterFactoryForTests((options) => { + createdOptions.push({ ...options }); + const limiter = new Bottleneck(options); + if (createdOptions.length === 1) injectDrainWedge(limiter); + return limiter; + }); + + const connectionId = "wedge-reenabled-conn"; + rateLimitManager.enableRateLimitProtection(connectionId); + rateLimitManager.updateFromHeaders( + "openai", + connectionId, + { + "x-ratelimit-limit-requests": "100", + "x-ratelimit-remaining-requests": "1", + "x-ratelimit-reset-requests": "60s", + }, + 200, + "gpt-4o" + ); + await waitForCondition( + async () => + (await rateLimitManager.__getLimiterStateForTests("openai", connectionId, "gpt-4o")) + ?.reservoir === 1, + "learned reservoir was not applied before disable/re-enable" + ); + + const pending = rateLimitManager.withRateLimit( + "openai", + connectionId, + "gpt-4o", + async () => "must-not-run" + ); + await waitForCondition( + () => rateLimitManager.getRateLimitStatus("openai", connectionId).queued === 1, + "disable/re-enable caller did not enter the wedged queue" + ); + await settleWithin( + rateLimitManager.__runLimiterWatchdogForTests(Date.now() + 11_000), + "disable/re-enable wedge cleanup did not finish" + ); + await expectWedgeError(pending); + + rateLimitManager.disableRateLimitProtection(connectionId); + rateLimitManager.enableRateLimitProtection(connectionId); + assert.equal( + await rateLimitManager.withRateLimit("openai", connectionId, "gpt-4o", async () => "reenabled"), + "reenabled" + ); + assert.equal(createdOptions[1].reservoir, 60); + assert.equal(createdOptions[1].minTime, 0); +}); + +test("idle-capacity watchdog preserves a legitimate exhausted-reservoir queue", async () => { + await rateLimitManager.applyRequestQueueSettings({ + ...resilienceSettings.DEFAULT_RESILIENCE_SETTINGS.requestQueue, + autoEnableApiKeyProviders: false, + maxWaitMs: 240_000, requestsPerMinute: 1, - concurrentRequests: 10, + concurrentRequests: 1, 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 limiter: TestBottleneck | null = null; + rateLimitManager.__setLimiterFactoryForTests((options) => { + limiter = new Bottleneck(options) as TestBottleneck; + return limiter; + }); + rateLimitManager.enableRateLimitProtection("zero-reservoir-conn"); + assert.equal( + await rateLimitManager.withRateLimit( + "openai", + "zero-reservoir-conn", + "gpt-4o", + async () => "first" + ), + "first" ); - let secondCalls = 0; - await assert.rejects( - rateLimitManager.withRateLimit("anthropic", "failed-dispatch-b", null, async () => { - secondCalls++; - }), - (error: { code?: string }) => error.code === "RATE_LIMIT_QUEUE_TIMEOUT" + const pending = rateLimitManager.withRateLimit( + "openai", + "zero-reservoir-conn", + "gpt-4o", + async () => "after-refresh" ); - assert.equal(secondCalls, 0, "a dispatched failure still counts against the RPM window"); + await waitForCondition( + () => rateLimitManager.getRateLimitStatus("openai", "zero-reservoir-conn").queued === 1, + "the exhausted reservoir did not queue the follow-up" + ); + + await settleWithin( + rateLimitManager.__runLimiterWatchdogForTests(Date.now() + 150_000), + "zero-reservoir watchdog scan did not finish" + ); + assert.equal( + rateLimitManager.getRateLimitStatus("openai", "zero-reservoir-conn").queued, + 1, + "a zero-reservoir wait must survive regardless of elapsed time" + ); + + assert.ok(limiter); + await limiter.incrementReservoir(1); + assert.equal(await pending, "after-refresh"); +}); + +test("idle-capacity watchdog preserves a real Bottleneck minTime wait", async () => { + await rateLimitManager.applyRequestQueueSettings({ + ...resilienceSettings.DEFAULT_RESILIENCE_SETTINGS.requestQueue, + autoEnableApiKeyProviders: false, + maxWaitMs: 240_000, + requestsPerMinute: 0, + concurrentRequests: 1, + minTimeBetweenRequestsMs: 100, + maxQueueDepth: 0, + }); + + rateLimitManager.enableRateLimitProtection("min-time-conn"); + await rateLimitManager.withRateLimit("openai", "min-time-conn", "gpt-4o", async () => "first"); + const pending = rateLimitManager.withRateLimit( + "openai", + "min-time-conn", + "gpt-4o", + async () => "after-min-time" + ); + await waitForCondition( + () => rateLimitManager.getRateLimitStatus("openai", "min-time-conn").running === 1, + "Bottleneck did not place the minTime-delayed job in RUNNING" + ); + + await settleWithin( + rateLimitManager.__runLimiterWatchdogForTests(Date.now() + 150_000), + "minTime watchdog scan did not finish" + ); + assert.equal( + rateLimitManager.getRateLimitStatus("openai", "min-time-conn").running, + 1, + "a legitimate RUNNING minTime delay must not be evicted" + ); + assert.equal(await pending, "after-min-time"); +}); + +test("events from an evicted limiter cannot erase replacement queue progress", async () => { + await rateLimitManager.applyRequestQueueSettings({ + ...resilienceSettings.DEFAULT_RESILIENCE_SETTINGS.requestQueue, + autoEnableApiKeyProviders: false, + maxWaitMs: 240_000, + requestsPerMinute: 0, + concurrentRequests: 1, + minTimeBetweenRequestsMs: 0, + maxQueueDepth: 0, + }); + + const limiters: InstanceType[] = []; + rateLimitManager.__setLimiterFactoryForTests((options) => { + const limiter = new Bottleneck(options); + limiters.push(limiter); + if (limiters.length === 2) injectDrainWedge(limiter); + return limiter; + }); + + const connectionId = "stale-listener-conn"; + rateLimitManager.enableRateLimitProtection(connectionId); + const { promise: oldGate, resolve: releaseOld } = Promise.withResolvers(); + const oldExecuting = rateLimitManager.withRateLimit( + "openai", + connectionId, + "gpt-4o", + async () => { + await oldGate; + return "old-first"; + } + ); + await waitForCondition( + () => limiters[0]?.counts().EXECUTING === 1, + "the old limiter did not begin executing" + ); + const oldQueued = rateLimitManager.withRateLimit( + "openai", + connectionId, + "gpt-4o", + async () => "old-second" + ); + await waitForCondition( + () => limiters[0]?.counts().QUEUED === 1, + "the old limiter did not queue its second job" + ); + + rateLimitManager.refreshConnectionRateLimits(connectionId, {}); + const replacementPending = rateLimitManager.withRateLimit( + "openai", + connectionId, + "gpt-4o", + async () => "must-not-run" + ); + await waitForCondition( + () => rateLimitManager.getRateLimitStatus("openai", connectionId).queued === 1, + "the replacement limiter did not establish its queue" + ); + + releaseOld(); + assert.equal(await oldExecuting, "old-first"); + assert.equal(await oldQueued, "old-second"); + + await settleWithin( + rateLimitManager.__runLimiterWatchdogForTests(Date.now() + 11_000), + "stale-listener watchdog cleanup did not finish" + ); + await expectWedgeError(replacementPending); +}); + +test("watchdog ticks are serialized while an eligibility check is in flight", async () => { + const { promise: checkGate, resolve: releaseCheck } = Promise.withResolvers(); + let checks = 0; + rateLimitManager.__setLimiterFactoryForTests((options) => { + const limiter = injectDrainWedge(new Bottleneck(options)); + const originalCheck = limiter.check.bind(limiter); + limiter.check = async (weight) => { + checks++; + await checkGate; + return originalCheck(weight); + }; + return limiter; + }); + + const connectionId = "serialized-watchdog-conn"; + rateLimitManager.enableRateLimitProtection(connectionId); + const pending = rateLimitManager.withRateLimit( + "openai", + connectionId, + "gpt-4o", + async () => "must-not-run" + ); + await waitForCondition( + () => rateLimitManager.getRateLimitStatus("openai", connectionId).queued === 1, + "the serialized-watchdog fixture did not queue" + ); + + const now = Date.now() + 11_000; + const firstTick = rateLimitManager.__runLimiterWatchdogForTests(now); + const secondTick = rateLimitManager.__runLimiterWatchdogForTests(now); + await waitForCondition(() => checks === 1, "the first tick did not reach limiter.check()"); + releaseCheck(); + await settleWithin( + Promise.all([firstTick, secondTick]), + "serialized watchdog scans did not finish" + ); + await expectWedgeError(pending); + assert.equal(checks, 1, "overlapping watchdog calls must share one scan"); +}); + +test("application errors resembling Bottleneck failures remain untouched", async () => { + rateLimitManager.enableRateLimitProtection("lookalike-error-conn"); + for (const message of [ + "This job timed out after 240000 ms.", + "rate-limit-watchdog-wedge-reset", + ]) { + const applicationError = new Error(message); + await assert.rejects( + rateLimitManager.withRateLimit("openai", "lookalike-error-conn", "gpt-4o", async () => { + throw applicationError; + }), + (error) => error === applicationError + ); + } }); test("withRateLimit forwards AbortController DOMException without mutating it", async () => { @@ -462,130 +793,6 @@ 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( @@ -617,11 +824,6 @@ 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", @@ -645,20 +847,6 @@ 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", @@ -762,21 +950,18 @@ test("withRateLimit rejects cleanly when the caller aborts with the default DOME isActive: true, }); rateLimitManager.enableRateLimitProtection(String(connection.id)); - const controller = new AbortController(); - // Mirror how a real executor call behaves: it settles once the signal it - // was handed aborts, so this job doesn't dangle forever in Bottleneck once - // withRateLimit's own Promise.race settles via the abort path below. - const settlesOnAbort = (signal) => - new Promise((_resolve, reject) => { - signal.addEventListener("abort", () => reject(signal.reason), { once: true }); - }); - const pending = rateLimitManager.withRateLimit( "openai", String(connection.id), "gpt-4o", - () => settlesOnAbort(controller.signal), + () => { + const { promise, reject } = Promise.withResolvers(); + controller.signal.addEventListener("abort", () => reject(controller.signal.reason), { + once: true, + }); + return promise; + }, controller.signal ); diff --git a/tests/unit/rate-limit-queue-timeout-message-4165.test.ts b/tests/unit/rate-limit-queue-timeout-message-4165.test.ts deleted file mode 100644 index 685c01e67d..0000000000 --- a/tests/unit/rate-limit-queue-timeout-message-4165.test.ts +++ /dev/null @@ -1,148 +0,0 @@ -/** - * #4165 — surface a clear error when the request-queue (Bottleneck) drops a job. - * - * 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 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"; -import fs from "node:fs"; -import os from "node:os"; -import path from "node:path"; - -const TEST_DATA_DIR = fs.mkdtempSync(path.join(os.tmpdir(), "omniroute-rl-queue-timeout-")); -process.env.DATA_DIR = TEST_DATA_DIR; - -const core = await import("../../src/lib/db/core.ts"); -const resilienceSettings = await import("../../src/lib/resilience/settings.ts"); -const rateLimitManager = await import("../../open-sse/services/rateLimitManager.ts"); - -function wait(ms: number) { - return new Promise((resolve) => setTimeout(resolve, ms)); -} - -// Leave enough scheduling headroom for a loaded CI/devbox while keeping the -// executing callback longer than the queue-only budget. The actual queued-job -// case stays short because it controls dispatch deterministically. -const DISPATCHED_QUEUE_BUDGET_MS = 2_000; -const QUEUED_QUEUE_BUDGET_MS = 250; - -test.afterEach(async () => { - await rateLimitManager.__resetRateLimitManagerForTests(); -}); - -test.after(() => { - core.resetDbInstance(); - fs.rmSync(TEST_DATA_DIR, { recursive: true, force: true }); -}); - -// A dispatched provider call may run longer than maxWaitMs without being killed. -async function triggerQueueTimeout() { - await rateLimitManager.applyRequestQueueSettings({ - ...resilienceSettings.DEFAULT_RESILIENCE_SETTINGS.requestQueue, - autoEnableApiKeyProviders: false, - concurrentRequests: 1, - requestsPerMinute: 100000, - minTimeBetweenRequestsMs: 0, - maxWaitMs: DISPATCHED_QUEUE_BUDGET_MS, - }); - const connectionId = "conn-dispatched-timeout"; - rateLimitManager.enableRateLimitProtection(connectionId); - - let dispatched = false; - const result = await rateLimitManager.withRateLimit( - "test-provider", - connectionId, - null, - async () => { - dispatched = true; - await wait(DISPATCHED_QUEUE_BUDGET_MS + 250); - return "should-not-reach"; - } - ); - return { dispatched, result }; -} - -async function triggerQueuedTimeout() { - await rateLimitManager.applyRequestQueueSettings({ - ...resilienceSettings.DEFAULT_RESILIENCE_SETTINGS.requestQueue, - autoEnableApiKeyProviders: false, - concurrentRequests: 1, - requestsPerMinute: 0, - minTimeBetweenRequestsMs: 0, - maxWaitMs: QUEUED_QUEUE_BUDGET_MS, - }); - const connectionId = "conn-queued-timeout"; - rateLimitManager.enableRateLimitProtection(connectionId); - - let resolveFirstExecuting: () => void = () => undefined; - const firstExecuting = new Promise((resolve) => { - resolveFirstExecuting = resolve; - }); - let releaseFirst: () => void = () => undefined; - const first = rateLimitManager.withRateLimit("test-provider", connectionId, null, async () => { - resolveFirstExecuting(); - await new Promise((resolve) => { - releaseFirst = resolve; - }); - }); - await firstExecuting; - - let caught: unknown; - let queuedDispatched = false; - try { - await rateLimitManager.withRateLimit("test-provider", connectionId, null, async () => { - queuedDispatched = true; - return "should-not-dispatch"; - }); - assert.fail("expected the queued job to expire"); - } catch (error) { - caught = error; - } finally { - releaseFirst(); - await first; - } - return { caught, queuedDispatched }; -} - -test("#4165 a dispatched provider call is not killed by the queue budget", async () => { - const execution = await triggerQueueTimeout(); - assert.equal(execution.dispatched, true, "the callback must enter execution"); - assert.equal(execution.result, "should-not-reach"); -}); - -test("#4165 queue expiry surfaces a clear local error", async () => { - const result = await triggerQueuedTimeout(); - assert.ok(result.caught instanceof Error, "queue expiry must reject with an Error"); - assert.equal(result.queuedDispatched, false, "an expired queued callback must never dispatch"); - const caught = result.caught 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 () => { - await rateLimitManager.applyRequestQueueSettings({ - ...resilienceSettings.DEFAULT_RESILIENCE_SETTINGS.requestQueue, - autoEnableApiKeyProviders: false, - concurrentRequests: 1, - requestsPerMinute: 100000, - minTimeBetweenRequestsMs: 0, - maxWaitMs: 5000, - }); - rateLimitManager.enableRateLimitProtection("conn-fast"); - - const result = await rateLimitManager.withRateLimit( - "openai", - "conn-fast", - "gpt-4o", - async () => "ok" - ); - assert.equal(result, "ok"); -}); diff --git a/tests/unit/rate-limit-wedge-recovery.test.ts b/tests/unit/rate-limit-wedge-recovery.test.ts index cd70a47934..7f9bfe865e 100644 --- a/tests/unit/rate-limit-wedge-recovery.test.ts +++ b/tests/unit/rate-limit-wedge-recovery.test.ts @@ -101,23 +101,3 @@ test("stop({ dropWaitingJobs: true }) rejects a genuinely queued (nothing-runnin "expected Bottleneck to reject with our dropErrorMessage verbatim" ); }); - -test("watchdog wedge branch uses stop({ dropWaitingJobs: true }), not disconnect()", async () => { - const source = await import("node:fs/promises").then((fs) => - fs.readFile(new URL("../../open-sse/services/rateLimitManager.ts", import.meta.url), "utf8") - ); - - const wedgeBlockStart = source.indexOf("WEDGED:"); - assert.ok(wedgeBlockStart >= 0, "expected to find the WEDGED log line in rateLimitManager.ts"); - const wedgeBlock = source.slice(wedgeBlockStart, wedgeBlockStart + 1500); - - assert.ok( - wedgeBlock.includes("stop({ dropWaitingJobs: true"), - "wedge-recovery branch must call stop({ dropWaitingJobs: true }) so orphaned queued jobs reject " + - "promptly instead of hanging until the outer per-target timeout (live incident 1784465227489-a2cbc0)" - ); - assert.ok( - !/limiter\.disconnect\(\)/.test(wedgeBlock), - "wedge-recovery branch must not still call disconnect() — it doesn't reject queued jobs" - ); -}); diff --git a/tests/unit/ratelimit-admission-control-6593.test.ts b/tests/unit/ratelimit-admission-control-6593.test.ts index 6eacf52696..521d6c6561 100644 --- a/tests/unit/ratelimit-admission-control-6593.test.ts +++ b/tests/unit/ratelimit-admission-control-6593.test.ts @@ -19,6 +19,7 @@ import os from "node:os"; import path from "node:path"; import { checkQueueAdmission } from "../../open-sse/services/rateLimitManager/admission.ts"; +import { getTrustedLocalRateLimitError } from "../../open-sse/services/rateLimitManager/errors.ts"; const TEST_DATA_DIR = fs.mkdtempSync(path.join(os.tmpdir(), "omniroute-rl-admission-")); process.env.DATA_DIR = TEST_DATA_DIR; @@ -78,6 +79,10 @@ test("#6593 checkQueueAdmission: rejects with a typed error at/over the cap", () // also risks tripping the whole-provider circuit breaker for a purely local // admission decision. assert.equal(err?.status, 429); + assert.deepEqual(getTrustedLocalRateLimitError(err), { + code: "RATE_LIMIT_QUEUE_FULL", + status: 429, + }); assert.match(err?.message ?? "", /maxQueueDepth/); assert.match(err?.message ?? "", /openai\/gpt-4o/); @@ -106,16 +111,26 @@ test("#6593 withRateLimit: fast-fails once the queue is at the configured maxQue // Job 1 occupies the single concurrent slot. Poll (not a fixed sleep) until // Bottleneck has actually dispatched it, since QUEUED -> EXECUTING takes a // few event-loop ticks, not one. - const job1 = rateLimitManager.withRateLimit("openai", "conn-admission-cap", "gpt-4o", async () => { - await wait(150); - return "job1"; - }); + const job1 = rateLimitManager.withRateLimit( + "openai", + "conn-admission-cap", + "gpt-4o", + async () => { + await wait(150); + return "job1"; + } + ); await pollUntil(() => (status()?.executing ?? 0) + (status()?.running ?? 0) >= 1); // Job 2 has to wait behind job1 -> occupies the one allowed queue slot (QUEUED=1). - const job2 = rateLimitManager.withRateLimit("openai", "conn-admission-cap", "gpt-4o", async () => { - return "job2"; - }); + const job2 = rateLimitManager.withRateLimit( + "openai", + "conn-admission-cap", + "gpt-4o", + async () => { + return "job2"; + } + ); await pollUntil(() => (status()?.queued ?? 0) >= 1); // Job 3 arrives while QUEUED (1) is already at maxQueueDepth (1) -> fast-rejected. @@ -124,6 +139,10 @@ test("#6593 withRateLimit: fast-fails once the queue is at the configured maxQue (err: Error & { code?: string; status?: number }) => { assert.equal(err.code, "RATE_LIMIT_QUEUE_FULL"); assert.equal(err.status, 429); + assert.deepEqual(getTrustedLocalRateLimitError(err), { + code: "RATE_LIMIT_QUEUE_FULL", + status: 429, + }); assert.match(err.message, /maxQueueDepth/); return true; } @@ -161,10 +180,7 @@ test("#6593 withRateLimit: default maxQueueDepth=0 preserves unbounded-queue beh test("#6593 DEFAULT_REQUEST_QUEUE_MAX_WAIT_MS is 15s absent RATE_LIMIT_MAX_WAIT_MS", () => { assert.equal(process.env.RATE_LIMIT_MAX_WAIT_MS, undefined); assert.equal(resilienceSettings.DEFAULT_REQUEST_QUEUE_MAX_WAIT_MS, 15000); - assert.equal( - resilienceSettings.DEFAULT_RESILIENCE_SETTINGS.requestQueue.maxWaitMs, - 15000 - ); + assert.equal(resilienceSettings.DEFAULT_RESILIENCE_SETTINGS.requestQueue.maxWaitMs, 15000); }); test("#6593 DEFAULT_REQUEST_QUEUE_MAX_DEPTH defaults to 0 (disabled) absent an env override", () => {