pjandro/OmniRoute
1
0
Fork 0
mirror of https://github.com/diegosouzapw/OmniRoute.git synced 2026-08-22 23:22:09 +03:00
Code Issues Packages Projects Releases Wiki Activity

Merge pull request #10528 from excessivechaos/fix/direct-dispatcher-timeout-10214

Browse Source 
fix(network): bound direct-path response-start timeout and retry on fresh socket (#10214)
This commit is contained in:
Diego Rodrigues de Sa e Souza
2026-08-20 19:55:57 -03:00
committed by GitHub
parent 9935f80971 142ae93498
commit ce6249cbb7
6 changed files with 267 additions and 38 deletions
Download Patch File Download Diff File Expand all files Collapse all files

8
.env.example
View File

@@ -1361,6 +1361,14 @@ CURSOR_USER_AGENT="Cursor/3.4"
# FETCH_BODY_TIMEOUT_MS=600000 # Time to receive full response body
# FETCH_CONNECT_TIMEOUT_MS=30000 # TCP connection establishment (default: 30s)
# FETCH_KEEPALIVE_TIMEOUT_MS=4000 # Keep-alive socket idle timeout (default: 4s)
# OMNIROUTE_DIRECT_HEADERS_TIMEOUT_MS=30000 # Bounded response-start window per direct
# # (no-proxy) attempt (#10214). A silently-dropped
# # pooled keep-alive socket surfaces no transport
# # error, so without this bound a direct request can
# # stall until undici's headersTimeout (600s) or the
# # caller's deadline; on expiry the request retries
# # once on a fresh no-keep-alive socket. 0 disables
# # the bound (default: 30000 = 30s).
# Default timeout (ms) for src/shared/utils/fetchTimeout.ts. Acts as the
# fallback when FETCH_TIMEOUT_MS is unset. Default: 120000 (2 min).

1
changelog.d/fixes/10528-direct-dispatcher-response-start-timeout.md Normal file
View File

@@ -0,0 +1 @@
- **fix(network):** direct (no-proxy) egress now bounds each attempt's response-start window (default 30s, `OMNIROUTE_DIRECT_HEADERS_TIMEOUT_MS`) and retries once on a fresh no-keep-alive socket, so a silently-dropped pooled keep-alive connection can no longer stall direct providers (opencode-go, command-code) until a service restart ([#10214](https://github.com/diegosouzapw/OmniRoute/issues/10214))

1
docs/reference/ENVIRONMENT.md
View File

@@ -732,6 +732,7 @@ REQUEST_TIMEOUT_MS (global override)
| `OMNIROUTE_AGENT_GOAL_STREAM_RECOVERY` | `true` | Enable early stream recovery automatically for detected `/goal` agent runs. Set `false`/`0`/`off` to disable the goal-specific opt-in. This can only ADD recovery on top of the operator default — it never overrides an explicit `STREAM_RECOVERY_ENABLED`/DB settings opt-out. |
| `OMNIROUTE_CODEX_DROP_NONSTANDARD_EVENTS` | _(off)_ | Strip non-standard `codex.*` SSE events (e.g. `codex.rate_limits`) that break the OpenAI SDK's `responses.stream()` with a 502. Set `true`/`1`/`yes` to enable. |
| `FETCH_HEADERS_TIMEOUT_MS` | = `FETCH_TIMEOUT_MS` | Time to receive response headers. |
| `OMNIROUTE_DIRECT_HEADERS_TIMEOUT_MS` | `30000` (30s) | Maximum response-start wait (ms) for each direct no-proxy attempt. A timeout retries once on a fresh socket; set `0` to disable the bound and retain the previous behavior. |
| `FETCH_BODY_TIMEOUT_MS` | = `FETCH_TIMEOUT_MS` | Time to receive the full response body. |
| `FETCH_CONNECT_TIMEOUT_MS` | `30000` | TCP connection establishment timeout. |
| `FETCH_KEEPALIVE_TIMEOUT_MS` | `4000` | Keep-alive socket idle timeout. |

77
open-sse/utils/directResponseStartTimeout.ts Normal file
View File

@@ -0,0 +1,77 @@
type DirectFetchOptions = RequestInit & { dispatcher?: unknown };
type DirectFetch = (
input: RequestInfo | URL,
options: DirectFetchOptions
) => Promise<Response>;
const DEFAULT_DIRECT_HEADERS_TIMEOUT_MS = 30_000;
const DIRECT_RESPONSE_START_TIMEOUT_CODE = "DIRECT_RESPONSE_START_TIMEOUT";
export function resolveDirectHeadersTimeoutMs(
env: Record<string, string | undefined> = process.env
): number {
const raw = env.OMNIROUTE_DIRECT_HEADERS_TIMEOUT_MS;
if (raw == null || raw.trim() === "") return DEFAULT_DIRECT_HEADERS_TIMEOUT_MS;
const parsed = Number(raw);
return Number.isFinite(parsed) && parsed > 0 ? Math.floor(parsed) : 0;
}
function createDirectResponseStartTimeout(timeoutMs: number): Error & { code: string } {
const err = new Error(
`Direct response did not start within ${timeoutMs}ms — retrying on a fresh socket`
) as Error & { code: string };
err.name = "TimeoutError";
err.code = DIRECT_RESPONSE_START_TIMEOUT_CODE;
return err;
}
export function isDirectResponseStartTimeout(err: unknown): boolean {
return (
!!err &&
typeof err === "object" &&
"code" in err &&
err.code === DIRECT_RESPONSE_START_TIMEOUT_CODE
);
}
function mergeAbortSignals(
primary: AbortSignal | null | undefined,
secondary: AbortSignal
): AbortSignal {
if (!primary) return secondary;
if (primary.aborted) return primary;
const controller = new AbortController();
const onPrimaryAbort = () => controller.abort(primary.reason);
const onSecondaryAbort = () => controller.abort(secondary.reason);
const cleanup = () => {
primary.removeEventListener("abort", onPrimaryAbort);
secondary.removeEventListener("abort", onSecondaryAbort);
};
primary.addEventListener("abort", onPrimaryAbort, { once: true });
secondary.addEventListener("abort", onSecondaryAbort, { once: true });
controller.signal.addEventListener("abort", cleanup, { once: true });
return controller.signal;
}
export async function directFetchWithBoundedResponseStart(
input: RequestInfo | URL,
options: DirectFetchOptions,
fetchImpl: DirectFetch,
timeoutMs: number
): Promise<Response> {
if (!timeoutMs || timeoutMs <= 0) return fetchImpl(input, options);
const attemptController = new AbortController();
const timer = setTimeout(
() => attemptController.abort(createDirectResponseStartTimeout(timeoutMs)),
timeoutMs
);
timer.unref?.();
try {
return await fetchImpl(input, {
...options,
signal: mergeAbortSignals(options.signal, attemptController.signal),
});
} finally {
clearTimeout(timer);
}
}

75
open-sse/utils/proxyFetch.ts
View File

@@ -19,6 +19,11 @@ import {
isControlPlaneProxyDirectFallbackEnabled,
isFeatureFlagEnabled,
} from "@/shared/utils/featureFlags";
import {
directFetchWithBoundedResponseStart,
isDirectResponseStartTimeout,
resolveDirectHeadersTimeoutMs,
} from "./directResponseStartTimeout.ts";
// #9100: relay egress (Vercel / Deno / Cloudflare edge functions) used to go
// through bare `originalFetch` — NO connection pooling, NO timeout, NO retry.
@@ -154,7 +159,6 @@ type TlsFingerprintStore = {
provider?: string | null;
sessionScope?: string;
};
/**
* #5217 (Gap-secondary): a mutable sink that records the proxy actually applied
* by `runWithProxyContext` for the in-flight request. Executors that pin their
@@ -802,15 +806,7 @@ async function patchedFetch(
(deps.nativeFetch as FetchWithDispatcher | undefined) ?? originalFetchWithDispatcher;
return _nativeFetch(input, options);
}
// Direct connection (no proxy) — use undici with custom dispatcher for timeout control.
// Falls back to original native fetch if dispatcher initialization fails (#1054).
// Retries once on transient dispatcher errors before falling back (fix: proxyfetch-undici-retry).
//
// Non-replayable body guard: if the body is stream-like (ReadableStream/Blob)
// or the input is a Request that carries a body, the first dispatcher attempt
// owns that body. Retrying or falling back to native fetch would replay a
// consumed/locked body and can mask the original transport error with
// "Response body object should not be disturbed or locked".
// Direct undici path: bound response-start, fresh-socket retry, and body guard.
const hasNonReplayableBody = requestHasNonReplayableBody(input, options);
const maxAttempts = hasNonReplayableBody ? 1 : 2;
const _undiciDirect =
@@ -818,32 +814,44 @@ async function patchedFetch(
const _nativeFallback =
(deps.nativeFetch as FetchWithDispatcher | undefined) ?? originalFetchWithDispatcher;
let lastDispatcherError: unknown = null;
const directHeadersTimeoutMs = resolveDirectHeadersTimeoutMs();
let targetHostForLogs = "";
try {
targetHostForLogs = new URL(targetUrl).host;
} catch {
// ignore — logging is best-effort
}
for (let attempt = 0; attempt < maxAttempts; attempt++) {
try {
return await _undiciDirect(input, {
...options,
// #4252: first attempt uses the pooled keep-alive dispatcher; a retry
// (after a transient socket error) uses the no-keep-alive dispatcher so
// it opens a FRESH socket instead of grabbing another stale pooled one
// — the burst pattern was the retry re-hitting a dead pooled socket and
// then falling through to native fetch (which also pools) → 502.
dispatcher: attempt === 0 ? getDefaultDispatcher() : getRetryDispatcher(),
});
return await directFetchWithBoundedResponseStart(
input,
{
...options,
dispatcher: attempt === 0 ? getDefaultDispatcher() : getRetryDispatcher(),
},
_undiciDirect,
directHeadersTimeoutMs
);
} catch (dispatcherError) {
if (isDirectResponseStartTimeout(dispatcherError)) {
if (attempt === 0 && maxAttempts > 1) {
console.warn(
`[ProxyFetch] Direct response-start timeout (${directHeadersTimeoutMs}ms) on pooled dispatcher — retrying on fresh no-keep-alive dispatcher: ${targetHostForLogs}`
);
lastDispatcherError = dispatcherError;
continue;
}
throw dispatcherError;
}
const msg =
dispatcherError instanceof Error ? dispatcherError.message : String(dispatcherError);
// CAUTION: Do NOT fallback to native fetch if the error is a version mismatch (invalid onRequestStart)
// because the native fetch will definitely fail with the undici v8 dispatcher.
if (msg.includes("onRequestStart")) {
console.error(
`[ProxyFetch] Fatal version mismatch: Dispatcher (v8) vs Fetch (v6/native). Hardware upgrade or SOCKS5 config isolation required. Error: ${msg}`
);
throw dispatcherError;
}
// Only retry/fallback for connection/dispatcher errors, not HTTP errors.
// Prefer the .code property when available (more stable across undici
// versions than message-string matching); fall back to substring match
// for errors that lack a structured code.
// Retry/fallback only for connection errors, never HTTP errors.
tagProxyUnreachable(dispatcherError);
const errCode = (dispatcherError as { code?: unknown })?.code;
if (
@@ -854,10 +862,7 @@ async function patchedFetch(
msg.includes("UND_ERR")
) {
if (attempt === 0 && maxAttempts > 1) {
// First failure — retry once after a short backoff before giving up.
// Delay is OMNIROUTE_RETRY_BACKOFF_MS (default 10ms): a fixed backoff
// beats random jitter here because the retry opens a fresh socket, so
// jitter was pure added latency with no herd benefit.
// Retry after a short fixed backoff on a fresh socket.
lastDispatcherError = dispatcherError;
await new Promise((r) => setTimeout(r, RETRY_BACKOFF_MS));
continue;
@@ -873,7 +878,7 @@ async function patchedFetch(
throw tagProxyUnreachable(dispatcherError);
}
// All attempts exhausted — try proxy fallback before native fetch
// Exhausted attempts: try proxy fallback before native fetch.
if (
!tlsDirectFallback &&
source === "direct" &&
@@ -899,20 +904,14 @@ async function patchedFetch(
}
}
}
// Preserve original phrase intact for monitoring: "Undici dispatcher failed, falling back to native fetch"
// #4252: append the flattened err.cause (code/syscall/errno/address) — the bare
// "fetch failed" message hides what actually broke, making bursts undiagnosable.
// Preserve the original monitoring phrase and append the transport cause.
console.warn(
`[ProxyFetch] Undici dispatcher failed, falling back to native fetch (after retry): ${describeFetchCause(dispatcherError)}`
);
try {
return await _nativeFallback(input, options);
} catch (nativeError) {
// #4252: both the undici dispatcher AND native fetch failed. Surface BOTH
// causes (server log) and tag the propagated error so the combo executor sees
// a diagnosable failure IMMEDIATELY instead of a bare "fetch failed" — the
// latter left jobs sitting until the 30s semaphore queue timeout, which then
// tripped the circuit breaker.
// Surface both dispatcher and native causes immediately.
const detail = `dispatcher=[${describeFetchCause(dispatcherError)}] native=[${describeFetchCause(nativeError)}]`;
console.warn(`[ProxyFetch] native fetch fallback ALSO failed: ${detail}`);
if (nativeError instanceof Error) {

143
tests/unit/proxyfetch-direct-response-start-timeout-10214.test.ts Normal file
View File

@@ -0,0 +1,143 @@
/**
* #10214 — Direct (no-proxy) requests stall on a silently-dropped pooled
* keep-alive socket until the caller's deadline or a service restart.
*
* The default direct dispatcher pools keep-alive sockets for up to
* `fetchKeepAliveTimeoutMs` (4 s). A socket that silently drops (half-open, no
* RST) surfaces NO transport error — undici's headersTimeout (600 s default) is
* the only guard, so the existing fresh-socket retry (which fires on
* UND_ERR/ECONNRESET/fetch-failed) never triggers. Observed live: opencode-go
* and command-code stall 100% of routed requests until `systemctl restart`.
*
* The fix bounds the response-start window per direct attempt
* (`OMNIROUTE_DIRECT_HEADERS_TIMEOUT_MS`, default 30 s) and retries once on the
* fresh no-keep-alive dispatcher (a brand-new socket) when the pooled attempt
* times out — converting the zombie-socket stall into a clean failover.
*/
import { test } from "node:test";
import assert from "node:assert/strict";
import { proxyFetch } from "../../open-sse/utils/proxyFetch.ts";
import { getDefaultDispatcher, getRetryDispatcher } from "../../open-sse/utils/proxyDispatcher.ts";
const DIRECT_RESPONSE_START_TIMEOUT_CODE = "DIRECT_RESPONSE_START_TIMEOUT";
/** Simulates a silent half-open pooled socket: the request never resolves, but
* observes the abort signal like real undici does (rejects with the reason). */
function hangingFetch(capture: {
calls: number;
dispatchers: unknown[];
}): (input: RequestInfo | URL, init?: RequestInit) => Promise<Response> {
return (input, init) => {
capture.calls++;
capture.dispatchers.push((init as { dispatcher?: unknown } | undefined)?.dispatcher);
return new Promise<Response>((_, reject) => {
const signal = init?.signal;
signal?.addEventListener(
"abort",
() =>
reject(signal.reason instanceof Error ? signal.reason : new Error(String(signal.reason))),
{ once: true }
);
// never resolve — the upstream accepted the connection but sends nothing
});
};
}
function withFastTimeout<T>(fn: () => Promise<T>): Promise<T> {
process.env.OMNIROUTE_DIRECT_HEADERS_TIMEOUT_MS = "50";
return fn().finally(() => {
delete process.env.OMNIROUTE_DIRECT_HEADERS_TIMEOUT_MS;
});
}
test("#10214 a response-start timeout on the pooled attempt retries on the FRESH no-keep-alive dispatcher", async () => {
const capture = { calls: 0, dispatchers: [] as unknown[] };
const mockUndici = async (input: RequestInfo | URL, init?: RequestInit): Promise<Response> => {
capture.calls++;
capture.dispatchers.push((init as { dispatcher?: unknown } | undefined)?.dispatcher);
if (capture.calls === 1) {
// First attempt hits a silently-dead pooled socket — hang, no error.
return new Promise<Response>((_, reject) => {
init?.signal?.addEventListener("abort", () => reject(init.signal!.reason), { once: true });
});
}
return new Response("ok", { status: 200 });
};
const mockNative = async (): Promise<Response> =>
new Response("native-should-not-fire", { status: 200 });
const res = await withFastTimeout(() =>
proxyFetch(
"https://opencode.ai/zen/go/v1/chat/completions",
{ method: "POST" },
{ undiciFetch: mockUndici, nativeFetch: mockNative }
)
);
assert.equal(capture.calls, 2, "pooled attempt times out and must retry once");
assert.equal(await res.text(), "ok");
// The regression guard: attempt 0 used the pooled keep-alive dispatcher; the
// retry used the fresh no-keep-alive dispatcher — a DIFFERENT instance, so the
// retry opens a brand-new socket that cannot be the zombie.
assert.equal(
capture.dispatchers[0],
getDefaultDispatcher(),
"first attempt must use the pooled default dispatcher"
);
assert.equal(
capture.dispatchers[1],
getRetryDispatcher(),
"timeout retry must use the fresh no-keep-alive dispatcher"
);
assert.notEqual(capture.dispatchers[0], capture.dispatchers[1]);
});
test("#10214 when the fresh-dispatcher retry also stalls, the timeout surfaces (no native fallback)", async () => {
const capture = { calls: 0, dispatchers: [] as unknown[] };
const mockUndici = hangingFetch(capture);
const mockNative = async (): Promise<Response> =>
new Response("native-should-not-fire", { status: 200 });
await assert.rejects(
withFastTimeout(() =>
proxyFetch(
"https://opencode.ai/zen/go/v1/chat/completions",
{ method: "POST" },
{ undiciFetch: mockUndici, nativeFetch: mockNative }
)
),
(err: unknown) => {
assert.equal(
(err as { code?: unknown }).code,
DIRECT_RESPONSE_START_TIMEOUT_CODE,
"final failure must be the classified direct response-start timeout"
);
return true;
}
);
assert.equal(capture.calls, 2, "both attempts must have been made");
assert.equal(capture.dispatchers[0], getDefaultDispatcher());
assert.equal(capture.dispatchers[1], getRetryDispatcher());
});
test("#10214 a healthy fast response is untouched by the bound (single attempt, no retry)", async () => {
const capture = { calls: 0, dispatchers: [] as unknown[] };
const mockUndici = async (input: RequestInfo | URL, init?: RequestInit): Promise<Response> => {
capture.calls++;
capture.dispatchers.push((init as { dispatcher?: unknown } | undefined)?.dispatcher);
return new Response("ok", { status: 200 });
};
const res = await withFastTimeout(() =>
proxyFetch(
"https://opencode.ai/zen/go/v1/chat/completions",
{ method: "POST" },
{ undiciFetch: mockUndici }
)
);
assert.equal(capture.calls, 1, "healthy request must not retry");
assert.equal(capture.dispatchers[0], getDefaultDispatcher());
assert.equal(await res.text(), "ok");
});