diff --git a/open-sse/services/combo/targetTimeoutRunner.ts b/open-sse/services/combo/targetTimeoutRunner.ts index 09a80559b2..4eb6cb8b68 100644 --- a/open-sse/services/combo/targetTimeoutRunner.ts +++ b/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 { const { handleSingleModel, comboTargetTimeoutMs, log } = deps; + ensureDiagnosticListener(); return async ( b: Record, modelStr: string, @@ -46,11 +117,18 @@ export function buildTargetTimeoutRunner(deps: { const timeoutPromise = new Promise((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) { diff --git a/open-sse/services/rateLimitManager.ts b/open-sse/services/rateLimitManager.ts index 3582eec9c0..a7bdeb2b09 100644 --- a/open-sse/services/rateLimitManager.ts +++ b/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); diff --git a/tests/unit/combo-target-timeout-runner.test.ts b/tests/unit/combo-target-timeout-runner.test.ts index e75fee746b..a058a94552 100644 --- a/tests/unit/combo-target-timeout-runner.test.ts +++ b/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((_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(() => {}), // 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"); +}); diff --git a/tests/unit/rateLimitManager-queue-timeout.test.ts b/tests/unit/rateLimitManager-queue-timeout.test.ts index be0a124941..aa57fc3426 100644 --- a/tests/unit/rateLimitManager-queue-timeout.test.ts +++ b/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" + ); +});