diff --git a/src/runtime/webcore/fetch/FetchTasklet.rs b/src/runtime/webcore/fetch/FetchTasklet.rs index b40e105b6511..ee6f348cf430 100644 --- a/src/runtime/webcore/fetch/FetchTasklet.rs +++ b/src/runtime/webcore/fetch/FetchTasklet.rs @@ -1152,12 +1152,7 @@ impl FetchTasklet { // mark to wait until deinit self.is_waiting_abort = self.result.has_more; self.abort_reason.set(&global_object, check_result); - self.signal_store.aborted.store(true, Ordering::Relaxed); - self.tracker.did_cancel(&self.global_this); - // we need to abort the request - if let Some(http_) = self.http.as_mut() { - http::http_thread().schedule_shutdown(http_); - } + self.abort_task(); self.result.fail = Some(err!("ERR_TLS_CERT_ALTNAME_INVALID")); return false; } @@ -1177,11 +1172,7 @@ impl FetchTasklet { let hostname_err_result = global_object.try_take_exception().unwrap(); self.is_waiting_abort = self.result.has_more; self.abort_reason.set(&global_object, hostname_err_result); - self.signal_store.aborted.store(true, Ordering::Relaxed); - self.tracker.did_cancel(&self.global_this); - if let Some(http_) = self.http.as_mut() { - http::http_thread().schedule_shutdown(http_); - } + self.abort_task(); self.result.fail = Some(err!("ERR_TLS_CERT_ALTNAME_INVALID")); return false; } @@ -1202,13 +1193,7 @@ impl FetchTasklet { // mark to wait until deinit self.is_waiting_abort = self.result.has_more; self.abort_reason.set(&global_object, check_result); - self.signal_store.aborted.store(true, Ordering::Relaxed); - self.tracker.did_cancel(&self.global_this); - - // we need to abort the request - if let Some(http_) = self.http.as_mut() { - http::http_thread().schedule_shutdown(http_); - } + self.abort_task(); self.result.fail = Some(err!("ERR_TLS_CERT_ALTNAME_INVALID")); return false; } @@ -1622,6 +1607,9 @@ impl FetchTasklet { if this.signal_store.body_receive_mode() == BodyReceiveMode::Ignore { return; } + // reader.cancel() / body.cancel() aborts the fetch so the server sees the + // close (Node/Deno/browsers abort unconditionally). abort_task() is idempotent. + this.abort_task(); this.ignore_remaining_response_body(false); } @@ -1766,14 +1754,21 @@ impl FetchTasklet { } // enabling streaming will make the http thread to drain into the main thread (aka stop buffering) // without a stream ref, response body or response instance alive it will just ignore the result + // An aborted fetch is already shutting down; don't re-arm receive/resume + // draining, which would read the rest of an unbounded body and hold the + // socket open (drain_events resumes before shutdowns). + let aborted = self.signal_store.aborted.load(Ordering::Relaxed); if self .signal_store .set_receive_mode_terminal(BodyReceiveMode::Ignore) + && !aborted { self.schedule_receive_resume(); } if let Some(http_) = self.http.as_mut() { - http_.enable_response_body_streaming(); + if !aborted { + http_.enable_response_body_streaming(); + } } // we should not keep the process alive if we are ignoring the body let _ = self.javascript_vm; @@ -2254,7 +2249,12 @@ impl FetchTasklet { } pub(crate) fn abort_task(&mut self) { - self.signal_store.aborted.store(true, Ordering::Relaxed); + // Idempotent: reader.cancel() and an AbortSignal can both reach here for + // the same fetch. Only the first abort enqueues a shutdown; a second + // would append a redundant ShutdownMessage for an already-closing socket. + if self.signal_store.aborted.swap(true, Ordering::Relaxed) { + return; + } self.tracker.did_cancel(&self.global_this); if let Some(http_) = self.http.as_mut() { diff --git a/test/js/web/fetch/fetch-backpressure.test.ts b/test/js/web/fetch/fetch-backpressure.test.ts index 8e5b9973353e..91583033bb84 100644 --- a/test/js/web/fetch/fetch-backpressure.test.ts +++ b/test/js/web/fetch/fetch-backpressure.test.ts @@ -300,21 +300,17 @@ describe.concurrent("fetch() receive backpressure — buffered consumers are not }); describe.concurrent("fetch() receive backpressure — streaming consumer shapes", () => { - test("reader.cancel() resumes the socket so the abandoned body drains", async () => { + test("reader.cancel() mid-stream lets a subsequent request complete", async () => { await using server = await serve("h1"); const r1 = await fetch(server.url, { keepalive: true }); const reader = r1.body!.getReader(); await reader.read(); - await Bun.sleep(50); await reader.cancel(); - // The cancelled body must drain before the connection is poolable; - // poll server.sent() instead of asserting a connection count. - while (server.sent() < TOTAL) await Bun.sleep(5); + // reader.cancel() aborts the in-flight request (#33227), closing the + // connection; the client must recover so a later request still completes. + // The abort-vs-drain behavior itself is asserted in regression/issue/33227. const buf = await (await fetch(server.url, { keepalive: true })).arrayBuffer(); - expect({ byteLength: buf.byteLength, sent: server.sent() }).toEqual({ - byteLength: TOTAL, - sent: 2 * TOTAL, - }); + expect(buf.byteLength).toBe(TOTAL); }); test("res.body.tee() both branches drain", async () => { diff --git a/test/regression/issue/33227.test.ts b/test/regression/issue/33227.test.ts new file mode 100644 index 000000000000..77b965bf2891 --- /dev/null +++ b/test/regression/issue/33227.test.ts @@ -0,0 +1,45 @@ +import { sleep } from "bun"; +import { expect, test } from "bun:test"; + +// https://github.com/oven-sh/bun/issues/33227 +test("reader.cancel() aborts the fetch and triggers the server's stream cancel", async () => { + const { promise: sawAbort, resolve: onAbort } = Promise.withResolvers(); + const { promise: sawCancel, resolve: onCancel } = Promise.withResolvers(); + const encoder = new TextEncoder(); + + using server = Bun.serve({ + port: 0, + fetch(request) { + request.signal.addEventListener("abort", () => onAbort()); + let count = 0; + return new Response( + new ReadableStream({ + // Stream forever (paced) so the body is always in flight: the unfixed + // drain path keeps the connection open and actively reading, so the + // server never sees a close unless the cancel actually aborts. + async pull(controller) { + controller.enqueue(encoder.encode(`data: ${count++}\n\n`)); + await sleep(10); + }, + cancel() { + onCancel(); + }, + }), + { headers: { "Content-Type": "text/event-stream" } }, + ); + }, + }); + + const res = await fetch(`http://localhost:${server.port}`, { method: "POST" }); + const reader = res.body!.getReader(); + + const first = await reader.read(); + expect(first.done).toBe(false); + expect(first.value!.byteLength).toBeGreaterThan(0); + + await reader.cancel(); + + // Cancelling the reader must abort the fetch and close the connection, so the + // server observes request.signal's abort and runs the body stream's cancel(). + await Promise.all([sawAbort, sawCancel]); +});