Bun.serve: send the completed responses of a pipelined read before JavaScript runs again

packages/bun-uws/src/AsyncSocket.h

@@ -140,6 +140,17 @@ struct AsyncSocket {
         return getLoopData()->findCorkSlot(this) != LoopData::INVALID_CORK_SLOT;
     }
 
+    /* Sends what the cork buffer holds for this socket. The socket stays corked. */
+    void sendCorked() {
+        LoopData *loopData = getLoopData();
+        int slot = loopData->findCorkSlot(this);
+        if (slot == LoopData::INVALID_CORK_SLOT || loopData->getCorkSlot(slot)->offset == 0) {
+            return;
+        }
+        uncork();
+        cork();
+    }
+
     /* Returns a suitable buffer for temporary assemblation of send data */
     std::pair<char *, SendBufferAttribute> getSendBuffer(size_t size) {
         LoopData *loopData = getLoopData();

packages/bun-uws/src/HttpResponse.h

@@ -115,10 +115,11 @@ struct HttpResponse : public AsyncSocket<SSL> {
      *
      * The socket onData is parsing gets onData's uncork and close gate once the
      * read is consumed: false. A Bun.serve response leaves the cork to onData,
-     * so the responses to requests pipelined in one read share one send(). A
-     * node:http response is still sent now: its 'finish' event and end()
-     * callback run before onData gets control back and expect the bytes to be
-     * out.
+     * so the responses to requests pipelined in one read share one send(). Bun
+     * sends them earlier, with sendCorked(), when the handler of a later
+     * request is about to run JavaScript. A node:http response is still sent
+     * now: its 'finish' event and end() callback run before onData gets control
+     * back and expect the bytes to be out.
      *
      * Any other socket (an async handler completing, possibly inside another
      * socket's parse window via a drained microtask) is uncorked here and gets

src/runtime/server/mod.rs

@@ -745,6 +745,9 @@ impl<const SSL: bool, const DEBUG: bool> NewServer<SSL, DEBUG> {
             None
         };
 
+        // Earlier responses of this read leave before JavaScript runs (the 413 above runs none).
+        resp_ref.send_corked();
+
         server.on_pending_request();
 
         req.set_yield(false);

src/runtime/server/server_body.rs

@@ -3266,6 +3266,8 @@ where
             resp.end_without_body(true);
             return;
         }
+        // Same as `prepare_js_request_context`, which this path does not use.
+        resp.send_corked();
         let _entered = this.vm().enter_event_loop_scope_without_checkpoint();
         this.on_pending_request();
         req.set_yield(false);

src/uws_sys/Response.rs

@@ -184,6 +184,11 @@ impl<const SSL: bool> Response<SSL> {
         c::uws_res_uncork(Self::ssl_flag(), self.as_raw())
     }
 
+    /// Sends what the cork buffer holds for this socket. The socket stays corked.
+    pub fn send_corked(&mut self) {
+        c::uws_res_send_corked(Self::ssl_flag(), self.as_raw())
+    }
+
     pub(crate) fn pause(&mut self) {
         c::uws_res_pause(Self::ssl_flag(), self.as_raw())
     }
@@ -1175,6 +1180,7 @@ pub mod c {
             is_ipv6: &mut bool,
         ) -> usize;
         pub(crate) safe fn uws_res_uncork(ssl: i32, res: &mut uws_res);
+        pub(crate) safe fn uws_res_send_corked(ssl: i32, res: &mut uws_res);
         pub(crate) fn uws_res_end(
             ssl: i32,
             res: *mut uws_res,

src/uws_sys/libuwsockets.cpp

@@ -1525,6 +1525,20 @@ size_t uws_req_get_header(uws_req_t *res, const char *lower_case_header,
     }
   }
 
+  void uws_res_send_corked(int ssl, uws_res_r res)
+  {
+    if (ssl)
+    {
+      uWS::HttpResponse<true> *uwsRes = (uWS::HttpResponse<true> *)res;
+      uwsRes->sendCorked();
+    }
+    else
+    {
+      uWS::HttpResponse<false> *uwsRes = (uWS::HttpResponse<false> *)res;
+      uwsRes->sendCorked();
+    }
+  }
+
   void us_socket_mark_needs_more_not_ssl(uws_res_r res)
   {
     us_socket_r s = (us_socket_t *)res;

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

@@ -15,6 +15,7 @@ import {
   isIPv6,
   isLinux,
   isPosix,
+  isWindows,
   libcPathForDlopen,
   runFixtureMaxRSS,
   tempDir,
@@ -27,10 +28,12 @@ import { join, resolve } from "path";
 // import app_jsx from "./app.jsx";
 import { heapStats } from "bun:jsc";
 import { spawn } from "child_process";
+import { once } from "node:events";
 import net from "node:net";
 import { networkInterfaces } from "node:os";
 import { Duplex } from "node:stream";
 import nodeTls from "node:tls";
+import { Worker } from "node:worker_threads";
 import { tmpdir } from "os";
 
 let renderToReadableStream: any = null;
@@ -4706,8 +4709,9 @@ it("serves a TLS connection whose handshake completes after a graceful stop()",
   }
 });
 
-// Requests that arrive in one read are dispatched from one onData() call. The
-// socket is corked for the whole call, so their responses leave together.
+// HTTP/1.1 requests that arrive in one read are dispatched from one onData()
+// call. The socket is corked for the whole call. Completed responses share one
+// send(), and they leave before the handler of a later request runs JavaScript.
 describe("requests pipelined in one read", () => {
   const get = (path: string, extraHeaders = "") => `GET ${path} HTTP/1.1\r\nHost: x\r\n${extraHeaders}\r\n`;
 
@@ -4752,8 +4756,10 @@ describe("requests pipelined in one read", () => {
   const echoPath = (req: Request) => new Response(new URL(req.url).pathname);
 
   // Each send() is one TCP segment on loopback (TCP_NODELAY). The client counts
-  // the segments with TCP_INFO.
-  it.skipIf(!isLinux && !isAndroid)("are answered with one send()", async () => {
+  // the segments with TCP_INFO. A handler that runs JavaScript first sends the
+  // responses completed before it. Responses that need no JavaScript (static
+  // routes) collect until then, or until the read is consumed.
+  it.skipIf(!isLinux && !isAndroid)("share one send() until a handler runs JavaScript", async () => {
     await using proc = Bun.spawn({
       cmd: [
         bunExe(),
@@ -4773,9 +4779,8 @@ describe("requests pipelined in one read", () => {
             if (len[0] < 156) throw new Error("tcp_info has no tcpi_data_segs_in: " + len[0] + " bytes");
             return new DataView(info.buffer).getUint32(152, true);
           }
-          const depth = 8;
-          const request = "GET / HTTP/1.1\\r\\nHost: x\\r\\n\\r\\n";
-          async function run(options) {
+          const request = path => "GET " + path + " HTTP/1.1\\r\\nHost: x\\r\\n\\r\\n";
+          async function run(options, paths = ["/", "/", "/", "/", "/", "/", "/", "/"]) {
             const server = Bun.serve({ port: 0, hostname: "127.0.0.1", ...options });
             const done = Promise.withResolvers();
             let reply = "";
@@ -4785,11 +4790,11 @@ describe("requests pipelined in one read", () => {
               port: server.port,
               socket: {
                 open(socket) {
-                  socket.write(Buffer.alloc(request.length * depth, request));
+                  socket.write(paths.map(request).join(""));
                 },
                 data(socket, chunk) {
                   reply += chunk.toString("latin1");
-                  if (responses() === depth && reply.endsWith("hello")) done.resolve();
+                  if (responses() === paths.length && reply.endsWith("hello")) done.resolve();
                 },
                 error(socket, error) {
                   done.reject(error);
@@ -4805,10 +4810,54 @@ describe("requests pipelined in one read", () => {
             server.stop(true);
             return result;
           }
+          // The 413 for a body over maxRequestBodySize runs no JavaScript. A client that reads
+          // the 100 Continue alone starts an upload that the server then resets.
+          async function runTooLarge() {
+            const server = Bun.serve({
+              port: 0,
+              hostname: "127.0.0.1",
+              maxRequestBodySize: 1024,
+              fetch: () => new Response("hello"),
+            });
+            const done = Promise.withResolvers();
+            let reply = "";
+            let segments = 0;
+            await Bun.connect({
+              hostname: "127.0.0.1",
+              port: server.port,
+              socket: {
+                open(socket) {
+                  socket.write(
+                    "POST / HTTP/1.1\\r\\nHost: x\\r\\nExpect: 100-continue\\r\\nContent-Length: 2000000\\r\\n\\r\\n",
+                  );
+                },
+                data(socket, chunk) {
+                  reply += chunk.toString("latin1");
+                  segments = dataSegmentsIn(socket.fd);
+                },
+                error(socket, error) {
+                  done.reject(error);
+                },
+                close() {
+                  done.resolve();
+                },
+              },
+            });
+            await done.promise;
+            server.stop(true);
+            return { statuses: reply.split("\\r\\n").filter(line => line.startsWith("HTTP/1.1 ")), segments };
+          }
           console.log(JSON.stringify({
             fetch: await run({ fetch: () => new Response("hello", { headers: { "X-Custom": "1" } }) }),
             route: await run({ routes: { "/": () => new Response("hello") } }),
             static: await run({ routes: { "/": new Response("hello") } }),
+            // One send() before each "/js" handler, and one when the read is consumed:
+            // [static static static] [js static static] [js static]
+            mixed: await run(
+              { routes: { "/static": new Response("hello"), "/js": () => new Response("hello") } },
+              ["/static", "/static", "/static", "/js", "/static", "/static", "/js", "/static"],
+            ),
+            tooLarge: await runTooLarge(),
           }));
         `,
       ],
@@ -4818,14 +4867,100 @@ describe("requests pipelined in one read", () => {
     const [stdout, exitCode] = await Promise.all([proc.stdout.text(), proc.exited]);
     expect({ result: stdout.trim() ? JSON.parse(stdout) : stdout, exitCode }).toEqual({
       result: {
-        fetch: { responses: 8, segments: 1 },
-        route: { responses: 8, segments: 1 },
+        fetch: { responses: 8, segments: 8 },
+        route: { responses: 8, segments: 8 },
         static: { responses: 8, segments: 1 },
+        mixed: { responses: 8, segments: 3 },
+        tooLarge: { statuses: ["HTTP/1.1 100 Continue", "HTTP/1.1 413 Request Entity Too Large"], segments: 1 },
       },
       exitCode: 0,
     });
   });
 
+  // On Windows the exit resets the connection (the client reads ECONNRESET), and
+  // the reset can discard what the client has not read yet.
+  it.skipIf(isWindows)("delivers the completed responses when a later handler ends the process", async () => {
+    await using proc = Bun.spawn({
+      cmd: [
+        bunExe(),
+        "-e",
+        `
+          const server = Bun.serve({
+            port: 0,
+            hostname: "127.0.0.1",
+            routes: { "/static": new Response("/static") },
+            fetch(req) {
+              const { pathname } = new URL(req.url);
+              if (pathname === "/exit") process.exit(0);
+              return new Response(pathname);
+            },
+          });
+          console.log(server.port);
+        `,
+      ],
+      env: bunEnv,
+      stdout: "pipe",
+      stderr: "inherit",
+    });
+    let stdout = "";
+    for await (const chunk of proc.stdout) {
+      stdout += Buffer.from(chunk).toString();
+      if (stdout.includes("\n")) break;
+    }
+    // The static response needs no JavaScript. It is still in the cork buffer when "/exit" is dispatched.
+    const reply = await exchange(plain(Number(stdout)), get("/a") + get("/static") + get("/exit"));
+    expect(parseResponses(reply)).toEqual([
+      { status: "HTTP/1.1 200 OK", body: "/a" },
+      { status: "HTTP/1.1 200 OK", body: "/static" },
+    ]);
+    expect(await proc.exited).toBe(0);
+  });
+
+  it("sends the completed responses before a later handler blocks", async () => {
+    // The "/wait" handler blocks the server's thread until the client has the
+    // responses before it, so the server runs in a worker.
+    using dir = tempDir("serve-pipelined-blocked-handler", {
+      "server.mjs": `
+        import { parentPort, workerData as released } from "node:worker_threads";
+        const server = Bun.serve({
+          port: 0,
+          hostname: "127.0.0.1",
+          routes: { "/static": new Response("/static") },
+          fetch(req) {
+            const { pathname } = new URL(req.url);
+            if (pathname !== "/wait") return new Response(pathname);
+            return new Response(Atomics.wait(released, 0, 0, 2000) === "timed-out" ? "timed-out" : "released");
+          },
+        });
+        parentPort.postMessage(server.port);
+      `,
+    });
+    const released = new Int32Array(new SharedArrayBuffer(4));
+    const worker = new Worker(join(String(dir), "server.mjs"), { workerData: released });
+    try {
+      const [port] = await once(worker, "message");
+      // The static response needs no JavaScript. It is still in the cork buffer when "/wait" is dispatched.
+      const reply = await exchange(
+        plain(port),
+        get("/a") + get("/static") + get("/wait", "Connection: close\r\n"),
+        reply => {
+          if (parseResponses(reply).length >= 2 && reply.includes("\r\n\r\n/static")) {
+            Atomics.store(released, 0, 1);
+            Atomics.notify(released, 0);
+          }
+          return false;
+        },
+      );
+      expect(parseResponses(reply)).toEqual([
+        { status: "HTTP/1.1 200 OK", body: "/a" },
+        { status: "HTTP/1.1 200 OK", body: "/static" },
+        { status: "HTTP/1.1 200 OK", body: "released" },
+      ]);
+    } finally {
+      await worker.terminate();
+    }
+  });
+
   // The tests below pass without the batching too. They cover the paths that
   // have to send the cork buffer before they close or hand over the socket.
 
@@ -4933,6 +5068,39 @@ describe("requests pipelined in one read", () => {
     expect(rest[1]).toEndWith("\r\n\r\n" + frame);
   });
 
+  it("sends the completed responses before an upgrade that open() terminates", async () => {
+    using server = Bun.serve({
+      port: 0,
+      hostname: "127.0.0.1",
+      fetch(req, server) {
+        if (req.url.endsWith("/ws")) {
+          if (server.upgrade(req)) return;
+          return new Response("upgrade failed", { status: 500 });
+        }
+        return echoPath(req);
+      },
+      websocket: {
+        open(ws) {
+          ws.terminate();
+        },
+        message() {},
+      },
+    });
+    const reply = await exchange(
+      plain(server.port),
+      get("/a") +
+        get("/b") +
+        get(
+          "/ws",
+          "Connection: Upgrade\r\nUpgrade: websocket\r\nSec-WebSocket-Version: 13\r\nSec-WebSocket-Key: dGhlIHNhbXBsZSBub25jZQ==\r\n",
+        ),
+    );
+    expect(parseResponses(reply).slice(0, 2)).toEqual([
+      { status: "HTTP/1.1 200 OK", body: "/a" },
+      { status: "HTTP/1.1 200 OK", body: "/b" },
+    ]);
+  });
+
   it("sends the completed response before a graceful stop() closes the idle connection", async () => {
     using server = Bun.serve({
       port: 0,