node:http: answer every request past maxRequestsPerSocket with its own 503

src/js/node/_http_server.ts

@@ -879,12 +879,9 @@
           const requestCount = (socket._requestCount || 0) + 1;
           socket._requestCount = requestCount;
           http_res._maxRequestsPerSocket = server.maxRequestsPerSocket;
-          // At (or beyond) the limit the response advertises Connection:
-          // close, like Node.js - including the over-limit 503 dropRequest
-          // answer, which would otherwise claim keep-alive right before the
-          // socket is destroyed. Closing the socket here instead would race
-          // already-pipelined requests, which still need to be dispatched so
-          // they can be answered with 503 via dropRequest.
+          // At (or beyond) the limit the response advertises Connection: close,
+          // like Node.js - including the over-limit 503 dropRequest answer,
+          // which would otherwise claim keep-alive.
           http_res.maxRequestsOnConnectionReached = server.maxRequestsPerSocket <= requestCount;
           if (server.maxRequestsPerSocket < requestCount) {
             reachedRequestsLimit = true;
@@ -975,16 +972,16 @@
         }
 
         if (reachedRequestsLimit) {
+          // Like Node.js's parserOnIncoming: every request past the limit gets
+          // its own 'dropRequest' + 503. Closing the socket now (or on this
+          // response's "finish") would drop the requests pipelined behind it.
           server.emit("dropRequest", http_req, socket);
           http_res.writeHead(503);
           http_res.end();
-          if (isPipelined) {
-            // The 503 is queued behind the in-flight responses; the connection
-            // closes once it has been written, like Node.js (Connection: close).
-            http_res[kMustCloseConnection] = true;
-          } else {
-            socket.destroy();
+          if (!socket[kEndAfterDroppedRequests]) {
+            socket[kEndAfterDroppedRequests] = true;
+            setImmediate(endSocketAfterDroppedRequests, socket);
           }
         } else if (is_upgrade) {
           // Hand the raw socket over to the 'upgrade' listener, like Node.js.
           // Without a body the message is already complete: bytes that arrived
@@ -2410,6 +2407,27 @@
 }
 
 const kMustCloseConnection = Symbol("kMustCloseConnection");
+const kEndAfterDroppedRequests = Symbol("kEndAfterDroppedRequests");
+
+// Runs once the current read has been fully parsed, so every request that was
+// already pipelined behind the dropped one has been answered with its own 503.
+function endSocketAfterDroppedRequests(socket) {
+  if (!socket || socket.destroyed || socket.writableEnded) return;
+  // An earlier response on this connection has not finished yet: its queued
+  // 503s are still waiting to be written, so close after the last one instead.
+  const queue = socket[kPipelinedResponses];
+  if (queue?.length) {
+    queue[queue.length - 1][kMustCloseConnection] = true;
+    return;
+  }
+  const inFlight = socket._httpMessage;
+  if (inFlight && !inFlight.finished) {
+    inFlight[kMustCloseConnection] = true;
+    return;
+  }
+  socket.end();
+}
+
 function stopServerResponsePerf(this: any) {
   if (this[kServerResponseStatistics] && hasObserver("http")) {
     stopPerf(this, kServerResponseStatistics, {

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 () => {
+  // 127.0.0.1 rather than "localhost": where the resolver prefers IPv6 the
+  // server binds ::1 while the client dials 127.0.0.1 and the connect is
+  // refused. exampleSite() already binds 127.0.0.1.
+  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: {

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

@@ -3657,29 +3657,57 @@
   }
 });
 
-it("the over-limit 503 advertises Connection: close, not keep-alive", async () => {
+const rawGet = (path: string) => `GET ${path} HTTP/1.1\r\nHost: x\r\n\r\n`;
+const rawPost = (path: string, body: string) =>
+  `POST ${path} HTTP/1.1\r\nHost: x\r\nContent-Length: ${body.length}\r\n\r\n${body}`;
+
+// Responses whose header block has fully arrived.
+const responsesIn = (raw: string) => Math.min(raw.split("HTTP/1.1 ").length - 1, raw.split("\r\n\r\n").length - 1);
+
+// Pipelines every request into one segment. `until` "responses" hangs up as
+// soon as they have all arrived; "serverClose" waits for the server to hang up
+// instead, so a test can assert it did. Resolving on "close" in either case
+// means a server that tears the connection down mid-pipeline yields the
+// truncated bytes (an assertion failure) rather than hanging until the test
+// times out.
+function sendRequests(
+  port: number,
+  requests: string[],
+  until: "responses" | "serverClose" = "responses",
+): Promise<{ raw: string; serverEnded: boolean }> {
+  const { promise, resolve, reject } = Promise.withResolvers<{ raw: string; serverEnded: boolean }>();
+  const socket = connect(port, "127.0.0.1");
+  let data = "";
+  let serverEnded = false;
+  socket.on("data", chunk => {
+    data += chunk;
+    if (until === "responses" && responsesIn(data) >= requests.length) {
+      socket.destroy();
+      resolve({ raw: data, serverEnded });
+    }
+  });
+  socket.on("end", () => (serverEnded = true));
+  socket.on("close", () => resolve({ raw: data, serverEnded }));
+  socket.on("error", reject);
+  socket.on("connect", () => socket.write(requests.join("")));
+  return promise;
+}
+
+it.concurrent("the over-limit 503 advertises Connection: close, not keep-alive", async () => {
   // Node sets maxRequestsOnConnectionReached unconditionally
   // (maxRequestsPerSocket <= count), so the dropRequest 503 carries
-  // Connection: close instead of advertising keep-alive right before the
-  // socket is destroyed.
+  // Connection: close instead of advertising keep-alive.
   const server = createServer((req, res) => res.end("ok"));
   server.maxRequestsPerSocket = 1;
   try {
     server.listen(0, "127.0.0.1");
     await once(server, "listening");
     const { port } = server.address() as AddressInfo;
 
-    const out = await new Promise<string>((resolve, reject) => {
-      const socket = connect(port, "127.0.0.1");
-      let data = "";
-      socket.on("data", chunk => (data += chunk));
-      socket.on("close", () => resolve(data));
-      socket.on("error", reject);
-      // Two pipelined requests: the second exceeds maxRequestsPerSocket.
-      socket.write("GET / HTTP/1.1\r\nHost: x\r\n\r\n" + "GET / HTTP/1.1\r\nHost: x\r\n\r\n");
-    });
+    // Two pipelined requests: the second exceeds maxRequestsPerSocket.
+    const { raw } = await sendRequests(port, [rawGet("/a"), rawGet("/b")]);
 
-    const second = out.slice(out.indexOf("HTTP/1.1 503"));
+    const second = raw.slice(raw.indexOf("HTTP/1.1 503"));
     expect(second).toContain("HTTP/1.1 503");
     expect(second).toContain("Connection: close");
     expect(second).not.toContain("keep-alive");
@@ -3688,6 +3716,122 @@
   }
 });
 
+it.concurrent("every pipelined request past maxRequestsPerSocket gets its own 503 and dropRequest", async () => {
+  // Node answers each over-limit request with a 503 and emits 'dropRequest'
+  // for it, so a client that pipelined several requests at once is not left
+  // with a torn-down connection and no response.
+  const drops: { url: string; isIncomingMessage: boolean; socket: unknown }[] = [];
+  const server = createServer((req, res) => {
+    req.resume();
+    res.end("ok");
+  });
+  server.maxRequestsPerSocket = 1;
+  server.on("dropRequest", (req, socket) =>
+    drops.push({ url: req.url, isIncomingMessage: req instanceof IncomingMessage, socket }),
+  );
+  try {
+    server.listen(0, "127.0.0.1");
+    await once(server, "listening");
+    const { port } = server.address() as AddressInfo;
+
+    // Three requests pipelined into one segment: /b and /c are both over the
+    // limit of 1, so both must be answered.
+    const { raw } = await sendRequests(port, [rawGet("/a"), rawGet("/b"), rawGet("/c")]);
+
+    expect([...raw.matchAll(/HTTP\/1\.1 (\d{3})/g)].map(m => m[1])).toEqual(["200", "503", "503"]);
+    expect(drops.map(({ url, isIncomingMessage }) => ({ url, isIncomingMessage }))).toEqual([
+      { url: "/b", isIncomingMessage: true },
+      { url: "/c", isIncomingMessage: true },
+    ]);
+    // Both drops report the one connection they arrived on.
+    expect(drops[0].socket).toBeDefined();
+    expect(drops[1].socket).toBe(drops[0].socket);
+  } finally {
+    server.close();
+  }
+});
+
+it.concurrent("a dropped request carrying a body does not stall the rest of the pipeline", async () => {
+  // The 503'd request's body still has to come off the wire, otherwise the
+  // parser never reaches the request pipelined behind it.
+  const drops: string[] = [];
+  const server = createServer((req, res) => {
+    req.resume();
+    res.end("ok");
+  });
+  server.maxRequestsPerSocket = 1;
+  server.on("dropRequest", req => drops.push(req.url));
+  try {
+    server.listen(0, "127.0.0.1");
+    await once(server, "listening");
+    const { port } = server.address() as AddressInfo;
+
+    const body = "hello=world";
+    const requests = [rawPost("/a", body), rawPost("/b", body), rawPost("/c", body)];
+    const { raw } = await sendRequests(port, requests);
+
+    expect([...raw.matchAll(/HTTP\/1\.1 (\d{3})/g)].map(m => m[1])).toEqual(["200", "503", "503"]);
+    expect(drops).toEqual(["/b", "/c"]);
+  } finally {
+    server.close();
+  }
+});
+
+it.concurrent("the connection is ended only after the pipelined requests have been answered", async () => {
+  // Node leans on keepAliveTimeout to reap the connection; Bun hangs up once
+  // the current read has been answered - but not before the requests already
+  // sitting in the read buffer have each gotten their own 503.
+  const drops: string[] = [];
+  const server = createServer((req, res) => {
+    req.resume();
+    res.end("ok");
+  });
+  server.maxRequestsPerSocket = 1;
+  server.on("dropRequest", req => drops.push(req.url));
+  try {
+    server.listen(0, "127.0.0.1");
+    await once(server, "listening");
+    const { port } = server.address() as AddressInfo;
+
+    // The client never hangs up, so reaching "close" at all means the server did.
+    const { raw, serverEnded } = await sendRequests(port, [rawGet("/a"), rawGet("/b"), rawGet("/c")], "serverClose");
+
+    expect(serverEnded).toBe(true);
+    expect([...raw.matchAll(/HTTP\/1\.1 (\d{3})/g)].map(m => m[1])).toEqual(["200", "503", "503"]);
+    expect(drops).toEqual(["/b", "/c"]);
+  } finally {
+    server.close();
+  }
+});
+
+it.concurrent(
+  "over-limit 503s queued behind an async response are written before the connection ends",
+  async () => {
+    // When the first response finishes asynchronously, the over-limit 503s are
+    // buffered in the response pipeline; closing before the last one flushes
+    // would drop it.
+    const drops: string[] = [];
+    const server = createServer((req, res) => {
+      req.resume();
+      setImmediate(() => res.end("ok"));
+    });
+    server.maxRequestsPerSocket = 1;
+    server.on("dropRequest", req => drops.push(req.url));
+    try {
+      server.listen(0, "127.0.0.1");
+      await once(server, "listening");
+      const { port } = server.address() as AddressInfo;
+
+      const { raw } = await sendRequests(port, [rawGet("/a"), rawGet("/b"), rawGet("/c")], "serverClose");
+
+      expect([...raw.matchAll(/HTTP\/1\.1 (\d{3})/g)].map(m => m[1])).toEqual(["200", "503", "503"]);
+      expect(drops).toEqual(["/b", "/c"]);
+    } finally {
+      server.close();
+    }
+  },
+);
+
 it("a non-200 CONNECT through a proxy that holds the connection open is destroyed client-side", async () => {
   // cleanupAndPropagate deliberately defers destroy to req.onSocket for
   // status-code tunnel failures; oncreate must forward the socket so