fetch: release the per-packet body scratch when the stream stays parked

src/http/AsyncHTTP.rs

@@ -81,6 +81,7 @@ const fn noop_callback() -> HTTPClientResultCallback {
         ctx: core::ptr::null_mut(),
         function: noop_result_callback,
         release_at_shutdown: None,
+        release_idle_scratch: None,
     }
 }
 

src/http/HTTPThread.rs

@@ -134,6 +134,8 @@ pub struct HttpThread {
     /// — the request socket never reaches a terminal state once the JS thread
     /// stops driving the world, so the box would otherwise strand.
     pub(crate) in_flight: Vec<NonNull<crate::ThreadlocalAsyncHttp<'static>>>,
+    /// `timer_read()` at the last `sweep_idle_scratch`. HTTP-thread-only.
+    last_scratch_sweep_ns: u64,
 }
 
 impl HttpThread {
@@ -187,6 +189,7 @@ impl HttpThread {
             timer: Instant::now(),
             lazy_libdeflater: None,
             in_flight: Vec::new(),
+            last_scratch_sweep_ns: 0,
         }
     }
 }
@@ -812,7 +815,27 @@ impl HttpThread {
         self.deferred_tasks.insert(first_that_yields, http);
     }
 
+    /// Once per timeout sweep, a paused body consumer's request drops its per-packet scratch.
+    fn sweep_idle_scratch(&mut self) {
+        /// `LIBUS_TIMEOUT_GRANULARITY` (packages/bun-usockets/src/libusockets.h), in ns.
+        const SWEEP_NS: u64 = 4 * 1_000_000_000;
+        let now = self.timer_read();
+        if now.wrapping_sub(self.last_scratch_sweep_ns) < SWEEP_NS {
+            return;
+        }
+        self.last_scratch_sweep_ns = now;
+        for nn in &self.in_flight {
+            // SAFETY: every entry is a live in-flight request that only this thread touches; see
+            // `in_flight`.
+            let client = unsafe { &mut (*nn.as_ptr()).async_http.client };
+            if client.signals.is_receive_paused() {
+                client.release_idle_scratch();
+            }
+        }
+    }
+
     pub(crate) fn drain_events(&mut self) {
+        self.sweep_idle_scratch();
         // Process any pending writes **before** aborting.
         self.drain_queued_receive_resumes();
         self.drain_queued_writes();

src/http/lib.rs

@@ -385,6 +385,15 @@ const PREALLOCATE_MAX: usize = 1024 * 1024 * 256;
 /// bounded for long-lived streaming responses.
 pub const DECODED_BODY_RETAIN_CAP: usize = 512 * 1024;
 
+/// Clears a delivered per-chunk scratch, keeping up to [`DECODED_BODY_RETAIN_CAP`] for the next packet.
+pub fn recycle_scratch(list: &mut Vec<u8>) {
+    if list.capacity() > DECODED_BODY_RETAIN_CAP {
+        *list = Vec::new();
+    } else {
+        list.clear();
+    }
+}
+
 /// Whether the experimental Alt-Svc-driven HTTP/3 upgrade is enabled at all
 /// (CLI flag or env var). Used on its own to gate `H3.AltSvc.record` — a
 /// response that arrived over a request shape h3 can't serve (proxy, sendfile,
@@ -625,6 +634,8 @@ pub struct HTTPClientResultCallback {
     /// for the ack. `None` ⇒ no-op (the default for callers whose `ctx`
     /// is process-lifetime or whose code path never reaches `global_exit`).
     pub(crate) release_at_shutdown: Option<unsafe fn(*mut ())>,
+    /// Optional, same contract as `release_at_shutdown`; see `HTTPClient::release_idle_scratch`.
+    pub(crate) release_idle_scratch: Option<unsafe fn(*mut ())>,
 }
 
 impl HTTPClientResultCallback {
@@ -648,6 +659,7 @@ impl HTTPClientResultCallback {
                 >(callback)
             },
             release_at_shutdown: None,
+            release_idle_scratch: None,
         }
     }
 
@@ -660,6 +672,11 @@ impl HTTPClientResultCallback {
         cb.release_at_shutdown = Some(release);
         cb
     }
+
+    pub fn with_idle_scratch_release(mut self, release: unsafe fn(*mut ())) -> Self {
+        self.release_idle_scratch = Some(release);
+        self
+    }
 }
 
 // Exists for heap stats reasons.
@@ -4448,10 +4465,8 @@ impl<'a> HTTPClient<'a> {
         if has_more {
             result.body = decoded_body.list.as_slice();
             callback.run(parent, result);
-            if decoded_body.list.capacity() <= DECODED_BODY_RETAIN_CAP {
-                decoded_body.list.clear();
-                self.state.decoded_body = decoded_body;
-            }
+            recycle_scratch(&mut decoded_body.list);
+            self.state.decoded_body = decoded_body;
             self.maybe_pause_receive(socket);
             // Only a paused consumer asks for the rest.
             self.state.has_pending_compressed() && !self.signals.is_receive_paused()
@@ -4491,9 +4506,40 @@ impl<'a> HTTPClient<'a> {
         }
         result.body = decoded_body.list.as_slice();
         callback.run(parent, result);
-        if decoded_body.list.capacity() <= DECODED_BODY_RETAIN_CAP {
-            decoded_body.list.clear();
-            self.state.decoded_body = decoded_body;
+        recycle_scratch(&mut decoded_body.list);
+        self.state.decoded_body = decoded_body;
+    }
+
+    /// The body consumer is paused (`HTTPThread::sweep_idle_scratch`): drops the empty per-packet scratch.
+    pub(crate) fn release_idle_scratch(&mut self) {
+        bun_core::scoped_log!(fetch, "release idle scratch {}", self.async_http_id);
+        if self.state.decoded_body.list.is_empty() {
+            self.state.decoded_body = MutableString::default();
+        }
+        if self.state.compressed_body.list.is_empty() {
+            self.state.compressed_body = MutableString::default();
+        }
+        if let Some(mut stream) = self.h2 {
+            // SAFETY: `h2` points at this client's stream until the session detaches it.
+            let body_buffer = unsafe { &mut stream.as_mut().body_buffer };
+            if body_buffer.is_empty() {
+                *body_buffer = Vec::new();
+            }
+        }
+        if let Some(mut stream) = self.h3 {
+            // SAFETY: `h3` points at this client's stream until the session detaches it.
+            let body_buffer = unsafe { &mut stream.as_mut().body_buffer };
+            if body_buffer.is_empty() {
+                *body_buffer = Vec::new();
+            }
+        }
+        // The consumer's callback is the embedding `AsyncHTTP`'s, not the client's wrapper.
+        let parent = self.parent_async_http();
+        // SAFETY: this client is in `HTTPThread::in_flight`, so its `AsyncHTTP` and consumer are live.
+        let callback = unsafe { (*parent).result_callback };
+        if let Some(release) = callback.release_idle_scratch {
+            // SAFETY: see above.
+            unsafe { release(callback.ctx) };
         }
     }
 

src/runtime/webcore/fetch/FetchTasklet.rs

@@ -774,12 +774,7 @@ impl FetchTasklet {
             if buffer_reset.get() {
                 // SAFETY: `self` outlives this defer (it's a local in this fn) and no other
                 // borrow of scheduled_response_buffer is live at scope exit / `?` unwind.
-                let list = unsafe { &mut (*scheduled_buf).list };
-                if list.capacity() > http::DECODED_BODY_RETAIN_CAP {
-                    *list = Vec::new();
-                } else {
-                    list.clear();
-                }
+                http::recycle_scratch(unsafe { &mut (*scheduled_buf).list });
             }
         }
 
@@ -1803,6 +1798,18 @@ impl FetchTasklet {
         self.unpark_body_stream();
     }
 
+    /// HTTP thread, the stream is paused: drops the hop buffer's capacity once the JS side took its bytes.
+    ///
+    /// SAFETY: `this` is the live `result_callback.ctx`; the HTTP thread's ref keeps it alive.
+    unsafe fn release_idle_scratch(this: *mut ()) {
+        let this = Self::from_raw_mut(this.cast::<FetchTasklet>());
+        this.mutex.lock();
+        if this.scheduled_response_buffer.list.is_empty() {
+            this.scheduled_response_buffer = MutableString::default();
+        }
+        this.mutex.unlock();
+    }
+
     /// The other half of this rule is in `callback` (HTTP thread).
     fn after_body_chunk_delivered(&self, bytes: &crate::webcore::ByteStream) {
         use crate::webcore::byte_stream::AfterDelivery;
@@ -2149,7 +2156,8 @@ impl FetchTasklet {
                 // contract `callback` documents.
                 FetchTasklet::callback,
                 FetchTasklet::release_at_shutdown,
-            ),
+            )
+            .with_idle_scratch_release(FetchTasklet::release_idle_scratch),
             fetch_options.redirect_type,
             http::async_http::Options {
                 http_proxy: proxy,

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

@@ -2,7 +2,7 @@
 // HTTP thread from buffering the entire response in memory.
 import { S3Client } from "bun";
 import { describe, expect, test } from "bun:test";
-import { bunEnv, bunExe, forEachLine, isASAN, isDebug, isWindows, tempDir, tls } from "harness";
+import { bunEnv, bunExe, forEachLine, isASAN, isDebug, isWindows, tempDir, tls, withoutAsanQuarantine } from "harness";
 import { once } from "node:events";
 import { statSync } from "node:fs";
 import { stat } from "node:fs/promises";
@@ -513,6 +513,110 @@ import { createServer as createTlsServer } from "node:tls";
   );
 });
 
+// A body that nothing reads costs what it holds: the untaken bytes (the mark plus one recv). The
+// per-packet scratch on both sides of the HTTP-thread hop (`decoded_body`, the tasklet's staging
+// buffer) is reused while the stream flows, and has to go once the stream parks; kept, it pinned one
+// empty recv each (about 1 MiB per held response on top of the bytes, #47009). The HTTP thread
+// drops it at its next sweep (every 4 s), so the client waits for the drop.
+describe.concurrent("fetch() receive backpressure — memory of a held body stream", () => {
+  // A loopback recv is up to 512 KiB and the scratch is sized by it, so the cost is flat from about
+  // 1 MiB of body up; 4 MiB keeps the drain at the end short in a debug build.
+  const N = 24;
+  const BODY = 4 * 1024 * 1024;
+  // MiB per held response after the stream parks and a full collection, on Linux loopback, with
+  // these four running at once. Measured with the scratch kept, then released, [0 reads, 1 read]
+  // (debug is the debug+ASAN build):
+  //   h1       release 2.0-2.2, 1.9      -> 1.1-1.2, 0.95      debug 2.7, 2.5   -> 1.5, 1.3-1.35
+  //   h1-gzip  release 2.1-2.2, 2.4-2.5  -> 1.4-1.45, 1.25-1.4 debug 2.6, 3.0   -> 1.8, 2.0-2.1
+  // What stays is the untaken bytes (a stream nothing read holds what was staged before the reader
+  // attached plus one recv; a gzip one also the decoder's input and its budgeted output), the stream
+  // and the reader. The bounds sit between the two.
+  const LIMIT_MIB = {
+    "h1": isASAN || isDebug ? [2.05, 1.9] : [1.6, 1.4],
+    "h1-gzip": isASAN || isDebug ? [2.25, 2.5] : [1.75, 1.9],
+  };
+
+  for (const kind of ["h1", "h1-gzip"] as const) {
+    for (const reads of [0, 1]) {
+      const limit = LIMIT_MIB[kind][reads];
+      test.skipIf(process.platform !== "linux")(
+        `${N} held ${kind} streams, ${reads} read() each: under ${limit} MiB per response, and the rest still arrives`,
+        async () => {
+          // The server lives here, not in the measured process: its write queue is not the client's.
+          await using server = await serve(kind, BODY / CHUNK);
+          const script = /* js */ `
+            const N = ${N}, READS = ${reads}, TOTAL = ${BODY}, LIMIT = ${limit};
+            const url = ${JSON.stringify(server.url)};
+            const rss = () => process.memoryUsage.rss() / 1048576;
+            // RSS after a full collection, once it held still for three samples in a row (or the
+            // last sample at the deadline, so a drifting RSS still reports a number).
+            async function settled(deadline = Date.now() + 10_000) {
+              Bun.gc(true);
+              for (let last = rss(), stable = 0; ; ) {
+                await Bun.sleep(20);
+                Bun.gc(true);
+                const now = rss();
+                stable = Math.abs(now - last) < 0.5 ? stable + 1 : 0;
+                last = now;
+                if (stable >= 3 || Date.now() > deadline) return now;
+              }
+            }
+            // Every body is "A" repeated; a chunk that is not is foreign (a memcmp per chunk).
+            const A = Buffer.alloc(1 << 20, 65);
+            let foreign = 0;
+            const take = chunk => {
+              const buf = Buffer.from(chunk.buffer, chunk.byteOffset, chunk.byteLength);
+              for (let off = 0; off < buf.length; off += A.length) {
+                const end = Math.min(off + A.length, buf.length);
+                if (buf.compare(A, 0, end - off, off, end) !== 0) foreign++;
+              }
+              return chunk.byteLength;
+            };
+            const base = await settled();
+            const held = await Promise.all(Array.from({ length: N }, async () => {
+              const reader = (await fetch(url)).body.getReader();
+              let got = 0;
+              for (let i = 0; i < READS; i++) got += take((await reader.read()).value);
+              return { reader, got };
+            }));
+            // The scratch goes at the HTTP thread's next sweep: wait for it.
+            let perResponse;
+            for (const deadline = Date.now() + 10_000; ; ) {
+              perResponse = (await settled(deadline) - base) / N;
+              if (perResponse < LIMIT || Date.now() > deadline) break;
+              await Bun.sleep(100);
+            }
+            // Un-park: every held body still arrives whole and intact.
+            const totals = await Promise.all(held.map(async ({ reader, got }) => {
+              for (;;) { const { done, value } = await reader.read(); if (done) return got; got += take(value); }
+            }));
+            process.stdout.write(JSON.stringify({ perResponse, foreign, short: totals.filter(t => t !== TOTAL).length }));
+          `;
+          await using proc = Bun.spawn({
+            cmd: [bunExe(), "-e", script],
+            env: withoutAsanQuarantine(bunEnv),
+            stdout: "pipe",
+            stderr: "pipe",
+          });
+          const [stdout, stderr, exitCode] = await Promise.all([proc.stdout.text(), proc.stderr.text(), proc.exited]);
+          if (!stdout) throw new Error(`exited ${exitCode}: ${stderr}`);
+          const { perResponse, foreign, short } = JSON.parse(stdout);
+          expect({ short, foreign, stderr, perResponse }).toEqual({
+            short: 0,
+            foreign: 0,
+            stderr: "",
+            perResponse: expect.any(Number),
+          });
+          expect(perResponse).toBeLessThan(limit);
+          expect(exitCode).toBe(0);
+        },
+        // 24 streams settle, wait for a sweep and drain under ASAN.
+        30_000,
+      );
+    }
+  }
+});
+
 // h2 advertises a 16 MiB initial per-stream window (LOCAL_INITIAL_WINDOW_SIZE),
 // so withholding WINDOW_UPDATE only takes effect past that. Asserting a tight
 // RSS bound for h2 needs that window lowered, which is a separate change.