diff --git a/.env.example b/.env.example index 6a443a9c2d..899ba5b0e7 100644 --- a/.env.example +++ b/.env.example @@ -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). diff --git a/changelog.d/fixes/10528-direct-dispatcher-response-start-timeout.md b/changelog.d/fixes/10528-direct-dispatcher-response-start-timeout.md new file mode 100644 index 0000000000..9354b02822 --- /dev/null +++ b/changelog.d/fixes/10528-direct-dispatcher-response-start-timeout.md @@ -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)) diff --git a/docs/reference/ENVIRONMENT.md b/docs/reference/ENVIRONMENT.md index 3d0e74ff32..1104618012 100644 --- a/docs/reference/ENVIRONMENT.md +++ b/docs/reference/ENVIRONMENT.md @@ -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. | diff --git a/open-sse/utils/directResponseStartTimeout.ts b/open-sse/utils/directResponseStartTimeout.ts new file mode 100644 index 0000000000..90e7b6a04a --- /dev/null +++ b/open-sse/utils/directResponseStartTimeout.ts @@ -0,0 +1,77 @@ +type DirectFetchOptions = RequestInit & { dispatcher?: unknown }; +type DirectFetch = ( + input: RequestInfo | URL, + options: DirectFetchOptions +) => Promise; + +const DEFAULT_DIRECT_HEADERS_TIMEOUT_MS = 30_000; +const DIRECT_RESPONSE_START_TIMEOUT_CODE = "DIRECT_RESPONSE_START_TIMEOUT"; + +export function resolveDirectHeadersTimeoutMs( + env: Record = 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 { + 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); + } +} diff --git a/open-sse/utils/proxyFetch.ts b/open-sse/utils/proxyFetch.ts index 56920ccef5..e8e3ea26f8 100644 --- a/open-sse/utils/proxyFetch.ts +++ b/open-sse/utils/proxyFetch.ts @@ -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) { diff --git a/tests/unit/proxyfetch-direct-response-start-timeout-10214.test.ts b/tests/unit/proxyfetch-direct-response-start-timeout-10214.test.ts new file mode 100644 index 0000000000..fa3a789612 --- /dev/null +++ b/tests/unit/proxyfetch-direct-response-start-timeout-10214.test.ts @@ -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 { + return (input, init) => { + capture.calls++; + capture.dispatchers.push((init as { dispatcher?: unknown } | undefined)?.dispatcher); + return new Promise((_, 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(fn: () => Promise): Promise { + 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 => { + 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((_, reject) => { + init?.signal?.addEventListener("abort", () => reject(init.signal!.reason), { once: true }); + }); + } + return new Response("ok", { status: 200 }); + }; + const mockNative = async (): Promise => + 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 => + 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 => { + 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"); +});