diff --git a/src/http/AsyncHTTP.rs b/src/http/AsyncHTTP.rs index c540234348de..95c4221aef64 100644 --- a/src/http/AsyncHTTP.rs +++ b/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::(), ); - 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); + } } } diff --git a/src/http/lib.rs b/src/http/lib.rs index 30b56609467d..6feb19ce082e 100644 --- a/src/http/lib.rs +++ b/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; } diff --git a/test/js/web/fetch/fetch-backpressure.test.ts b/test/js/web/fetch/fetch-backpressure.test.ts index cabc70188c5f..9fdacc0a0113 100644 --- a/test/js/web/fetch/fetch-backpressure.test.ts +++ b/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);