node:http: don't leave the server socket Duplex corked across kept-alive requests

src/js/node/_http_server.ts

@@ -1066,8 +1066,6 @@ Server.prototype[kRealListen] = function (tls, port, host, socketPath, reusePort
           }
         }
 
-        socket.cork();
-
         if (isPipelined) {
           // Completion of a queued response is tracked through the pipeline
           // (advanceResponsePipeline) and its native handle, not this dispatch.
@@ -3218,6 +3216,13 @@ ServerResponse.prototype.end = function (chunk, encoding, callback) {
     }
   }
   this._header = " ";
+  // Like Node's OutgoingMessage.prototype.end: fully uncork the connection.
+  const resSocket = this[fakeSocketSymbol];
+  const resSocketWritableState = resSocket?._writableState;
+  if (resSocketWritableState && resSocketWritableState.corked) {
+    resSocketWritableState.corked = 1;
+    resSocket.uncork();
+  }
   const req = this.req;
   if (!req._consuming && !req?._readableState?.resumeScheduled) {
     req._dump();

test/js/node/http/node-http-connect.test.ts

@@ -783,6 +783,123 @@ describe("Should be compatible with node.js", () => {
   });
 });
 
+// A ServerResponse writes through the native handle, not the socket Duplex,
+// so a cork() left on the Duplex after a kept-alive request would make a
+// 'connect'/'upgrade' handler's socket.write() buffer forever. Covers the
+// proxy-chain / puppeteer-with-proxy failure in #12213.
+describe("CONNECT/Upgrade on a kept-alive connection can write to the socket", () => {
+  async function runKeepAliveHandoff(method: "CONNECT" | "Upgrade", writeAsync: boolean) {
+    await using server = http.createServer((req, res) => {
+      res.end("body");
+    }) as http.Server & AsyncDisposable;
+    let corkedAtHandoff: number | undefined;
+    let writableLengthAfterWrite: number | undefined;
+    const handler = (req: http.IncomingMessage, socket: net.Socket) => {
+      corkedAtHandoff = socket.writableCorked;
+      let clientEnded = false;
+      let responded = false;
+      const respond = () => {
+        socket.write(
+          method === "CONNECT"
+            ? "HTTP/1.1 200 Connection Established\r\n\r\n"
+            : "HTTP/1.1 101 Switching Protocols\r\nUpgrade: raw\r\nConnection: Upgrade\r\n\r\n",
+        );
+        // write() only (no end), matching an open proxy tunnel; Writable.end's
+        // force-uncork would flush the leaked cork and mask the regression.
+        socket.write("tunnel-data");
+        writableLengthAfterWrite = socket.writableLength;
+        responded = true;
+        if (clientEnded) socket.end();
+      };
+      // End once the client FINs so the received promise below never hangs.
+      // The client's FIN can arrive before a setImmediate-deferred respond() on
+      // Windows, so end() is deferred until both have happened.
+      socket.on("end", () => {
+        clientEnded = true;
+        if (responded) socket.end();
+      });
+      if (writeAsync) setImmediate(respond);
+      else respond();
+    };
+    server.on(method === "CONNECT" ? "connect" : "upgrade", handler);
+    server.listen(0, "127.0.0.1");
+    await once(server, "listening");
+    const { port } = server.address() as AddressInfo;
+
+    const received = await new Promise<string>((resolve, reject) => {
+      const sock = net.connect({ port, host: "127.0.0.1" });
+      sock.on("error", reject);
+      let buf = "";
+      let phase = 0;
+      sock.on("connect", () => {
+        sock.write("GET / HTTP/1.1\r\nHost: x\r\n\r\n");
+      });
+      sock.on("data", d => {
+        buf += d.toString();
+        if (phase === 0 && buf.includes("body")) {
+          phase = 1;
+          buf = "";
+          sock.end(
+            method === "CONNECT"
+              ? "CONNECT example.com:443 HTTP/1.1\r\nHost: example.com:443\r\n\r\n"
+              : "GET / HTTP/1.1\r\nHost: x\r\nUpgrade: raw\r\nConnection: Upgrade\r\n\r\n",
+          );
+        }
+      });
+      sock.on("close", () => resolve(buf));
+    });
+
+    expect({ received, corkedAtHandoff, writableLengthAfterWrite }).toEqual({
+      received: expect.stringContaining("tunnel-data"),
+      corkedAtHandoff: 0,
+      writableLengthAfterWrite: 0,
+    });
+  }
+
+  for (const method of ["CONNECT", "Upgrade"] as const) {
+    for (const writeAsync of [false, true]) {
+      test(`${method}, ${writeAsync ? "async" : "sync"} write`, async () => {
+        await runKeepAliveHandoff(method, writeAsync);
+      });
+    }
+  }
+
+  // The dispatch path left the connection Duplex corked after every 'request'
+  // emit (proxy-chain hit this via CONNECT on a kept-alive socket above, but
+  // writing to req.socket directly after res.end() is the same leak).
+  test("req.socket is not corked across 'request' dispatches", async () => {
+    const seen: number[] = [];
+    await using server = http.createServer((req, res) => {
+      seen.push(req.socket.writableCorked);
+      res.cork();
+      res.end("body");
+      // Node's OutgoingMessage.end() fully uncorks the connection.
+      seen.push(req.socket.writableCorked);
+    }) as http.Server & AsyncDisposable;
+    server.listen(0, "127.0.0.1");
+    await once(server, "listening");
+    const { port } = server.address() as AddressInfo;
+
+    let bodies = 0;
+    await new Promise<void>((resolve, reject) => {
+      const sock = net.connect({ port, host: "127.0.0.1" });
+      sock.on("error", reject);
+      sock.on("close", () => (bodies === 3 ? resolve() : reject(new Error(`closed after ${bodies}/3 responses`))));
+      sock.on("connect", () => sock.write("GET / HTTP/1.1\r\nHost: x\r\n\r\n"));
+      let buf = "";
+      sock.on("data", d => {
+        buf += d.toString();
+        while (buf.includes("body")) {
+          buf = buf.slice(buf.indexOf("body") + 4);
+          if (++bodies < 3) sock.write("GET / HTTP/1.1\r\nHost: x\r\n\r\n");
+          else sock.end();
+        }
+      });
+    });
+    expect(seen).toEqual([0, 0, 0, 0, 0, 0]);
+  });
+});
+
 // Windows: after FIN on a CONNECT-tunnel socket, AFD's level-triggered
 // UV_DISCONNECT used to re-derive EOF and bounce the poll between 0 and
 // WRITABLE forever (pins the poll_cb allow_half_open arm).