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

src/js/internal/http.ts

@@ -101,6 +101,18 @@ const kCloseCallback = Symbol("closeCallback");
 
 const kEmptyObject = Object.freeze(Object.create(null));
 
+// node:_http_server's pipelined-response machinery, registered onto this
+// object when that module initializes. internal/http1_server_fallback drives
+// the same per-connection queue over foreign duplex sockets through it,
+// without putting the helpers on the user-visible node:_http_server surface.
+// The fallback requires node:http (which loads _http_server) before reading
+// these, so they are always populated by then.
+const http1ServerPipeline: {
+  queuePipelinedResponse?: (socket: unknown, res: unknown, isAncient: boolean) => void;
+  advanceResponsePipeline?: (server: unknown, socket: unknown) => void;
+  abortQueuedPipelinedResponses?: (socket: unknown) => void;
+} = {};
+
 export const enum ClientRequestEmitState {
   socket = 1,
   prefinish = 2,
@@ -618,6 +630,7 @@ export {
   headerStateSymbol,
   headersSymbol,
   headersTuple,
+  http1ServerPipeline,
   isAbortError,
   isTlsSymbol,
   kAbortController,

src/js/internal/http1_server_fallback.ts

@@ -235,7 +235,9 @@ function createHttp1FallbackResponseHandle(socket, shouldKeepAlive, keepAliveTim
 function connectionListenerHTTP1(server, socket, options) {
   const http = require("node:http");
   const { HTTPParser, prepareError, calculateLenientFlags, continueExpression } = require("node:_http_common");
-  const { kHandle: kHttp1ResponseHandle } = require("internal/http");
+  const { kHandle: kHttp1ResponseHandle, http1ServerPipeline } = require("internal/http");
+  // Populated by node:_http_server, which the require("node:http") above loads.
+  const { queuePipelinedResponse, advanceResponsePipeline, abortQueuedPipelinedResponses } = http1ServerPipeline;
   const { allMethods } = process.binding("http_parser");
 
   const http1Options = options.http1Options || {};
@@ -326,12 +328,22 @@ function connectionListenerHTTP1(server, socket, options) {
       }
     };
     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
@@ -444,6 +456,10 @@ function connectionListenerHTTP1(server, socket, options) {
   socket.once("end", onHttp1SocketEnd);
   socket.once("close", () => {
     connections.delete(socket);
+    // Like the native socket's close path: responses (and their requests)
+    // still queued behind the in-flight one when the connection dies are
+    // aborted, so they emit 'close' instead of hanging forever.
+    abortQueuedPipelinedResponses(socket);
     try {
       parser.close();
     } catch {}

src/js/node/_http_server.ts

@@ -67,6 +67,7 @@ const {
   kOutHeaders,
   onDataIncomingMessage,
   validateMsecs,
+  http1ServerPipeline,
 } = require("internal/http");
 const { FakeSocket } = require("internal/http/FakeSocket");
 const NumberIsNaN = Number.isNaN;
@@ -948,16 +949,7 @@ Server.prototype[kRealListen] = function (tls, port, host, socketPath, reusePort
           // 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.
@@ -1731,28 +1723,7 @@ const NodeHTTPServerSocket = class Socket extends NetSocket {
     // Pipelined responses (and their requests) that were still queued behind
     // the in-flight response are aborted, like Node.js's socketOnClose
     // (abortIncoming + abortOutgoing).
-    const pipelined = this[kPipelinedResponses];
-    const pipelinedLength = pipelined ? pipelined.length : 0;
-    if (pipelinedLength) {
-      this[kPipelinedResponses] = undefined;
-      for (let i = 0; i < pipelinedLength; i++) {
-        const queuedRes = pipelined[i];
-        const queuedReq = queuedRes.req;
-        if (queuedReq && !queuedReq.destroyed) {
-          queuedReq[kHandle] = undefined;
-          if (queuedReq.listenerCount("error") > 0) {
-            queuedReq.destroy(new ConnResetException("aborted"));
-          } else {
-            queuedReq.destroy();
-          }
-        }
-        if (!queuedRes.destroyed) {
-          queuedRes.destroy();
-        } else if (!queuedRes._closed) {
-          process.nextTick(emitCloseNT, queuedRes);
-        }
-      }
-    }
+    abortQueuedPipelinedResponses(this);
 
     // Node's server connection socket emits 'close' whenever the TCP
     // connection closes, even with no request in flight (this also covers
@@ -2566,6 +2537,52 @@ function advancePipelineIfIdleNT(server, socket) {
   }
 }
 
+// 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);
+}
+
+// When the connection dies with pipelined responses still queued behind the
+// in-flight one, abort them and their requests, like Node.js's socketOnClose
+// (abortIncoming). Runs from the native socket's close path and from the
+// http1 fallback's socket 'close' listener.
+function abortQueuedPipelinedResponses(socket) {
+  const pipelined = socket[kPipelinedResponses];
+  const pipelinedLength = pipelined ? pipelined.length : 0;
+  if (pipelinedLength) {
+    socket[kPipelinedResponses] = undefined;
+    for (let i = 0; i < pipelinedLength; i++) {
+      const queuedRes = pipelined[i];
+      const queuedReq = queuedRes.req;
+      if (queuedReq && !queuedReq.destroyed) {
+        queuedReq[kHandle] = undefined;
+        if (queuedReq.listenerCount("error") > 0) {
+          queuedReq.destroy(new ConnResetException("aborted"));
+        } else {
+          queuedReq.destroy();
+        }
+      }
+      if (!queuedRes.destroyed) {
+        queuedRes.destroy();
+      } else if (!queuedRes._closed) {
+        process.nextTick(emitCloseNT, queuedRes);
+      }
+    }
+  }
+}
+
 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 +2601,6 @@ function advanceResponsePipeline(server, socket) {
   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 +2611,36 @@ function advanceResponsePipeline(server, socket) {
     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;
@@ -4013,6 +4043,12 @@ function ensureReadableStreamController(run) {
   );
 }
 
+// Share the pipelining machinery with internal/http1_server_fallback through
+// internal/http instead of the user-visible module exports.
+http1ServerPipeline.queuePipelinedResponse = queuePipelinedResponse;
+http1ServerPipeline.advanceResponsePipeline = advanceResponsePipeline;
+http1ServerPipeline.abortQueuedPipelinedResponses = abortQueuedPipelinedResponses;
+
 export default {
   Server,
   ServerResponse,

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

@@ -4139,3 +4139,109 @@ 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);
+  }
+});
+
+it("connectionListener aborts queued pipelined responses when the connection dies", async () => {
+  // Like Node's socketOnClose (abortIncoming) and the native socket's close
+  // path: a response still queued behind the in-flight one when the socket
+  // closes is destroyed together with its request, so both emit 'close'
+  // instead of hanging forever.
+  const closedEvents: string[] = [];
+  const { promise: aborted, resolve: onAborted } = Promise.withResolvers<void>();
+  let closesPending = 2;
+  const onQueuedClose = (tag: string) => {
+    closedEvents.push(tag);
+    if (--closesPending === 0) onAborted();
+  };
+  const server = createServer((req, res) => {
+    // /a never responds, so its response keeps the socket and /b stays queued.
+    if (req.url !== "/b") return;
+    req.on("close", () => onQueuedClose("reqB"));
+    res.on("close", () => onQueuedClose("resB"));
+    // Kill the connection while /b is queued behind /a.
+    serverSide.destroy();
+  });
+  const [clientSide, serverSide] = duplexPair();
+  server.emit("connection", serverSide);
+  clientSide.write("GET /a HTTP/1.1\r\nHost: x\r\n\r\nGET /b HTTP/1.1\r\nHost: x\r\n\r\n");
+  await aborted;
+  expect(closedEvents.sort()).toEqual(["reqB", "resB"]);
+  clientSide.destroy();
+});