diff --git a/.env.example b/.env.example index 0ca0ae476f..479c0fe3b8 100644 --- a/.env.example +++ b/.env.example @@ -1576,9 +1576,13 @@ APP_LOG_TO_FILE=true # Also configurable from Dashboard > Settings > Feature Flags. # OMNIROUTE_CONTROL_PLANE_PROXY_DIRECT_FALLBACK=false -# Rate limit maximum wait time before failing a request (ms). Default: 120000 (2 min) +# Rate limit maximum wait time before failing a request (ms). Default: 15000 (15s) # Used by: open-sse/services/rateLimitManager.ts -# RATE_LIMIT_MAX_WAIT_MS=120000 +# RATE_LIMIT_MAX_WAIT_MS=15000 + +# Rate limit queue admission cap: reject with 429 queue_full once this many requests +# are already queued (0 = disabled/unbounded, the default). Used by: open-sse/services/rateLimitManager.ts +# RATE_LIMIT_MAX_QUEUE_DEPTH=0 # Force the auto-enable rate limit safety net on/off regardless of the persisted # Dashboard setting. Used by: open-sse/services/rateLimitManager.ts. diff --git a/changelog.d/features/6593-ratelimit-admission-control.md b/changelog.d/features/6593-ratelimit-admission-control.md new file mode 100644 index 0000000000..de933fab8c --- /dev/null +++ b/changelog.d/features/6593-ratelimit-admission-control.md @@ -0,0 +1 @@ +- **feat(sse):** rate-limit request queue admission control — `resilienceSettings.requestQueue.maxQueueDepth` (default `0` = disabled, opt-in 0–100000) fast-rejects a request with a typed `RATE_LIMIT_QUEUE_FULL` error once the local per-provider+connection queue already holds `maxQueueDepth` requests, instead of growing the queue unboundedly; the factory default for `requestQueue.maxWaitMs` (how long a request may wait before being dropped) also fell from 120s to 15s so a saturated queue fails fast (#6593 — thanks @chirag127). diff --git a/docs/architecture/RESILIENCE_GUIDE.md b/docs/architecture/RESILIENCE_GUIDE.md index 70257372cf..799cdfdd71 100644 --- a/docs/architecture/RESILIENCE_GUIDE.md +++ b/docs/architecture/RESILIENCE_GUIDE.md @@ -217,6 +217,47 @@ Bounded by `comboCooldownWait` (`enabled`, `maxWaitMs` 5s, `maxAttempts` 2, --- +## 5. Request Queue Admission Control (v3.8.49 · issue #6593) + +**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). + +**`maxQueueDepth` — opt-in admission cap (new).** `resilienceSettings.requestQueue.maxQueueDepth` +bounds how many requests may sit queued (not yet dispatched) for one +provider+connection at once. When the queue already holds `maxQueueDepth` +requests, a new request is fast-rejected with a typed +`code: "RATE_LIMIT_QUEUE_FULL"` error **before** it ever reaches `limiter.schedule()` +— so the rejection is cheap and happens ahead of any downstream +prompt-compression / translation work for that request. Default `0` = +disabled, preserving the existing unbounded-queue behavior; bounded 0–100000. +Override via `RATE_LIMIT_MAX_QUEUE_DEPTH` (env) or +`resilienceSettings.requestQueue.maxQueueDepth` (dashboard/API patch). + +The admission check itself is a pure function +(`open-sse/services/rateLimitManager/admission.ts::checkQueueAdmission`) so +it is unit-testable without a real Bottleneck limiter. + +> The RFC that opened #6593 also proposed a `bypassCompressionOnRateLimit` +> flag. This repo's `open-sse/services/compression/` pipeline is +> prompt/context compression on the outbound LLM request (`chatCore.ts`, +> 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 +> 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 +> reordering risk. + +--- + ## Other Resilience Features - **18 routing strategies** (priority, weighted, round-robin, context-relay, fill-first, p2c, random, least-used, cost-optimized, reset-aware, reset-window, headroom, strict-random, auto, lkgp, context-optimized, fusion, pipeline) — see [AUTO-COMBO.md](../routing/AUTO-COMBO.md). diff --git a/docs/reference/ENVIRONMENT.md b/docs/reference/ENVIRONMENT.md index 8832e68d6a..9b39ccee2e 100644 --- a/docs/reference/ENVIRONMENT.md +++ b/docs/reference/ENVIRONMENT.md @@ -848,7 +848,8 @@ Anthropic-compatible provider instead. | `PROXY_AUTO_REMOVE` | `false` | `src/lib/proxyHealth/scheduler.ts` | Set `true` to let the scheduler auto-remove proxies after repeated consecutive failures. | | `PROXY_AUTO_REMOVE_AFTER` | `3` | `src/lib/proxyHealth/scheduler.ts` | Consecutive failures before the scheduler auto-removes a proxy (when `PROXY_AUTO_REMOVE=true`). | | `OMNIROUTE_CONTROL_PLANE_PROXY_DIRECT_FALLBACK` | `false` | `src/shared/constants/featureFlagDefinitions.ts` | Allow OAuth and provider validation flows to bypass a pinned proxy and connect directly when proxy reachability pre-checks fail. Effective precedence is Feature Flags DB override > env var > default. | -| `RATE_LIMIT_MAX_WAIT_MS` | `120000` (2 min) | `open-sse/services/rateLimitManager.ts` | Max time to wait on a 429 before failing the request. | +| `RATE_LIMIT_MAX_WAIT_MS` | `15000` (15s) | `open-sse/services/rateLimitManager.ts` | Max time to wait on a 429 before failing the request. | +| `RATE_LIMIT_MAX_QUEUE_DEPTH` | `0` (disabled) | `open-sse/services/rateLimitManager.ts` | Queue admission cap: reject with a 429 `queue_full` once this many requests are already queued. `0` = unbounded (default). | | `RATE_LIMIT_AUTO_ENABLE` | _(unset)_ | `open-sse/services/rateLimitManager.ts` | Force the auto-enable rate limit safety net on/off regardless of the persisted Dashboard setting. Accepts `true`/`1`/`on` to force on, `false`/`0`/`off` to force off. | | `PROVIDER_COOLDOWN_ENABLED` | _(unset → off)_ | `open-sse/services/providerCooldownTracker.ts` | Opt-in global cross-request provider/connection cooldown tracking. OFF by default (overlaps Connection Cooldown / Provider Circuit Breaker). Accepts `true`/`1`/`on` to enable. | | `PROVIDER_COOLDOWN_MIN_MS` | `5000` | `open-sse/services/providerCooldownTracker.ts` | Minimum cooldown (ms) before a failed provider/connection is retried. Scaled exponentially with consecutive failures. Only used when `PROVIDER_COOLDOWN_ENABLED`. | diff --git a/open-sse/services/rateLimitManager.ts b/open-sse/services/rateLimitManager.ts index 2a667028af..290678b503 100644 --- a/open-sse/services/rateLimitManager.ts +++ b/open-sse/services/rateLimitManager.ts @@ -24,6 +24,7 @@ import { parseResetTime, toPlainHeaders, } from "./rateLimitManager/headers"; +import { checkQueueAdmission } from "./rateLimitManager/admission"; interface LearnedLimitEntry { provider: string; @@ -547,6 +548,21 @@ export async function withRateLimit(provider, connectionId, model, fn, signal = const maxWaitMs = currentRequestQueueSettings.maxWaitMs; const scheduleOpts = maxWaitMs && maxWaitMs > 0 ? { expiration: maxWaitMs } : {}; + // Issue #6593: opt-in admission cap — fast-reject before Bottleneck's + // schedule() (and before any downstream compression/prompt work runs) when + // the queue is already at/over maxQueueDepth. Default 0 = disabled. + const admissionErr = checkQueueAdmission( + limiter.counts().QUEUED, + currentRequestQueueSettings.maxQueueDepth, + model ? `${provider}/${model}` : provider + ); + if (admissionErr) { + logRateLimit( + `🚧 [RATE-LIMIT] ${getLimiterKey(provider, connectionId, model)} — queue full, rejecting fast (maxQueueDepth=${currentRequestQueueSettings.maxQueueDepth})` + ); + throw admissionErr; + } + try { if (signal) { let abortListener: (() => void) | undefined; diff --git a/open-sse/services/rateLimitManager/admission.ts b/open-sse/services/rateLimitManager/admission.ts new file mode 100644 index 0000000000..d9ace7bf13 --- /dev/null +++ b/open-sse/services/rateLimitManager/admission.ts @@ -0,0 +1,48 @@ +/** + * rateLimitManager/admission — queue-depth admission check (pure). + * + * `maxQueueDepth` (RequestQueueSettings, issue #6593) is an opt-in admission + * cap on the local rate-limit queue: when set (>0), a request that would be + * queued behind `maxQueueDepth` already-queued jobs is fast-rejected before + * it ever reaches Bottleneck's `schedule()`, instead of growing the queue + * unboundedly. Default `0` = disabled, preserving today's behavior exactly. + * + * Extracted as a pure function (no Bottleneck/limiter dependency) so it is + * unit-testable without spinning up a real limiter. + * + * @module services/rateLimitManager/admission + */ + +export interface QueueFullError extends Error { + code: "RATE_LIMIT_QUEUE_FULL"; + status: 429; +} + +/** + * Returns a typed `RATE_LIMIT_QUEUE_FULL` error when `queuedCount` is at or + * above `maxQueueDepth`, or `null` when admission should proceed (cap + * disabled, i.e. `maxQueueDepth <= 0`, or the queue has room). + */ +export function checkQueueAdmission( + queuedCount: number, + maxQueueDepth: number, + identity: string +): QueueFullError | null { + if (!maxQueueDepth || maxQueueDepth <= 0) return null; + if (queuedCount < maxQueueDepth) return null; + + const err = new Error( + `Request rejected: the local rate-limit queue for ${identity} already holds ${queuedCount} ` + + `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; +} diff --git a/src/lib/resilience/settings.ts b/src/lib/resilience/settings.ts index f2f0a70b65..e5aad43d75 100644 --- a/src/lib/resilience/settings.ts +++ b/src/lib/resilience/settings.ts @@ -34,8 +34,15 @@ export type { } from "./settings/types"; export const DEFAULT_REQUEST_QUEUE_MAX_WAIT_MS = (() => { - const parsed = Number(process.env.RATE_LIMIT_MAX_WAIT_MS || "120000"); - return Number.isFinite(parsed) && parsed > 0 ? Math.trunc(parsed) : 120000; + const parsed = Number(process.env.RATE_LIMIT_MAX_WAIT_MS || "15000"); + return Number.isFinite(parsed) && parsed > 0 ? Math.trunc(parsed) : 15000; +})(); + +// Issue #6593: opt-in admission cap on the local rate-limit queue depth. +// Default 0 = disabled (unbounded queue, today's behavior unchanged). +export const DEFAULT_REQUEST_QUEUE_MAX_DEPTH = (() => { + const parsed = Number(process.env.RATE_LIMIT_MAX_QUEUE_DEPTH || "0"); + return Number.isFinite(parsed) && parsed >= 0 ? Math.trunc(parsed) : 0; })(); export const DEFAULT_RESILIENCE_SETTINGS: ResilienceSettings = { @@ -45,6 +52,7 @@ export const DEFAULT_RESILIENCE_SETTINGS: ResilienceSettings = { minTimeBetweenRequestsMs: DEFAULT_API_LIMITS.minTimeBetweenRequests, concurrentRequests: DEFAULT_API_LIMITS.concurrentRequests, maxWaitMs: DEFAULT_REQUEST_QUEUE_MAX_WAIT_MS, + maxQueueDepth: DEFAULT_REQUEST_QUEUE_MAX_DEPTH, }, connectionCooldown: { oauth: { @@ -172,6 +180,7 @@ function buildLegacyFallback(settings: JsonRecord): ResilienceSettings { { min: 1, max: 10_000 } ), maxWaitMs: DEFAULT_RESILIENCE_SETTINGS.requestQueue.maxWaitMs, + maxQueueDepth: DEFAULT_RESILIENCE_SETTINGS.requestQueue.maxQueueDepth, }, connectionCooldown: { oauth: normalizeLegacyConnectionCooldownProfile( diff --git a/src/lib/resilience/settings/normalize.ts b/src/lib/resilience/settings/normalize.ts index 9325873439..405777e9f8 100644 --- a/src/lib/resilience/settings/normalize.ts +++ b/src/lib/resilience/settings/normalize.ts @@ -107,6 +107,10 @@ export function normalizeRequestQueueSettings( min: 1, max: 24 * 60 * 60 * 1000, }); + const maxQueueDepth = toInteger(record.maxQueueDepth, fallback.maxQueueDepth, { + min: 0, + max: 100_000, + }); return { autoEnableApiKeyProviders: toBoolean( @@ -117,6 +121,7 @@ export function normalizeRequestQueueSettings( minTimeBetweenRequestsMs, concurrentRequests, maxWaitMs, + maxQueueDepth, }; } diff --git a/src/lib/resilience/settings/types.ts b/src/lib/resilience/settings/types.ts index 395f8dd698..61e2031906 100644 --- a/src/lib/resilience/settings/types.ts +++ b/src/lib/resilience/settings/types.ts @@ -17,6 +17,13 @@ export interface RequestQueueSettings { minTimeBetweenRequestsMs: number; concurrentRequests: number; maxWaitMs: number; + /** + * Issue #6593: opt-in admission cap on the local rate-limit queue. When the + * queue already holds `maxQueueDepth` requests, a new request is + * fast-rejected (429 `queue_full`) instead of joining the queue. Default 0 + * = disabled, preserving the unbounded-queue behavior. Bounded 0-100000. + */ + maxQueueDepth: number; } export interface ConnectionCooldownProfileSettings { diff --git a/src/shared/validation/schemas/settings.ts b/src/shared/validation/schemas/settings.ts index 9fdd8b6caa..8f115d7f28 100644 --- a/src/shared/validation/schemas/settings.ts +++ b/src/shared/validation/schemas/settings.ts @@ -41,6 +41,7 @@ export const requestQueueSettingsSchema = z minTimeBetweenRequestsMs: z.number().int().min(0).optional(), concurrentRequests: z.number().int().min(1).optional(), maxWaitMs: z.number().int().min(1).optional(), + maxQueueDepth: z.number().int().min(0).max(100_000).optional(), }) .strict(); diff --git a/tests/unit/limiter-lifecycle.test.ts b/tests/unit/limiter-lifecycle.test.ts index 7d71dbbabf..c70981d90c 100644 --- a/tests/unit/limiter-lifecycle.test.ts +++ b/tests/unit/limiter-lifecycle.test.ts @@ -227,6 +227,7 @@ test("request queue refresh treats zero limits as unbounded for existing limiter requestsPerMinute: 0, concurrentRequests: 0, minTimeBetweenRequestsMs: 0, + maxQueueDepth: 0, }); assert.equal( diff --git a/tests/unit/ratelimit-admission-control-6593.test.ts b/tests/unit/ratelimit-admission-control-6593.test.ts new file mode 100644 index 0000000000..05eefd0c2b --- /dev/null +++ b/tests/unit/ratelimit-admission-control-6593.test.ts @@ -0,0 +1,174 @@ +/** + * #6593 — rate-limit queue admission control (maxQueueDepth) + 15s default. + * + * RFC bundles 2 grounded changes to the local rate-limit request queue: + * 1. `RequestQueueSettings.maxQueueDepth` — opt-in admission cap (default 0 + * = disabled). When the queue already holds `maxQueueDepth` requests, a + * new request is fast-rejected with a typed `RATE_LIMIT_QUEUE_FULL` + * error instead of joining Bottleneck's queue. + * 2. `DEFAULT_REQUEST_QUEUE_MAX_WAIT_MS` lowered 120000 -> 15000. + * + * (Item #3 from the RFC, `bypassCompressionOnRateLimit`, has no matching + * code path in this repo's compression pipeline — see the plan-file — and is + * intentionally not implemented.) + */ +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"; + +import { checkQueueAdmission } from "../../open-sse/services/rateLimitManager/admission.ts"; + +const TEST_DATA_DIR = fs.mkdtempSync(path.join(os.tmpdir(), "omniroute-rl-admission-")); +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)); +} + +// Bottleneck moves a job from QUEUED -> RUNNING/EXECUTING across a few event +// loop ticks (not synchronously on schedule()), so a fixed sleep is flaky +// under load. Poll the live counts instead of guessing a wall-clock delay. +async function pollUntil( + predicate: () => boolean, + { timeoutMs = 2000, intervalMs = 5 } = {} +): Promise { + const deadline = Date.now() + timeoutMs; + while (!predicate()) { + if (Date.now() > deadline) { + throw new Error(`pollUntil: condition not met within ${timeoutMs}ms`); + } + await wait(intervalMs); + } +} + +test.afterEach(async () => { + await rateLimitManager.__resetRateLimitManagerForTests(); +}); + +test.after(() => { + core.resetDbInstance(); + fs.rmSync(TEST_DATA_DIR, { recursive: true, force: true }); +}); + +// --- Pure unit tests for the extracted admission check ------------------- + +test("#6593 checkQueueAdmission: disabled (maxQueueDepth<=0) always admits", () => { + assert.equal(checkQueueAdmission(0, 0, "openai"), null); + assert.equal(checkQueueAdmission(1000, 0, "openai"), null); + assert.equal(checkQueueAdmission(5, -1, "openai"), null); +}); + +test("#6593 checkQueueAdmission: admits while queued count is below the cap", () => { + assert.equal(checkQueueAdmission(0, 2, "openai"), null); + assert.equal(checkQueueAdmission(1, 2, "openai"), null); +}); + +test("#6593 checkQueueAdmission: rejects with a typed error at/over the cap", () => { + const err = checkQueueAdmission(2, 2, "openai/gpt-4o"); + assert.ok(err, "expected a queue_full error at the cap"); + assert.equal(err?.code, "RATE_LIMIT_QUEUE_FULL"); + // #7649: must carry an explicit HTTP status of 429 (not 502) — chatCore's + // generic fallback reads `error.status` and otherwise defaults to 502, which + // also risks tripping the whole-provider circuit breaker for a purely local + // admission decision. + assert.equal(err?.status, 429); + assert.match(err?.message ?? "", /maxQueueDepth/); + assert.match(err?.message ?? "", /openai\/gpt-4o/); + + const overErr = checkQueueAdmission(5, 2, "openai"); + assert.ok(overErr, "expected a queue_full error over the cap"); + assert.equal(overErr?.code, "RATE_LIMIT_QUEUE_FULL"); + assert.equal(overErr?.status, 429); +}); + +// --- Integration: wired into withRateLimit() ------------------------------ + +test("#6593 withRateLimit: fast-fails once the queue is at the configured maxQueueDepth", async () => { + await rateLimitManager.applyRequestQueueSettings({ + ...resilienceSettings.DEFAULT_RESILIENCE_SETTINGS.requestQueue, + autoEnableApiKeyProviders: false, + concurrentRequests: 1, + requestsPerMinute: 100000, + minTimeBetweenRequestsMs: 0, + maxWaitMs: 5000, + maxQueueDepth: 1, + }); + rateLimitManager.enableRateLimitProtection("conn-admission-cap"); + const statusKey = "openai:conn-admission-cap"; + const status = () => rateLimitManager.getAllRateLimitStatus()[statusKey]; + + // 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"; + }); + 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"; + }); + await pollUntil(() => (status()?.queued ?? 0) >= 1); + + // Job 3 arrives while QUEUED (1) is already at maxQueueDepth (1) -> fast-rejected. + await assert.rejects( + rateLimitManager.withRateLimit("openai", "conn-admission-cap", "gpt-4o", async () => "job3"), + (err: Error & { code?: string; status?: number }) => { + assert.equal(err.code, "RATE_LIMIT_QUEUE_FULL"); + assert.equal(err.status, 429); + assert.match(err.message, /maxQueueDepth/); + return true; + } + ); + + assert.equal(await job1, "job1"); + assert.equal(await job2, "job2"); +}); + +test("#6593 withRateLimit: default maxQueueDepth=0 preserves unbounded-queue behavior", async () => { + await rateLimitManager.applyRequestQueueSettings({ + ...resilienceSettings.DEFAULT_RESILIENCE_SETTINGS.requestQueue, + autoEnableApiKeyProviders: false, + concurrentRequests: 1, + requestsPerMinute: 100000, + minTimeBetweenRequestsMs: 0, + maxWaitMs: 5000, + maxQueueDepth: 0, + }); + rateLimitManager.enableRateLimitProtection("conn-unbounded"); + + const jobs = [1, 2, 3, 4].map((n) => + rateLimitManager.withRateLimit("openai", "conn-unbounded", "gpt-4o", async () => { + await wait(15); + return `job${n}`; + }) + ); + + const results = await Promise.all(jobs); + assert.deepEqual(results, ["job1", "job2", "job3", "job4"]); +}); + +// --- Default maxWaitMs lowered 120000 -> 15000 ---------------------------- + +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 + ); +}); + +test("#6593 DEFAULT_REQUEST_QUEUE_MAX_DEPTH defaults to 0 (disabled) absent an env override", () => { + assert.equal(process.env.RATE_LIMIT_MAX_QUEUE_DEPTH, undefined); + assert.equal(resilienceSettings.DEFAULT_REQUEST_QUEUE_MAX_DEPTH, 0); + assert.equal(resilienceSettings.DEFAULT_RESILIENCE_SETTINGS.requestQueue.maxQueueDepth, 0); +});