http: route identity chunked bodies through the multi-packet decode path

bench/snippets/fetch-chunked-small.mjs

@@ -0,0 +1,76 @@
+// Benchmark fetch() decoding small Transfer-Encoding: chunked bodies that
+// arrive with their headers in a single read. Exercises the <=16 KiB
+// dispatch in the HTTP client's chunked-body handler for both identity and
+// gzip Content-Encoding.
+import { bench, group, run } from "../runner.mjs";
+import net from "node:net";
+import zlib from "node:zlib";
+
+function chunked(buf) {
+  return Buffer.concat([
+    Buffer.from(buf.length.toString(16) + "\r\n"),
+    buf,
+    Buffer.from("\r\n0\r\n\r\n"),
+  ]);
+}
+
+function makeReply(body, gzip) {
+  const payload = gzip ? zlib.gzipSync(body, { level: 6 }) : body;
+  return Buffer.concat([
+    Buffer.from(
+      "HTTP/1.1 200 OK\r\n" +
+        "Transfer-Encoding: chunked\r\n" +
+        (gzip ? "Content-Encoding: gzip\r\n" : "") +
+        "Connection: keep-alive\r\n" +
+        "\r\n",
+    ),
+    chunked(payload),
+  ]);
+}
+
+const sizes = [256, 4096, 15 * 1024];
+const bodies = Object.fromEntries(sizes.map(n => [n, Buffer.alloc(n, "x")]));
+const replies = {};
+for (const n of sizes) {
+  replies[`i${n}`] = makeReply(bodies[n], false);
+  replies[`g${n}`] = makeReply(bodies[n], true);
+}
+
+const server = net.createServer(sock => {
+  sock.setNoDelay(true);
+  let pending = Buffer.alloc(0);
+  sock.on("data", chunk => {
+    pending = Buffer.concat([pending, chunk]);
+    while (true) {
+      const end = pending.indexOf("\r\n\r\n");
+      if (end < 0) break;
+      const head = pending.subarray(0, end).toString("latin1");
+      pending = pending.subarray(end + 4);
+      const m = head.match(/^GET \/(\w+)/);
+      sock.write(replies[m[1]]);
+    }
+  });
+});
+
+await new Promise(r => server.listen(0, r));
+const base = `http://127.0.0.1:${server.address().port}`;
+
+// Warm the keep-alive connection so the first iteration doesn't pay connect.
+for (const key of Object.keys(replies)) await fetch(`${base}/${key}`).then(r => r.arrayBuffer());
+
+for (const n of sizes) {
+  group(`chunked ${n}B`, () => {
+    bench("identity → arrayBuffer()", async () => {
+      const r = await fetch(`${base}/i${n}`);
+      await r.arrayBuffer();
+    });
+    bench("gzip → arrayBuffer()", async () => {
+      const r = await fetch(`${base}/g${n}`);
+      await r.arrayBuffer();
+    });
+  });
+}
+
+await run();
+server.close();
+process.exit(0);

src/http/lib.rs

@@ -4781,8 +4781,17 @@ impl<'a> HTTPClient<'a> {
         &mut self,
         incoming_data: &[u8],
     ) -> crate::Result<bool> {
+        // The single-packet path decodes into a 16 KiB scratch so a small
+        // compressed body can be handed straight to `decompress_bytes` without
+        // ever allocating `compressed_body`. For identity bodies that shortcut
+        // doesn't exist: scratch copy + `append_slice_exact` is one memcpy more
+        // than `_from_multiple_packets`' append + decode-in-tail + truncate, so
+        // route identity bodies there unconditionally.
         let small_len = 16 * 1024usize;
-        if incoming_data.len() <= small_len && self.state.get_body_buffer().list.is_empty() {
+        if self.state.encoding.is_compressed()
+            && incoming_data.len() <= small_len
+            && self.state.get_body_buffer().list.is_empty()
+        {
             self.handle_response_body_chunked_encoding_from_single_packet(incoming_data)
         } else {
             self.handle_response_body_chunked_encoding_from_multiple_packets(incoming_data)

test/js/web/fetch/fetch-retry-chunked.test.ts

@@ -4,9 +4,10 @@
 // body_out_str == None state itself is not deterministically reachable from
 // fetch().
 
-import { expect, test } from "bun:test";
+import { describe, expect, test } from "bun:test";
 import type { AddressInfo } from "node:net";
 import net from "node:net";
+import zlib from "node:zlib";
 
 // The server drops every third request without responding, so the client
 // that adopted the pooled socket observes on_close with response_stage ==
@@ -59,3 +60,68 @@
     await new Promise<void>(r => server.close(() => r()));
   }
 });
+
+// The <=16 KiB chunked-body dispatcher routes identity bodies through the
+// append + decode-in-tail path and compressed bodies through the scratch
+// decode path. These pin both paths for a body that fits in one read and for
+// one that is split so the first read yields -2 (needs more data).
+describe("small chunked body decode", () => {
+  function chunked(buf: Buffer): Buffer {
+    return Buffer.concat([Buffer.from(buf.length.toString(16) + "\r\n"), buf, Buffer.from("\r\n0\r\n\r\n")]);
+  }
+
+  async function serve(reply: Buffer, fn: (url: string) => Promise<void>): Promise<void> {
+    const sockets = new Set<net.Socket>();
+    const server = net.createServer(sock => {
+      sockets.add(sock);
+      sock.on("close", () => sockets.delete(sock));
+      sock.on("error", () => {});
+      sock.once("data", () => sock.write(reply));
+    });
+    await new Promise<void>((resolve, reject) => {
+      server.once("error", reject);
+      server.listen(0, "127.0.0.1", resolve);
+    });
+    const { port } = server.address() as AddressInfo;
+    try {
+      await fn(`http://127.0.0.1:${port}/`);
+    } finally {
+      for (const s of sockets) s.destroy();
+      await new Promise<void>(r => server.close(() => r()));
+    }
+  }
+
+  function reply(body: Buffer, extraHeader: string): Buffer {
+    return Buffer.concat([
+      Buffer.from(
+        "HTTP/1.1 200 OK\r\n" +
+          "Transfer-Encoding: chunked\r\n" +
+          "Connection: keep-alive\r\n" +
+          extraHeader +
+          "\r\n",
+      ),
+      chunked(body),
+    ]);
+  }
+
+  for (const n of [1, 256, 4096, 15 * 1024]) {
+    const body = Buffer.alloc(n, "x");
+    const gz = zlib.gzipSync(body, { level: 6 });
+
+    test.concurrent(`identity ${n}B`, async () => {
+      await serve(reply(body, ""), async url => {
+        const res = await fetch(url);
+        expect(res.status).toBe(200);
+        expect(Buffer.from(await res.arrayBuffer()).equals(body)).toBe(true);
+      });
+    });
+
+    test.concurrent(`gzip ${n}B`, async () => {
+      await serve(reply(gz, "Content-Encoding: gzip\r\n"), async url => {
+        const res = await fetch(url);
+        expect(res.status).toBe(200);
+        expect(Buffer.from(await res.arrayBuffer()).equals(body)).toBe(true);
+      });
+    });
+  }
+});