fetch: release the in-flight request slot while a response body is paused for backpressure

src/http/AsyncHTTP.rs

@@ -780,6 +780,7 @@ impl<'a> AsyncHTTP<'a> {
                 }
                 let elapsed = (*this).elapsed;
                 bun_core::scoped_log!(AsyncHTTP, "onAsyncHTTPCallback: {:?}", elapsed);
+                let slot_released_on_pause = (*this).client.flags.released_active_slot;
                 callback.run(async_http, result);
 
                 // SAFETY: `async_http` is the `async_http` field of a
@@ -807,8 +808,10 @@ impl<'a> AsyncHTTP<'a> {
                     std::alloc::Layout::new::<ThreadlocalAsyncHTTP>(),
                 );
 
-                let active_requests = ACTIVE_REQUESTS_COUNT.fetch_sub(1, Ordering::Relaxed);
-                debug_assert!(active_requests > 0);
+                if !slot_released_on_pause {
+                    let active_requests = ACTIVE_REQUESTS_COUNT.fetch_sub(1, Ordering::Relaxed);
+                    debug_assert!(active_requests > 0);
+                }
             }
         }
 

src/http/lib.rs

@@ -198,6 +198,12 @@ pub struct Flags {
     /// once on a fresh connection but never loops.
     pub h3_retried: bool,
     pub is_node_http_client: bool,
+    /// Set by `maybe_pause_receive` when it decrements `ACTIVE_REQUESTS_COUNT`
+    /// for a backpressure pause (JS has not pulled the buffered body yet);
+    /// cleared by `resume_receive` when it re-increments. Checked in
+    /// `on_async_http_callback_raw`'s terminal branch so a request that ends
+    /// while still paused (abort/close/GC) skips the second decrement.
+    pub released_active_slot: bool,
 }
 
 impl Default for Flags {
@@ -221,6 +227,7 @@ impl Default for Flags {
             force_http3: false,
             h3_retried: false,
             is_node_http_client: false,
+            released_active_slot: false,
         }
     }
 }
@@ -4116,6 +4123,20 @@ impl<'a> HTTPClient<'a> {
         self.state.flags.receive_paused = true;
         socket.set_timeout(0);
         let _ = socket.pause_stream();
+        // A socket parked for JS-side backpressure is no longer doing I/O, so it
+        // must not count against `MAX_SIMULTANEOUS_REQUESTS`: otherwise `max`
+        // retained-but-unread Responses permanently starve every later fetch
+        // (any origin) until GC finalizes them. `resume_receive` re-acquires
+        // the slot; the terminal callback checks `released_active_slot` so a
+        // request that ends while still paused is not double-decremented. Runs
+        // on the HTTP thread inside `tick()`, so the next loop iteration's
+        // `drain_events` observes the freed slot without an explicit wakeup.
+        if !self.flags.released_active_slot {
+            self.flags.released_active_slot = true;
+            let prev = crate::async_http::ACTIVE_REQUESTS_COUNT
+                .fetch_sub(1, core::sync::atomic::Ordering::Relaxed);
+            debug_assert!(prev > 0);
+        }
         bun_core::scoped_log!(fetch, "pause receive {}", self.async_http_id);
     }
 
@@ -4124,6 +4145,14 @@ impl<'a> HTTPClient<'a> {
             return;
         }
         self.state.flags.receive_paused = false;
+        // Re-acquire the slot released in `maybe_pause_receive`. Admission is
+        // one-way (the request is already in flight), so this may briefly push
+        // the count above `max`; `drain_events` only gates NEW requests on it.
+        if self.flags.released_active_slot {
+            self.flags.released_active_slot = false;
+            crate::async_http::ACTIVE_REQUESTS_COUNT
+                .fetch_add(1, core::sync::atomic::Ordering::Relaxed);
+        }
         if socket.is_closed() {
             return;
         }

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

@@ -379,3 +379,77 @@ describe.concurrent("fetch() receive backpressure — streaming consumer shapes"
     }
   });
 });
+
+// Runs in a child so BUN_CONFIG_MAX_HTTP_REQUESTS can be capped without
+// affecting the rest of the suite. Before the fix the retain loop wedged at
+// the cap and both probes reported HUNG, even the one to a fresh origin.
+test("retained unread Responses do not starve later fetch() requests", async () => {
+  const fixture = /* js */ `
+    import { createServer } from "node:http";
+    import { once } from "node:events";
+
+    const body = Buffer.alloc(512 * 1024, 97);
+    async function mk() {
+      const s = createServer((_, w) => { w.setHeader("content-length", body.length); w.end(body); });
+      s.listen(0, "127.0.0.1");
+      await once(s, "listening");
+      return s;
+    }
+    const s1 = await mk(), s2 = await mk();
+    const O1 = "http://127.0.0.1:" + s1.address().port;
+    const O2 = "http://127.0.0.1:" + s2.address().port;
+
+    const race = (p, ms) => Promise.race([
+      p.then(v => ({ ok: true, v })),
+      new Promise(r => setTimeout(() => r({ ok: false }), ms)),
+    ]);
+
+    const hold = [];
+    let wedgedAt = 0;
+    for (let i = 1; i <= 12; i++) {
+      const r = await race(fetch(O1 + "/x" + i), 3000);
+      if (!r.ok) { wedgedAt = i; break; }
+      hold.push(r.v);
+    }
+
+    const same   = await race(fetch(O1 + "/same").then(r => r.arrayBuffer()), 3000);
+    const second = await race(fetch(O2 + "/second").then(r => r.arrayBuffer()), 3000);
+
+    // Draining a retained Response must re-acquire the slot and let the
+    // body finish so the counter stays balanced.
+    const drained = hold[0] ? (await hold[0].arrayBuffer()).byteLength : 0;
+
+    process.stdout.write(JSON.stringify({
+      wedgedAt,
+      retained: hold.length,
+      same: same.ok,
+      second: second.ok,
+      drained,
+    }));
+
+    hold.length = 0;
+    s1.closeAllConnections(); s1.close();
+    s2.closeAllConnections(); s2.close();
+  `;
+
+  await using proc = Bun.spawn({
+    cmd: [bunExe(), "-e", fixture],
+    env: { ...bunEnv, BUN_CONFIG_MAX_HTTP_REQUESTS: "4" },
+    stdout: "pipe",
+    stderr: "pipe",
+  });
+  const [stdout, stderr, exitCode] = await Promise.all([proc.stdout.text(), proc.stderr.text(), proc.exited]);
+  const filteredStderr = stderr
+    .split("\n")
+    .filter(l => l && !l.startsWith("WARNING: ASAN interferes"))
+    .join("\n");
+  expect(filteredStderr).toBe("");
+  expect(JSON.parse(stdout)).toEqual({
+    wedgedAt: 0,
+    retained: 12,
+    same: true,
+    second: true,
+    drained: 512 * 1024,
+  });
+  expect(exitCode).toBe(0);
+}, 30_000);