Files
OmniRoute/tests/unit/executor-base-utils.test.ts
Beexly 706dc75c13 fix(chatCore): stop executeWithUpstreamStartTimeout leaking its abortPromise listener (hedge-cancelled process exit) (#12406)
* fix(sse): stop mergeAbortSignals from leaking abort listeners

mergeAbortSignals() attached "abort" listeners to its primary/secondary
signals but never removed them once the merged signal settled. Every
executor fetch attempt calls this (fetchWithStartTimeout, once per
URL/retry), so a busy combo request accumulated one live listener per
call on the long-lived combo/client signal. A leaked listener still
fires when that signal is later aborted (e.g. a hedge cancellation
arriving after this merge's own caller already finished), for a merged
output nothing is watching anymore.

Mirrors the already-correct self-cleaning pattern in
open-sse/utils/directResponseStartTimeout.ts's local mergeAbortSignals.

Regression test measures listener growth across repeated merges of the
same long-lived signal: 25 merges leaked exactly 25 listeners pre-fix,
0 post-fix.

(cherry picked from commit 07969655147d3236969b38bfb41280ab4fb52b79)

* fix(server): stop the crash guard re-throwing combo abort reasons

Production crash 2026-08-31 (omniroute.log): on a client disconnect,
handleDisconnect aborted the combo controller and a late abort listener
threw the abort reason on an empty stack:

    Error [AbortError]: hedge-cancelled
        at ... AbortController.abort ... handleDisconnect
    file:///.../src/shared/utils/httpClientAbortGuard.mjs:130  throw err;

isClientAbortError() only knew Node's stream codes and "aborted", so
shouldSwallowUncaught() said false and the guard re-threw, taking the
whole server down.

- Port upstream's AbortError line (name "AbortError" + abort-flavoured
  message) so request_signal_aborted / DOMException aborts are absorbed.
- Add an exact-message match for the combo abort reasons from
  open-sse/services/combo/comboAbortReasons.ts ("hedge-cancelled",
  "combo-per-model-timeout"), name-agnostic because the raw reason is a
  plain Error that only gets name="AbortError" stamped on the way out.
  A losing hedge / stalled target is never a server fault. Inlined so
  this .mjs stays dependency-free for scripts/dev/run-next.mjs.

Tests: port upstream's guard tests, add the exact crash shape, a
child-process replay of the crash (dies pre-fix, survives post-fix), a
genuine-error case that must still crash, and a sync check against
comboAbortReasons.ts. The child-process helper passes a file:// URL, not
a bare path, so the tests run on Windows.

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
(cherry picked from commit 90c9bce8c474b60c37cc63f4421d50feae4c0ad2)

* fix(chatCore): stop executeWithUpstreamStartTimeout leaking its abortPromise listener

Root cause of the 2026-08-31 production exit (Error [AbortError]:
hedge-cancelled), verified by mapping the crash frames in
.build/next/server/chunks/13721.js back to this file:

- The abortPromise abort listener registered on the long-lived client /
  stream signal was never removed in the finally block (only abortListener
  and timeoutAbortListener were), so every executor attempt (and every
  retry) leaked one listener onto that signal.
- Promise.race only subscribes to abortPromise/timeoutPromise once the
  array literal has been evaluated. When execute() threw synchronously the
  race never ran, abortPromise was orphaned, and the next hedge
  cancellation / client disconnect aborted the signal with the string
  reason streamHandler.ts forwards; createAbortError() rebuilt it as an
  AbortError-named Error and rejected a promise nothing awaited. That
  unhandledRejection reached the process crash guard, which re-threw it as
  an uncaughtException and exited with code 7.

Keep a handle to the listener and remove it with the others, and mark the
two race-loser promises as handled so a synchronous throw from execute()
can never orphan them. Race semantics are unchanged (the race still
observes their rejections).

Regression tests: (1) a resolving execute leaves the listener count on
the client signal unchanged; (2) a synchronously throwing execute leaks
no listener and a later abort with the string "hedge-cancelled" produces
no unhandledRejection. Both fail against the previous implementation.

Note: commit 079696551 (mergeAbortSignals cleanup) is correct listener
hygiene but is not on this crash path; this is the fix for the incident.

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
(cherry picked from commit e68a50ad854a1945c23e747da4ef15a820cc148d)

* fix(server): absorb raw string abort reasons in the crash guard; document the verified crash path

Follow-ups from the adversarial review of 90c9bce8c:

- 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 reach
  process level as a bare string. isClientAbortError() returned false for
  every non-object, which would still have exited the process. Absorb the
  combo abort reasons and the stream-handler disconnect reasons when they
  arrive as strings.
- Correct the mechanism comment: the 2026-08-31 exit was a leaked
  upstreamTimeouts.ts abortPromise listener rejecting a promise nothing
  awaited (unhandledRejection), escalated by this guard, not a listener
  throwing synchronously. The leak is fixed at the source in the previous
  commit; this guard remains the last-resort net.
- Reword the inlining rationale (plain node launcher, no reliance on
  type-stripping for the .ts constants module).
- Tests: the child-process replay now also exercises the
  unhandledRejection route with the exact production error shape and with
  raw string reasons; add unit coverage for string reasons and non-object
  look-alikes.

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
(cherry picked from commit 696fcc8fe1b9b9bc43e6e5f5f5e44b619a86e68a)

---------

Co-authored-by: Claude Fable 5.1 <noreply@anthropic.com>
Co-authored-by: diegosouzapw <8016841+diegosouzapw@users.noreply.github.com>
Co-authored-by: Beexly <Beexly@users.noreply.github.com>
2026-09-18 13:13:06 -03:00

186 lines
7.8 KiB
TypeScript

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");
test("mergeUpstreamExtraHeaders skips null/undefined extra", () => {
const h: Record<string, string> = { Authorization: "Bearer x" };
base.mergeUpstreamExtraHeaders(h, null);
assert.deepEqual(h, { Authorization: "Bearer x" });
base.mergeUpstreamExtraHeaders(h, undefined);
assert.deepEqual(h, { Authorization: "Bearer x" });
});
test("mergeUpstreamExtraHeaders merges string key-value pairs", () => {
const h: Record<string, string> = {};
base.mergeUpstreamExtraHeaders(h, { "X-Custom": "val1", "X-Other": "val2" });
assert.equal(h["X-Custom"], "val1");
assert.equal(h["X-Other"], "val2");
});
test("mergeUpstreamExtraHeaders overrides user-agent via setUserAgentHeader", () => {
const h: Record<string, string> = { "User-Agent": "old", "user-agent": "old" };
base.mergeUpstreamExtraHeaders(h, { "user-agent": "new-agent" });
assert.equal(h["User-Agent"], "new-agent");
assert.equal(h["user-agent"], "new-agent");
});
test("mergeUpstreamExtraHeaders skips empty keys", () => {
const h: Record<string, string> = {};
base.mergeUpstreamExtraHeaders(h, { "": "val", "X-Valid": "ok" });
assert.equal(h[""], undefined);
assert.equal(h["X-Valid"], "ok");
});
test("mergeUpstreamExtraHeaders skips non-string values", () => {
const h: Record<string, string> = {};
base.mergeUpstreamExtraHeaders(h, { "X-Num": 123 as any, "X-Bool": true as any });
assert.equal(h["X-Num"], undefined);
assert.equal(h["X-Bool"], undefined);
});
test("getCustomUserAgent returns null for null/undefined", () => {
assert.equal(base.getCustomUserAgent(null), null);
assert.equal(base.getCustomUserAgent(undefined), null);
});
test("getCustomUserAgent returns null for empty string", () => {
assert.equal(base.getCustomUserAgent({ customUserAgent: "" }), null);
assert.equal(base.getCustomUserAgent({ customUserAgent: " " }), null);
});
test("getCustomUserAgent returns trimmed user agent", () => {
assert.equal(base.getCustomUserAgent({ customUserAgent: " MyAgent/1.0 " }), "MyAgent/1.0");
});
test("getCustomUserAgent returns null for non-string customUserAgent", () => {
assert.equal(base.getCustomUserAgent({ customUserAgent: 123 }), null);
});
test("setUserAgentHeader sets User-Agent casing", () => {
const h: Record<string, string> = {};
base.setUserAgentHeader(h, "TestAgent/1.0");
assert.equal(h["User-Agent"], "TestAgent/1.0");
});
test("setUserAgentHeader overwrites existing", () => {
const h: Record<string, string> = { "User-Agent": "old", "user-agent": "old" };
base.setUserAgentHeader(h, "NewAgent/2.0");
assert.equal(h["User-Agent"], "NewAgent/2.0");
assert.equal(h["user-agent"], "NewAgent/2.0");
});
test("applyConfiguredUserAgent does nothing when no custom user agent", () => {
const h: Record<string, string> = { "User-Agent": "default" };
base.applyConfiguredUserAgent(h, null);
assert.equal(h["User-Agent"], "default");
});
test("applyConfiguredUserAgent applies custom user agent", () => {
const h: Record<string, string> = { "User-Agent": "default" };
base.applyConfiguredUserAgent(h, { customUserAgent: "Custom/1.0" });
assert.equal(h["User-Agent"], "Custom/1.0");
});
test("mergeAbortSignals returns secondary if primary already aborted", () => {
const c1 = new AbortController();
const c2 = new AbortController();
c1.abort(new Error("primary aborted"));
const merged = base.mergeAbortSignals(c1.signal, c2.signal);
assert.ok(merged.aborted);
});
test("mergeAbortSignals returns primary if secondary already aborted", () => {
const c1 = new AbortController();
const c2 = new AbortController();
c2.abort(new Error("secondary aborted"));
const merged = base.mergeAbortSignals(c1.signal, c2.signal);
assert.ok(merged.aborted);
});
test("mergeAbortSignals aborts when primary fires", () => {
const c1 = new AbortController();
const c2 = new AbortController();
const merged = base.mergeAbortSignals(c1.signal, c2.signal);
assert.ok(!merged.aborted);
c1.abort(new Error("primary"));
assert.ok(merged.aborted);
});
test("mergeAbortSignals aborts when secondary fires", () => {
const c1 = new AbortController();
const c2 = new AbortController();
const merged = base.mergeAbortSignals(c1.signal, c2.signal);
assert.ok(!merged.aborted);
c2.abort(new Error("secondary"));
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");
assert.deepEqual(result, body);
});
test("sanitizeReasoningEffortForProvider preserves xhigh unless explicitly unsupported", () => {
const body = { reasoning_effort: "xhigh" };
const result = base.sanitizeReasoningEffortForProvider(body, "openai", "gpt-4o") as any;
assert.equal(result.reasoning_effort, "xhigh");
});
test("sanitizeReasoningEffortForProvider preserves high effort", () => {
const body = { reasoning_effort: "high" };
const result = base.sanitizeReasoningEffortForProvider(body, "openai", "gpt-4o") as any;
assert.equal(result.reasoning_effort, "high");
});
test("sanitizeReasoningEffortForProvider preserves medium effort", () => {
const body = { reasoning_effort: "medium" };
const result = base.sanitizeReasoningEffortForProvider(body, "openai", "gpt-4o") as any;
assert.equal(result.reasoning_effort, "medium");
});
test("sanitizeReasoningEffortForProvider returns non-object body as-is", () => {
assert.equal(base.sanitizeReasoningEffortForProvider(null, "openai", "gpt-4o"), null);
assert.equal(base.sanitizeReasoningEffortForProvider("string", "openai", "gpt-4o"), "string");
assert.equal(base.sanitizeReasoningEffortForProvider(42, "openai", "gpt-4o"), 42);
});