fix(sse): surface undici err.cause on dispatcher failure (#4252)

open-sse/utils/proxyFetch.ts

@@ -32,6 +32,54 @@ type FetchWithDispatcher = (
   init?: FetchWithDispatcherOptions
 ) => Promise<Response>;
 
+/**
+ * Flatten a fetch error's `cause` chain (and any Happy-Eyeballs `AggregateError`
+ * sub-errors) into a single diagnostic line: code/syscall/errno/address:port + a
+ * truncated message. undici/native both reject with a bare `TypeError: fetch failed`
+ * whose real reason hides in `.cause`; surfacing it is what makes dispatcher-failure
+ * bursts (#4252) diagnosable. Never includes a stack trace (Rule #12). Pure + testable.
+ */
+export function describeFetchCause(err: unknown): string {
+  const parts: string[] = [];
+  const seen = new Set<unknown>();
+  let cur: unknown = err;
+  for (let depth = 0; cur && depth < 5 && !seen.has(cur); depth++) {
+    seen.add(cur);
+    const e = cur as Record<string, unknown>;
+    const seg = [
+      typeof e.name === "string" && e.name !== "Error" ? e.name : null,
+      typeof e.message === "string" ? e.message.slice(0, 160) : null,
+      e.code != null ? `code=${String(e.code)}` : null,
+      e.syscall != null ? `syscall=${String(e.syscall)}` : null,
+      e.errno != null ? `errno=${String(e.errno)}` : null,
+      e.address != null
+        ? `address=${String(e.address)}${e.port != null ? `:${String(e.port)}` : ""}`
+        : null,
+    ]
+      .filter(Boolean)
+      .join(" ");
+    if (seg) parts.push(seg);
+    if (Array.isArray(e.errors)) {
+      for (const sub of (e.errors as unknown[]).slice(0, 4)) {
+        const s = (sub ?? {}) as Record<string, unknown>;
+        const subSeg = [
+          s.code != null ? `code=${String(s.code)}` : null,
+          s.syscall != null ? `syscall=${String(s.syscall)}` : null,
+          s.address != null
+            ? `address=${String(s.address)}${s.port != null ? `:${String(s.port)}` : ""}`
+            : null,
+        ]
+          .filter(Boolean)
+          .join(" ");
+        if (subSeg) parts.push(`↳ ${subSeg}`);
+        else if (typeof s.message === "string") parts.push(`↳ ${s.message.slice(0, 80)}`);
+      }
+    }
+    cur = e.cause;
+  }
+  return parts.join(" | ") || String(err);
+}
+
 /** Injectable dependencies for testability (Approach B DI). */
 export type ProxyFetchDeps = {
   undiciFetch?: FetchWithDispatcher;
@@ -424,10 +472,26 @@ 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.
           console.warn(
-            `[ProxyFetch] Undici dispatcher failed, falling back to native fetch (after retry): ${msg}`
+            `[ProxyFetch] Undici dispatcher failed, falling back to native fetch (after retry): ${describeFetchCause(dispatcherError)}`
           );
-          return _nativeFallback(input, options);
+          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.
+            const detail = `dispatcher=[${describeFetchCause(dispatcherError)}] native=[${describeFetchCause(nativeError)}]`;
+            console.warn(`[ProxyFetch] native fetch fallback ALSO failed: ${detail}`);
+            if (nativeError instanceof Error) {
+              (nativeError as Error & { proxyFetchDetail?: string }).proxyFetchDetail = detail;
+            }
+            throw nativeError;
+          }
         }
         throw dispatcherError;
       }

tests/unit/proxyfetch-undici-retry.test.ts

@@ -9,7 +9,7 @@
 
 import { test } from "node:test";
 import assert from "node:assert/strict";
-import { proxyFetch } from "../../open-sse/utils/proxyFetch.ts";
+import { proxyFetch, describeFetchCause } 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 };
@@ -154,3 +154,81 @@ test("does not retry when body is a ReadableStream (non-replayable body)", async
   assert.equal(nativeCalls, 1, "native fallback fires after single undici attempt");
   assert.equal(await res.text(), "native-stream-fallback");
 });
+
+// ── #4252: surface err.cause so silent "fetch failed" dispatcher bursts are diagnosable ──
+// The bare undici/native "fetch failed" message hides the real cause (ECONNRESET vs
+// UND_ERR_CONNECT_TIMEOUT vs ENETUNREACH). describeFetchCause flattens the cause chain
+// (and Happy-Eyeballs AggregateError sub-errors) into one diagnostic line — never a stack.
+
+test("#4252 describeFetchCause flattens the cause chain (code/syscall/errno/address)", () => {
+  const cause = Object.assign(new Error("read ECONNRESET"), {
+    code: "ECONNRESET",
+    syscall: "read",
+    errno: -104,
+    address: "1.2.3.4",
+    port: 443,
+  });
+  const top = Object.assign(new TypeError("fetch failed"), { cause });
+  const desc = describeFetchCause(top);
+  assert.match(desc, /fetch failed/);
+  assert.match(desc, /code=ECONNRESET/);
+  assert.match(desc, /syscall=read/);
+  assert.match(desc, /address=1\.2\.3\.4:443/);
+  assert.doesNotMatch(desc, /\n\s+at /); // never leaks a stack trace (Rule #12)
+});
+
+test("#4252 describeFetchCause expands AggregateError sub-errors (Happy-Eyeballs)", () => {
+  const sub1 = Object.assign(new Error("connect ECONNREFUSED ::1:443"), {
+    code: "ECONNREFUSED",
+    address: "::1",
+    port: 443,
+  });
+  const sub2 = Object.assign(new Error("connect ETIMEDOUT 1.2.3.4:443"), {
+    code: "ETIMEDOUT",
+    address: "1.2.3.4",
+    port: 443,
+  });
+  const agg = new AggregateError([sub1, sub2], "all connection attempts failed");
+  const top = Object.assign(new TypeError("fetch failed"), { cause: agg });
+  const desc = describeFetchCause(top);
+  assert.match(desc, /code=ECONNREFUSED/);
+  assert.match(desc, /address=::1:443/);
+  assert.match(desc, /code=ETIMEDOUT/);
+});
+
+test("#4252 describeFetchCause falls back to String(err) when nothing is structured", () => {
+  assert.equal(describeFetchCause("boom"), "boom");
+});
+
+test("#4252 both undici AND native fetch fail → rejects fast with cause detail attached", async () => {
+  const dispErr = Object.assign(new Error("fetch failed"), {
+    code: "UND_ERR_SOCKET",
+    cause: Object.assign(new Error("other side closed"), { code: "UND_ERR_SOCKET" }),
+  });
+  const nativeErr = Object.assign(new Error("fetch failed"), {
+    cause: Object.assign(new Error("connect ECONNREFUSED 1.2.3.4:443"), {
+      code: "ECONNREFUSED",
+      syscall: "connect",
+    }),
+  });
+  const mockUndici = async () => {
+    throw dispErr;
+  };
+  const mockNative = async () => {
+    throw nativeErr;
+  };
+  await assert.rejects(
+    proxyFetch(
+      "https://example.invalid/x",
+      { method: "GET" },
+      { undiciFetch: mockUndici, nativeFetch: mockNative }
+    ),
+    (err: Error & { proxyFetchDetail?: string }) => {
+      assert.ok(err.proxyFetchDetail, "propagated error must carry proxyFetchDetail");
+      assert.match(err.proxyFetchDetail, /dispatcher=\[/);
+      assert.match(err.proxyFetchDetail, /native=\[/);
+      assert.match(err.proxyFetchDetail, /ECONNREFUSED/);
+      return true;
+    }
+  );
+});