diff --git a/open-sse/handlers/chatCore.ts b/open-sse/handlers/chatCore.ts index 6ebc6595f5..9317c55d98 100644 --- a/open-sse/handlers/chatCore.ts +++ b/open-sse/handlers/chatCore.ts @@ -945,6 +945,14 @@ export async function handleChatCore({ const requestedModel = typeof body?.model === "string" && body.model.trim().length > 0 ? body.model : model; const startTime = Date.now(); + // Per-request trace id + checkpoint helper. Lets us see exactly which await + // a hung request was sitting on in `[STAGE_TRACE]` log lines. + const traceId = Math.random().toString(36).slice(2, 8); + const trace = (label: string, extra?: Record) => { + const elapsed = Date.now() - startTime; + const suffix = extra ? ` ${JSON.stringify(extra)}` : ""; + log?.info?.("STAGE_TRACE", `${traceId} ${label} t=${elapsed}ms${suffix}`); + }; const persistFailureUsage = (statusCode: number, errorCode?: string | null) => { saveRequestUsage({ provider: provider || "unknown", @@ -1543,6 +1551,8 @@ export async function handleChatCore({ } } + trace("post_injection", { provider, model }); + // Translate request (pass reqLogger for intermediate logging) // ── Proactive Context Compression (Phase 4) ── // Check if context exceeds 70% of limit and compress proactively before sending to provider. @@ -1976,6 +1986,8 @@ export async function handleChatCore({ return createErrorResult(statusCode, message); } + trace("post_translation"); + // Extract toolNameMap for response translation (Claude OAuth) const translatedToolNameMap = translatedBody._toolNameMap; const nativeClaudeToolNameMap = isClaudePassthrough @@ -2222,6 +2234,10 @@ export async function handleChatCore({ } } + trace("pre_semaphore", { + semaphoreKey: accountSemaphoreKey, + max: accountSemaphoreMaxConcurrency, + }); const acquireAccountSemaphoreRelease = accountSemaphoreKey && accountSemaphoreMaxConcurrency != null ? await acquireAccountSemaphore(accountSemaphoreKey, { @@ -2229,17 +2245,21 @@ export async function handleChatCore({ signal: streamController.signal, }) : () => {}; + trace("post_semaphore"); try { + trace("pre_rate_limit"); const rawResult = await withRateLimit( provider, connectionId, modelToCall, async () => { + trace("inside_rate_limit"); let attempts = 0; const maxAttempts = provider === "qwen" ? 3 : 1; while (attempts < maxAttempts) { + trace("pre_executor", { attempt: attempts }); const res = await executor.execute({ model: modelToCall, body: bodyToSend, @@ -2252,6 +2272,7 @@ export async function handleChatCore({ clientHeaders: buildExecutorClientHeaders(clientRawRequest?.headers, userAgent), onCredentialsRefreshed, }); + trace("post_executor", { status: res?.response?.status }); // Qwen 429 strict quota backoff (wait 1.5s, 3s and retry) if ( diff --git a/open-sse/services/rateLimitManager.ts b/open-sse/services/rateLimitManager.ts index d0acd2410f..aa52ace948 100644 --- a/open-sse/services/rateLimitManager.ts +++ b/open-sse/services/rateLimitManager.ts @@ -68,6 +68,34 @@ let initialized = false; let currentRequestQueueSettings: RequestQueueSettings = DEFAULT_RESILIENCE_SETTINGS.requestQueue; +// Watchdog: detect Bottleneck limiters that are wedged (queue has work, but no +// jobs are dispatched). When the reservoir/refresh state desyncs from reality, +// this catches it and force-resets so traffic isn't stuck forever. +const lastDispatchAt = new Map(); +let watchdogInterval: ReturnType | null = null; +const WATCHDOG_INTERVAL_MS = 30_000; +// Threshold has to exceed any *legitimate* gap between dispatches: +// - default reservoirRefreshInterval is 60s +// - adaptive minTime can climb to ~60s for 1-RPM providers (see updateFromHeaders) +// 120s gives a 2× margin against both, while still catching the actual wedge +// case we observed (queue stalled for 3+ minutes with no progress). +const WEDGE_THRESHOLD_MS = 120_000; + +/** + * Env-var override for the auto-enable safety net. Highest priority — wins + * over the persisted dashboard setting. Use to disable in an incident without + * needing dashboard access. + * RATE_LIMIT_AUTO_ENABLE=false → never auto-enable + * RATE_LIMIT_AUTO_ENABLE=true → force on regardless of dashboard + * (unset) → use dashboard setting + */ +function isAutoEnableActive(settings: RequestQueueSettings): boolean { + const env = process.env.RATE_LIMIT_AUTO_ENABLE?.trim().toLowerCase(); + if (env === "false" || env === "0" || env === "off") return false; + if (env === "true" || env === "1" || env === "on") return true; + return settings.autoEnableApiKeyProviders; +} + function buildLimiterDefaults() { return { maxConcurrent: currentRequestQueueSettings.concurrentRequests, @@ -113,23 +141,17 @@ function reconcileEnabledConnections( } if ( - requestQueueSettings.autoEnableApiKeyProviders && + isAutoEnableActive(requestQueueSettings) && getProviderCategory(provider) === "apikey" && isActive ) { nextEnabledConnections.add(connectionId); autoCount++; - const key = `${provider}:${connectionId}`; - if (!limiters.has(key)) { - limiters.set( - key, - new Bottleneck({ - ...buildLimiterDefaults(), - id: key, - }) - ); - } + // 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. + getLimiter(provider, connectionId); } } @@ -149,6 +171,43 @@ function reconcileEnabledConnections( }; } +function watchdogTick() { + const now = Date.now(); + for (const [key, limiter] of Array.from(limiters)) { + const counts = limiter.counts(); + if (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; + + console.warn( + `🚨 [RATE-LIMIT] WEDGED: ${key} queued=${counts.QUEUED} running=0 executing=0 stalled=${stalledMs}ms — force-resetting` + ); + limiters.delete(key); + lastDispatchAt.delete(key); + trackAsyncOperation(limiter.stop({ dropWaitingJobs: true })); + } +} + +export function startRateLimitWatchdog(): void { + if (watchdogInterval) return; + watchdogInterval = setInterval(watchdogTick, WATCHDOG_INTERVAL_MS); + watchdogInterval.unref?.(); +} + +export function stopRateLimitWatchdog(): void { + if (!watchdogInterval) return; + clearInterval(watchdogInterval); + watchdogInterval = null; +} + function trackAsyncOperation(promise: Promise): Promise { pendingAsyncOperations.add(promise); promise.finally(() => { @@ -184,6 +243,10 @@ export async function initializeRateLimits() { // Load persisted learned limits await loadPersistedLimits(); + + // Watchdog runs unconditionally — cheap, only fires when something is + // actually wedged. + startRateLimitWatchdog(); } catch (err) { console.error("[RATE-LIMIT] Failed to load settings:", err.message); } @@ -209,12 +272,15 @@ export function enableRateLimitProtection(connectionId) { */ export function disableRateLimitProtection(connectionId) { enabledConnections.delete(connectionId); - // Clean up limiters for this connection - for (const [key] of limiters) { + // 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)) { if (key.includes(connectionId)) { const limiter = limiters.get(key); - limiter?.disconnect(); limiters.delete(key); + lastDispatchAt.delete(key); + if (limiter) trackAsyncOperation(limiter.stop({ dropWaitingJobs: true })); } } } @@ -259,8 +325,15 @@ function getLimiter(provider, connectionId, model = null) { ); } }); + // 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()); + }); limiters.set(key, limiter); + lastDispatchAt.set(key, Date.now()); } return limiters.get(key); diff --git a/src/app/api/admin/concurrency/route.ts b/src/app/api/admin/concurrency/route.ts new file mode 100644 index 0000000000..6bbf79b05f --- /dev/null +++ b/src/app/api/admin/concurrency/route.ts @@ -0,0 +1,24 @@ +import { NextResponse } from "next/server"; +import { getAllRateLimitStatus } from "@omniroute/open-sse/services/rateLimitManager.ts"; +import { + getStats as getSemaphoreStats, + resetAll as resetAllSemaphores, +} from "@omniroute/open-sse/services/accountSemaphore.ts"; + +export async function GET() { + return NextResponse.json({ + timestamp: new Date().toISOString(), + rateLimits: getAllRateLimitStatus(), + semaphores: getSemaphoreStats(), + }); +} + +export async function POST(request: Request) { + const url = new URL(request.url); + const action = url.searchParams.get("action"); + if (action === "reset-semaphores") { + resetAllSemaphores(); + return NextResponse.json({ ok: true, action }); + } + return NextResponse.json({ error: "unknown action" }, { status: 400 }); +} diff --git a/tests/unit/rate-limit-manager.test.ts b/tests/unit/rate-limit-manager.test.ts index d505a49add..908e752325 100644 --- a/tests/unit/rate-limit-manager.test.ts +++ b/tests/unit/rate-limit-manager.test.ts @@ -208,6 +208,41 @@ test("rate limit manager parses retry hints from response bodies and locks model assert.equal(rateLimitManager.getRateLimitStatus("openai", "conn-body").active, true); }); +test("RATE_LIMIT_AUTO_ENABLE env var overrides dashboard auto-enable setting", async () => { + const conn = await providersDb.createProviderConnection({ + provider: "openai", + authType: "apikey", + name: "Env Override", + apiKey: "sk-env", + isActive: true, + }); + + // Dashboard says auto-enable on, but env says off → off wins + const original = process.env.RATE_LIMIT_AUTO_ENABLE; + process.env.RATE_LIMIT_AUTO_ENABLE = "false"; + try { + await rateLimitManager.initializeRateLimits(); + assert.equal(rateLimitManager.isRateLimitEnabled(conn.id), false); + } finally { + if (original === undefined) delete process.env.RATE_LIMIT_AUTO_ENABLE; + else process.env.RATE_LIMIT_AUTO_ENABLE = original; + } + + // Reset and verify the opposite: env=true forces on even when dashboard would be off + await rateLimitManager.__resetRateLimitManagerForTests(); + process.env.RATE_LIMIT_AUTO_ENABLE = "true"; + try { + await rateLimitManager.applyRequestQueueSettings({ + ...resilienceSettings.DEFAULT_RESILIENCE_SETTINGS.requestQueue, + autoEnableApiKeyProviders: false, + }); + assert.equal(rateLimitManager.isRateLimitEnabled(conn.id), true); + } finally { + if (original === undefined) delete process.env.RATE_LIMIT_AUTO_ENABLE; + else process.env.RATE_LIMIT_AUTO_ENABLE = original; + } +}); + test("rate limit manager recomputes auto-enabled API key connections when queue settings change", async () => { const autoConnection = await providersDb.createProviderConnection({ provider: "openai",