fix(proxyFetch): retry once on undici dispatcher failure before native fallback (#2222)

Integrated into release/v3.8.0

open-sse/utils/proxyFetch.ts

@@ -25,6 +25,12 @@ type FetchWithDispatcher = (
   init?: FetchWithDispatcherOptions
 ) => Promise<Response>;
 
+/** Injectable dependencies for testability (Approach B DI). */
+export type ProxyFetchDeps = {
+  undiciFetch?: FetchWithDispatcher;
+  nativeFetch?: (input: RequestInfo | URL, init?: RequestInit) => Promise<Response>;
+};
+
 type PatchState = {
   originalFetch: typeof globalThis.fetch;
   proxyContext: AsyncLocalStorage<unknown>;
@@ -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<Response>)(input, options);
+    const _undiciDispatcher =
+      deps.undiciFetch ?? (undiciFetch as unknown as (...args: unknown[]) => Promise<Response>);
+    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<Response>)(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<string, unknown>).getReader === "function" || // ReadableStream
+        typeof (_bodyUnknown as Record<string, unknown>).stream === "function"); // Blob
+    const maxAttempts = bodyIsStream ? 1 : 2;
+    const _undiciDirect =
+      deps.undiciFetch ?? (undiciFetch as unknown as (...args: unknown[]) => Promise<Response>);
+    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<Response>)(input, {
+    const _undiciProxy =
+      deps.undiciFetch ?? (undiciFetch as unknown as (...args: unknown[]) => Promise<Response>);
+    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<Response> {
+  return patchedFetch(input, options as FetchWithDispatcherOptions, deps);
+}
+
 if (!isCloud && !patchState.isPatched) {
   globalThis.fetch = patchedFetch;
   patchState.isPatched = true;

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<Response> => {
+    undiciCalls++;
+    throw makeUndiciError("fetch failed");
+  };
+
+  const mockNative = async (
+    _input: RequestInfo | URL,
+    _init?: RequestInit
+  ): Promise<Response> => {
+    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<Response> => {
+    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<Response> => {
+    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<Response> => {
+    undiciCalls++;
+    throw makeUndiciError("fetch failed");
+  };
+
+  const mockNative = async (
+    _input: RequestInfo | URL,
+    _init?: RequestInit
+  ): Promise<Response> => {
+    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");
+});