node:http: queue pipelined responses on fallback connections instead of throwing ERR_HTTP_SOCKET_ASSIGNED

src/js/internal/http1_server_fallback.ts

@@ -235,6 +235,7 @@
 function connectionListenerHTTP1(server, socket, options) {
   const http = require("node:http");
   const { HTTPParser, prepareError, calculateLenientFlags, continueExpression } = require("node:_http_common");
+  const { queuePipelinedResponse, advanceResponsePipeline } = require("node:_http_server");
   const { kHandle: kHttp1ResponseHandle } = require("internal/http");
   const { allMethods } = process.binding("http_parser");
 
@@ -326,12 +327,22 @@
       }
     };
     res[kHttp1ResponseHandle] = handle;
-    res.assignSocket(socket);
-    // node's resOnFinish: release the socket once the response completes so the next
-    // keep-alive request's response can attach (assignSocket throws
-    // ERR_HTTP_SOCKET_ASSIGNED while a previous response is still assigned).
+    // Node's parserOnIncoming outgoing queue: while the previous response is
+    // still assigned (its deferred 'finish' has not detached it yet -
+    // pipelined requests always land here), this response is queued and its
+    // write()/end() buffer until the finish path below assigns it the socket.
+    // assignSocket would throw ERR_HTTP_SOCKET_ASSIGNED.
+    if (socket._httpMessage) {
+      queuePipelinedResponse(socket, res, versionMajor < 1 || versionMinor < 1);
+    } else {
+      res.assignSocket(socket);
+    }
+    // node's resOnFinish: release the socket once the response completes so
+    // the next keep-alive request's response can attach, then hand it to the
+    // next queued pipelined response (replaying whatever it buffered).
     res.on("finish", function onFallbackResponseFinish() {
       this.detachSocket(socket);
+      advanceResponsePipeline(server, socket);
     });
 
     // Node's parserOnIncoming Expect routing (the native dispatcher applies the

src/js/node/_http_server.ts

@@ -948,16 +948,7 @@
           // Node.js, this response is queued (res.socket === null) and its
           // writes are buffered until the in-flight response finishes and the
           // pipeline assigns it the socket (advanceResponsePipeline).
-          http_res[kPipelinedQueuedState] = {
-            ops: [],
-            bytes: 0,
-            headerBytes: 0,
-            needDrain: false,
-            ended: false,
-            isAncient: !!isAncientHTTP,
-            socket,
-          };
-          (socket[kPipelinedResponses] ??= []).push(http_res);
+          queuePipelinedResponse(socket, http_res, !!isAncientHTTP);
           // A pipelined dispatch can arrive after the previous response finished and detached
           // (bytes still flushing keep it pending), leaving nothing in flight to advance the
           // queue. Kick the pipeline once this dispatch settles.
@@ -2566,6 +2557,23 @@
   }
 }
 
+// Like the dispatcher's pipelined branch and Node.js's parserOnIncoming
+// outgoing queue: park the response behind the connection's in-flight one.
+// Its write()/end() buffer (kPipelinedQueuedState) until
+// advanceResponsePipeline assigns it the socket and replays them.
+function queuePipelinedResponse(socket, res, isAncient) {
+  res[kPipelinedQueuedState] = {
+    ops: [],
+    bytes: 0,
+    headerBytes: 0,
+    needDrain: false,
+    ended: false,
+    isAncient,
+    socket,
+  };
+  (socket[kPipelinedResponses] ??= []).push(res);
+}
+
 function advanceResponsePipeline(server, socket) {
   // The previous response on this connection closed it (Connection: close,
   // HTTP/1.0, maxRequestsPerSocket): like Node.js's resOnFinish, advancing
@@ -2584,7 +2592,6 @@
   res[kPipelinedQueuedState] = undefined;
   releasePipelineOutgoingData(socket, queued.bytes);
   const handle = res[kHandle];
-  const socketHandle = socket[kHandle];
 
   if (res.destroyed || !handle) {
     // The queued response was destroyed before it could be sent; the
@@ -2595,22 +2602,36 @@
     return;
   }
 
-  if (
-    !socketHandle ||
-    socket.destroyed ||
-    !socketHandle.startPipelinedResponse(handle, !!queued.isAncient, !requestShouldKeepAlive(res.req))
-  ) {
-    // The connection is already gone; the socket close path destroys queued
-    // responses, but make sure this (already dequeued) one is not skipped.
-    if (!res.destroyed) {
-      res.destroy();
+  if (socket instanceof NodeHTTPServerSocket) {
+    const socketHandle = socket[kHandle];
+    if (
+      !socketHandle ||
+      socket.destroyed ||
+      !socketHandle.startPipelinedResponse(handle, !!queued.isAncient, !requestShouldKeepAlive(res.req))
+    ) {
+      // The connection is already gone; the socket close path destroys queued
+      // responses, but make sure this (already dequeued) one is not skipped.
+      if (!res.destroyed) {
+        res.destroy();
+      }
+      return;
     }
-    return;
-  }
 
-  if (res.assignSocket === ServerResponse.prototype.assignSocket) {
-    assignSocketInternal(res, socket);
+    if (res.assignSocket === ServerResponse.prototype.assignSocket) {
+      assignSocketInternal(res, socket);
+    } else {
+      res.assignSocket(socket);
+    }
   } else {
+    // internal/http1_server_fallback connection (a foreign duplex): each
+    // response's JS handle writes to the socket itself, so there is no native
+    // socket handle to switch. The prototype assignSocket also installs the
+    // 'close' listener a plain stream needs. Clear `finished` before the
+    // assignment: assignSocket's _flush would otherwise emit 'prefinish' for
+    // an already-ended queued response before its ops have been replayed.
+    if (queued.ended) {
+      res.finished = false;
+    }
     res.assignSocket(socket);
   }
   socket[kRequest] = res.req;
@@ -4017,4 +4038,7 @@
   Server,
   ServerResponse,
   kConnectionsCheckingInterval,
+  // Pipelining internals shared with internal/http1_server_fallback.
+  queuePipelinedResponse,
+  advanceResponsePipeline,
 };

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

@@ -4139,3 +4139,81 @@ it("connectionListener hands off Upgrade and CONNECT like Node", async () => {
     expect(serverSide.destroyed).toBe(true);
   }
 });
+
+it("connectionListener queues pipelined responses like Node", async () => {
+  // Both requests parse in one socket 'data' event, so the second one's
+  // headers complete while the first response is still assigned (its 'finish'
+  // detach is deferred a tick). The second response must be queued, not
+  // assigned - assignSocket throws ERR_HTTP_SOCKET_ASSIGNED - and its output
+  // must be held back until the first response completes.
+  async function exchange(handler: (req: any, res: any) => void, requestBytes: string, done: (buf: string) => boolean) {
+    const server = createServer(handler);
+    const [clientSide, serverSide] = duplexPair();
+    server.emit("connection", serverSide);
+    try {
+      return await new Promise<string>((resolve, reject) => {
+        let buf = "";
+        clientSide.on("data", d => {
+          buf += d;
+          if (done(buf)) resolve(buf);
+        });
+        clientSide.on("error", reject);
+        clientSide.on("close", () => reject(new Error("closed before expected output: " + buf)));
+        clientSide.write(requestBytes);
+      });
+    } finally {
+      clientSide.destroy();
+      serverSide.destroy();
+    }
+  }
+
+  // Handler finishes after the current tick: the second request is dispatched
+  // while the first response is still in flight.
+  {
+    const out = await exchange(
+      (req, res) => setImmediate(() => res.end("ok:" + req.url)),
+      "GET /a HTTP/1.1\r\nHost: x\r\n\r\nGET /b HTTP/1.1\r\nHost: x\r\n\r\n",
+      buf => buf.includes("ok:/a") && buf.includes("ok:/b"),
+    );
+    expect(out.indexOf("ok:/a")).toBeLessThan(out.indexOf("ok:/b"));
+    expect(out.match(/HTTP\/1\.1 200/g)).toHaveLength(2);
+  }
+
+  // The queued response is written (write + end, chunked) before the first
+  // one finishes: its bytes must still go out second, after the first body.
+  {
+    let releaseFirst!: () => void;
+    const firstGate = new Promise<void>(resolve => (releaseFirst = resolve));
+    const out = await exchange(
+      (req, res) => {
+        if (req.url === "/a") {
+          firstGate.then(() => res.end("first"));
+        } else {
+          res.write("second-part1");
+          res.end("second-part2");
+          releaseFirst();
+        }
+      },
+      "GET /a HTTP/1.1\r\nHost: x\r\n\r\nGET /b HTTP/1.1\r\nHost: x\r\n\r\n",
+      buf => buf.includes("second-part2") && buf.endsWith("0\r\n\r\n"),
+    );
+    expect(out.indexOf("first")).toBeLessThan(out.indexOf("second-part1"));
+    expect(out.indexOf("second-part1")).toBeLessThan(out.indexOf("second-part2"));
+    const second = out.slice(out.indexOf("HTTP/1.1 200", 1));
+    expect(second).toContain("Transfer-Encoding: chunked");
+  }
+
+  // Three pipelined requests answered synchronously: each finish hands the
+  // socket to the next queued response.
+  {
+    const out = await exchange(
+      (req, res) => res.end("r:" + req.url + ";"),
+      "GET /1 HTTP/1.1\r\nHost: x\r\n\r\nGET /2 HTTP/1.1\r\nHost: x\r\n\r\nGET /3 HTTP/1.1\r\nHost: x\r\n\r\n",
+      buf => buf.includes("r:/3;"),
+    );
+    expect(out.indexOf("r:/1;")).toBeGreaterThanOrEqual(0);
+    expect(out.indexOf("r:/1;")).toBeLessThan(out.indexOf("r:/2;"));
+    expect(out.indexOf("r:/2;")).toBeLessThan(out.indexOf("r:/3;"));
+    expect(out.match(/HTTP\/1\.1 200/g)).toHaveLength(3);
+  }
+});