fix: a fast upstream failure hung the client instead of answering 502

handleMessages installs an abort so a client that gives up mid-SSE frees the
upstream. It was keyed on clientReq's "close" — which Node emits when the
request BODY has been consumed, i.e. on every request, immediately — so it
aborted while the client was still sitting there, and the forwardRequest catch
opens with `if (aborted) return`. Nothing was written back.

Measured in reverse mode on the real /v1/messages path, against an upstream that
refuses instantly, which is what a dead local hop does:

  before   body at once -> HANG (6s client timeout)   body delayed -> HANG
  after    body at once -> 502 in 9ms                 body delayed -> 502 in 52ms

Keyed on clientRes's close instead: it fires when the response finishes OR the
connection is destroyed, so pairing it with writableEnded separates "we
answered" from "the client hung up". Same change in handlePassthrough, which
carried the identical line.

This was found by measuring an exposure I had already dismissed as too risky to
touch. The previous commit recorded it as a latent defect blocking a different
fix; it is not latent, it is on the most ordinary upstream failure there is.

WHAT THE TESTS DO AND DO NOT GUARD, because the difference matters:

  the 502 case dies when the listener is reverted to clientReq — mutation-checked
  the no-leak case does NOT die when the listener is deleted outright

Two attempts at the second: client takes a frame then leaves (the pipe tears the
upstream down by itself), and an upstream that accepts and never answers so no
pipe exists (still freed). Both passed with the listener removed. So the listener
may be doing nothing that socket teardown does not already do. It stays — "I
could not demonstrate it matters" is not "it does not matter" — and the case is
labelled as pinning the PROPERTY, not guarding the listener, so nobody reads it
as coverage it is not.

ALSO: the relayed probe added in 70ff998 dialled the real api.anthropic.com,
because that test never set CACHE_FIX_PROXY_UPSTREAM and the default is the live
host. That is the trap integrated.conf line 20 already warns about, and it took
CI red on node 22 while bafabae with identical proxy code was green. It now runs
against a local 418, on its own instance — pointing config.upstream at loopback
for the whole case makes the CONNECT half read the tunnel target as the upstream
and stop blind-tunnelling it, which failed the fail-open assertion for an
unrelated reason.

Ref cnighswonger#304

Co-Authored-By: Claude <noreply@anthropic.com>

Author: codeslake

proxy/server.mjs

@@ -151,7 +151,22 @@ async function handleMessages(clientReq, clientRes) {
   // Bootstrap (handleBootstrap) doesn't install this because its response is
   // a single non-SSE JSON payload — aborting on clientReq close prematurely
   // would race the response write on fast-failure paths (e.g. ECONNREFUSED).
-  clientReq.on("close", () => {
+  // THE RESPONSE'S close, NOT THE REQUEST'S. Node emits "close" on an
+  // IncomingMessage when the request BODY has been consumed — which is every
+  // request, immediately, not only the ones where the client left. So this
+  // aborted while the client was still sitting there waiting, and the catch
+  // below opens with `if (aborted) return`, so nothing was ever written back.
+  //
+  // Measured in reverse mode against an upstream that refuses instantly, which
+  // is what a dead local hop does: POST /v1/messages HUNG for the client's full
+  // 6s timeout instead of answering 502. Both body shapes, sent-at-once and
+  // sent-delayed. That is a live session stalling on the most ordinary upstream
+  // failure there is.
+  //
+  // clientRes's close fires when the response is finished OR the connection is
+  // destroyed, so pairing it with writableEnded separates the two: ended means
+  // we answered, not-ended means the client hung up and the upstream should go.
+  clientRes.on("close", () => {
     if (!clientRes.writableEnded) abortController.abort();
   });
 
@@ -481,7 +496,11 @@ function handleNotFound(_req, res) {
 // /v1/messages arrives there), so its 404 contract is unchanged.
 async function handlePassthrough(clientReq, clientRes) {
   const abortController = new AbortController();
-  clientReq.on("close", () => { if (!clientRes.writableEnded) abortController.abort(); });
+  // clientRes, not clientReq — see handleMessages' matching comment. The request
+  // object's "close" fires when its body is consumed, so this aborted every
+  // request the instant it arrived and the catch below then returned without
+  // writing anything.
+  clientRes.on("close", () => { if (!clientRes.writableEnded) abortController.abort(); });
 
   const method = (clientReq.method || "GET").toUpperCase();
   const body = (method === "GET" || method === "HEAD") ? null : await collectBody(clientReq);

test/proxy-forward-attach-fallback.test.mjs

@@ -256,6 +256,15 @@ test("CONNECT falls open to a direct dial, unless CACHE_FIX_REQUIRE_HOP says oth
   // Where a direct dial lands. Reaching it is the fail-OPEN outcome.
   const direct = net.createServer((sock) => { seen.push("DIRECT"); sock.destroy(); });
   const directPort = await listen(direct);
+  // A LOCAL upstream that answers unmistakably. Without it config.upstream is
+  // the real https://api.anthropic.com and the relayed probe below dials it for
+  // real — measured as a CI regression: the run that added this probe went red
+  // on node 22 while the identical code without it was green, and
+  // integrated.conf line 20 already warns that an unproxied test here "hangs on
+  // this network until it times out". 418 is a status nothing else in this
+  // chain produces, so reaching it cannot be confused with a refusal.
+  const upstream = http.createServer((_q, r) => { r.writeHead(418); r.end("teapot"); });
+  await new Promise((r) => upstream.listen(0, "127.0.0.1", r));
   // A hop address with nothing behind it: the whole chain refuses.
   const deadHop = net.createServer();
   const deadPort = await listen(deadHop);
@@ -311,6 +320,14 @@ test("CONNECT falls open to a direct dial, unless CACHE_FIX_REQUIRE_HOP says oth
     // aborted=true, writableEnded=false, no response, client timed out at 10s.
     // Change this assertion the day that abort listener distinguishes "body
     // done" from "client gone".
+    // ITS OWN INSTANCE. Pointing config.upstream at loopback for the whole case
+    // breaks the CONNECT half above — the forward proxy then reads the tunnel
+    // target 127.0.0.1:<port> as the upstream host and stops blind-tunnelling
+    // it, so `seen` came back empty and the fail-open assertion failed for a
+    // reason that had nothing to do with fail-open.
+    await handle.close();
+    process.env.CACHE_FIX_PROXY_UPSTREAM = `http://127.0.0.1:${upstream.address().port}`;
+    handle = await startProxy({ port: 0, watch: false });
     const relayed = await new Promise((resolve) => {
       const r = http.request({ host: "127.0.0.1", port: handle.port, method: "POST",
                                path: "/v1/messages", headers: { "content-type": "application/json" } },
@@ -319,14 +336,15 @@ test("CONNECT falls open to a direct dial, unless CACHE_FIX_REQUIRE_HOP says oth
       r.setTimeout(4_000, () => { r.destroy(); resolve("TIMEOUT"); });
       r.end("{}");
     });
-    assert.notEqual(relayed, 502,
-      "the relayed path now refuses under CACHE_FIX_REQUIRE_HOP — good, but the " +
-      "comment above and this assertion both describe the OLD state; update them");
+    assert.equal(relayed, 418,
+      `the relayed path answered ${relayed} instead of reaching the upstream. 502 ` +
+      `means CACHE_FIX_REQUIRE_HOP now covers it — good, but the comment above and ` +
+      `this assertion both describe the OLD state, so update them together`);
 
   } finally {
     restoreEnv(saved);
     if (handle) await handle.close();
-    direct.close();
+    direct.close(); upstream.close();
     try { rmSync(caDir, { recursive: true, force: true }); } catch {}
   }
 });

test/proxy-server.test.mjs

@@ -939,3 +939,98 @@ describe("/health hop reporting", () => {
     }
   });
 });
+
+// THE CLIENT-ABANDON ABORT, IN BOTH DIRECTIONS.
+//
+// handleMessages installs an abort so a client that gives up mid-SSE frees the
+// upstream. It was keyed on clientReq's "close" — which Node emits when the
+// request BODY is consumed, i.e. on every request, immediately — so it aborted
+// while the client was still waiting, and the forwardRequest catch opens with
+// `if (aborted) return`. Nothing was written back.
+//
+// Measured before the fix, reverse mode, upstream refusing instantly (what a
+// dead local hop does): POST /v1/messages hung for the client's full timeout
+// instead of answering 502, for both an at-once and a delayed body.
+//
+// BOTH CASES OR NEITHER. Asserting only the 502 would pass against a build that
+// deleted the listener outright, which leaks an upstream connection for every
+// abandoned stream — a worse bug, and invisible until the box runs out of
+// sockets.
+describe("client-abandon abort", () => {
+  const ENV = ["CACHE_FIX_PROXY_UPSTREAM", "CACHE_FIX_FORWARD_PROXY", "CACHE_FIX_CA_DIR",
+               "CACHE_FIX_FALLBACK_PROXIES", "CACHE_FIX_UPSTREAM_PROXY",
+               "HTTPS_PROXY", "https_proxy", "HTTP_PROXY", "http_proxy"];
+  const save = () => Object.fromEntries(ENV.map((k) => [k, process.env[k]]));
+  const restore = (s) => { for (const [k, v] of Object.entries(s)) {
+    if (v === undefined) delete process.env[k]; else process.env[k] = v; } };
+
+  it("answers 502 when the upstream refuses, instead of hanging the client", async () => {
+    const saved = save();
+    const dead = await freePort();               // nothing listening: instant ECONNREFUSED
+    let h;
+    try {
+      for (const k of ENV) delete process.env[k];
+      process.env.CACHE_FIX_PROXY_UPSTREAM = `http://127.0.0.1:${dead}`;
+      h = await startProxy({ port: 0, watch: false });
+      const got = await new Promise((resolve) => {
+        const r = http.request({ host: "127.0.0.1", port: h.port, method: "POST",
+                                 path: "/v1/messages", headers: { "content-type": "application/json" } },
+          (res) => { res.resume(); res.on("end", () => resolve(res.statusCode)); });
+        r.on("error", (e) => resolve(`ERR:${e.code}`));
+        r.setTimeout(6_000, () => { r.destroy(); resolve("HANG"); });
+        r.end(JSON.stringify({ model: "x", messages: [] }));
+      });
+      assert.equal(got, 502,
+        `a refusing upstream produced ${got} — the client was never answered, ` +
+        `which is a live session stalling on the most ordinary upstream failure`);
+    } finally { restore(saved); if (h) await h.close(); }
+  });
+
+  // NO UPSTREAM LEAK WHEN A CLIENT WALKS AWAY — the PROPERTY, and deliberately
+  // not a claim about which mechanism provides it.
+  //
+  // I could not build a case that dies when the abort listener is deleted. Two
+  // tries: letting the client take a frame then leave (the pipe tears the
+  // upstream down on its own), and an upstream that accepts and never answers
+  // so no pipe exists (still freed). Both passed with the listener removed
+  // outright. So the listener may be doing nothing here that socket teardown
+  // does not already do — which would make it pure liability, since keying it
+  // on clientReq is what hung every fast upstream failure.
+  //
+  // It stays, because "I could not demonstrate it matters" is not "it does not
+  // matter", and removing it is a bigger change than this evidence supports.
+  // This case pins the property so a future refactor that DOES introduce a leak
+  // is caught, and says plainly that it is not a guard on the listener.
+  it("frees an upstream that has not answered yet when the client walks away", async () => {
+    const saved = save();
+    let liveUpstream = 0;
+    const upstream = http.createServer(() => { /* accept, never respond */ });
+    upstream.on("connection", (sock) => {
+      liveUpstream++;
+      sock.on("close", () => { liveUpstream--; });
+    });
+    await new Promise((r) => upstream.listen(0, "127.0.0.1", r));
+    let h;
+    try {
+      for (const k of ENV) delete process.env[k];
+      process.env.CACHE_FIX_PROXY_UPSTREAM = `http://127.0.0.1:${upstream.address().port}`;
+      h = await startProxy({ port: 0, watch: false });
+      await new Promise((resolve) => {
+        const r = http.request({ host: "127.0.0.1", port: h.port, method: "POST",
+                                 path: "/v1/messages", headers: { "content-type": "application/json" } },
+          (res) => { res.resume(); });
+        r.on("error", () => {});
+        r.end(JSON.stringify({ model: "x", messages: [] }));
+        // Long enough for the proxy to have dialled and be WAITING on the
+        // upstream — the state this case is about — then walk away.
+        setTimeout(() => { r.destroy(); resolve(); }, 400);
+      });
+      assert.ok(liveUpstream > 0 || true, "");   // the count below is the assertion
+      for (let i = 0; i < 40 && liveUpstream > 0; i++) await new Promise((r) => setTimeout(r, 50));
+      assert.equal(liveUpstream, 0,
+        `the client walked away while the upstream had not answered, and ${liveUpstream} ` +
+        `upstream connection(s) stayed open — one leak per abandoned request, with no ` +
+        `pipe in place to tear it down`);
+    } finally { restore(saved); if (h) await h.close(); await new Promise((r) => upstream.close(r)); }
+  });
+});