From 44a04df4f60d5e1397ab333cfb44e88615519307 Mon Sep 17 00:00:00 2001 From: Anton <39598727+NomenAK@users.noreply.github.com> Date: Thu, 14 May 2026 05:21:16 +0200 Subject: [PATCH] fix(rateLimit): never .stop() during runtime reset, evict cache instead (#2218) Integrated into release/v3.8.0 --- open-sse/services/rateLimitManager.ts | 85 ++++++++++--- tests/unit/limiter-lifecycle.test.ts | 164 ++++++++++++++++++++++++++ 2 files changed, 234 insertions(+), 15 deletions(-) create mode 100644 tests/unit/limiter-lifecycle.test.ts diff --git a/open-sse/services/rateLimitManager.ts b/open-sse/services/rateLimitManager.ts index aa52ace948..31af3c2fb9 100644 --- a/open-sse/services/rateLimitManager.ts +++ b/open-sse/services/rateLimitManager.ts @@ -192,14 +192,31 @@ function watchdogTick() { ); limiters.delete(key); lastDispatchAt.delete(key); - trackAsyncOperation(limiter.stop({ dropWaitingJobs: true })); + // Do NOT call limiter.stop() — it permanently rejects future .schedule() calls with + // "This limiter has been stopped". In-flight requests still holding a reference to + // the old instance cannot be redirected to a new one, causing spurious 502 bursts. + // Call disconnect() (not stop()) to release Bottleneck's internal heartbeat timer + // without poisoning the queue for any remaining in-flight jobs. This prevents the + // heartbeat-timer memory leak observed when many limiters are evicted at runtime. + // getLimiter() lazily allocates a fresh Bottleneck on the next call. + trackAsyncOperation(limiter.disconnect()); } } +let shutdownHandlersRegistered = false; + export function startRateLimitWatchdog(): void { if (watchdogInterval) return; watchdogInterval = setInterval(watchdogTick, 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 + // subprocess IPC teardown — the test suite does not call startRateLimitWatchdog(). + if (!shutdownHandlersRegistered) { + shutdownHandlersRegistered = true; + process.once("SIGTERM", shutdownLimiters); + process.once("SIGINT", shutdownLimiters); + } } export function stopRateLimitWatchdog(): void { @@ -208,6 +225,26 @@ export function stopRateLimitWatchdog(): void { watchdogInterval = null; } +/** + * 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. + */ +function shutdownLimiters(): void { + for (const limiter of limiters.values()) { + limiter.stop({ dropWaitingJobs: false }); + } + limiters.clear(); + lastDispatchAt.clear(); +} + +// Only register shutdown handlers when there are active limiters to shut down. +// Guard with once() so repeated registrations (e.g. test resets) don't stack. +// Note: these are registered lazily in startRateLimitWatchdog() to avoid +// interfering with test runner subprocess IPC teardown. + function trackAsyncOperation(promise: Promise): Promise { pendingAsyncOperations.add(promise); promise.finally(() => { @@ -272,15 +309,19 @@ export function enableRateLimitProtection(connectionId) { */ export function disableRateLimitProtection(connectionId) { enabledConnections.delete(connectionId); - // Clean up limiters for this connection. Use stop({dropWaitingJobs:true}) - // instead of disconnect() so any queued promises actually reject — disconnect - // shuts the limiter down without draining the queue, leaking stuck callers. - for (const [key] of Array.from(limiters)) { + // Evict limiters for this connection from the cache. Do NOT call limiter.stop() — + // it permanently rejects future .schedule() calls with "This limiter has been stopped", + // and in-flight requests holding a reference to the old instance would fail with 502. + // Call disconnect() (not stop()) to release Bottleneck's internal heartbeat timer + // without permanently poisoning the instance for any remaining in-flight jobs. + // Eviction-only would leak the heartbeat timer until GC; disconnect() releases it + // synchronously so the runtime memory footprint stays flat under heavy connection churn. + // .stop() is reserved exclusively for SIGTERM/SIGINT shutdown (see shutdownLimiters). + for (const [key, limiter] of Array.from(limiters)) { if (key.includes(connectionId)) { - const limiter = limiters.get(key); limiters.delete(key); lastDispatchAt.delete(key); - if (limiter) trackAsyncOperation(limiter.stop({ dropWaitingJobs: true })); + trackAsyncOperation(limiter.disconnect()); } } } @@ -517,14 +558,18 @@ 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)` ); - // Stop the limiter and drop all waiting jobs so they fail immediately - // instead of hanging in the queue until reservoir refreshes (which can - // be hours for providers like Codex with long rate limit windows). - // This lets upstream callers (e.g. LiteLLM) trigger fallback to other providers. - // Delete from the Map first so follow-up learning from the same error body - // can materialize a fresh limiter immediately. + // 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); - trackAsyncOperation(limiter.stop({ dropWaitingJobs: true })); + trackAsyncOperation(limiter.disconnect()); return; } @@ -685,12 +730,19 @@ export async function __resetRateLimitManagerForTests() { persistTimer = null; } + // Collect and await all disconnect() Promises so Bottleneck's internal + // yieldLoop(0) calls settle before the next test starts. Not awaiting + // these can cause the Node.js test runner IPC channel to receive a + // corrupted message when the pending Promise fires during IPC serialization. + const disconnectPromises: Promise[] = []; for (const limiter of limiters.values()) { - limiter.disconnect(); + disconnectPromises.push(limiter.disconnect()); } limiters.clear(); enabledConnections.clear(); initialized = false; + lastDispatchAt.clear(); + shutdownHandlersRegistered = false; for (const key of Object.keys(learnedLimits)) { delete learnedLimits[key]; @@ -699,6 +751,9 @@ export async function __resetRateLimitManagerForTests() { if (pendingAsyncOperations.size > 0) { await Promise.allSettled(Array.from(pendingAsyncOperations)); } + if (disconnectPromises.length > 0) { + await Promise.allSettled(disconnectPromises); + } } export async function __getLimiterStateForTests(provider, connectionId, model = null) { diff --git a/tests/unit/limiter-lifecycle.test.ts b/tests/unit/limiter-lifecycle.test.ts new file mode 100644 index 0000000000..1f6bf66cf9 --- /dev/null +++ b/tests/unit/limiter-lifecycle.test.ts @@ -0,0 +1,164 @@ +/** + * TDD regression test — Section B fix: + * limiter lifecycle: no .stop() during runtime reset, evict cache only. + * + * Bug: calling .stop() on a Bottleneck instance permanently rejects future + * .schedule() calls with "This limiter has been stopped". In-flight requests + * holding a reference to the now-stopped limiter cannot be redirected, causing + * spurious 502 bursts during container recreation / model registry refresh. + * + * Observed incidents (2026-05-12): + * - xiaomi-mimo: 13x burst at 17:14:28 (reset 3s) + * - mistral: 13x burst at 15:42:36 (reset 3s) + * - claude: 1 hit at 19:01:00 (post reboot) + * + * Design note: tests B and C use `wait(0)` instead of `wait(50)` to avoid + * creating a long-running Promise that can interfere with the Node.js v25 + * test runner IPC serialization window. + */ +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-limiter-lifecycle-")); +process.env.DATA_DIR = TEST_DATA_DIR; + +const core = await import("../../src/lib/db/core.ts"); +const rateLimitManager = await import("../../open-sse/services/rateLimitManager.ts"); + +function wait(ms) { + return new Promise((resolve) => setTimeout(resolve, ms)); +} + +async function resetStorage() { + core.resetDbInstance(); + fs.rmSync(TEST_DATA_DIR, { recursive: true, force: true }); + fs.mkdirSync(TEST_DATA_DIR, { recursive: true }); +} + +test.beforeEach(async () => { + await resetStorage(); +}); + +test.afterEach(async () => { + await rateLimitManager.__resetRateLimitManagerForTests(); +}); + +test.after(async () => { + await rateLimitManager.__resetRateLimitManagerForTests(); + core.resetDbInstance(); + fs.rmSync(TEST_DATA_DIR, { recursive: true, force: true }); +}); + +/** + * Core regression: after disableRateLimitProtection + enableRateLimitProtection, + * the next withRateLimit must succeed on a fresh limiter instance. + * + * Bug vector: disableRateLimitProtection() called limiter.stop({dropWaitingJobs:true}). + */ +test("after disable+re-enable, withRateLimit must succeed without stopped-limiter error", async () => { + const provider = "openai"; + const connectionId = "lifecycle-test-conn-a"; + + rateLimitManager.enableRateLimitProtection(connectionId); + const r1 = await rateLimitManager.withRateLimit( + provider, + connectionId, + null, + async () => "job-1" + ); + assert.equal(r1, "job-1"); + + // Reset cycle: disable then re-enable (hot-reload / container recreation scenario) + rateLimitManager.disableRateLimitProtection(connectionId); + rateLimitManager.enableRateLimitProtection(connectionId); + + let error = null; + let r2 = null; + try { + r2 = await rateLimitManager.withRateLimit(provider, connectionId, null, async () => "job-2"); + } catch (err) { + error = err; + } + + assert.equal( + error, + null, + "Expected no error after disable+re-enable, but got: " + (error && error.message) + ); + assert.equal(r2, "job-2", "post-reset request must return its value"); +}); + +/** + * In-flight safety: a job started BEFORE disable must still complete. + * Uses wait(0) (immediate tick) to minimize cross-test async interference. + * + * Bug vector: disableRateLimitProtection() called limiter.stop({dropWaitingJobs:true}). + */ +test("in-flight job before disable must complete without stopped-limiter error", async () => { + const provider = "openai"; + const connectionId = "lifecycle-test-conn-b"; + + rateLimitManager.enableRateLimitProtection(connectionId); + + // Start job (completes in next tick), disable immediately + const jobPromise = rateLimitManager.withRateLimit(provider, connectionId, null, () => + wait(0).then(() => "in-flight-ok") + ); + + // Disable before the job resolves (it's queued/executing in Bottleneck) + rateLimitManager.disableRateLimitProtection(connectionId); + + let error = null; + let result = null; + try { + result = await jobPromise; + } catch (err) { + error = err; + } + + assert.equal( + error, + null, + "In-flight job must not throw after disable, but got: " + (error && error.message) + ); + assert.equal(result, "in-flight-ok", "in-flight job must return its value"); +}); + +/** + * 429 teardown: after a 429 evicts the limiter, the next request must succeed. + * + * Bug vector: updateFromHeaders() 429 path called limiter.stop() before this fix. + */ +test("after 429 teardown, next withRateLimit must get a fresh limiter and succeed", async () => { + const provider = "openai"; + const connectionId = "lifecycle-test-conn-c"; + + rateLimitManager.enableRateLimitProtection(connectionId); + await rateLimitManager.withRateLimit(provider, connectionId, null, async () => "pre-429"); + + // Simulate 429 — evicts the limiter from cache + rateLimitManager.updateFromHeaders(provider, connectionId, { "retry-after": "1s" }, 429, null); + + let error = null; + let result = null; + try { + result = await rateLimitManager.withRateLimit( + provider, + connectionId, + null, + async () => "post-429" + ); + } catch (err) { + error = err; + } + + assert.equal( + error, + null, + "Post-429 request must not throw, but got: " + (error && error.message) + ); + assert.equal(result, "post-429", "post-429 request must return its value"); +});