From 9d70b743b10a16ab4cbc5b5980c34191b307934f Mon Sep 17 00:00:00 2001 From: robobun <117481402+robobun@users.noreply.github.com> Date: Tue, 4 Aug 2026 04:28:17 +0000 Subject: [PATCH 1/3] Bun.serve: cancel the piped upstream fetch when the client aborts while backpressured Follow-up to #36087, carrying over the abort-path fix from the now-closed #35547 that the SinkHandle rewrite did not pick up. on_abort handles this.sink (the JS-stream sink path) but not this.byte_stream (the native SinkHandle::ServerResponse path). When a client aborts while the ByteStream sink is paused for backpressure, the upstream FetchTasklet stays in Paused with no wake path and no cancellation, and the ref taken when the sink was installed is never released (end_chunk only runs via sink.end(), which requires a drain that never comes on a closed socket). on_abort now calls cancel_from_sink() on the held ByteStream (detaches the sink and closes the producer, aborting the upstream fetch), deinits the Strong, and drops the sink-install ref. end_chunk clears this.byte_stream so the field tracks whether that ref is still held, keeping the two release sites mutually exclusive. --- src/runtime/server/RequestContext.rs | 8 ++ ...e-fetch-body-abort-backpressure-fixture.ts | 11 ++ ...erve-fetch-body-abort-backpressure.test.ts | 102 ++++++++++++++++++ 3 files changed, 121 insertions(+) create mode 100644 test/js/bun/http/serve-fetch-body-abort-backpressure-fixture.ts create mode 100644 test/js/bun/http/serve-fetch-body-abort-backpressure.test.ts diff --git a/src/runtime/server/RequestContext.rs b/src/runtime/server/RequestContext.rs index ea47e3d5a730..bd808a886ab5 100644 --- a/src/runtime/server/RequestContext.rs +++ b/src/runtime/server/RequestContext.rs @@ -1496,6 +1496,13 @@ where return; } + if let Some(byte_stream) = this.byte_stream.take() { + bun_ptr::BackRef::from(byte_stream).cancel_from_sink(None); + any_js_calls.set(true); + this.response_body_readable_stream_ref.with_mut(|s| s.deinit()); + this.deref(); + } + // if we can, free the request now. if this.is_dead_request() { this.finalize_without_deinit(); @@ -3385,6 +3392,7 @@ where // SAFETY: caller passes the live `*mut RequestContext` stored as the // sink ctx; `_ref` keeps it alive for this call. let this = unsafe { &*this }; + this.byte_stream.set(None); if this.is_aborted_or_ended() { return; diff --git a/test/js/bun/http/serve-fetch-body-abort-backpressure-fixture.ts b/test/js/bun/http/serve-fetch-body-abort-backpressure-fixture.ts new file mode 100644 index 000000000000..950affea4b67 --- /dev/null +++ b/test/js/bun/http/serve-fetch-body-abort-backpressure-fixture.ts @@ -0,0 +1,11 @@ +// Proxy under test: returns a fetch() Response whose body is the native +// ByteStream from the upstream. +const upstream = process.argv[2]; + +const proxy = Bun.serve({ + port: 0, + idleTimeout: 255, + fetch: () => fetch(upstream), +}); + +console.log(JSON.stringify({ proxyPort: proxy.port })); diff --git a/test/js/bun/http/serve-fetch-body-abort-backpressure.test.ts b/test/js/bun/http/serve-fetch-body-abort-backpressure.test.ts new file mode 100644 index 000000000000..c11c6ddd1a49 --- /dev/null +++ b/test/js/bun/http/serve-fetch-body-abort-backpressure.test.ts @@ -0,0 +1,102 @@ +import { expect, test } from "bun:test"; +import { bunEnv, bunExe } from "harness"; +import net from "node:net"; +import { join } from "node:path"; + +// A client that aborts while Bun.serve is holding an upstream fetch() body +// paused for backpressure must not leave the upstream parked: on_abort has +// to cancel the ByteStream sink so the proxy tears down its connection to +// the upstream. +test("client abort while a fetch() body is backpressured cancels the upstream", async () => { + const CHUNK = 256 * 1024; + const CAP_CHUNKS = 256; + + let pulls = 0; + let cancelled = false; + + await using upstream = Bun.serve({ + port: 0, + idleTimeout: 255, + fetch() { + return new Response( + new ReadableStream({ + async pull(controller) { + controller.enqueue(new Uint8Array(CHUNK)); + pulls++; + if (pulls >= CAP_CHUNKS) return controller.close(); + if (pulls % 32 === 0) await Bun.sleep(0); + }, + cancel() { + cancelled = true; + }, + }), + { headers: { "content-length": String(CHUNK * CAP_CHUNKS) } }, + ); + }, + }); + + await using proxy = Bun.spawn({ + cmd: [ + bunExe(), + join(import.meta.dir, "serve-fetch-body-abort-backpressure-fixture.ts"), + `http://127.0.0.1:${upstream.port}/`, + ], + env: bunEnv, + stdout: "pipe", + stderr: "inherit", + }); + + const reader = proxy.stdout.getReader(); + let head = ""; + while (!head.includes("\n")) { + const { value, done } = await reader.read(); + if (done) throw new Error("proxy exited before reporting port"); + head += Buffer.from(value).toString("utf8"); + } + reader.releaseLock(); + const { proxyPort } = JSON.parse(head.slice(0, head.indexOf("\n"))); + + const failed = Promise.withResolvers(); + proxy.exited.then(code => failed.reject(new Error(`proxy exited early (code ${code})`))); + + const socket = net.connect(proxyPort, "127.0.0.1"); + try { + const stalled = Promise.withResolvers(); + socket.on("error", e => failed.reject(e)); + socket.on("connect", () => socket.write("GET / HTTP/1.1\r\nHost: x\r\nConnection: close\r\n\r\n")); + socket.once("data", () => { + socket.pause(); + stalled.resolve(); + }); + await Promise.race([stalled.promise, failed.promise]); + + // Wait until the upstream pull count stops growing (backpressure engaged) + // or the upstream produces the whole capped body (no backpressure at all). + let lastPulls = -1; + let stableTurns = 0; + while (pulls < CAP_CHUNKS && stableTurns < 12) { + await Bun.sleep(25); + if (pulls === lastPulls) stableTurns++; + else { + stableTurns = 0; + lastPulls = pulls; + } + } + expect(pulls).toBeLessThan(CAP_CHUNKS / 2); + } finally { + socket.removeAllListeners("error"); + socket.on("error", () => {}); + socket.destroy(); + } + + // The proxy's on_abort should cancel its fetch to the upstream; the + // upstream's serve then cancels the ReadableStream. Poll for that signal. + for (let i = 0; i < 200 && !cancelled; i++) await Bun.sleep(10); + + expect({ cancelled, pullsUnderCap: pulls < CAP_CHUNKS, pulls }).toMatchObject({ + cancelled: true, + pullsUnderCap: true, + }); + expect(proxy.exitCode).toBeNull(); + expect(proxy.signalCode).toBeNull(); +}); From 9ccdf0d6668b776a6c934afaec118f0a4b80b936 Mon Sep 17 00:00:00 2001 From: "autofix-ci[bot]" <114827586+autofix-ci[bot]@users.noreply.github.com> Date: Tue, 4 Aug 2026 04:30:47 +0000 Subject: [PATCH 2/3] [autofix.ci] apply automated fixes --- src/runtime/server/RequestContext.rs | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/src/runtime/server/RequestContext.rs b/src/runtime/server/RequestContext.rs index bd808a886ab5..c87d306617d8 100644 --- a/src/runtime/server/RequestContext.rs +++ b/src/runtime/server/RequestContext.rs @@ -1499,7 +1499,8 @@ where if let Some(byte_stream) = this.byte_stream.take() { bun_ptr::BackRef::from(byte_stream).cancel_from_sink(None); any_js_calls.set(true); - this.response_body_readable_stream_ref.with_mut(|s| s.deinit()); + this.response_body_readable_stream_ref + .with_mut(|s| s.deinit()); this.deref(); } From 96b99d40770e8819ed48f8b70a2dceeed178dba7 Mon Sep 17 00:00:00 2001 From: robobun <117481402+robobun@users.noreply.github.com> Date: Tue, 4 Aug 2026 04:45:48 +0000 Subject: [PATCH 3/3] end_chunk: drop the now-dead byte_stream.set(None) in the error branch --- src/runtime/server/RequestContext.rs | 1 - 1 file changed, 1 deletion(-) diff --git a/src/runtime/server/RequestContext.rs b/src/runtime/server/RequestContext.rs index c87d306617d8..b9b04b6f1018 100644 --- a/src/runtime/server/RequestContext.rs +++ b/src/runtime/server/RequestContext.rs @@ -3407,7 +3407,6 @@ where if !this.flags.has_written_status() { let global_this = this.server().global_this(); let js_err = err.to_js(global_this); - this.byte_stream.set(None); this.response_body_readable_stream_ref .with_mut(|s| s.deinit()); this.run_error_handler(js_err);