Bun.serve: keep the connection-close mark across a pipelined request

packages/bun-uws/src/HttpResponseData.h

@@ -89,6 +89,10 @@ struct HttpResponseData : AsyncSocketData<SSL>, HttpParser {
         HTTP_WRITE_CALLED = 2, // used
         HTTP_END_CALLED = 4, // used
         HTTP_RESPONSE_PENDING = 8, // used
+        /* Close once the response in flight has completed and flushed (HTTP/1.0
+         * or Connection: close request, or a response ended with closeConnection).
+         * Connection-scoped: a request pipelined behind the one that set it must
+         * not make the connection persistent again. */
         HTTP_CONNECTION_CLOSE = 16, // used
         HTTP_WROTE_CONTENT_LENGTH_HEADER = 32, // used
         HTTP_WROTE_DATE_HEADER = 64, // used
@@ -156,7 +160,7 @@ struct HttpResponseData : AsyncSocketData<SSL>, HttpParser {
          * There is one HttpResponseData per socket, reused by every request on a
          * keep-alive connection, so starting a new response clears the rest of the
          * word (resetResponseState) - these have to survive that. */
-        HTTP_CONNECTION_SCOPED = HTTP_NODE_PARSING_STOPPED | HTTP_NODE_READS_PAUSED
+        HTTP_CONNECTION_SCOPED = HTTP_CONNECTION_CLOSE | HTTP_NODE_PARSING_STOPPED | HTTP_NODE_READS_PAUSED
             | HTTP_NODE_TUNNEL_AFTER_BODY | HTTP_NODE_RECEIVED_FIN | HTTP_CLOSE_WHEN_IDLE,
     };
 

test/js/bun/http/serve.test.ts

@@ -619,6 +619,103 @@ it.each([
   expect(response.slice(response.indexOf("\r\n\r\n") + 4)).toBe("/helloooo");
 });
 
+// RFC 9112 9.3 / 9.6: an HTTP/1.0 request (Bun never answers one with
+// keep-alive), a request carrying Connection: close, or a response carrying
+// Connection: close makes the connection non-persistent. That mark belongs to
+// the connection, so an HTTP/1.1 request pipelined behind such a request in the
+// same TCP segment must not clear it: the server still has to close once it has
+// answered what it dispatched.
+describe("a request pipelined behind a non-persistent request does not keep the connection alive", () => {
+  const keepAliveFirst = "GET /first HTTP/1.1\r\nHost: x\r\n\r\n";
+  const pipelinedSecond = "GET /second HTTP/1.1\r\nHost: x\r\n\r\n";
+  const probe = "GET /probe HTTP/1.1\r\nHost: x\r\n\r\n";
+
+  // The handler must answer synchronously: Bun.serve only dispatches a
+  // pipelined request when the previous response has already completed
+  // (otherwise it closes the connection at the second request instead).
+  function startServer(firstResponseHeaders?: Record<string, string>) {
+    return Bun.serve({
+      port: 0,
+      hostname: "127.0.0.1",
+      fetch(req) {
+        const pathname = new URL(req.url).pathname;
+        // The ";" terminates the marker: the next response's status line
+        // follows the body directly on the wire.
+        return new Response("body:" + pathname + ";", {
+          headers: pathname === "/first" ? firstResponseHeaders : undefined,
+        });
+      },
+    });
+  }
+
+  // Writes both requests in one segment. Once the response to the pipelined
+  // request has arrived, sends a probe request: a server that closed the
+  // connection can never answer it, a server that was talked back into
+  // keep-alive answers it. Settles on whichever of the two happens.
+  async function pipelineThenProbe(port: number, first: string) {
+    const { promise, resolve } = Promise.withResolvers<"server closed" | "probe answered">();
+    let received = "";
+    let probeSent = false;
+    const socket = net.connect(port, "127.0.0.1");
+    socket.setEncoding("latin1");
+    socket.on("connect", () => socket.write(first + pipelinedSecond));
+    socket.on("data", chunk => {
+      received += chunk;
+      if (received.includes("body:/probe;")) {
+        resolve("probe answered");
+      } else if (!probeSent && received.includes("body:/second;")) {
+        probeSent = true;
+        socket.write(probe);
+      }
+    });
+    socket.on("end", () => resolve("server closed"));
+    socket.on("close", () => resolve("server closed"));
+    // ECONNRESET / EPIPE once the server has closed; "close" follows.
+    socket.on("error", () => {});
+    const outcome = await promise;
+    socket.destroy();
+    return { outcome, firstBody: received.match(/body:[^;]*;/)?.[0] ?? null };
+  }
+
+  it.each([
+    { label: "HTTP/1.0 request", first: "GET /first HTTP/1.0\r\nHost: x\r\n\r\n" },
+    {
+      label: "HTTP/1.0 request with Connection: keep-alive",
+      first: "GET /first HTTP/1.0\r\nHost: x\r\nConnection: keep-alive\r\n\r\n",
+    },
+    {
+      label: "HTTP/1.0 request with Connection: close",
+      first: "GET /first HTTP/1.0\r\nHost: x\r\nConnection: close\r\n\r\n",
+    },
+    {
+      label: "HTTP/1.1 request with Connection: close",
+      first: "GET /first HTTP/1.1\r\nHost: x\r\nConnection: close\r\n\r\n",
+    },
+    {
+      label: "HTTP/1.1 request answered with Connection: close",
+      first: keepAliveFirst,
+      firstResponseHeaders: { connection: "close" },
+    },
+  ])(
+    "$label + pipelined HTTP/1.1 request: the server closes the connection",
+    async ({ first, firstResponseHeaders }) => {
+      using server = startServer(firstResponseHeaders);
+      expect(await pipelineThenProbe(server.port, first)).toEqual({
+        outcome: "server closed",
+        firstBody: "body:/first;",
+      });
+    },
+  );
+
+  it("control: a persistent HTTP/1.1 request + pipelined HTTP/1.1 request keeps the connection alive", async () => {
+    using server = startServer();
+    expect(await pipelineThenProbe(server.port, keepAliveFirst)).toEqual({
+      outcome: "probe answered",
+      firstBody: "body:/first;",
+    });
+  });
+});
+
 describe("streaming", () => {
   describe("error handler", () => {
     it("throw on pull renders headers, does not call error handler", async () => {