node:http: close an upgraded socket when the client FINs before the request body completes

packages/bun-uws/src/HttpContext.h

@@ -752,15 +752,24 @@ struct HttpContext {
             /* CONNECT/Upgrade tunnels allow half-open: the peer finishing its
              * writable side ends the JS socket's readable side ('end' event) but
              * the server can keep writing until it ends the socket itself, like
-             * Node's http server (allowHalfOpen: true). This includes an accepted
-             * Upgrade whose body never completed (HTTP_NODE_TUNNEL_AFTER_BODY): the
-             * EOF ends the upgrade socket, exactly like Node's UpgradeStream. */
-            if (httpResponseData->isConnectRequest || (httpResponseData->state & HttpResponseData<SSL>::HTTP_NODE_TUNNEL_AFTER_BODY)) {
+             * Node's http server (allowHalfOpen: true). */
+            if (httpResponseData->isConnectRequest) {
                 if (httpResponseData->socketData && httpContextData->onSocketData) {
                     httpContextData->onSocketData(httpResponseData->socketData, SSL, s, "", 0, true);
                 }
                 return s;
             }
+            /* An accepted Upgrade whose body never completed is not a tunnel yet:
+             * Node's UpgradeStream wraps a socket that still has socketOnEnd
+             * attached, so a mid-body FIN ends the raw socket and the
+             * UpgradeStream destroys with it. Staying half-open here would
+             * strand the response's body-read ref and the server's
+             * pending-request count. onClose() runs the tunnel-after-body
+             * cleanup (socketData last=true + inStream last=true). */
+            if (httpResponseData->state & HttpResponseData<SSL>::HTTP_NODE_TUNNEL_AFTER_BODY) {
+                asyncSocket->uncorkWithoutSending();
+                return asyncSocket->close();
+            }
 
             if (httpContextData->onClientError && !(httpResponseData->state & HttpResponseData<SSL>::HTTP_NODE_PARSING_STOPPED)
                 && (httpResponseData->hasBufferedPartialRequestHeaders()

src/js/node/_http_server.ts

@@ -2002,6 +2002,13 @@ const NodeHTTPServerSocket = class Socket extends NetSocket {
       if (handle) {
         const flushed = handle.write(_chunk, _encoding);
         if (!flushed && handle.ondrain) {
+          // The native socket is already gone (the tunnel's 'end' fires before
+          // the JS Duplex is destroyed): there will be no drain to release the
+          // callback, so fail now like Node's socket.write() does.
+          if (handle.closed) {
+            _callback($ERR_STREAM_WRITE_AFTER_END());
+            return false;
+          }
           // Streaming mode (CONNECT tunnels): wait for the native drain
           // callback before completing the write.
           this.#pendingCallback = _callback;

test/js/node/http/node-http-upgrade-body-fin.test.ts

@@ -0,0 +1,69 @@
+import { expect, test } from "bun:test";
+import { bunEnv, bunExe } from "harness";
+
+test("upgrade with a body: a mid-body client FIN closes the upgraded socket and lets server.close() complete", async () => {
+  // Like Node's UpgradeStream: a FIN before the request body completes closes
+  // the upgraded socket instead of leaving the connection half-open.
+  const fixture = /* js */ `
+    const http = require("node:http");
+    const net = require("node:net");
+    const { once } = require("node:events");
+
+    (async () => {
+      const server = http.createServer();
+      let upgraded;
+      const gotUpgrade = new Promise(r => { upgraded = r; });
+      let socketClosed;
+      const gotSocketClose = new Promise(r => { socketClosed = r; });
+      const events = [];
+      server.on("upgrade", (req, socket, head) => {
+        events.push("upgrade");
+        req.on("data", () => {});
+        socket.on("end", () => {
+          events.push("end");
+          socket.write("late", err => events.push("write-cb:" + (err ? err.code : "ok")));
+        });
+        socket.on("error", err => events.push("error:" + err.code));
+        socket.on("close", () => { events.push("close"); socketClosed(); });
+        upgraded();
+      });
+      server.listen(0, "127.0.0.1");
+      await once(server, "listening");
+
+      const c = net.connect(server.address().port, "127.0.0.1");
+      c.on("error", () => {});
+      await once(c, "connect");
+      c.write("GET / HTTP/1.1\\r\\nHost: x\\r\\nUpgrade: foo\\r\\nConnection: Upgrade\\r\\nContent-Length: 100\\r\\n\\r\\npartial");
+      await gotUpgrade;
+      // Half-close the client; the body (100 bytes) is never completed.
+      c.end();
+      await gotSocketClose;
+      c.destroy();
+
+      // server.close() only resolves once pending_requests has reached zero.
+      await new Promise((resolve, reject) => {
+        server.close(err => err ? reject(err) : resolve());
+      });
+      console.log(JSON.stringify(events));
+      process.exit(0);
+    })().catch(err => { console.error(err); process.exit(1); });
+  `;
+  await using proc = Bun.spawn({
+    cmd: [bunExe(), "-e", fixture],
+    env: bunEnv,
+    stderr: "pipe",
+    timeout: 20_000,
+  });
+  const [stdout, stderr, exitCode] = await Promise.all([proc.stdout.text(), proc.stderr.text(), proc.exited]);
+  expect({ stdout: stdout.trim(), stderr, exitCode }).toEqual({
+    stdout: JSON.stringify([
+      "upgrade",
+      "end",
+      "write-cb:ERR_STREAM_WRITE_AFTER_END",
+      "error:ERR_STREAM_WRITE_AFTER_END",
+      "close",
+    ]),
+    stderr: expect.not.stringContaining("error"),
+    exitCode: 0,
+  });
+}, 30_000);