fetch: deliver response body bytes as a borrowed slice; reserve Content-Length for buffered consumers

src/runtime/webcore/fetch/FetchTasklet.rs

@@ -50,6 +50,9 @@
 
 bun_output::declare_scope!(FetchTasklet, visible);
 
+/// Upper bound on the Content-Length-driven `reserve_exact` in `callback()`.
+const SCHEDULED_PRERESERVE_MAX: usize = 256 * 1024 * 1024;
+
 use http::signals::BodyReceiveMode;
 
 #[derive(bun_ptr::ThreadSafeRefCounted)]
@@ -115,6 +118,12 @@
     // Custom Hostname
     pub(crate) hostname: Option<Box<[u8]>>,
     pub(crate) is_waiting_body: bool,
+    /// Set by `on_start_buffering_callback` (JS thread) and read by
+    /// `callback()` (HTTP thread, under `mutex`): the body is being
+    /// accumulated in `scheduled_response_buffer` for a buffered consumer.
+    /// Distinguishes that path from `drop_backpressure_if_unobserved`, which
+    /// also sets `BufferAll` but still delivers per chunk.
+    pub(crate) is_buffering_body: AtomicBool,
     pub(crate) is_waiting_abort: bool,
     pub(crate) is_waiting_request_stream_start: bool,
     pub(crate) mutex: Mutex,
@@ -1703,6 +1712,9 @@
     ) {
         let this = Self::from_ctx(ctx);
         this.readable_stream_ref = ReadableStreamStrong::init(readable, global_this);
+        // A ByteStream now drains scheduled_response_buffer per chunk; undo any
+        // buffered-consumer reservation request so callback() stops growing it.
+        this.is_buffering_body.store(false, Ordering::Release);
     }
 
     fn on_start_streaming_http_response_body_callback(ctx: *mut c_void) -> DrainResult {
@@ -1805,6 +1817,7 @@
     fn on_start_buffering_callback(ctx: *mut c_void) {
         let this = Self::from_ctx(ctx);
         this.poll_ref.ref_(bun_io::js_vm_ctx());
+        this.is_buffering_body.store(true, Ordering::Release);
         if this
             .signal_store
             .set_receive_mode_terminal(BodyReceiveMode::BufferAll)
@@ -2032,6 +2045,7 @@
             upgraded_connection: fetch_options.upgraded_connection,
             hostname: fetch_options.hostname,
             is_waiting_body: false,
+            is_buffering_body: AtomicBool::new(false),
             is_waiting_abort: false,
             is_waiting_request_stream_start: false,
             mutex: Mutex::new(),
@@ -2595,20 +2609,46 @@
             }
         } else {
             if success {
-                bun_core::handle_oom(
-                    task_ref
-                        .scheduled_response_buffer
-                        .write(task_ref.response_buffer.list.as_slice()),
-                );
+                let scheduled = &mut task_ref.scheduled_response_buffer;
+                let incoming = &mut task_ref.response_buffer;
+                if scheduled.list.capacity() == 0 {
+                    // `body_out_str` aliases the field, not the Vec's heap
+                    // pointer (asserted above), so swapping here is invisible
+                    // to the HTTP client.
+                    core::mem::swap(scheduled, incoming);
+                }
+                // Grow to Content-Length once so the per-packet append below
+                // doesn't leave the ~2x doubling over-capacity that the
+                // ArrayBuffer would adopt. Gated on `is_buffering_body`
+                // (set by `on_start_buffering_callback`), not the raw
+                // `BufferAll` mode: `drop_backpressure_if_unobserved` also
+                // sets `BufferAll` while still draining per chunk.
+                if task_ref.is_buffering_body.load(Ordering::Acquire) {
+                    if let http::BodySize::ContentLength(n) = task_ref.body_size {
+                        if n > scheduled.list.capacity() {
+                            let additional = n
+                                .min(SCHEDULED_PRERESERVE_MAX)
+                                .saturating_sub(scheduled.list.len());
+                            let _ = scheduled.list.try_reserve_exact(additional);
+                        }
+                    }
+                }
+                if !incoming.list.is_empty() {
+                    bun_core::handle_oom(scheduled.write(incoming.list.as_slice()));
+                }
                 if task_ref.result.has_more && !task_ref.scheduled_response_buffer.list.is_empty() {
                     let _ = task_ref.signal_store.try_transition_receive_mode(
                         BodyReceiveMode::AutoPause,
                         BodyReceiveMode::Paused,
                     );
                 }
             }
-            // reset for reuse
-            task_ref.response_buffer.reset();
+            if task_ref.result.has_more {
+                // reset for reuse
+                task_ref.response_buffer.reset();
+            } else {
+                task_ref.response_buffer = MutableString::default();
+            }
         }
 
         if let Err(has_schedule_callback) = task_ref.has_schedule_callback.compare_exchange(

test/js/web/fetch/fetch-leak.test.ts

@@ -2,6 +2,7 @@ import { describe, expect, test } from "bun:test";
 import { bunEnv, bunExe, tls as COMMON_CERT, gc, isASAN, isCI, isDebug } from "harness";
 import { once } from "node:events";
 import { createServer } from "node:http";
+import net from "node:net";
 import { join } from "node:path";
 
 describe("fetch doesn't leak", () => {
@@ -962,3 +963,66 @@ test("aborting in-flight streaming fetch() responses does not retain the buffere
   expect(stderr).not.toContain("LEAK");
   expect(exitCode).toBe(0);
 });
+
+test("fetch().arrayBuffer() of a large Content-Length body peaks at ~1x the payload", async () => {
+  // The per-packet handoff in FetchTasklet::callback appended each socket read
+  // into scheduled_response_buffer via extend_from_slice, so the Vec grew by
+  // amortized doubling. For a body just past a doubling step that left the
+  // allocation the ArrayBuffer adopts at ~2x the payload, and the intermediate
+  // reallocations spiked ru_maxrss well past 2x. Reserving Content-Length
+  // exactly up front keeps it to a single allocation that is moved through to
+  // the ArrayBuffer.
+  //
+  // Body size is 128 MiB + 1 MiB so the old doubling growth would have crossed
+  // the 128 -> 256 step, making the unfixed peak reliably > 2x body.
+  const bodyBytes = 129 * 1024 * 1024;
+  const chunk = Buffer.alloc(256 * 1024, "abcdefghij");
+  const server = net.createServer(socket => {
+    socket.once("data", () => {
+      socket.write(`HTTP/1.1 200 OK\r\nContent-Length: ${bodyBytes}\r\nConnection: close\r\n\r\n`);
+      let sent = 0;
+      const pump = () => {
+        while (sent < bodyBytes) {
+          const n = Math.min(chunk.length, bodyBytes - sent);
+          sent += n;
+          if (!socket.write(n === chunk.length ? chunk : chunk.subarray(0, n))) {
+            socket.once("drain", pump);
+            return;
+          }
+        }
+        socket.end();
+      };
+      pump();
+    });
+    socket.on("error", () => {});
+  });
+  await once(server.listen(0, "127.0.0.1"), "listening");
+  const { port } = server.address();
+
+  try {
+    await using proc = Bun.spawn({
+      cmd: [bunExe(), "--smol", join(import.meta.dir, "fetch-buffer-peak-fixture.ts")],
+      env: {
+        ...bunEnv,
+        SERVER: `http://127.0.0.1:${port}/`,
+        BODY_BYTES: String(bodyBytes),
+      },
+      stdout: "pipe",
+      stderr: "pipe",
+    });
+    const [stdout, stderr, exitCode] = await Promise.all([proc.stdout.text(), proc.stderr.text(), proc.exited]);
+    console.log(stdout.trim());
+    expect(stderr).toBe("");
+    const { bodyMB, rssBeforeMB, rssAfterMB } = JSON.parse(stdout.trim());
+    expect(bodyMB).toBe(129);
+    // Unfixed: the doubling reallocations (retained by ASAN quarantine / mimalloc
+    //          page cache) leave RSS at >= 2x body over the pre-fetch resident
+    //          set (release linux ~2.9x, debug+ASAN default quarantine ~2.7x).
+    // Fixed:   a single exact allocation, ~1.0x body.
+    const delta = rssAfterMB - rssBeforeMB;
+    expect(delta).toBeLessThan(1.5 * bodyMB);
+    expect(exitCode).toBe(0);
+  } finally {
+    server.close();
+  }
+}, 60000);