fix(chatCore): stop executeWithUpstreamStartTimeout leaking its abortPromise listener (hedge-cancelled process exit)

changelog.d/fixes/12406-upstream-timeouts-abort-listener-leak.md

@@ -0,0 +1 @@
+- **fix(chatCore):** stop `executeWithUpstreamStartTimeout` leaking its abortPromise listener onto the long-lived client/stream signal, and stop `mergeAbortSignals` leaking per-attempt abort listeners, so a later hedge cancellation or client disconnect cannot reject an orphaned promise and take the process down (`Error [AbortError]: hedge-cancelled`). The crash guard also absorbs combo abort reasons (`hedge-cancelled`, `combo-per-model-timeout`) and raw string disconnect reasons as a last-resort net ([#12406](https://github.com/diegosouzapw/OmniRoute/pull/12406) — thanks @Beexly)

open-sse/executors/base.ts

@@ -130,6 +130,8 @@ import { sanitizeReasoningEffortForProvider } from "./base/reasoningEffort.ts";
 // Reasoning-effort sanitation extracted to a pure leaf; re-exported for external
 // importers (mimoThinking service + tests) that import it from "./base.ts".
 export { sanitizeReasoningEffortForProvider } from "./base/reasoningEffort.ts";
+import { mergeAbortSignals } from "./base/mergeAbortSignals.ts";
+export { mergeAbortSignals } from "./base/mergeAbortSignals.ts";
 
 /**
  * Sanitizes a custom API path to prevent path traversal attacks.
@@ -229,29 +231,6 @@ export type CountTokensInput = {
   signal?: AbortSignal | null;
 };
 
-export function mergeAbortSignals(primary: AbortSignal, secondary: AbortSignal): AbortSignal {
-  const controller = new AbortController();
-
-  const abortFrom = (source: AbortSignal) => {
-    if (!controller.signal.aborted) {
-      controller.abort(source.reason);
-    }
-  };
-
-  if (primary.aborted) {
-    abortFrom(primary);
-    return controller.signal;
-  }
-  if (secondary.aborted) {
-    abortFrom(secondary);
-    return controller.signal;
-  }
-
-  primary.addEventListener("abort", () => abortFrom(primary), { once: true });
-  secondary.addEventListener("abort", () => abortFrom(secondary), { once: true });
-  return controller.signal;
-}
-
 import {
   hasActiveClaudeThinking,
   readNestedThinkingBudget,

open-sse/executors/base/mergeAbortSignals.ts

@@ -0,0 +1,47 @@
+// Self-cleaning abort-signal merge. Extracted from base.ts so the frozen
+// executor host does not grow past its file-size cap for this leak fix.
+
+/**
+ * Merge two abort signals into one that fires when either does.
+ *
+ * The `primary`/`secondary` abort listeners registered below MUST be removed once the
+ * merged controller settles — otherwise they outlive this call and stay attached to
+ * whichever input signal is longer-lived (typically the combo/client signal, which
+ * stays open for the whole request while this merge is re-done per executor fetch
+ * attempt/retry). A hedge cancellation or client disconnect arriving after this
+ * particular merge's caller has already moved on then still fires the listener,
+ * detached from anything that's still awaiting it. Mirrors the already-correct
+ * self-cleaning pattern in `open-sse/utils/directResponseStartTimeout.ts`'s local
+ * `mergeAbortSignals` (#hedge-cancelled-abort-listener-leak).
+ */
+export function mergeAbortSignals(primary: AbortSignal, secondary: AbortSignal): AbortSignal {
+  const controller = new AbortController();
+
+  const cleanup = () => {
+    primary.removeEventListener("abort", onPrimaryAbort);
+    secondary.removeEventListener("abort", onSecondaryAbort);
+  };
+
+  const abortFrom = (source: AbortSignal) => {
+    if (!controller.signal.aborted) {
+      controller.abort(source.reason);
+    }
+    cleanup();
+  };
+
+  const onPrimaryAbort = () => abortFrom(primary);
+  const onSecondaryAbort = () => abortFrom(secondary);
+
+  if (primary.aborted) {
+    abortFrom(primary);
+    return controller.signal;
+  }
+  if (secondary.aborted) {
+    abortFrom(secondary);
+    return controller.signal;
+  }
+
+  primary.addEventListener("abort", onPrimaryAbort, { once: true });
+  secondary.addEventListener("abort", onSecondaryAbort, { once: true });
+  return controller.signal;
+}

open-sse/handlers/chatCore/upstreamTimeouts.ts

@@ -123,10 +123,7 @@ export function getExecutorTimeoutMs(
     // Defensive backstop for direct callers: resolveConnectionTimeoutMs is the
     // gate (it rejects out-of-range values so the chain falls through); this
     // clamp only caps values a future caller could pass unvetted.
-    return Math.min(
-      Math.max(0, Math.floor(connectionTimeoutMs)),
-      MAX_PROVIDER_SPECIFIC_TIMEOUT_MS
-    );
+    return Math.min(Math.max(0, Math.floor(connectionTimeoutMs)), MAX_PROVIDER_SPECIFIC_TIMEOUT_MS);
   }
   const modelOverride = resolveModelTimeoutOverride(provider, model);
   if (modelOverride !== undefined) return modelOverride;
@@ -271,15 +268,30 @@ export async function executeWithUpstreamStartTimeout<T>({
     }, timeoutMs);
   });
 
+  let abortPromiseListener: (() => void) | null = null;
   const abortPromise = new Promise<never>((_, reject) => {
-    signal.addEventListener("abort", () => reject(createAbortError(signal)), { once: true });
+    abortPromiseListener = () => reject(createAbortError(signal));
+    signal.addEventListener("abort", abortPromiseListener, { once: true });
   });
+  // Promise.race only subscribes to timeoutPromise/abortPromise once the array
+  // literal below has been evaluated. If execute() throws synchronously the race
+  // never runs, both promises are orphaned, and a later abort of the long-lived
+  // client signal surfaces as an unhandledRejection. That was the 2026-08-31
+  // production exit: a hedge sibling won after a client disconnect, the leaked
+  // listener below rebuilt the string reason as an AbortError, and nothing was
+  // awaiting the promise it rejected. Marking them handled keeps the race
+  // semantics (it still observes the rejections) while closing that path.
+  abortPromise.catch(() => {});
+  timeoutPromise.catch(() => {});
 
   try {
     return await Promise.race([execute(combinedController.signal), timeoutPromise, abortPromise]);
   } finally {
     if (timeoutId) clearTimeout(timeoutId);
     if (abortListener) signal.removeEventListener("abort", abortListener);
+    // Never removed before this fix: one listener leaked onto the client signal
+    // per call (chatCore.ts invokes this once per executor attempt, plus retries).
+    if (abortPromiseListener) signal.removeEventListener("abort", abortPromiseListener);
     if (timeoutAbortListener) {
       timeoutController.signal.removeEventListener("abort", timeoutAbortListener);
     }

src/shared/utils/httpClientAbortGuard.mjs

@@ -48,12 +48,31 @@
  * @module
  */
 
+/**
+ * Abort reasons raised by combo target dispatch. Mirror of
+ * open-sse/services/combo/comboAbortReasons.ts — keep the two in sync.
+ */
+const COMBO_ABORT_REASONS = new Set(["hedge-cancelled", "combo-per-model-timeout"]);
+
+/**
+ * Raw string reasons open-sse/utils/streamHandler.ts passes to
+ * handleDisconnect() / abortController.abort() when the client goes away.
+ */
+const CLIENT_DISCONNECT_REASONS = new Set(["request_signal_aborted", "client_closed", "cancelled"]);
+
 /**
  * @param {unknown} err
  * @returns {boolean} true when `err` represents a client closing the
  *   connection rather than a server-side fault.
  */
 export function isClientAbortError(err) {
+  // open-sse/utils/streamHandler.ts aborts the stream controller with a RAW
+  // STRING reason (getClientAbortReason / handleDisconnect), and undici rejects
+  // with signal.reason verbatim, so a cancellation can surface at the process
+  // level as a bare string rather than an Error object.
+  if (typeof err === "string") {
+    return COMBO_ABORT_REASONS.has(err) || CLIENT_DISCONNECT_REASONS.has(err);
+  }
   if (!err || typeof err !== "object") return false;
   const e = /** @type {NodeJS.ErrnoException} */ (err);
   // Node emits `Error: aborted` (no code) from http.Server#abortIncoming.
@@ -65,6 +84,23 @@ export function isClientAbortError(err) {
   // `Error: aborted` — an emitter-left 'error' event on any of these used to
   // kill the process (#fix-dev-server-aborted).
   if (e.name === "AbortError" && /abort/i.test(String(e.message))) return true;
+  // Combo dispatch cancels a losing hedged target / a stalled target by
+  // aborting with `new Error(reason)` for one of the reasons in
+  // open-sse/services/combo/comboAbortReasons.ts (targetTimeoutRunner.ts).
+  // streamHandler.ts forwards that reason to the stream controller as a plain
+  // string, and a leaked abort listener in
+  // open-sse/handlers/chatCore/upstreamTimeouts.ts (executeWithUpstreamStartTimeout)
+  // rebuilt it via createAbortError() as an AbortError-named Error to reject a
+  // promise nothing was awaiting. That unhandledRejection reached this guard,
+  // which re-threw it as an uncaughtException. Production exit 2026-08-31:
+  //   Error [AbortError]: hedge-cancelled
+  // The listener leak is fixed at the source; this stays as the last-resort net.
+  // A sibling target winning / a target stalling is never a server fault, so
+  // match the exact reason text whatever `name` the thrower stamped on it.
+  // Inlined rather than imported from comboAbortReasons.ts: this file runs
+  // under plain node (scripts/dev/run-next.mjs, no tsx) and must not depend on
+  // type-stripping being available for that .ts module.
+  if (COMBO_ABORT_REASONS.has(String(e.message))) return true;
   switch (e.code) {
     case "ERR_STREAM_PREMATURE_CLOSE":
     case "ECONNRESET":

tests/unit/chatcore-upstream-timeouts.test.ts

@@ -1,10 +1,12 @@
 import test from "node:test";
 import assert from "node:assert/strict";
+import { getEventListeners } from "node:events";
 
 import {
   createBodyTimeoutError,
   createUpstreamStartTimeoutError,
   createAbortError,
+  executeWithUpstreamStartTimeout,
   computeBillableTokens,
   getExecutorTimeoutMs,
   normalizeExecutorResult,
@@ -57,3 +59,66 @@ test("normalizeExecutorResult rejects malformed executor output", () => {
     /must contain a Response/
   );
 });
+test("executeWithUpstreamStartTimeout leaves no abort listener on the client signal after a resolving execute", async () => {
+  const client = new AbortController();
+  const before = getEventListeners(client.signal, "abort").length;
+  const result = await executeWithUpstreamStartTimeout({
+    executor: {},
+    provider: "test-provider",
+    model: "test-model",
+    connectionTimeoutMs: 5_000,
+    signal: client.signal,
+    execute: async () => "ok",
+  });
+  assert.equal(result, "ok");
+  assert.equal(
+    getEventListeners(client.signal, "abort").length,
+    before,
+    "every listener registered for the race must be removed once it settles"
+  );
+});
+
+test("executeWithUpstreamStartTimeout: a synchronously throwing execute cannot orphan abortPromise (2026-08-31 hedge-cancelled exit)", async () => {
+  // Production shape: execute() threw before Promise.race subscribed, so the
+  // abortPromise listener stayed on the long-lived client signal with nobody
+  // awaiting its rejection. The next hedge cancellation / client disconnect
+  // then aborted that signal with a string reason and the rejection became an
+  // unhandledRejection: `Error [AbortError]: hedge-cancelled`.
+  const client = new AbortController();
+  const before = getEventListeners(client.signal, "abort").length;
+  await assert.rejects(
+    executeWithUpstreamStartTimeout({
+      executor: {},
+      provider: "test-provider",
+      model: "test-model",
+      connectionTimeoutMs: 5_000,
+      signal: client.signal,
+      execute: () => {
+        throw new Error("sync failure before the race subscribed");
+      },
+    }),
+    /sync failure before the race subscribed/
+  );
+  assert.equal(
+    getEventListeners(client.signal, "abort").length,
+    before,
+    "no listener may leak onto the client signal when execute() throws synchronously"
+  );
+
+  let unhandled: unknown = null;
+  const onUnhandled = (reason: unknown) => {
+    unhandled = reason;
+  };
+  process.on("unhandledRejection", onUnhandled);
+  try {
+    client.abort("hedge-cancelled");
+    await new Promise((resolve) => setTimeout(resolve, 25));
+  } finally {
+    process.off("unhandledRejection", onUnhandled);
+  }
+  assert.equal(
+    unhandled,
+    null,
+    "a late abort must not surface as an unhandledRejection: " + String(unhandled)
+  );
+});

tests/unit/executor-base-utils.test.ts

@@ -1,5 +1,7 @@
 import test from "node:test";
 import assert from "node:assert/strict";
+import { getEventListeners } from "node:events";
+import type { EventEmitter } from "node:events";
 
 const base = await import("../../open-sse/executors/base.ts");
 
@@ -116,6 +118,42 @@ test("mergeAbortSignals aborts when secondary fires", () => {
   assert.ok(merged.aborted);
 });
 
+// Regression: mergeAbortSignals used to attach its "abort" listeners to `primary`/
+// `secondary` and never remove them. Every executor fetch attempt calls this (once
+// per URL/retry — open-sse/executors/base.ts's fetchWithStartTimeout), so a busy
+// combo request accumulated one live listener per call on the long-lived combo/
+// client signal. Each leaked listener still fires when that signal is aborted later
+// (e.g. a hedge cancellation — COMBO_HEDGE_CANCELLED_REASON, "hedge-cancelled" —
+// arriving after this particular merge's own caller already finished), for a
+// `merged` output nothing is watching anymore. Proven here by counting real
+// listener growth (via node:events' getEventListeners) across repeated merges of
+// the SAME long-lived primary signal — exactly the fetchWithStartTimeout pattern.
+test("mergeAbortSignals removes its listeners once the merged signal settles (no leak across repeated merges)", () => {
+  const countAbortListeners = (target: AbortSignal) =>
+    getEventListeners(target as unknown as EventEmitter, "abort").length;
+
+  const longLived = new AbortController();
+  const before = countAbortListeners(longLived.signal);
+
+  for (let i = 0; i < 25; i++) {
+    const perAttempt = new AbortController();
+    const merged = base.mergeAbortSignals(longLived.signal, perAttempt.signal);
+    assert.ok(!merged.aborted);
+    // Simulate this attempt finishing on its own (its own timeout/abort fires),
+    // the way each fetchWithStartTimeout call's timeoutController does.
+    perAttempt.abort(new Error(`attempt-${i}-timeout`));
+    assert.ok(merged.aborted);
+  }
+
+  const after = countAbortListeners(longLived.signal);
+  assert.ok(
+    after <= before + 1,
+    `expected mergeAbortSignals to clean up its "abort" listener on the long-lived ` +
+      `signal after each merge settles; listener count grew from ~${before} to ${after} ` +
+      `across 25 merges (leak)`
+  );
+});
+
 test("sanitizeReasoningEffortForProvider passes through body without reasoning_effort", () => {
   const body = { model: "gpt-4o", temperature: 0.7 };
   const result = base.sanitizeReasoningEffortForProvider(body, "openai", "gpt-4o");