fix(rate-limit): watchdog, env override, and stage tracing to prevent silent wedges (#1828)

Integrated into release/v3.7.7

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<string, unknown>) => {
+    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 (

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<string, number>();
+let watchdogInterval: ReturnType<typeof setInterval> | 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<T>(promise: Promise<T>): Promise<T> {
   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);

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 });
+}

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",