From 29b9f27919dcb0c9b16bda3b85d7d2be581628bc Mon Sep 17 00:00:00 2001 From: Anton <39598727+NomenAK@users.noreply.github.com> Date: Thu, 14 May 2026 05:08:35 +0200 Subject: [PATCH] fix(proxyFetch): retry once on undici dispatcher failure before native fallback (#2222) Integrated into release/v3.8.0 --- open-sse/utils/proxyFetch.ts | 115 ++++++++++++++---- tests/unit/proxyfetch-undici-retry.test.ts | 133 +++++++++++++++++++++ 2 files changed, 225 insertions(+), 23 deletions(-) create mode 100644 tests/unit/proxyfetch-undici-retry.test.ts diff --git a/open-sse/utils/proxyFetch.ts b/open-sse/utils/proxyFetch.ts index 6ba4b02999..ddb4e59d20 100644 --- a/open-sse/utils/proxyFetch.ts +++ b/open-sse/utils/proxyFetch.ts @@ -25,6 +25,12 @@ type FetchWithDispatcher = ( init?: FetchWithDispatcherOptions ) => Promise; +/** Injectable dependencies for testability (Approach B DI). */ +export type ProxyFetchDeps = { + undiciFetch?: FetchWithDispatcher; + nativeFetch?: (input: RequestInfo | URL, init?: RequestInit) => Promise; +}; + type PatchState = { originalFetch: typeof globalThis.fetch; proxyContext: AsyncLocalStorage; @@ -202,12 +208,18 @@ export async function runWithProxyContext(proxyConfig, fn) { }); } -async function patchedFetch(input: RequestInfo | URL, options: FetchWithDispatcherOptions = {}) { +async function patchedFetch( + input: RequestInfo | URL, + options: FetchWithDispatcherOptions = {}, + deps: ProxyFetchDeps = {} +) { if (options?.dispatcher) { // When a dispatcher is present, we MUST use the undici library fetch // to ensure version compatibility. Node 22 built-in fetch (undici v6) // is incompatible with undici v8 dispatchers (missing onRequestStart, etc.) - return (undiciFetch as unknown as (...args: unknown[]) => Promise)(input, options); + const _undiciDispatcher = + deps.undiciFetch ?? (undiciFetch as unknown as (...args: unknown[]) => Promise); + return _undiciDispatcher(input, options); } const targetUrl = getTargetUrl(input); @@ -243,34 +255,78 @@ async function patchedFetch(input: RequestInfo | URL, options: FetchWithDispatch } // Direct connection (no proxy) — use undici with custom dispatcher for timeout control. // Falls back to original native fetch if dispatcher initialization fails (#1054). - try { - return await (undiciFetch as unknown as (...args: unknown[]) => Promise)(input, { - ...options, - dispatcher: getDefaultDispatcher(), - }); - } catch (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}` - ); + // Retries once on transient dispatcher errors before falling back (fix: proxyfetch-undici-retry). + // + // ReadableStream/Blob body guard: if the body is non-replayable, skip the retry because + // the first attempt drains the stream; a second attempt would silently send an empty body. + // ReadableStream check: cast through unknown to avoid explicit-any budget (T11). + const _bodyUnknown = options.body as unknown; + const bodyIsStream = + _bodyUnknown !== null && + _bodyUnknown !== undefined && + typeof _bodyUnknown === "object" && + (typeof (_bodyUnknown as Record).getReader === "function" || // ReadableStream + typeof (_bodyUnknown as Record).stream === "function"); // Blob + const maxAttempts = bodyIsStream ? 1 : 2; + const _undiciDirect = + deps.undiciFetch ?? (undiciFetch as unknown as (...args: unknown[]) => Promise); + const _nativeFallback = + (deps.nativeFetch as FetchWithDispatcher | undefined) ?? originalFetchWithDispatcher; + let lastDispatcherError: unknown = null; + for (let attempt = 0; attempt < maxAttempts; attempt++) { + try { + return await _undiciDirect(input, { + ...options, + dispatcher: getDefaultDispatcher(), + }); + } catch (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. + const errCode = (dispatcherError as { code?: string })?.code; + if ( + msg.includes("fetch failed") || + errCode === "ECONNREFUSED" || + msg.includes("ECONNREFUSED") || + (errCode !== undefined && errCode.startsWith("UND_ERR")) || + msg.includes("UND_ERR") + ) { + if (attempt === 0 && maxAttempts > 1) { + // First failure — retry once with a short jittered delay before giving up. + lastDispatcherError = dispatcherError; + await new Promise((r) => setTimeout(r, 25 + Math.random() * 50)); + continue; + } + // All attempts exhausted — fall back to native fetch. + // Preserve original phrase intact for monitoring: "Undici dispatcher failed, falling back to native fetch" + console.warn( + `[ProxyFetch] Undici dispatcher failed, falling back to native fetch (after retry): ${msg}` + ); + return _nativeFallback(input, options); + } throw dispatcherError; } - // Only fallback for connection/dispatcher errors, not HTTP errors - if (msg.includes("fetch failed") || msg.includes("ECONNREFUSED") || msg.includes("UND_ERR")) { - console.warn(`[ProxyFetch] Undici dispatcher failed, falling back to native fetch: ${msg}`); - return originalFetchWithDispatcher(input, options); - } - throw dispatcherError; } + // Should not be reached, but satisfy TypeScript control-flow. + throw lastDispatcherError; } try { const dispatcher = createProxyDispatcher(proxyUrl); - return await (undiciFetch as unknown as (...args: unknown[]) => Promise)(input, { + const _undiciProxy = + deps.undiciFetch ?? (undiciFetch as unknown as (...args: unknown[]) => Promise); + return await _undiciProxy(input, { ...options, dispatcher, }); @@ -281,6 +337,19 @@ async function patchedFetch(input: RequestInfo | URL, options: FetchWithDispatch } } +/** + * Named export for proxyFetch — identical to the patched globalThis.fetch but + * accepts an optional ProxyFetchDeps for unit test dependency injection. + * Production code should use globalThis.fetch (or the default export) instead. + */ +export async function proxyFetch( + input: RequestInfo | URL, + options: RequestInit = {}, + deps: ProxyFetchDeps = {} +): Promise { + return patchedFetch(input, options as FetchWithDispatcherOptions, deps); +} + if (!isCloud && !patchState.isPatched) { globalThis.fetch = patchedFetch; patchState.isPatched = true; diff --git a/tests/unit/proxyfetch-undici-retry.test.ts b/tests/unit/proxyfetch-undici-retry.test.ts new file mode 100644 index 0000000000..f5a0e91d19 --- /dev/null +++ b/tests/unit/proxyfetch-undici-retry.test.ts @@ -0,0 +1,133 @@ +// Tests for the undici dispatcher retry logic in proxyFetch. +// +// Approach B (dependency injection): proxyFetch.ts exports a named `proxyFetch` function +// that accepts optional `deps: ProxyFetchDeps` for injecting mock undici/native fetch +// implementations. This makes retry count assertions precise and deterministic without +// requiring mock.module() (which is unavailable in this Node 25 / tsx/ESM setup). +// +// All three tests MUST fail when the retry loop is removed from proxyFetch.ts (sanity check). + +import { test } from "node:test"; +import assert from "node:assert/strict"; +import { proxyFetch } from "../../open-sse/utils/proxyFetch.ts"; + +function makeUndiciError(msg = "fetch failed", code = "UND_ERR_SOCKET"): Error { + const err = new Error(msg) as Error & { code?: string }; + err.code = code; + return err; +} + +test("undici is called exactly twice then native fallback fires once (both undici attempts fail)", async () => { + let undiciCalls = 0; + let nativeCalls = 0; + + const mockUndici = async ( + _input: RequestInfo | URL, + _init?: RequestInit + ): Promise => { + undiciCalls++; + throw makeUndiciError("fetch failed"); + }; + + const mockNative = async ( + _input: RequestInfo | URL, + _init?: RequestInit + ): Promise => { + nativeCalls++; + return new Response("native-fallback-body", { status: 200 }); + }; + + const res = await proxyFetch( + "https://example.invalid/test", + { method: "GET" }, + { undiciFetch: mockUndici, nativeFetch: mockNative } + ); + + assert.equal(undiciCalls, 2, "undici must be called exactly twice (initial + retry)"); + assert.equal( + nativeCalls, + 1, + "native fallback must fire exactly once after both undici attempts fail" + ); + assert.equal(await res.text(), "native-fallback-body"); +}); + +test("retry-succeeds: undici fails once then succeeds, native fallback is NOT invoked", async () => { + let undiciCalls = 0; + let nativeCalls = 0; + + const mockUndici = async ( + _input: RequestInfo | URL, + _init?: RequestInit + ): Promise => { + undiciCalls++; + if (undiciCalls === 1) { + throw makeUndiciError("fetch failed"); + } + return new Response("undici-retry-success", { status: 200 }); + }; + + const mockNative = async ( + _input: RequestInfo | URL, + _init?: RequestInit + ): Promise => { + nativeCalls++; + return new Response("should-not-be-called", { status: 200 }); + }; + + const res = await proxyFetch( + "https://example.invalid/test", + { method: "GET" }, + { undiciFetch: mockUndici, nativeFetch: mockNative } + ); + + assert.equal( + undiciCalls, + 2, + "undici must be called exactly twice (initial fail + retry success)" + ); + assert.equal(nativeCalls, 0, "native fallback must NOT be invoked when retry succeeds"); + assert.equal(await res.text(), "undici-retry-success"); +}); + +test("does not retry when body is a ReadableStream (non-replayable body)", async () => { + let undiciCalls = 0; + let nativeCalls = 0; + + const mockUndici = async ( + _input: RequestInfo | URL, + _init?: RequestInit + ): Promise => { + undiciCalls++; + throw makeUndiciError("fetch failed"); + }; + + const mockNative = async ( + _input: RequestInfo | URL, + _init?: RequestInit + ): Promise => { + nativeCalls++; + return new Response("native-stream-fallback", { status: 200 }); + }; + + const stream = new ReadableStream({ + start(controller) { + controller.enqueue(new Uint8Array([1, 2, 3])); + controller.close(); + }, + }); + + const res = await proxyFetch( + "https://example.invalid/test", + { method: "POST", body: stream }, + { undiciFetch: mockUndici, nativeFetch: mockNative } + ); + + assert.equal( + undiciCalls, + 1, + "undici must NOT retry when body is a ReadableStream (called exactly once)" + ); + assert.equal(nativeCalls, 1, "native fallback fires after single undici attempt"); + assert.equal(await res.text(), "native-stream-fallback"); +});