fix(combo): prevent unhandledRejection from per-model-timeout abort (#10846)

Merged — locally validated (11/11 focused tests across both new test files, typecheck:core clean after a 1-char fix pushed to this branch: ComboLogger.error is optional in combo/types.ts so the defensive race-catch needed log.error?.(...) — TS2722 otherwise). Solid production diagnosis (47 unhandledRejections traced to the orphaned race loser). Thanks!

open-sse/services/combo/targetTimeoutRunner.ts

@@ -19,6 +19,76 @@ import type { HandleSingleModel, SingleModelTarget, ComboLogger } from "./types.
 /** Stable internal classification for OmniRoute's own combo per-target timer. */
 export const COMBO_TARGET_TIMEOUT_CODE = "combo_target_timeout";
 
+/**
+ * Diagnostic: track recent combo-per-model-timeout abort errors so an
+ * unhandledRejection handler can attribute the stack trace to a specific model
+ * and timeout value. Ring buffer of 4 — concurrent per-model timeouts are rare
+ * but possible (e.g. hedge + per-target timeout on different targets).
+ */
+const CONTEXT_RING_SIZE = 4;
+const lastTimeoutContexts: Array<{
+  modelStr: string;
+  timeoutMs: number;
+  abortError: Error;
+  timestamp: number;
+}> = [];
+let contextRingIndex = 0;
+
+function recordTimeoutContext(ctx: {
+  modelStr: string;
+  timeoutMs: number;
+  abortError: Error;
+  timestamp: number;
+}): void {
+  if (lastTimeoutContexts.length < CONTEXT_RING_SIZE) {
+    lastTimeoutContexts.push(ctx);
+  } else {
+    lastTimeoutContexts[contextRingIndex] = ctx;
+    contextRingIndex = (contextRingIndex + 1) % CONTEXT_RING_SIZE;
+  }
+}
+
+/** Retrieve (and clear) all pending combo-per-model-timeout diagnostic contexts. */
+export function drainLastTimeoutContexts(): typeof lastTimeoutContexts {
+  const out = lastTimeoutContexts.splice(0);
+  contextRingIndex = 0;
+  return out;
+}
+
+/**
+ * Install a persistent unhandledRejection listener that logs combo-per-model-timeout
+ * diagnostics. Call once at module load. The listener stays installed permanently —
+ * it only acts on combo-per-model-timeout rejections and returns early for everything
+ * else, so there is no handler leak and no remove/re-install race window.
+ */
+let diagnosticInstalled = false;
+function ensureDiagnosticListener(): void {
+  if (diagnosticInstalled) return;
+  diagnosticInstalled = true;
+  process.on("unhandledRejection", (reason: unknown) => {
+    try {
+      const isComboTimeout =
+        reason instanceof Error && reason.message === COMBO_PER_MODEL_TIMEOUT_REASON;
+      if (!isComboTimeout) return;
+      const contexts = drainLastTimeoutContexts();
+      // Log the full stack trace so the next production incident is diagnosable.
+      // Without this, Node's default unhandledRejection warning shows only
+      // "Error: combo-per-model-timeout" with no caller context.
+      const summary =
+        contexts.length > 0
+          ? contexts.map((c) => `  model=${c.modelStr} timeout=${c.timeoutMs}ms`).join("\n")
+          : "  (no context recorded)";
+      console.error(
+        "[COMBO-TIMEOUT-DIAGNOSTIC] unhandledRejection from combo per-model timeout.\n" +
+          `${summary}\n` +
+          `  abortError stack:\n${reason.stack ?? reason}`
+      );
+    } catch {
+      // Diagnostic logging failed — never let this break the process.
+    }
+  });
+}
+
 export function buildTargetTimeoutRunner(deps: {
   handleSingleModel: HandleSingleModel;
   comboTargetTimeoutMs: number;
@@ -29,6 +99,7 @@ export function buildTargetTimeoutRunner(deps: {
   target?: SingleModelTarget
 ) => Promise<Response> {
   const { handleSingleModel, comboTargetTimeoutMs, log } = deps;
+  ensureDiagnosticListener();
   return async (
     b: Record<string, unknown>,
     modelStr: string,
@@ -46,11 +117,18 @@ export function buildTargetTimeoutRunner(deps: {
     const timeoutPromise = new Promise<Response>((resolve) => {
       timeoutId = setTimeout(() => {
         timedOut = true;
+        const abortErr = new Error(COMBO_PER_MODEL_TIMEOUT_REASON);
+        recordTimeoutContext({
+          modelStr,
+          timeoutMs: comboTargetTimeoutMs,
+          abortError: abortErr,
+          timestamp: Date.now(),
+        });
         log.warn(
           "COMBO",
           `Model ${modelStr} exceeded ${comboTargetTimeoutMs}ms timeout — falling back`
         );
-        timeoutController.abort(new Error(COMBO_PER_MODEL_TIMEOUT_REASON));
+        timeoutController.abort(abortErr);
         // HTTP 504 (not proprietary 524): this is OmniRoute's own per-target timer.
         // Typed as combo_target_timeout so request-scoped classification can keep the
         // connection eligible for fallback instead of treating it like Cloudflare 524
@@ -88,6 +166,13 @@ export function buildTargetTimeoutRunner(deps: {
       }
     }
     try {
+      // Both branches of the race resolve (never reject): the inner
+      // handleSingleModel call has a .catch() that converts rejections into
+      // responses, and timeoutPromise always resolves. A defensive outer
+      // .catch() guards against unexpected throws in the .catch() handler
+      // itself (e.g. a broken Error.prototype.message getter) — without
+      // this, such a throw would surface as an unhandledRejection tagged
+      // "combo-per-model-timeout" in production logs.
       return await Promise.race([
         handleSingleModel(b, modelStr, targetWithSignal).catch((err) => {
           if (timedOut) {
@@ -99,7 +184,13 @@ export function buildTargetTimeoutRunner(deps: {
           return errorResponse(502, err?.message ?? "Upstream model error");
         }),
         timeoutPromise,
-      ]);
+      ]).catch((raceErr) => {
+        // Defensive: should never fire — both race branches always resolve.
+        // Include the error message so the root cause is not masked.
+        const detail = raceErr instanceof Error ? raceErr.message : String(raceErr);
+        log.error?.("COMBO", `Unexpected rejection in combo timeout race for ${modelStr}: ${detail}`);
+        return errorResponse(502, `Combo timeout dispatch error: ${detail}`);
+      });
     } finally {
       clearTimeout(timeoutId);
       if (parentHedgeSignal && onParentHedgeAbort) {

open-sse/services/rateLimitManager.ts

@@ -9,10 +9,7 @@
  */
 
 import Bottleneck from "bottleneck";
-import {
-  applyBottleneckDoExpirePatch,
-  applyBottleneckHeartbeatPatch,
-} from "./bottleneckPatch.ts";
+import { applyBottleneckDoExpirePatch, applyBottleneckHeartbeatPatch } from "./bottleneckPatch.ts";
 import { parseRetryAfterFromBody } from "./accountFallback.ts";
 import { getAntigravityQuotaFamily } from "./antigravityQuotaFamily.ts";
 import { getProviderCategory } from "../config/providerRegistry.ts";
@@ -550,12 +547,7 @@ export async function withRateLimit(provider, connectionId, model, fn, signal =
   // Proactive sliding-window fallback for header-less providers with a declared cap
   // (Fase 8.2). No-op unless PROVIDER_DEFAULT_RATE_LIMITS has an entry for `provider`.
   const maxWaitMs = resolveRequestQueueMaxWaitMs(provider);
-  await awaitProviderDefaultSlot(
-    provider,
-    connectionId,
-    signal,
-    maxWaitMs
-  );
+  await awaitProviderDefaultSlot(provider, connectionId, signal, maxWaitMs);
 
   const limiter = getLimiter(provider, connectionId, model);
   // Bottleneck's `expiration` starts only after a job leaves QUEUED. The
@@ -607,7 +599,14 @@ export async function withRateLimit(provider, connectionId, model, fn, signal =
       }
 
       try {
-        return await Promise.race([limiter.schedule(scheduleOpts, fn), abortPromise]);
+        // Race the work against the abort signal. When abort wins, fn is still
+        // running inside Bottleneck's limiter — its eventual rejection must not
+        // surface as an unhandledRejection. The .catch(noop) silences only the
+        // orphaned branch; the real rejection comes from abortPromise.
+        const scheduled = limiter.schedule(scheduleOpts, fn);
+        scheduled.catch(() => {}); // prevent unhandledRejection when abort wins
+        abortPromise.catch(() => {}); // prevent unhandledRejection when scheduled wins
+        return await Promise.race([scheduled, abortPromise]);
       } finally {
         if (abortListener) {
           signal.removeEventListener("abort", abortListener);

tests/unit/combo-target-timeout-runner.test.ts

@@ -1,6 +1,9 @@
 import { test } from "node:test";
 import assert from "node:assert/strict";
-import { buildTargetTimeoutRunner } from "../../open-sse/services/combo/targetTimeoutRunner.ts";
+import {
+  buildTargetTimeoutRunner,
+  drainLastTimeoutContexts,
+} from "../../open-sse/services/combo/targetTimeoutRunner.ts";
 import type { ComboLogger, SingleModelTarget } from "../../open-sse/services/combo/types.ts";
 
 const noopLog: ComboLogger = { warn() {}, info() {}, error() {}, debug() {} };
@@ -85,3 +88,119 @@ test("hedge do parent já abortado propaga o abort ao filho", async () => {
   await runner({}, "m", parentTarget);
   assert.equal(sawAbort, true);
 });
+
+test("rejection from handleSingleModel after timeout does not leak as unhandledRejection", async () => {
+  // Simulate: timeout fires, handleSingleModel later rejects with the abort error.
+  // Before the fix, this rejection could escape as an unhandledRejection if the
+  // .catch() handler itself threw or if the promise chain had a gap.
+  let unhandledRejectionFired = false;
+  const handler = (reason: unknown) => {
+    if (reason instanceof Error && reason.message === "combo-per-model-timeout") {
+      unhandledRejectionFired = true;
+    }
+  };
+  process.on("unhandledRejection", handler);
+
+  const runner = buildTargetTimeoutRunner({
+    handleSingleModel: (_b, _m, target) =>
+      new Promise<Response>((_resolve, reject) => {
+        const sig = target?.modelAbortSignal;
+        sig?.addEventListener("abort", () => {
+          // Simulate an upstream that rejects on abort (common pattern).
+          reject(new Error(sig.reason?.message ?? "aborted"));
+        });
+      }),
+    comboTargetTimeoutMs: 10,
+    log: noopLog,
+  });
+
+  const res = await runner({}, "test-model");
+  assert.equal(res.status, 504, "timeout must win the race");
+
+  // Drain microtasks — the rejected promise from handleSingleModel should be
+  // caught by the .catch() handler, not surface as unhandledRejection.
+  await new Promise((r) => setImmediate(r));
+  await new Promise((r) => setImmediate(r));
+
+  process.removeListener("unhandledRejection", handler);
+  assert.equal(
+    unhandledRejectionFired,
+    false,
+    "handleSingleModel rejection must be caught, not leak as unhandledRejection"
+  );
+});
+
+test("defensive outer .catch() handles unexpected throws in inner .catch()", async () => {
+  // Edge case: if the inner .catch() handler itself throws (e.g. a broken
+  // Error.prototype.message getter), the outer defensive .catch() must
+  // prevent an unhandledRejection.
+  let unhandledRejectionFired = false;
+  const handler = (reason: unknown) => {
+    if (reason instanceof Error && reason.message === "message getter exploded") {
+      unhandledRejectionFired = true;
+    }
+  };
+  process.on("unhandledRejection", handler);
+
+  const runner = buildTargetTimeoutRunner({
+    handleSingleModel: async () => {
+      const err = new Error("upstream-fail");
+      // Sabotage the message getter to throw in the .catch() handler.
+      Object.defineProperty(err, "message", {
+        get() {
+          throw new Error("message getter exploded");
+        },
+      });
+      throw err;
+    },
+    comboTargetTimeoutMs: 10000, // long enough that timeout doesn't fire
+    log: noopLog,
+  });
+
+  const res = await runner({}, "broken-model");
+  // The defensive outer .catch() should return a 502 instead of letting
+  // the throw escape.
+  assert.equal(res.status, 502, "defensive catch must return 502");
+  assert.match(
+    await res.text(),
+    /message getter exploded/,
+    "error detail must be included in response"
+  );
+
+  // Drain microtasks.
+  await new Promise((r) => setImmediate(r));
+  await new Promise((r) => setImmediate(r));
+
+  process.removeListener("unhandledRejection", handler);
+  assert.equal(
+    unhandledRejectionFired,
+    false,
+    "defensive catch must prevent unhandledRejection from inner .catch() throw"
+  );
+});
+
+test("drainLastTimeoutContexts returns and clears recorded contexts", async () => {
+  // Drain any leftover contexts from previous tests.
+  drainLastTimeoutContexts();
+
+  const runner = buildTargetTimeoutRunner({
+    handleSingleModel: () => new Promise<Response>(() => {}), // never resolves
+    comboTargetTimeoutMs: 10,
+    log: noopLog,
+  });
+
+  // Fire two timeouts to verify the ring buffer.
+  await runner({}, "model-a");
+  await runner({}, "model-b");
+
+  const contexts = drainLastTimeoutContexts();
+  assert.ok(contexts.length >= 1, "at least one context must be recorded");
+  assert.equal(contexts[contexts.length - 1].modelStr, "model-b");
+  assert.equal(contexts[contexts.length - 1].timeoutMs, 10);
+  assert.ok(contexts[contexts.length - 1].abortError instanceof Error);
+  assert.ok(contexts[contexts.length - 1].timestamp > 0);
+
+  // drain clears the buffer.
+  const second = drainLastTimeoutContexts();
+  assert.equal(second.length, 0, "second drain must return empty");
+});

tests/unit/rateLimitManager-queue-timeout.test.ts

@@ -36,3 +36,52 @@ test("multiple sequential withRateLimit calls work", async () => {
   ]);
   assert.deepEqual(results.sort(), ["a", "b"]);
 });
+
+test("abort signal rejection does not leak as unhandledRejection", async () => {
+  // Simulate the combo-per-model-timeout scenario: abort signal fires while
+  // fn is running inside Bottleneck's limiter. The abortPromise rejects and
+  // wins Promise.race, but fn's eventual rejection must be silently caught
+  // (not surface as unhandledRejection).
+  enableRateLimitProtection("test-queue-abort");
+
+  let unhandledRejectionFired = false;
+  const handler = (reason: unknown) => {
+    if (reason instanceof Error && reason.message === "combo-per-model-timeout") {
+      unhandledRejectionFired = true;
+    }
+  };
+  process.on("unhandledRejection", handler);
+
+  const ac = new AbortController();
+  const err = new Error("combo-per-model-timeout");
+
+  // Schedule a slow function, then abort mid-flight.
+  const promise = withRateLimit(
+    "openai",
+    "test-queue-abort",
+    "gpt-4",
+    async () => {
+      // Simulate work that respects the abort signal (like a fetch).
+      await new Promise((r) => setTimeout(r, 200));
+      throw err;
+    },
+    ac.signal
+  );
+
+  // Abort quickly so abortPromise wins the race.
+  setTimeout(() => ac.abort(err), 10);
+
+  // The withRateLimit call itself should reject (from abortPromise).
+  await assert.rejects(promise, (e: Error) => e.message === "combo-per-model-timeout");
+
+  // Give Bottleneck time to finish the orphaned job and let any
+  // unhandledRejection fire.
+  await new Promise((r) => setTimeout(r, 500));
+
+  process.removeListener("unhandledRejection", handler);
+  assert.equal(
+    unhandledRejectionFired,
+    false,
+    "fn rejection after abort must be silently caught, not leak as unhandledRejection"
+  );
+});