fetch: cap the GC-triggered drain of an abandoned Response body

src/http/Signals.rs

@@ -79,6 +79,13 @@ impl Signals {
             .map(bun_ptr::BackRef::from)
             .is_some_and(|a| a.load(Ordering::Acquire) == BodyReceiveMode::Paused as u8)
     }
+
+    #[inline]
+    pub fn is_receive_ignored(self) -> bool {
+        self.body_receive_mode
+            .map(bun_ptr::BackRef::from)
+            .is_some_and(|a| a.load(Ordering::Acquire) == BodyReceiveMode::Ignore as u8)
+    }
 }
 
 pub struct Store {

src/http/lib.rs

@@ -293,6 +293,10 @@ pub fn normalize_idle_timeout_seconds(raw: u64) -> c_uint {
 
 pub const END_OF_CHUNKED_HTTP1_1_ENCODING_RESPONSE_BODY: &[u8] = b"0\r\n\r\n";
 
+/// Wall-clock cap on the `BodyReceiveMode::Ignore` discard-drain (not re-armed
+/// per read in `on_data`).
+pub const IGNORE_DRAIN_TIMEOUT_SECONDS: c_uint = 5;
+
 /// HTTP-thread-only scratch buffer for building NUL-terminated hostnames.
 pub static TEMP_HOSTNAME: bun_core::RacyCell<[u8; 8192]> = bun_core::RacyCell::new([0; 8192]);
 
@@ -2098,7 +2102,7 @@ impl<'a> HTTPClient<'a> {
     }
 
     pub fn on_timeout<const IS_SSL: bool>(&mut self, socket: HttpSocket<IS_SSL>) {
-        if self.flags.disable_timeout {
+        if self.flags.disable_timeout && !self.signals.is_receive_ignored() {
             return;
         }
         bun_core::scoped_log!(fetch, "Timeout  {}\n", BStr::new(self.url.href));
@@ -3336,7 +3340,7 @@ impl<'a> HTTPClient<'a> {
             }
             RequestStage::Body => {
                 bun_core::scoped_log!(fetch, "send body");
-                if !self.state.flags.receive_paused {
+                if !self.state.flags.receive_paused && !self.signals.is_receive_ignored() {
                     self.set_timeout(&socket);
                 }
 
@@ -3818,7 +3822,9 @@ impl<'a> HTTPClient<'a> {
 
         if self.proxy_tunnel.is_some() {
             // if we have a tunnel we dont care about the other stages, we will just tunnel the data
-            self.set_timeout(&socket);
+            if !self.signals.is_receive_ignored() {
+                self.set_timeout(&socket);
+            }
             self.proxy_tunnel_mut().unwrap().receive(incoming_data);
             return;
         }
@@ -3837,7 +3843,7 @@ impl<'a> HTTPClient<'a> {
                 self.handle_on_data_headers::<IS_SSL>(incoming_data, ctx, socket);
             }
             ResponseStage::Body => {
-                if !self.state.flags.receive_paused {
+                if !self.state.flags.receive_paused && !self.signals.is_receive_ignored() {
                     self.set_timeout(&socket);
                 }
 
@@ -3855,7 +3861,7 @@ impl<'a> HTTPClient<'a> {
                 }
             }
             ResponseStage::BodyChunk => {
-                if !self.state.flags.receive_paused {
+                if !self.state.flags.receive_paused && !self.signals.is_receive_ignored() {
                     self.set_timeout(&socket);
                 }
 
@@ -4042,7 +4048,13 @@ impl<'a> HTTPClient<'a> {
         // timeout disabled and the body promise pending forever.
         let _ = socket.resume_stream();
         bun_core::scoped_log!(fetch, "resume receive {}", self.async_http_id);
-        self.set_timeout(&socket);
+        if self.signals.is_receive_ignored() {
+            // Not re-armed in `on_data` while ignored: absolute drain deadline.
+            // Proxy tunnels never pause so never reach this arm.
+            socket.set_timeout(IGNORE_DRAIN_TIMEOUT_SECONDS);
+        } else {
+            self.set_timeout(&socket);
+        }
     }
 
     pub fn drain_response_body<const IS_SSL: bool>(&mut self, socket: HttpSocket<IS_SSL>) {

src/runtime/webcore/fetch/FetchTasklet.rs

@@ -1,5 +1,5 @@
 use core::ffi::c_void;
-use core::sync::atomic::{AtomicBool, Ordering};
+use core::sync::atomic::{AtomicBool, AtomicUsize, Ordering};
 
 use bun_boringssl as boringssl;
 use bun_cares_sys::c_ares_draft as c_ares;
@@ -107,6 +107,10 @@ pub struct FetchTasklet {
     pub signals: Signals,
     pub signal_store: http::signals::Store,
     pub has_schedule_callback: AtomicBool,
+    /// Atomic mirror of `BodySize::ContentLength(n)` for the Weak finalizer
+    /// (`usize::MAX` = unknown/chunked). `body_size` itself is written by the
+    /// HTTP thread under `mutex`, which the finalizer cannot take.
+    pub response_content_length: AtomicUsize,
 
     // must be stored because AbortSignal stores reason weakly
     pub abort_reason: StrongOptional,
@@ -1896,6 +1900,7 @@ impl FetchTasklet {
             signals: Signals::default(),
             signal_store: http::signals::Store::default(),
             has_schedule_callback: AtomicBool::new(false),
+            response_content_length: AtomicUsize::new(usize::MAX),
             abort_reason: StrongOptional::empty(),
             check_server_identity: fetch_options.check_server_identity,
             reject_unauthorized: fetch_options.reject_unauthorized,
@@ -2394,6 +2399,9 @@ impl FetchTasklet {
         }
 
         task_ref.body_size = task_ref.result.body_size;
+        if let http::BodySize::ContentLength(n) = task_ref.body_size {
+            task_ref.response_content_length.store(n, Ordering::Release);
+        }
 
         let success = task_ref.result.is_success();
         // `result.body` always aliases `task_ref.response_buffer` (the
@@ -2502,7 +2510,26 @@ impl FetchTasklet {
     }
 }
 
+/// Drain-to-reuse cap for a GC-abandoned body (undici's `Readable.dump` is 128 KiB).
+const ABANDONED_RESPONSE_DRAIN_MAX_BYTES: usize = 256 * 1024;
+
 impl FetchTasklet {
+    /// Scenario 2b/3 of [`on_response_finalize`]. Runs inside a Weak finalizer:
+    /// no `JSCell` may be touched, so the close arm inlines `abort_task` minus
+    /// `tracker.did_cancel`.
+    fn abandon_response_body_from_finalizer(&mut self) {
+        // `self.mutex` cannot be taken here: this finalizer can re-enter on
+        // the JS thread while `on_progress_update` already holds it.
+        let drain = self.response_content_length.load(Ordering::Acquire)
+            <= ABANDONED_RESPONSE_DRAIN_MAX_BYTES;
+        if !drain && !self.signal_store.aborted.swap(true, Ordering::Relaxed) {
+            if let Some(http_) = self.http.as_mut() {
+                http::http_thread().schedule_shutdown(http_);
+            }
+        }
+        self.ignore_remaining_response_body(true);
+    }
+
     #[bun_uws::uws_callback(export = "Bun__FetchResponse_finalize", no_catch)]
     pub(crate) fn on_response_finalize(&mut self) {
         bun_output::scoped_log!(FetchTasklet, "onResponseFinalize");
@@ -2531,11 +2558,11 @@ impl FetchTasklet {
                 if let Some(promise) = locked.promise {
                     if promise.is_empty_or_undefined_or_null() {
                         // Scenario 2b.
-                        this.ignore_remaining_response_body(true);
+                        this.abandon_response_body_from_finalizer();
                     }
                 } else {
                     // Scenario 3.
-                    this.ignore_remaining_response_body(true);
+                    this.abandon_response_body_from_finalizer();
                 }
             }
         }

test/js/web/fetch/fetch-response-abandoned-drain.test.ts

@@ -0,0 +1,111 @@
+// When a fetch() Response is dropped without the body being read, the Weak
+// finalizer on the JS Response wrapper reaches
+// FetchTasklet::on_response_finalize (scenario 3: Locked body, no stream, no
+// promise). Previously that path unconditionally drained the remaining body so
+// the connection could be pooled, with no upper bound: an abandoned 32 MB
+// response was read in full and the socket reused. Now the drain is capped at
+// 256 KiB of remaining Content-Length; anything larger (or chunked/unknown)
+// closes the connection instead.
+
+import { describe, expect, test } from "bun:test";
+import { bunEnv, bunExe, isWindows } from "harness";
+
+// The server writes the body in 16 KiB chunks, pausing on backpressure so
+// `bodyWritten` tracks what the client actually pulled plus at most one kernel
+// send buffer. The client fetches, drops the Response, forces GC so the Weak
+// finalizer runs, and then waits until the server has either sent the whole
+// body or seen the socket close.
+function fixture(contentLength: number) {
+  return /* js */ `
+    const net = require("node:net");
+
+    const contentLength = ${contentLength};
+    const chunk = Buffer.alloc(16 * 1024, 0x61);
+    let bodyWritten = 0;
+    let conn;
+    const { promise: done, resolve: onDone } = Promise.withResolvers();
+
+    const server = net.createServer(socket => {
+      conn = socket;
+      socket.on("error", () => {});
+      socket.on("close", onDone);
+      let replied = false;
+      socket.on("data", () => {
+        if (replied) return;
+        replied = true;
+        socket.write(
+          "HTTP/1.1 200 OK\\r\\n" +
+            "Content-Length: " + contentLength + "\\r\\n" +
+            "Connection: keep-alive\\r\\n" +
+            "\\r\\n",
+        );
+        const writeMore = () => {
+          while (bodyWritten < contentLength && !socket.destroyed) {
+            const n = Math.min(chunk.length, contentLength - bodyWritten);
+            const ok = socket.write(n === chunk.length ? chunk : chunk.subarray(0, n));
+            bodyWritten += n;
+            if (!ok) return socket.once("drain", writeMore);
+          }
+          onDone();
+        };
+        writeMore();
+      });
+    });
+    await new Promise(r => server.listen(0, "127.0.0.1", r));
+    const port = server.address().port;
+
+    let collected = false;
+    const reg = new FinalizationRegistry(() => { collected = true; });
+    async function once() {
+      const res = await fetch("http://127.0.0.1:" + port + "/");
+      if (res.status !== 200) throw new Error("status " + res.status);
+      reg.register(res, 0);
+      // Drop the Response without touching .body / .text() etc.
+    }
+    await once();
+    while (!collected) {
+      Bun.gc(true);
+      await new Promise(r => setImmediate(r));
+    }
+
+    await done;
+    await new Promise(r => setImmediate(r));
+
+    console.log(bodyWritten + " " + (conn.destroyed ? "closed" : "open"));
+    conn.destroy();
+    server.close();
+  `;
+}
+
+async function run(contentLength: number) {
+  await using proc = Bun.spawn({
+    cmd: [bunExe(), "-e", fixture(contentLength)],
+    env: bunEnv,
+    stdout: "pipe",
+    stderr: "pipe",
+  });
+  const [stdout, stderr, exitCode] = await Promise.all([proc.stdout.text(), proc.stderr.text(), proc.exited]);
+  expect({ stderr, exitCode }).toEqual({ stderr: expect.any(String), exitCode: 0 });
+  const [bodyWritten, state] = stdout.trim().split(" ");
+  return { bodyWritten: Number(bodyWritten), state };
+}
+
+// Platform-independent code path; collectContinuously-style GC tests are slow
+// on Windows CI so keep this off there (matches fetch-response-finalizer-sweep).
+describe.skipIf(isWindows)("fetch: abandoned Response body drain policy", () => {
+  test("small body (64 KiB) is drained so the connection stays poolable", async () => {
+    const { bodyWritten, state } = await run(64 * 1024);
+    expect({ bodyWritten, state }).toEqual({ bodyWritten: 64 * 1024, state: "open" });
+  }, 60_000);
+
+  test("large body (32 MiB) is closed instead of drained in full", async () => {
+    const total = 32 * 1024 * 1024;
+    const { bodyWritten, state } = await run(total);
+    // The server front-loads one kernel send buffer before the client ever
+    // pauses, so `bodyWritten` includes that window on top of the 256 KiB cap.
+    // Without the cap the full 32 MiB is drained; a quarter of that is well
+    // above any loopback send buffer and well below the unbounded drain.
+    expect(bodyWritten).toBeLessThan(total / 4);
+    expect(state).toBe("closed");
+  }, 60_000);
+});