From dc2efa5e63b3799a4b014773ae23ed35429a83d1 Mon Sep 17 00:00:00 2001 From: Diego Rodrigues de Sa e Souza <8016841+diegosouzapw@users.noreply.github.com> Date: Sat, 4 Jul 2026 00:58:55 -0300 Subject: [PATCH] =?UTF-8?q?feat(resilience):=20throttle=20concurrent=20ups?= =?UTF-8?q?tream=20quota=20fetches=20=E2=80=94=20closes=20#6009=20(#6058)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Throttle concurrent upstream quota fetches (#6009). Integrated into release/v3.8.44. --- .env.example | 8 ++ CHANGELOG.md | 1 + docs/reference/ENVIRONMENT.md | 1 + open-sse/services/codexQuotaFetcher.ts | 8 ++ open-sse/services/quotaFetchThrottle.ts | 118 +++++++++++++++++++ tests/unit/quota-fetch-throttle-6009.test.ts | 84 +++++++++++++ 6 files changed, 220 insertions(+) create mode 100644 open-sse/services/quotaFetchThrottle.ts create mode 100644 tests/unit/quota-fetch-throttle-6009.test.ts diff --git a/.env.example b/.env.example index ba419dac43..fdfe5d0b0f 100644 --- a/.env.example +++ b/.env.example @@ -651,6 +651,14 @@ PROVIDER_LIMITS_SYNC_INTERVAL_MINUTES=70 # to opt out (restores fully concurrent fetches). Default: 1500 PROVIDER_LIMITS_SYNC_SPACING_MS=1500 +# Min interval (ms) between consecutive UPSTREAM quota fetches on the per-request +# preflight/monitor path (e.g. Codex /wham/usage), complementing the bulk-sync +# spacing above. Many accounts on one IP fetching quota in the same second can look +# like automation to the upstream and get an OAuth token revoked (#6009). This gate +# serializes genuine network calls (cache hits are unaffected). Set to 0 to disable. +# Default: 250 (clamped 0..5000). +# OMNIROUTE_QUOTA_FETCH_MIN_INTERVAL_MS=250 + # Delay (ms) before refreshing provider limits after a real usage event (e.g. a # completed request). Gives the upstream quota API time to register the consumption # before the dashboard polls. Default: 5000 diff --git a/CHANGELOG.md b/CHANGELOG.md index 7f79448f6a..b9c9c4a7f0 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -8,6 +8,7 @@ ### ✨ New Features +- **feat(resilience):** throttle upstream quota fetches on the per-request preflight path ([#6009](https://github.com/diegosouzapw/OmniRoute/issues/6009)) — a new global min-interval gate (`open-sse/services/quotaFetchThrottle.ts`) spaces the actual network calls made by the Codex quota fetcher so that many accounts on one IP no longer fetch quota in the same second (which, per `router-for-me/CLIProxyAPI#2385`, can get a Codex OAuth token revoked). Complements the existing bulk-sync spacing (`PROVIDER_LIMITS_SYNC_SPACING_MS`) which already serialized the periodic provider-limits sync — this covers the concurrent combo/preflight path it didn't. Cache hits are never delayed; fail-open (only ever awaits a timer). Configurable via `OMNIROUTE_QUOTA_FETCH_MIN_INTERVAL_MS` (default 250ms, clamped 0..5000; `0` disables). Regression guard: `tests/unit/quota-fetch-throttle-6009.test.ts` (5). (thanks @powellnorma) - **feat(autoCombo):** add **per-request Auto-Combo controls** via two headers ([#6024](https://github.com/diegosouzapw/OmniRoute/issues/6024) / [#6025](https://github.com/diegosouzapw/OmniRoute/issues/6025) / [#6023](https://github.com/diegosouzapw/OmniRoute/issues/6023)) — `X-OmniRoute-Mode` steers an `auto` combo's scoring for a single request (friendly presets `fast`/`balanced`/`quality`/`cheap`/`reliable`/`offline` **or** a raw mode-pack name; `balanced` forces the default weights), and `X-OmniRoute-Budget` sets a hard per-request USD cost ceiling. Both override the combo's stored config only for the request that carries them; unknown/garbage values are ignored so the saved config is preserved. The resolvers are pure (`open-sse/services/autoCombo/requestControls.ts`) and feed the engine's existing `config.modePack` / `config.budgetCap` inputs — no engine changes. Regression guard: `tests/unit/auto-combo-request-controls-6024.test.ts` (5). (thanks @chirag127) - **feat(providers):** add the **Kenari** OpenAI-compatible gateway (BYOK). Regression guard: `tests/unit/kenari.test.ts`. (thanks @doedja) - **feat(models):** add `claude-sonnet-5` to the Antigravity model catalog (alias mapping in `antigravityModelAliases.ts`). Regression guard: `tests/unit/antigravity-model-aliases.test.ts`. (thanks @anki1kr) diff --git a/docs/reference/ENVIRONMENT.md b/docs/reference/ENVIRONMENT.md index e45e4f1897..efe32cff43 100644 --- a/docs/reference/ENVIRONMENT.md +++ b/docs/reference/ENVIRONMENT.md @@ -413,6 +413,7 @@ detection above). | `MODEL_SYNC_INTERVAL_HOURS` | `24` | `src/shared/services/modelSyncScheduler.ts` | Model catalog sync interval in hours. | | `PROVIDER_LIMITS_SYNC_INTERVAL_MINUTES` | `70` | `src/server-init.ts` | Provider rate-limit and quota polling interval. | | `PROVIDER_LIMITS_SYNC_SPACING_MS` | `1500` | `src/lib/usage/providerLimits.ts` | Gap (ms) between consecutive OAuth quota fetches in a bulk sync; OAuth connections are fetched one at a time to avoid bursting an upstream. `0` opts out (concurrent). | +| `OMNIROUTE_QUOTA_FETCH_MIN_INTERVAL_MS` | `250` | `open-sse/services/quotaFetchThrottle.ts` | Min interval (ms) between consecutive upstream quota fetches on the per-request preflight/monitor path (e.g. Codex `/wham/usage`); spaces concurrent network calls so many accounts on one IP don't burst the upstream (#6009). Cache hits unaffected. `0` disables; clamped `0..5000`. | | `PROVIDER_LIMITS_POST_USAGE_REFRESH_DELAY_MS` | `5000` | `src/lib/usage/providerLimits.ts` | Delay (ms) before refreshing provider limits after a real usage event, giving the upstream quota API time to register consumption. | | `OMNIROUTE_DISABLE_BACKGROUND_SERVICES` | `false` | `src/instrumentation-node.ts` | Disable all background services (sync, pricing, model refresh). Useful for CI/test. | | `OMNIROUTE_ENABLE_RUNTIME_BACKGROUND_TASKS` | _(unset)_ | `src/lib/config/runtimeSettings.ts` | Force background tasks on under automated test detection. Set `1` to override the test heuristic. | diff --git a/open-sse/services/codexQuotaFetcher.ts b/open-sse/services/codexQuotaFetcher.ts index 0cb914e884..eb588ac3ba 100644 --- a/open-sse/services/codexQuotaFetcher.ts +++ b/open-sse/services/codexQuotaFetcher.ts @@ -24,6 +24,7 @@ import { } from "../config/codexQuotaScopes.ts"; import { registerQuotaFetcher, registerQuotaWindows, type QuotaInfo } from "./quotaPreflight.ts"; import { registerMonitorFetcher } from "./quotaMonitor.ts"; +import { throttleQuotaFetch } from "./quotaFetchThrottle.ts"; /** * Stable identifiers for Codex's quota windows. These match the quota keys @@ -227,6 +228,13 @@ export async function fetchCodexQuota( headers["chatgpt-account-id"] = meta.workspaceId; } + // #6009: space concurrent upstream quota fetches so N accounts on one IP do + // not all hit the provider in the same second (anti-fingerprint / avoids the + // Codex OAuth revocation reported in router-for-me/CLIProxyAPI#2385). Cache + // hits above never reach here; this only paces genuine network calls and is + // configurable (OMNIROUTE_QUOTA_FETCH_MIN_INTERVAL_MS, 0 = disabled). + await throttleQuotaFetch(); + const response = await fetch(CODEX_USAGE_URL, { method: "GET", headers, diff --git a/open-sse/services/quotaFetchThrottle.ts b/open-sse/services/quotaFetchThrottle.ts new file mode 100644 index 0000000000..d8791fc8d7 --- /dev/null +++ b/open-sse/services/quotaFetchThrottle.ts @@ -0,0 +1,118 @@ +/** + * quotaFetchThrottle.ts — global min-interval throttle for upstream quota fetches (#6009). + * + * Problem: when many accounts on one IP fetch provider quota, firing all the + * network calls in the same second looks like automation to the upstream and + * — per router-for-me/CLIProxyAPI#2385 — can get a Codex OAuth token revoked. + * + * Solution: serialize the *actual network calls* through a single gate that + * spaces each fetch start at least `minIntervalMs` (+ optional jitter) after the + * previous one. Cache hits never reach the gate (the fetcher returns early), so + * this only paces genuine upstream requests, and it is fail-open: `acquire()` + * only ever awaits a timer, it cannot throw the quota fetch off its fail-open path. + * + * `minIntervalMs = 0` disables throttling entirely (byte-identical to before). + */ + +const DEFAULT_MIN_INTERVAL_MS = 250; +const MAX_MIN_INTERVAL_MS = 5000; +const DEFAULT_JITTER_MS = 120; + +export interface ThrottleClock { + now: () => number; + sleep: (ms: number) => Promise; +} + +const realClock: ThrottleClock = { + now: () => Date.now(), + sleep: (ms: number) => new Promise((resolve) => setTimeout(resolve, ms)), +}; + +export interface MinIntervalThrottleOptions { + minIntervalMs: number; + jitterMs?: number; + clock?: ThrottleClock; + /** Injectable RNG (0..1) for deterministic jitter in tests. */ + rand?: () => number; +} + +/** + * Serializes callers so each `acquire()` resolves at least `minIntervalMs` + * (plus up to `jitterMs`) after the previous one. Concurrent callers queue in + * arrival order; the first through the gate is never delayed. + */ +export class MinIntervalThrottle { + private readonly minIntervalMs: number; + private readonly jitterMs: number; + private readonly clock: ThrottleClock; + private readonly rand: () => number; + private lastStart = 0; + private chain: Promise = Promise.resolve(); + + constructor(options: MinIntervalThrottleOptions) { + this.minIntervalMs = Math.max(0, options.minIntervalMs); + this.jitterMs = Math.max(0, options.jitterMs ?? 0); + this.clock = options.clock ?? realClock; + this.rand = options.rand ?? Math.random; + } + + async acquire(): Promise { + if (this.minIntervalMs <= 0) return; // throttling disabled — no serialization cost + const prev = this.chain; + let release!: () => void; + this.chain = new Promise((resolve) => { + release = resolve; + }); + try { + await prev; + const now = this.clock.now(); + if (this.lastStart !== 0) { + const jitter = this.jitterMs > 0 ? Math.floor(this.rand() * this.jitterMs) : 0; + const wait = this.lastStart + this.minIntervalMs + jitter - now; + if (wait > 0) await this.clock.sleep(wait); + } + this.lastStart = this.clock.now(); + } finally { + release(); + } + } +} + +/** + * Resolve the configured min interval (ms) from the environment, clamped to a + * sane range. Garbage / negative → default; above the ceiling → clamped. + */ +export function resolveQuotaFetchMinIntervalMs( + env: Record = process.env +): number { + const raw = env.OMNIROUTE_QUOTA_FETCH_MIN_INTERVAL_MS; + if (raw === undefined || raw === null || raw.trim() === "") return DEFAULT_MIN_INTERVAL_MS; + const n = Number(raw.trim()); + if (!Number.isFinite(n) || n < 0) return DEFAULT_MIN_INTERVAL_MS; + return Math.min(Math.round(n), MAX_MIN_INTERVAL_MS); +} + +// ─── Shared process-wide instance (used by the provider quota fetchers) ─────── + +let _sharedThrottle: MinIntervalThrottle | null = null; + +/** Lazily build (and memoize) the shared throttle from the current env config. */ +export function getQuotaFetchThrottle(): MinIntervalThrottle { + if (!_sharedThrottle) { + _sharedThrottle = new MinIntervalThrottle({ + minIntervalMs: resolveQuotaFetchMinIntervalMs(), + jitterMs: DEFAULT_JITTER_MS, + }); + } + return _sharedThrottle; +} + +/** Await the shared throttle gate before issuing an upstream quota fetch. */ +export function throttleQuotaFetch(): Promise { + return getQuotaFetchThrottle().acquire(); +} + +/** Test-only: reset the memoized shared throttle (e.g. after changing env). */ +export function resetQuotaFetchThrottle(): void { + _sharedThrottle = null; +} diff --git a/tests/unit/quota-fetch-throttle-6009.test.ts b/tests/unit/quota-fetch-throttle-6009.test.ts new file mode 100644 index 0000000000..46524b01e3 --- /dev/null +++ b/tests/unit/quota-fetch-throttle-6009.test.ts @@ -0,0 +1,84 @@ +import test from "node:test"; +import assert from "node:assert/strict"; + +// Regression guard for #6009 — forced delay/cooldown between upstream quota fetches. +// When many accounts on one IP fetch provider quota, firing them all in the same +// second looks suspicious to the upstream (per router-for-me/CLIProxyAPI#2385 this +// can get a Codex OAuth token revoked). MinIntervalThrottle serializes the actual +// network calls and spaces each start >= minIntervalMs after the previous one. + +const { MinIntervalThrottle, resolveQuotaFetchMinIntervalMs } = await import( + "../../open-sse/services/quotaFetchThrottle.ts" +); + +class FakeClock { + t = 1000; + slept: number[] = []; + now = () => this.t; + sleep = async (ms: number) => { + this.slept.push(ms); + this.t += ms; + }; +} + +test("#6009 spaces concurrent fetch starts by at least minIntervalMs", async () => { + const clock = new FakeClock(); + const throttle = new MinIntervalThrottle({ minIntervalMs: 250, jitterMs: 0, clock }); + + // Fire 4 concurrent acquisitions (simulates 4 accounts fetching at once). + await Promise.all(Array.from({ length: 4 }, () => throttle.acquire())); + + // First runs immediately; the other three each sleep the full 250ms gap, so + // the four network calls start at t = 1000, 1250, 1500, 1750 (spaced, not bursty). + assert.deepEqual(clock.slept, [250, 250, 250]); + assert.equal(clock.now(), 1750); +}); + +test("#6009 minIntervalMs=0 disables throttling (no sleeps)", async () => { + const clock = new FakeClock(); + const throttle = new MinIntervalThrottle({ minIntervalMs: 0, jitterMs: 0, clock }); + await Promise.all([throttle.acquire(), throttle.acquire(), throttle.acquire()]); + assert.deepEqual(clock.slept, []); +}); + +test("#6009 a single fetch is never delayed", async () => { + const clock = new FakeClock(); + const throttle = new MinIntervalThrottle({ minIntervalMs: 500, jitterMs: 0, clock }); + await throttle.acquire(); + assert.deepEqual(clock.slept, []); + assert.equal(clock.now(), 1000); +}); + +test("#6009 jitter adds a bounded extra spacing on top of the min gap", async () => { + const clock = new FakeClock(); + // Deterministic rand → always 0.5, so jitter = floor(0.5 * 100) = 50. + const throttle = new MinIntervalThrottle({ + minIntervalMs: 200, + jitterMs: 100, + clock, + rand: () => 0.5, + }); + await Promise.all(Array.from({ length: 3 }, () => throttle.acquire())); + // First call never sleeps; each subsequent waits gap = 200 + floor(0.5*100) = 250. + assert.deepEqual(clock.slept, [250, 250]); +}); + +test("#6009 env resolver clamps to sane bounds and defaults", () => { + assert.equal(resolveQuotaFetchMinIntervalMs({}), 250); // default + assert.equal(resolveQuotaFetchMinIntervalMs({ OMNIROUTE_QUOTA_FETCH_MIN_INTERVAL_MS: "0" }), 0); + assert.equal( + resolveQuotaFetchMinIntervalMs({ OMNIROUTE_QUOTA_FETCH_MIN_INTERVAL_MS: "1000" }), + 1000 + ); + // garbage / negative → default + assert.equal( + resolveQuotaFetchMinIntervalMs({ OMNIROUTE_QUOTA_FETCH_MIN_INTERVAL_MS: "abc" }), + 250 + ); + assert.equal(resolveQuotaFetchMinIntervalMs({ OMNIROUTE_QUOTA_FETCH_MIN_INTERVAL_MS: "-5" }), 250); + // absurdly high → clamped to max 5000 + assert.equal( + resolveQuotaFetchMinIntervalMs({ OMNIROUTE_QUOTA_FETCH_MIN_INTERVAL_MS: "999999" }), + 5000 + ); +});