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

src/js/internal/http.ts

@@ -101,6 +101,19 @@ const kCloseCallback = Symbol("closeCallback");
 
 const kEmptyObject = Object.freeze(Object.create(null));
 
+// node:_http_server registers its pipelined-response machinery here at module
+// initialization, letting internal/http1_server_fallback drive the same
+// per-connection queue without widening node:_http_server's exports. The
+// fallback loads node:http (and with it _http_server) before reading these.
+const http1ServerPipeline: {
+  queuePipelinedResponse?: (socket: unknown, res: unknown, isAncient: boolean) => void;
+  advanceResponsePipeline?: (server: unknown, socket: unknown) => void;
+  abortQueuedPipelinedResponses?: (socket: unknown) => void;
+  maybePauseFallbackReads?: (socket: unknown) => void;
+  resumeFallbackReadsOnDrain?: (socket: unknown) => void;
+  kMustCloseConnection?: symbol;
+} = {};
+
 export const enum ClientRequestEmitState {
   socket = 1,
   prefinish = 2,
@@ -618,6 +631,7 @@ export {
   headerStateSymbol,
   headersSymbol,
   headersTuple,
+  http1ServerPipeline,
   isAbortError,
   isTlsSymbol,
   kAbortController,

src/js/internal/http1_server_fallback.ts

@@ -235,7 +235,17 @@ 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 { ConnResetException } = require("internal/shared");
+  const { kHandle: kHttp1ResponseHandle, http1ServerPipeline } = require("internal/http");
+  // Populated by node:_http_server, which the require("node:http") above loads.
+  const {
+    queuePipelinedResponse,
+    advanceResponsePipeline,
+    abortQueuedPipelinedResponses,
+    maybePauseFallbackReads,
+    resumeFallbackReadsOnDrain,
+    kMustCloseConnection,
+  } = http1ServerPipeline;
   const { allMethods } = process.binding("http_parser");
 
   const http1Options = options.http1Options || {};
@@ -307,9 +317,11 @@ function connectionListenerHTTP1(server, socket, options) {
         return 2;
       }
     }
-    // The body is fed by the parser callbacks below; reading just resumes the socket.
+    // The body is fed by the parser callbacks below; reading just resumes the
+    // socket - unless the pipelining read gate paused it (the gate's release
+    // resumes it instead).
     req._read = function (_size) {
-      if (socket.readable) socket.resume();
+      if (!socket._paused && socket.readable) socket.resume();
     };
 
     const res = new ServerResponseClass(req);
@@ -326,14 +338,37 @@ 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: pipelined requests parse while
+    // the previous response is still assigned (its 'finish' detach is a tick
+    // away), so queue this response instead of letting assignSocket 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,
+    // then either end the connection (a response that advertised Connection:
+    // close must not be followed by another one - the close path aborts the
+    // queued responses) or hand the socket to the next queued pipelined
+    // response, replaying whatever it buffered.
     res.on("finish", function onFallbackResponseFinish() {
       this.detachSocket(socket);
+      if (this[kMustCloseConnection]) {
+        if (typeof socket.destroySoon === "function") {
+          socket.destroySoon();
+        } else if (!socket.writableEnded) {
+          socket.end();
+        }
+        return;
+      }
+      advanceResponsePipeline(server, socket);
     });
 
+    // Node's parserOnIncoming read gate: stop reading once the connection's
+    // outgoing side is backed up, so pipelined requests cannot flood it.
+    maybePauseFallbackReads(socket);
+
     // Node's parserOnIncoming Expect routing (the native dispatcher applies the
     // same at _http_server.ts's DISPATCH_HAS_EXPECT branch).
     const expect = req.headers.expect;
@@ -401,6 +436,7 @@ function connectionListenerHTTP1(server, socket, options) {
       socket.removeListener("data", onHttp1SocketData);
       socket.removeListener("error", onHttp1SocketErrorListener);
       socket.removeListener("end", onHttp1SocketEnd);
+      socket.removeListener("drain", onHttp1SocketDrain);
       connections.delete(socket);
       try {
         parser.close();
@@ -439,11 +475,31 @@ function connectionListenerHTTP1(server, socket, options) {
       socket.end();
     }
   }
+  // Node's socketOnDrain: a transport-backpressure pause lifts when the
+  // socket drains (a queued-bytes pause lifts from the pipeline advance).
+  function onHttp1SocketDrain() {
+    resumeFallbackReadsOnDrain(socket);
+  }
   socket.on("data", onHttp1SocketData);
   socket.on("error", onHttp1SocketErrorListener);
   socket.once("end", onHttp1SocketEnd);
+  socket.on("drain", onHttp1SocketDrain);
   socket.once("close", () => {
     connections.delete(socket);
+    // Like the native socket's close path (Node's socketOnClose ->
+    // abortIncoming): abort the in-flight request, then the responses (and
+    // requests) still queued behind it, so they all emit 'close'. The
+    // in-flight response's own 'close' comes from onServerResponseClose,
+    // installed by assignSocket.
+    const inflightReq = socket._httpMessage?.req;
+    if (inflightReq && !inflightReq.destroyed) {
+      if (inflightReq.listenerCount("error") > 0) {
+        inflightReq.destroy(new ConnResetException("aborted"));
+      } else {
+        inflightReq.destroy();
+      }
+    }
+    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
@@ -2537,6 +2508,28 @@ function pausePipelineReads(socket) {
   response.pauseReads();
 }
 
+// Node's parserOnIncoming read gate, for fallback connections (the native
+// sibling is the kOutgoingData check before pausePipelineReads at the
+// dispatcher): stop reading when the transport or the bytes buffered on
+// queued pipelined responses are backed up, so a pipelining client cannot
+// flood the connection's memory.
+function maybePauseFallbackReads(socket) {
+  if (socket._paused) return;
+  if (socket._writableState?.needDrain || (socket[kOutgoingData] ?? 0) >= socket.writableHighWaterMark) {
+    socket._paused = true;
+    socket.pause();
+  }
+}
+
+// Node's socketOnDrain: the transport drained, so resume reads unless queued
+// response bytes still hold the gate.
+function resumeFallbackReadsOnDrain(socket) {
+  if (socket._paused && (socket[kOutgoingData] ?? 0) <= socket.writableHighWaterMark) {
+    socket._paused = false;
+    socket.resume();
+  }
+}
+
 function addPipelineOutgoingData(queued, bytes) {
   const socket = queued.socket;
   socket[kOutgoingData] = (socket[kOutgoingData] ?? 0) + bytes;
@@ -2550,7 +2543,14 @@ function releasePipelineOutgoingData(socket, bytes) {
   socket[kOutgoingData] = outgoing > 0 ? outgoing : 0;
   if (socket._paused && outgoing <= socket.writableHighWaterMark) {
     socket._paused = false;
-    socket[kHandle]?.response?.resume();
+    const response = socket[kHandle]?.response;
+    if (response) {
+      response.resume();
+    } else if (!(socket instanceof NodeHTTPServerSocket)) {
+      // Fallback duplex paused by maybePauseFallbackReads: plain stream flow
+      // control is the only way to restart it.
+      socket.resume();
+    }
   }
 }
 
@@ -2566,6 +2566,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,33 +2630,48 @@ 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
     // connection cannot produce a response for this slot, so it is unusable.
+    // Deliberate divergence from Node v26, which assigns the destroyed
+    // message and wedges the connection until requestTimeout: an HTTP/1.1
+    // connection cannot skip a response slot, so reset it instead.
     if (!socket.destroyed) {
       socket.destroy();
     }
     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): the
+    // response's JS handle writes to the socket itself (nothing to switch
+    // natively), and the prototype assignSocket installs the 'close' listener
+    // a plain stream needs. Clear `finished` first, or assignSocket's _flush
+    // emits 'prefinish' before the ops replay below.
+    if (queued.ended) {
+      res.finished = false;
+    }
     res.assignSocket(socket);
   }
   socket[kRequest] = res.req;
@@ -4013,6 +4074,15 @@ 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;
+http1ServerPipeline.maybePauseFallbackReads = maybePauseFallbackReads;
+http1ServerPipeline.resumeFallbackReadsOnDrain = resumeFallbackReadsOnDrain;
+http1ServerPipeline.kMustCloseConnection = kMustCloseConnection;
+
 export default {
   Server,
   ServerResponse,

test/js/node/http/node-http-proxy.js

@@ -32,12 +32,15 @@ export async function run() {
     req.pipe(proxyRequest); // Use pipe instead of manual data handling
   });
 
-  proxyServer.listen(0, "localhost", async () => {
+  // Pin one address family: a bare "localhost" can bind ::1 while the client
+  // resolves 127.0.0.1 first (neither Node nor Bun falls back across
+  // families), which fails with ECONNREFUSED on dual-stack hosts.
+  proxyServer.listen(0, "127.0.0.1", async () => {
     const address = proxyServer.address();
 
     const options = {
       protocol: "http:",
-      hostname: "localhost",
+      hostname: "127.0.0.1",
       port: address.port,
       path: "/", // Change path to /
       headers: {