From 13378c54da55d7d979414f8b6b7cd4f22b7d27ee Mon Sep 17 00:00:00 2001 From: robobun <117481402+robobun@users.noreply.github.com> Date: Wed, 7 Oct 2026 17:41:15 +0000 Subject: [PATCH 1/2] ByteStream: report a stored producer error and keep the taken prefix with the buffer --- src/runtime/api/html_rewriter.rs | 2 +- src/runtime/server/RequestContext.rs | 21 +- src/runtime/webcore/ByteStream.rs | 240 ++++++++++-------- test/js/bun/http/bun-server.test.ts | 56 ++++ .../http/serve-stream-body-error-fixture.ts | 90 ++++++- .../bun/http/serve-stream-body-error.test.ts | 72 ++++++ .../s3/s3-write-to-file-sync-close.test.ts | 8 +- test/js/web/fetch/body-mixin-errors.test.ts | 74 ++++++ .../bytestream-partial-read-error-fixture.ts | 53 ++++ .../fetch/bytestream-partial-read-fixture.ts | 46 ++++ test/js/web/fetch/fetch.stream.test.ts | 36 +++ 11 files changed, 576 insertions(+), 122 deletions(-) create mode 100644 test/js/web/fetch/bytestream-partial-read-error-fixture.ts create mode 100644 test/js/web/fetch/bytestream-partial-read-fixture.ts diff --git a/src/runtime/api/html_rewriter.rs b/src/runtime/api/html_rewriter.rs index f00c557909bd..456befc3aa1b 100644 --- a/src/runtime/api/html_rewriter.rs +++ b/src/runtime/api/html_rewriter.rs @@ -1009,7 +1009,7 @@ impl RewriterPipe { /// Output emitted but not yet taken by a reader. fn unread_output(&self) -> BlobSizeType { let staged = self.output_buffer.get().len(); - let queued = self.output.get().map_or(0, |out| out.buffer.get().len()); + let queued = self.output.get().map_or(0, |out| out.buffered.held_len()); (staged + queued) as BlobSizeType } diff --git a/src/runtime/server/RequestContext.rs b/src/runtime/server/RequestContext.rs index 813b8502a282..756e67198a90 100644 --- a/src/runtime/server/RequestContext.rs +++ b/src/runtime/server/RequestContext.rs @@ -3224,7 +3224,10 @@ where let resp = this.resp.get().expect("infallible: resp bound"); // If we've received the complete body by the time this function is called // we can avoid streaming it and just send it all at once. - if byte_stream.has_received_last_chunk.get() { + // (A stored error goes through the attach path below instead.) + if byte_stream.has_received_last_chunk.get() + && !byte_stream.has_pending_error() + { let mut byte_list = byte_stream.drain(); this.blob.set(AnyBlob::from_array_list( byte_list.move_to_list_managed(), @@ -3267,10 +3270,12 @@ where Self::drain_response_buffer_and_metadata_corked, this.as_ctx_ptr(), ); - } else if matches!( - byte_stream.parent_const().producer.get(), - WebCore::streams::SourceHandle::HTMLRewriter(_) - ) { + } else if byte_stream.has_pending_error() + || matches!( + byte_stream.parent_const().producer.get(), + WebCore::streams::SourceHandle::HTMLRewriter(_) + ) + { // Defer status/headers to the first chunk/end // so a pre-first-byte handler failure can // still reach `error()`. @@ -3281,6 +3286,10 @@ where this.as_ctx_ptr(), ); } + // Runs `end_chunk`, which adopts the ref taken above and may free `this`. + if byte_stream.end_sink_with_pending_error() { + return; + } // Wake the producer after the older bytes are queued. byte_stream.signal_drained(); return; @@ -4146,7 +4155,7 @@ where bytes.on_data(WebCore::streams::Result::Temporary(borrowed)); // What `on_data` buffered; `on_stream_drained` resumes once it empties. - let buffered = bytes.buffer.get().len().saturating_sub(bytes.offset.get()); + let buffered = bytes.buffered_len(); if bytes.buffer_action.get().is_some() || (bytes.sink.get().is_some() && !bytes.sink_paused.get()) { diff --git a/src/runtime/webcore/ByteStream.rs b/src/runtime/webcore/ByteStream.rs index 019038c5c01f..3006d2f1cdfe 100644 --- a/src/runtime/webcore/ByteStream.rs +++ b/src/runtime/webcore/ByteStream.rs @@ -21,7 +21,7 @@ bun_output::declare_scope!(ByteStream, visible); /// across `ByteBlobLoader` / `FileReader`); the trait impl below auto-derefs /// to the `&self` inherent bodies. pub struct ByteStream { - pub(crate) buffer: JsCell>, + pub(crate) buffered: Buffered, pub(crate) has_received_last_chunk: Cell, pub(crate) pending: JsCell, pub(crate) done: Cell, @@ -30,7 +30,6 @@ pub struct ByteStream { // `pending_value: Strong`. Never freed by Rust. pub(crate) pending_buffer: Cell<*mut [u8]>, pub(crate) pending_value: JsCell, // jsc.Strong.Optional - pub offset: Cell, pub(crate) high_water_mark: blob::SizeType, /// Native sink this stream is piped into; `on_data` dispatches and honors `Writable`. pub(crate) sink: JsCell, @@ -43,7 +42,7 @@ pub struct ByteStream { impl Default for ByteStream { fn default() -> Self { Self { - buffer: JsCell::new(Vec::new()), + buffered: Buffered::default(), has_received_last_chunk: Cell::new(false), pending: JsCell::new(streams::Pending { result: streams::Result::Done, @@ -52,7 +51,6 @@ impl Default for ByteStream { done: Cell::new(false), pending_buffer: Cell::new(Self::empty_pending_buffer()), pending_value: JsCell::new(StrongOptional::empty()), - offset: Cell::new(0), high_water_mark: 0, sink: JsCell::new(SinkHandle::None), sink_paused: Cell::new(false), @@ -62,6 +60,86 @@ impl Default for ByteStream { } } +/// Bytes a producer delivered. A consumer already took `bytes[..consumed]`. +#[derive(Default)] +pub(crate) struct Buffered { + bytes: JsCell>, + consumed: Cell, +} + +impl Buffered { + /// Bytes delivered that no consumer has taken yet. + #[inline] + pub(crate) fn len(&self) -> usize { + self.bytes.get().len() - self.consumed.get() + } + + #[inline] + pub(crate) fn is_empty(&self) -> bool { + self.len() == 0 + } + + /// Bytes the buffer holds, the taken prefix included. The HTMLRewriter paces its input by this. + #[inline] + pub(crate) fn held_len(&self) -> usize { + self.bytes.get().len() + } + + /// Heap the buffer holds, including an already-taken prefix. + #[inline] + fn capacity(&self) -> usize { + self.bytes.get().capacity() + } + + /// Adopt `bytes`, of which a consumer already took the first `consumed`. + fn adopt(&self, bytes: Vec, consumed: usize) { + debug_assert!(consumed <= bytes.len()); + self.bytes.set(bytes); + self.consumed.set(consumed); + } + + /// Take the untaken bytes out. The buffer keeps no allocation. + fn take(&self) -> Vec { + let consumed = self.consumed.replace(0); + let mut bytes = self.bytes.replace(Vec::new()); + if consumed > 0 { + bytes.drain(..consumed); + } + bytes + } + + fn extend(&self, chunk: &[u8]) { + self.bytes.with_mut(|b| b.extend_from_slice(chunk)); + } + + /// Drop every buffered byte and release the allocation. + fn clear(&self) { + self.consumed.set(0); + if self.capacity() > 0 { + self.bytes.with_mut(|b| { + b.clear(); + b.shrink_to_fit(); + }); + } + } + + /// Copy up to `dest.len()` untaken bytes into `dest`. Returns how many it copied. + fn copy_out(&self, dest: &mut [u8]) -> usize { + self.bytes.with_mut(|b| { + let consumed = self.consumed.get(); + let to_write = (b.len() - consumed).min(dest.len()); + dest[..to_write].copy_from_slice(&b[consumed..][..to_write]); + if consumed + to_write == b.len() { + self.consumed.set(0); + b.clear(); + } else { + self.consumed.set(consumed + to_write); + } + to_write + }) + } +} + /// ReadableStream source backed by a ByteStream. pub(crate) type Source = readable_stream::NewSource; @@ -268,7 +346,7 @@ impl ByteStream { self.size_hint.set(estimated_size as blob::SizeType); } DrainResult::Owned { list, size_hint } => { - self.buffer.set(list); + self.buffered.adopt(list, 0); self.size_hint.set(size_hint as blob::SizeType); } DrainResult::Aborted => {} @@ -276,13 +354,15 @@ impl ByteStream { } fn on_start(&self) -> streams::Start { - if self.has_received_last_chunk.get() && self.buffer.get().is_empty() { - return streams::Start::Empty; - } - if self.has_received_last_chunk.get() { - let buffer = self.buffer.replace(Vec::new()); - return streams::Start::OwnedAndDone(Vec::::move_from_list(buffer)); + // `Empty` would close the stream without the pull that returns the stored error. + if self.has_pending_error() { + return streams::Start::Ready; + } + if self.buffered.is_empty() { + return streams::Start::Empty; + } + return streams::Start::OwnedAndDone(Vec::::move_from_list(self.buffered.take())); } if self.high_water_mark == 0 { @@ -333,7 +413,7 @@ impl ByteStream { /// Bytes delivered that no consumer has taken yet. #[inline] pub fn buffered_len(&self) -> usize { - self.buffer.get().len() - self.offset.get() + self.buffered.len() } /// Sink's drain ack: unpause, push buffered bytes, end if last chunk already arrived. @@ -348,9 +428,8 @@ impl ByteStream { return; } - if !self.buffer.get().is_empty() { - let buffered = self.buffer.replace(Vec::new()); - self.offset.set(0); + if !self.buffered.is_empty() { + let buffered = self.buffered.take(); let result = if self.has_received_last_chunk.get() { streams::Result::OwnedAndDone(buffered) } else { @@ -410,16 +489,9 @@ impl ByteStream { self.parent_const().producer.get().ready(None, None); } - /// Take the unread buffered bytes (`buffer[offset..]`) without signalling - /// the producer; the caller writes them to the sink before - /// [`Self::signal_drained`]. + /// Take the untaken bytes. The caller calls [`Self::signal_drained`] after it writes them. pub(crate) fn take_buffer(&self) -> Vec { - let consumed = self.offset.replace(0); - let mut list = self.buffer.replace(Vec::new()); - if consumed > 0 { - list.drain(..consumed); - } - Vec::::move_from_list(list) + Vec::::move_from_list(self.buffered.take()) } /// Called by native fast-paths after wiring `self.sink`: a consumer now @@ -502,10 +574,7 @@ impl ByteStream { self.signal_drained(); action.reject(global, err); - self.buffer.with_mut(|b| { - b.clear(); - b.shrink_to_fit(); - }); + self.buffered.clear(); self.pending.with_mut(|p| { p.result.release(); p.result = streams::Result::Done; @@ -526,7 +595,7 @@ impl ByteStream { return; }; - if self.buffer.get().capacity() == 0 && matches!(stream, streams::Result::Done) { + if self.buffered.capacity() == 0 && matches!(stream, streams::Result::Done) { bun_output::scoped_log!( ByteStream, "ByteStream.onData done and action.fulfill()" @@ -536,17 +605,17 @@ impl ByteStream { action.fulfill(self.parent_const().global_this(), &mut blob); return; } - if self.buffer.get().capacity() == 0 { + if self.buffered.capacity() == 0 { if let streams::Result::OwnedAndDone(mut owned) = stream { bun_output::scoped_log!( ByteStream, "ByteStream.onData owned_and_done and action.fulfill()" ); - // Move the owned Vec into `buffer` + // Move the owned Vec into the buffer // directly instead of round-tripping through `chunk` (which would borrow // `stream`). - self.buffer.set(owned.move_to_list_managed()); + self.buffered.adopt(owned.move_to_list_managed(), 0); let mut blob = self.to_any_blob().unwrap(); action.fulfill(self.parent_const().global_this(), &mut blob); return; @@ -558,8 +627,7 @@ impl ByteStream { "ByteStream.onData appendSlice and action.fulfill()" ); - self.buffer - .with_mut(|b| b.extend_from_slice(stream.slice())); + self.buffered.extend(stream.slice()); // The owned `Vec` // payload of `stream` is freed by its Drop glue at the explicit `drop` below // (Temporary* variants are non-owning `RawSlice` and so are left alone). @@ -568,8 +636,7 @@ impl ByteStream { action.fulfill(self.parent_const().global_this(), &mut blob); return; } else { - self.buffer - .with_mut(|b| b.extend_from_slice(stream.slice())); + self.buffered.extend(stream.slice()); // The owned `Vec` payload of // `stream` is freed by its Drop glue (Temporary* are non-owning `RawSlice`, left alone). drop(stream); @@ -581,7 +648,7 @@ impl ByteStream { let chunk = stream.slice(); if self.pending.get().state == streams::PendingState::Pending { - debug_assert!(self.buffer.get().is_empty()); + debug_assert!(self.buffered.is_empty()); // Re-derive the destination from the GC-rooted view instead of trusting the // raw pointer captured at pull time: JS can detach or transfer the backing // ArrayBuffer between the pull and the data arriving, leaving @@ -663,19 +730,19 @@ impl ByteStream { .unwrap_or_else(|_| panic!("Out of memory while copying request body")); } + /// Buffer `stream`, whose first `offset` bytes a pending pull already took. fn append(&self, stream: streams::Result, offset: usize) -> Result<(), bun_alloc::AllocError> { - if self.buffer.get().capacity() == 0 { + if self.buffered.capacity() == 0 { match stream { streams::Result::Owned(mut owned) | streams::Result::OwnedAndDone(mut owned) => { // `move_to_list_managed` moves the buffer, no copy. - self.buffer.set(owned.move_to_list_managed()); - self.offset.set(self.offset.get() + offset); + self.buffered.adopt(owned.move_to_list_managed(), offset); } streams::Result::TemporaryAndDone(temp) | streams::Result::Temporary(temp) => { let chunk = &temp.slice()[offset..]; let mut buf = Vec::with_capacity(chunk.len()); buf.extend_from_slice(chunk); - self.buffer.set(buf); + self.buffered.adopt(buf, 0); } streams::Result::Err(err) => { self.pending @@ -689,12 +756,10 @@ impl ByteStream { match stream { streams::Result::TemporaryAndDone(temp) | streams::Result::Temporary(temp) => { - self.buffer - .with_mut(|b| b.extend_from_slice(&temp.slice()[offset..])); + self.buffered.extend(&temp.slice()[offset..]); } streams::Result::OwnedAndDone(owned) | streams::Result::Owned(owned) => { - self.buffer - .with_mut(|b| b.extend_from_slice(&owned.slice()[offset..])); + self.buffered.extend(&owned.slice()[offset..]); // `owned: Vec` drops here. } streams::Result::Err(err) => { @@ -703,10 +768,7 @@ impl ByteStream { } // Erroring a stream discards queued chunks; drop the buffered // bytes now instead of retaining them off-heap until GC. - self.buffer.with_mut(|b| { - b.clear(); - b.shrink_to_fit(); - }); + self.buffered.clear(); self.pending .with_mut(|p| p.result = streams::Result::Err(err)); } @@ -729,42 +791,16 @@ impl ByteStream { debug_assert!(!buffer.is_empty()); debug_assert!(self.buffer_action.get().is_none()); - if !self.buffer.get().is_empty() { + if !self.buffered.is_empty() { debug_assert!(self.value().is_empty()); // == .zero - // R-2: confine the `&mut Vec` to a `with_mut` so no `JsCell` - // borrow escapes the copy. The result tuple drives the rest. - let (to_write, remaining_in_buffer_len) = self.buffer.with_mut(|b| { - let to_write = (b.len() - self.offset.get()).min(buffer.len()); - let remaining_in_buffer_len = to_write; // length of `this.buffer.items[this.offset..][0..to_write]` + let to_write = self.buffered.copy_out(buffer); + // Both sides are non-empty, so the copy moved at least one byte. + debug_assert!(to_write > 0); - buffer[..to_write].copy_from_slice(&b[self.offset.get()..][..to_write]); - - if self.offset.get() + to_write == b.len() { - self.offset.set(0); - b.clear(); - } else { - self.offset.set(self.offset.get() + to_write); - } - (to_write, remaining_in_buffer_len) - }); - - if self.buffer.get().is_empty() { + if self.buffered.is_empty() { self.signal_drained(); } - if self.has_received_last_chunk.get() && remaining_in_buffer_len == 0 { - self.buffer.with_mut(|b| { - b.clear(); - b.shrink_to_fit(); - }); - self.done.set(true); - - return streams::Result::IntoArrayAndDone(IntoArray { - value: view, - len: to_write as blob::SizeType, // @truncate - }); - } - return streams::Result::IntoArray(IntoArray { value: view, len: to_write as blob::SizeType, // @truncate @@ -774,7 +810,7 @@ impl ByteStream { if self.has_received_last_chunk.get() { // Surface a stored terminal error (set by `append(Err)` when no // reader was waiting) instead of silently reporting `Done`. - if matches!(self.pending.get().result, streams::Result::Err(_)) { + if self.has_pending_error() { return self .pending .with_mut(|p| core::mem::replace(&mut p.result, streams::Result::Done)); @@ -808,12 +844,7 @@ impl ByteStream { pub(crate) fn on_cancel(&self) { bun_jsc::mark_binding!(); let view = self.value(); - if self.buffer.get().capacity() > 0 { - self.buffer.with_mut(|b| { - b.clear(); - b.shrink_to_fit(); - }); - } + self.buffered.clear(); self.done.set(true); self.pending_value.with_mut(|pv| pv.deinit()); // A native sink wired to this stream must fail, not later see an EOF and commit what it @@ -847,7 +878,7 @@ impl ByteStream { fn memory_cost(&self) -> usize { // ReadableStreamSource covers @sizeOf(ByteStream) - self.buffer.get().capacity() + self.buffered.capacity() } /// NOTE: not `impl Drop` — `ByteStream` is the `context` payload of a `.classes.ts` @@ -860,12 +891,7 @@ impl ByteStream { /// `Box` provenance. fn finalize(&mut self) { bun_jsc::mark_binding!(); - if self.buffer.get().capacity() > 0 { - self.buffer.with_mut(|b| { - b.clear(); - b.shrink_to_fit(); - }); - } + self.buffered.clear(); self.pending_value.with_mut(|pv| pv.deinit()); if !self.done.get() { @@ -908,6 +934,11 @@ impl ByteStream { drained } + /// The producer failed before anything attached and [`Self::append`] stored its error. + pub(crate) fn has_pending_error(&self) -> bool { + matches!(self.pending.get().result, streams::Result::Err(_)) + } + /// Take a pre-attach `StreamResult::Err` stashed by [`Self::append`]. pub fn take_pending_error(&self) -> Option { self.pending.with_mut(|p| { @@ -922,9 +953,21 @@ impl ByteStream { }) } + /// End the attached sink with the stored producer error. Returns false if none is stored. + pub(crate) fn end_sink_with_pending_error(&self) -> bool { + let sink = *self.sink.get(); + debug_assert!(sink.is_some()); + let Some(err) = self.take_pending_error() else { + return false; + }; + self.detach_sink(Some(&err)); + sink.end(Some(err)); + true + } + pub(crate) fn to_any_blob(&self) -> Option { - if self.has_received_last_chunk.get() { - let buffer = self.buffer.replace(Vec::new()); + if self.has_received_last_chunk.get() && !self.has_pending_error() { + let buffer = self.buffered.take(); self.done.set(true); self.pending.with_mut(|p| { p.result.release(); @@ -954,10 +997,7 @@ impl ByteStream { err_js.ensure_still_alive(); self.pending.with_mut(|p| p.result = streams::Result::Done); self.done.set(true); - self.buffer.with_mut(|b| { - b.clear(); - b.shrink_to_fit(); - }); + self.buffered.clear(); return Ok(jsc::JSPromise::rejected_promise(cx.global(), err_js).to_js()); } diff --git a/test/js/bun/http/bun-server.test.ts b/test/js/bun/http/bun-server.test.ts index 949bec4432b9..2666109f4fda 100644 --- a/test/js/bun/http/bun-server.test.ts +++ b/test/js/bun/http/bun-server.test.ts @@ -2136,6 +2136,62 @@ test.concurrent("should be able to abrubtly close a upload request", async () => expect(await handlerDone.promise).toEqual({ error: "AbortError", received: chunk.byteLength }); }); +// Same failure, but delivered while nothing is reading: the handler takes +// req.body (which creates the native stream) and only starts reading after the +// client has gone away. The stored abort must surface from the first read; it +// used to come out as a clean `{ done: true }`, indistinguishable from a +// complete upload. +test("reading an upload whose client went away before the first read rejects", async () => { + const bodyTaken = Promise.withResolvers(); + const clientClosed = Promise.withResolvers(); + const outcome = Promise.withResolvers(); + using server = Bun.serve({ + port: 0, + async fetch(req) { + const body = req.body!; + bodyTaken.resolve(); + await clientClosed.promise; + try { + // The abort reaches the body on a later event-loop turn. Bun.inspect(req) + // lists the body stream while the body is still pending and stops listing + // it once the body holds the error; reading it to find out is the very + // thing under test. + const deadline = Date.now() + 10_000; + while (Bun.inspect(req).includes("ReadableStream")) { + if (Date.now() > deadline) throw new Error("the abort never reached the idle body"); + await Bun.sleep(1); + } + outcome.resolve( + await body + .getReader() + .read() + .then( + result => `resolved: ${JSON.stringify(result)}`, + error => `rejected: ${(error as Error).name}`, + ), + ); + } catch (error) { + outcome.reject(error); + } + return new Response("unreachable by the client"); + }, + }); + const socket = await Bun.connect({ + hostname: server.hostname, + port: server.port, + socket: { + open(socket) { + socket.write("POST / HTTP/1.1\r\nHost: localhost\r\nContent-Length: 100\r\n\r\npartial"); + }, + data() {}, + close: () => clientClosed.resolve(), + }, + }); + await bodyTaken.promise; + socket.end(); + expect(await outcome.promise).toBe("rejected: AbortError"); +}); + // This test is disabled because it can OOM the CI test.skip("should be able to stream huge amounts of data", async () => { const buf = Buffer.alloc(1024 * 1024 * 256); diff --git a/test/js/bun/http/serve-stream-body-error-fixture.ts b/test/js/bun/http/serve-stream-body-error-fixture.ts index 25e9ded60a14..10e8c5beb2ff 100644 --- a/test/js/bun/http/serve-stream-body-error-fixture.ts +++ b/test/js/bun/http/serve-stream-body-error-fixture.ts @@ -8,9 +8,11 @@ // argv[3..]: "development" to run the server with development: true; // "dev-server" to also give it an HTML route, which makes the // development server run a bake dev server (the requests below -// still go to fetch()) +// still go to fetch()); +// "no-error-handler" to run the server without an error() callback // -// Prints a single JSON object on stdout and exits 0. +// Prints a single JSON object on stdout. Exits 0, except that a run without +// error() reports the failure as unhandled, which makes Bun exit 1. import net from "node:net"; import devServerPage from "./serve-stream-body-error-fixture.html"; @@ -18,6 +20,7 @@ const variant = process.argv[2]; const flags = process.argv.slice(3); const development = flags.includes("development"); const devServer = flags.includes("dev-server"); +const withErrorHandler = !flags.includes("no-error-handler"); // Set by the mid-stream variants so they only error once their chunk has // provably reached the client socket, i.e. after the 200 response is committed @@ -196,10 +199,33 @@ async function fetchFromFailingUpstream(firstWrite: string) { }; } -// Bodies Bun.serve streams natively (no JS ReadableStream pump) whose producer -// fails after the status line is committed. error() can no longer answer, so -// the failure is reported and the connection is closed without the terminating -// chunk, whether or not a body byte went out first. +// The upstream failure reaches the fetch() Response on a later event-loop turn. +// Bun.inspect(res) lists the body stream while the body is still pending and +// stops listing it once the body holds the error; reading the stream to find +// out would consume the very state under test. By then the error has also been +// pushed into the native stream `res.body` made. +async function untilBodyFailed(res: Response) { + const deadline = Date.now() + 10_000; + while (Bun.inspect(res).includes("ReadableStream")) { + if (Date.now() > deadline) throw new Error("the upstream failure never reached the idle body"); + await Bun.sleep(1); + } +} + +// Lets an HTMLRewriter whose output nothing reads emit and fail: it works +// across event-loop turns while its output is unobserved. +async function turns() { + for (let i = 0; i < 8; i++) await new Promise(resolve => setImmediate(resolve)); +} + +const PARTIAL_BODY = "HTTP/1.1 200 OK\r\nContent-Length: 100\r\n\r\npartial"; + +// Bodies Bun.serve streams natively (no JS ReadableStream pump). The first four +// fail after the status line is committed: error() can no longer answer, so the +// failure is reported and the connection is closed without the terminating +// chunk, whether or not a body byte went out first. The `-before-render` ones +// fail while nothing reads them: the native stream holds the error, with no JS +// rejection anywhere, when Bun.serve renders the Response. const nativeBodies: Record Response | Promise> = { "rewriter-mid-stream-throw": () => rewriterResponse(() => { @@ -228,6 +254,46 @@ const nativeBodies: Record Response | Promise> = { promise.then(fail); return res; }, + // The stream exists (res.body was taken) but nothing reads it when the + // upstream dies. + "native-errored-before-render": async () => { + const { res, fail } = await fetchFromFailingUpstream(PARTIAL_BODY); + const body = res.body!; + fail(); + await untilBodyFailed(res); + return new Response(body); + }, + // Same, one producer further away: the failing fetch() body feeds an + // HTMLRewriter and it is the rewriter's output stream that holds the error. + // (The rewriter is given a wrapper so `res` itself still tracks the failure.) + "native-rewriter-errored-before-render": async () => { + const { res, fail } = await fetchFromFailingUpstream(PARTIAL_BODY); + const input = new Response(res.body, { headers: { "content-type": "text/html" } }); + const body = new HTMLRewriter().on("p", {}).transform(input).body!; + fail(); + await untilBodyFailed(res); + return new Response(body); + }, + // A consumer started the rewriter's output and let go of it without a read, + // so one pull view of the output sits in the stream's queue and the rest in + // the native stream, behind the count of bytes already taken. Then the + // rewrite fails. + "native-started-errored-before-render": async () => { + let input!: ReadableStreamDefaultController; + const res = new HTMLRewriter() + .on("p", { + element() { + throw new Error("boom"); + }, + }) + .transform(new Response(new ReadableStream({ start: c => void (input = c) }))); + res.body!.getReader().releaseLock(); + input.enqueue(new TextEncoder().encode(`${Buffer.alloc(400_000, "x").toString()}`)); + await turns(); + input.enqueue(new TextEncoder().encode("

x

")); + await turns(); + return res; + }, }; const source = sources[variant]; @@ -251,10 +317,14 @@ await using server = Bun.serve({ port: 0, development, ...(devServer ? { routes: { "/dev-server-page": devServerPage } } : {}), - error() { - errorCb++; - return new Response("err-body", { status: 500 }); - }, + // The body names the error that reached error(), so the wire pins both the + // channel and the error's identity. + error: withErrorHandler + ? (error: Error & { code?: string }) => { + errorCb++; + return new Response(`err-body:${error.code ?? error.message}`, { status: 500 }); + } + : undefined, fetch(req) { if (new URL(req.url).pathname === "/ok") { return new Response("ok"); diff --git a/test/js/bun/http/serve-stream-body-error.test.ts b/test/js/bun/http/serve-stream-body-error.test.ts index 80177caa5f09..1fd9fbe9aaa8 100644 --- a/test/js/bun/http/serve-stream-body-error.test.ts +++ b/test/js/bun/http/serve-stream-body-error.test.ts @@ -162,6 +162,78 @@ test.concurrent.each([ expect(stderr).toContain("boom"); }); +// Natively streamed bodies whose producer failed while nothing was reading them: +// the native stream holds the error, with no JS rejection anywhere, when +// Bun.serve renders the Response. Nothing is committed to the client at that +// point, so the failure goes to error(), as it does when the handler returns +// the failed fetch() Response itself or when such a producer fails before its +// first byte right after Bun.serve attaches (see end_chunk). These used to be +// sent as a complete `200` with `Content-Length: 0`, with the error reported +// nowhere in either mode. The started variant used to abort the whole process +// instead: `panic: range end index 262144 out of range for slice of length 0`. +// +// [variant, what error() received] +const failedBeforeRender = [ + ["native-errored-before-render", "ECONNRESET"], + ["native-rewriter-errored-before-render", "ECONNRESET"], + ["native-started-errored-before-render", "boom"], +] as const; + +for (const flags of [[], ["development"]]) { + const mode = flags.length ? "development" : "production"; + test.concurrent.each(failedBeforeRender)( + `%s in ${mode} mode: error() receives the stored error and answers the request`, + async (variant, received) => { + const { stdout, stderr, exitCode } = await runFixture(variant, ...flags); + expect({ result: JSON.parse(stdout), stderr, exitCode }).toEqual({ + result: { + statusLine: "HTTP/1.1 500 Internal Server Error", + cleanChunkedTerminator: false, + body: `err-body:${received}`, + errorCb: 1, + unhandled: 0, + secondStatusLine: "HTTP/1.1 200 OK", + }, + // error() handled it, so nothing is reported behind its back. + stderr: "", + exitCode: 0, + }); + }, + ); + + // Without an error() callback it is treated like any other unhandled handler + // failure: the default 500 goes out, the error is reported on stderr, and (as + // for a throwing handler) the process exit status becomes 1. The default body + // is mode-specific and not pinned. + test.concurrent( + `native stream errored before render, no error() (${mode}): default 500 and the error is reported`, + async () => { + const { stdout, stderr, exitCode } = await runFixture( + "native-errored-before-render", + ...flags, + "no-error-handler", + ); + const { statusLine, secondStatusLine, errorCb, unhandled } = JSON.parse(stdout); + expect({ + statusLine, + secondStatusLine, + errorCb, + unhandled, + // One report: its first line names the error. + reports: stderr.match(/^TypeError: ECONNRESET/gm), + exitCode, + }).toEqual({ + statusLine: "HTTP/1.1 500 Internal Server Error", + secondStatusLine: "HTTP/1.1 200 OK", + errorCb: 0, + unhandled: 0, + reports: ["TypeError: ECONNRESET"], + exitCode: 1, + }); + }, + ); +} + // The body errors after a chunk has already been flushed to the client. The // 200 is irrevocable at that point, but the connection must be closed without // the terminating `0\r\n\r\n` chunk (RFC 9112 section 7) so the client can diff --git a/test/js/bun/s3/s3-write-to-file-sync-close.test.ts b/test/js/bun/s3/s3-write-to-file-sync-close.test.ts index 51afa22d461d..6d6dad0b4b96 100644 --- a/test/js/bun/s3/s3-write-to-file-sync-close.test.ts +++ b/test/js/bun/s3/s3-write-to-file-sync-close.test.ts @@ -8,10 +8,8 @@ import path from "node:path"; // controller before assignToStream returns. This used to trip a debug // assertion that the signal was still live. // -// Today the credential error is swallowed by ByteStream::on_start() returning -// Start::Empty without checking pending.result, so write() resolves to 0. -// Accept either outcome here so a future fix that surfaces the error does not -// trip this test. +// The credential error is stored in the ByteStream before anything reads it. +// write() rejects with it, and nothing else reports it on stderr. test("Bun.file().write(S3 file) when the S3 stream closes synchronously", async () => { using dir = tempDir("s3-write-sync-close", {}); const dest = path.join(String(dir), "out.bin"); @@ -54,7 +52,7 @@ test("Bun.file().write(S3 file) when the S3 stream closes synchronously", async stderr: normalizeBunSnapshot(stderr), signalCode: proc.signalCode, }).toEqual({ - stdout: expect.stringMatching(/^(resolved|rejected:ERR_S3_MISSING_CREDENTIALS)$/), + stdout: "rejected:ERR_S3_MISSING_CREDENTIALS", stderr: "", signalCode: null, }); diff --git a/test/js/web/fetch/body-mixin-errors.test.ts b/test/js/web/fetch/body-mixin-errors.test.ts index 9dbf3953c93e..6287873fb58d 100644 --- a/test/js/web/fetch/body-mixin-errors.test.ts +++ b/test/js/web/fetch/body-mixin-errors.test.ts @@ -193,6 +193,80 @@ describe("body-mixin-errors", () => { }, ); + // The failure is delivered while `res.body` already exists but nothing is + // reading it, so it is held inside the native ByteStream behind the stream + // rather than on the Response. The server announces 100 bytes, sends part of + // them, and closes only after the test has materialized `res.body`, so the + // ordering does not depend on timing. Before the fix every consumer below + // saw a clean, empty body instead of the error. + async function withIdleErroredBody(fn: (body: ReadableStream) => Promise): Promise { + const sockets: net.Socket[] = []; + const server = net.createServer(socket => { + sockets.push(socket); + socket.resume(); + socket.on("error", () => {}); + socket.write("HTTP/1.1 200 OK\r\nContent-Length: 100\r\n\r\npartial"); + }); + server.listen(0, "127.0.0.1"); + await once(server, "listening"); + const { port } = server.address() as net.AddressInfo; + try { + const res = await fetch(`http://127.0.0.1:${port}/`); + const body = res.body!; + for (const socket of sockets) socket.end(); + // The failure reaches the body on a later event-loop turn. Bun.inspect(res) + // lists the body stream while the body is still pending and stops listing + // it once the body holds the error; reading the stream to find out would + // consume the very state under test. + const deadline = Date.now() + 10_000; + while (Bun.inspect(res).includes("ReadableStream")) { + if (Date.now() > deadline) throw new Error("the truncation never reached the idle body"); + await Bun.sleep(1); + } + return await fn(body); + } finally { + for (const socket of sockets) socket.destroy(); + await new Promise(r => server.close(() => r())); + } + } + + function expectTruncationError(err: unknown) { + expect(err).toBeInstanceOf(TypeError); + expect((err as any).code).toBe("ECONNRESET"); + } + + it.concurrent("fetch: reading a stream that errored while idle rejects instead of ending cleanly", async () => { + await withIdleErroredBody(async body => { + let err: unknown; + await body + .getReader() + .read() + .catch(e => (err = e)); + expectTruncationError(err); + }); + }); + + // text() never took the blob fast path and already rejected; it is listed so + // the whole family is pinned to one behaviour. + it.concurrent.each(["text", "arrayBuffer", "bytes", "blob", "json"] as const)( + "fetch: Response wrapping a stream that errored while idle rejects %s()", + async method => { + await withIdleErroredBody(async body => { + let err: unknown; + await new Response(body)[method]().catch((e: unknown) => (err = e)); + expectTruncationError(err); + }); + }, + ); + + it.concurrent("fetch: Request wrapping a stream that errored while idle rejects arrayBuffer()", async () => { + await withIdleErroredBody(async body => { + let err: unknown; + await new Request("http://example.com/", { method: "POST", body }).arrayBuffer().catch((e: unknown) => (err = e)); + expectTruncationError(err); + }); + }); + // Counts inbound TCP connections on a server that answers a chunked POST // once it sees the 0\r\n\r\n terminator. `expectConnections(n)` first sends // a probe request so every accept queued before it has been delivered by the diff --git a/test/js/web/fetch/bytestream-partial-read-error-fixture.ts b/test/js/web/fetch/bytestream-partial-read-error-fixture.ts new file mode 100644 index 000000000000..cdd09938d940 --- /dev/null +++ b/test/js/web/fetch/bytestream-partial-read-error-fixture.ts @@ -0,0 +1,53 @@ +// The same stale bookkeeping as bytestream-partial-read-fixture.ts, reached through the stream's +// own pull loop: a text-mode reader takes one pull view out of the buffered bytes, the producer +// fails, and the reader reads on. `append(Err)` dropped the buffer and kept the index, so the +// next pull's `drain()` aborted the process: +// +// panic: range end index 262144 out of range for slice of length 0 +// +// The producer is an HTMLRewriter whose handler throws on the second chunk, so every step runs on +// this thread, in order. Text mode matters: a byte reader's full view retires the pending view, +// and the next pull then skips `drain()`. +const VIEW = 256 * 1024; // nativeSourceDefaultChunkSize (BunStreamSource.cpp) +const tick = () => new Promise(resolve => setImmediate(resolve)); +// The rewriter emits and fails across turns while nothing reads its output. +const turns = async () => { + for (let i = 0; i < 8; i++) await tick(); +}; + +let threw = false; +let input!: ReadableStreamDefaultController; +const res = new HTMLRewriter() + .on("p", { + element() { + threw = true; + throw new Error("boom"); + }, + }) + .transform(new Response(new ReadableStream({ start: c => void (input = c) }))); + +// Starts the stream and lets go of it: the first pull waits for output, and with no lock on the +// output the rewriter does not hold its input. +const text = res.textStream(); +text.getReader().releaseLock(); +// The output of this chunk fills the pull view and leaves the rest behind the consumed prefix. +input.enqueue(new TextEncoder().encode(`${Buffer.alloc(400_000, "x").toString()}`)); +await turns(); +// The handler throws, so the output stream stores the failure while no pull is pending. +input.enqueue(new TextEncoder().encode("

x

")); +while (!threw) await tick(); +await turns(); + +// The view the pull took is queued on the stream. The bytes behind it went with the failed +// body, so the read after it has to report the failure. +const reader = text.getReader(); +const first = await reader.read(); +if (first.done || first.value.length !== VIEW) { + console.log(`BAD first read ${first.done ? "done" : first.value.length}, want ${VIEW}`); + process.exit(0); +} +const outcome = await reader.read().then( + r => `read returned ${r.done ? "done" : `${r.value.length} chars`}`, + (e: Error) => `rejected ${e.message}`, +); +console.log(outcome === "rejected boom" ? "ok" : `BAD ${outcome}`); diff --git a/test/js/web/fetch/bytestream-partial-read-fixture.ts b/test/js/web/fetch/bytestream-partial-read-fixture.ts new file mode 100644 index 000000000000..13d484b0a576 --- /dev/null +++ b/test/js/web/fetch/bytestream-partial-read-fixture.ts @@ -0,0 +1,46 @@ +// A consumer starts a native body stream and takes one pull view out of it, then hands the +// stream to node:stream, which drives the native source directly. The source used to report the +// bytes a reader had already taken as still buffered, so `drain()` read them out of a buffer that +// was no longer there and the process aborted: +// +// panic: range end index 262144 out of range for slice of length 0 +// +// The producer is an HTMLRewriter because it emits the output of one input chunk in one piece, so +// the split between the pull view and the bytes left behind is the same on every run. +import { Readable } from "node:stream"; + +const SIZE = 400_000; +const tick = () => new Promise(resolve => setImmediate(resolve)); +// The rewriter emits and ends across turns while nothing reads its output. +const turns = async () => { + for (let i = 0; i < 8; i++) await tick(); +}; + +// A pattern whose period (a prime) does not divide the pull view, so bytes handed over from +// the wrong offset cannot pass the tail check below. +const period = Buffer.alloc(4093); +for (let i = 0; i < period.length; i++) period[i] = 97 + (i % 26); +const payload = Buffer.alloc(SIZE, period); + +let input!: ReadableStreamDefaultController; +const res = new HTMLRewriter() + .on("nothing", { element() {} }) + .transform(new Response(new ReadableStream({ start: c => void (input = c) }))); + +res.body!.getReader().releaseLock(); // starts the stream, reads nothing +input.enqueue(payload); +await turns(); +input.close(); +await turns(); + +const chunks: Buffer[] = []; +for await (const chunk of Readable.fromWeb(res.body!)) chunks.push(Buffer.from(chunk)); +const got = Buffer.concat(chunks); + +// The native source still holds the end of the body. Whatever it hands over has to be that +// tail: never a byte a reader already took, and never nothing at all. +if (got.length > 0 && got.equals(payload.subarray(SIZE - got.length))) { + console.log("ok"); +} else { + console.log(`BAD ${got.length} of ${SIZE} bytes, not the tail`); +} diff --git a/test/js/web/fetch/fetch.stream.test.ts b/test/js/web/fetch/fetch.stream.test.ts index f4028be7b42a..476c09be4946 100644 --- a/test/js/web/fetch/fetch.stream.test.ts +++ b/test/js/web/fetch/fetch.stream.test.ts @@ -1468,6 +1468,42 @@ test.concurrent( }, ); +// A native body stream counts the bytes a reader already took as a consumed prefix of its +// buffer. Every path that dropped the buffer used to keep that index, so the next `drain()` +// read the prefix out of a buffer that was no longer there and aborted the process with +// `panic: range end index 262144 out of range for slice of length 0`. One fixture per route: +// node:stream taking the native source over, and the stream's own pull loop after the body +// failed. +test.concurrent("node:stream reading a started native body stream gets its tail instead of crashing", async () => { + await using proc = Bun.spawn({ + cmd: [bunExe(), join(import.meta.dir, "bytestream-partial-read-fixture.ts")], + env: bunEnv, + stdout: "pipe", + stderr: "pipe", + }); + + const [stdout, stderr, exitCode] = await Promise.all([proc.stdout.text(), proc.stderr.text(), proc.exited]); + + expect(stderr).toBe(""); + expect(stdout).toBe("ok\n"); + expect(exitCode).toBe(0); +}); + +test.concurrent("a body that fails after a partial read rejects the next read instead of crashing", async () => { + await using proc = Bun.spawn({ + cmd: [bunExe(), join(import.meta.dir, "bytestream-partial-read-error-fixture.ts")], + env: bunEnv, + stdout: "pipe", + stderr: "pipe", + }); + + const [stdout, stderr, exitCode] = await Promise.all([proc.stdout.text(), proc.stderr.text(), proc.exited]); + + expect(stderr).toBe(""); + expect(stdout).toBe("ok\n"); + expect(exitCode).toBe(0); +}); + // https://github.com/oven-sh/bun/issues/41439 // A flushed zstd chunk that decodes to more than 4096 bytes must reach the // reader in full. The decoder used to hand over 4096 bytes and keep the rest From b2cd22a3335ec3db666f380b111d04da4f7a01e3 Mon Sep 17 00:00:00 2001 From: robobun <117481402+robobun@users.noreply.github.com> Date: Wed, 7 Oct 2026 17:22:13 +0000 Subject: [PATCH 2/2] test: cover S3 and HTMLRewriter streams, and more readers, that are read only after they failed A stream that is taken first and read only after its producer failed holds the failure. New tests for the sources that had none: - S3File.stream(): no credentials (the download fails inside stream(), so no timing is involved), a body cut short, a missing key, a denied key. The first case runs every reader and sink: for await, a reader, pipeTo, tee, Response wrappers, Bun.write, a copy into S3, an upload through fetch, and a Bun.serve response. - HTMLRewriter output: a handler that throws and an upstream that resets while `.body` exists and nothing reads it. - fetch body: for await, pipeTo and tee beside the reader. --- test/js/bun/s3/s3-stream-error-gc.test.ts | 73 +++++++- .../bun/s3/s3-stream-stored-error-fixture.ts | 170 ++++++++++++++++++ test/js/web/fetch/body-mixin-errors.test.ts | 18 +- test/js/workerd/html-rewriter.test.js | 43 +++++ 4 files changed, 298 insertions(+), 6 deletions(-) create mode 100644 test/js/bun/s3/s3-stream-stored-error-fixture.ts diff --git a/test/js/bun/s3/s3-stream-error-gc.test.ts b/test/js/bun/s3/s3-stream-error-gc.test.ts index 404fe1778c10..1433e44386c7 100644 --- a/test/js/bun/s3/s3-stream-error-gc.test.ts +++ b/test/js/bun/s3/s3-stream-error-gc.test.ts @@ -1,5 +1,6 @@ import { expect, test } from "bun:test"; -import { bunEnv, bunExe, normalizeBunSnapshot } from "harness"; +import { bunEnv, bunExe, normalizeBunSnapshot, tempDir } from "harness"; +import { join } from "node:path"; test.concurrent("collecting file blobs with Buffer paths does not crash during GC sweep", async () => { // The S3/file blob store keeps the pinned path buffer until the wrapper is @@ -93,3 +94,73 @@ test("S3 stream error parked before consumption survives GC", async () => { } `); }); + +// An S3 stream whose download failed before anything read it holds the failure. It used to read +// as a complete, empty object: a late reader saw a clean end, a copy stored an empty object, an +// upload went out as a complete POST with no body, and Bun.serve answered an empty 200. +async function runStoredErrorFixture(mode: "stored" | "download") { + using dir = tempDir("s3-stream-stored-error", {}); + await using proc = Bun.spawn({ + cmd: [bunExe(), join(import.meta.dir, "s3-stream-stored-error-fixture.ts"), mode], + env: { + ...bunEnv, + S3_ACCESS_KEY_ID: undefined, + S3_SECRET_ACCESS_KEY: undefined, + S3_REGION: undefined, + S3_ENDPOINT: undefined, + S3_BUCKET: undefined, + S3_SESSION_TOKEN: undefined, + AWS_ACCESS_KEY_ID: undefined, + AWS_SECRET_ACCESS_KEY: undefined, + AWS_REGION: undefined, + AWS_ENDPOINT: undefined, + AWS_BUCKET: undefined, + AWS_SESSION_TOKEN: undefined, + // An inherited proxy would hijack the requests to the stub servers. + HTTP_PROXY: undefined, + HTTPS_PROXY: undefined, + http_proxy: undefined, + https_proxy: undefined, + }, + cwd: String(dir), + stdout: "pipe", + stderr: "pipe", + }); + const [stdout, stderr, exitCode] = await Promise.all([proc.stdout.text(), proc.stderr.text(), proc.exited]); + return { stdout, stderr, exitCode }; +} + +test.concurrent("S3 stream that failed before anything read it rejects every consumer and sink", async () => { + const { stdout, stderr, exitCode } = await runStoredErrorFixture("stored"); + const rejected = "rejected ERR_S3_MISSING_CREDENTIALS"; + + expect(stderr).toBe(""); + expect(JSON.parse(stdout)).toEqual({ + "for await": rejected, + "getReader().read()": rejected, + "pipeTo()": rejected, + "tee()": rejected, + "new Response(stream).bytes()": rejected, + "new Response(stream).blob()": rejected, + "Bun.write(path, new Response(stream))": rejected, + "client.write(key, new Response(stream))": rejected, + "Bun.write(s3file, new Response(stream))": rejected, + "requests the bucket got": [], + "fetch(url, { method: 'POST', body: stream })": rejected, + "complete requests the receiver got": 0, + "Bun.serve: return new Response(stream)": "500 error() got ERR_S3_MISSING_CREDENTIALS", + }); + expect(exitCode).toBe(0); +}); + +test.concurrent("S3 stream read only after its download failed on the wire rejects with the failure", async () => { + const { stdout, stderr, exitCode } = await runStoredErrorFixture("download"); + + expect(stderr).toBe(""); + expect(JSON.parse(stdout)).toEqual({ + cut: "rejected ConnectionClosed", + missing: "rejected NoSuchKey", + denied: "rejected AccessDenied", + }); + expect(exitCode).toBe(0); +}); diff --git a/test/js/bun/s3/s3-stream-stored-error-fixture.ts b/test/js/bun/s3/s3-stream-stored-error-fixture.ts new file mode 100644 index 000000000000..2b690736100e --- /dev/null +++ b/test/js/bun/s3/s3-stream-stored-error-fixture.ts @@ -0,0 +1,170 @@ +// Fixture for s3-stream-error-gc.test.ts. An S3 stream whose download failed before anything read +// it holds the failure. It used to read as a complete, empty object instead. +// +// Run it with no S3 credentials and no proxy in the environment, in a scratch directory. +// +// argv[2]: +// "stored" the download fails inside stream() (no credentials), so no timing is involved. +// Prints what every consumer and sink of such a stream got. +// "download" the download fails on the wire: a body cut short, a key that does not exist, a +// key that is denied. Prints what a read that starts after the failure got. +import { once } from "node:events"; +import type { AddressInfo } from "node:net"; +import net from "node:net"; +import path from "node:path"; + +type Body = ReadableStream; + +function outcome(promise: Promise, resolved: (value: T) => string): Promise { + return promise.then(resolved, (error: { code?: string; name?: string }) => `rejected ${error.code ?? error.name}`); +} + +async function count(stream: Body) { + let bytes = 0; + for await (const chunk of stream) bytes += chunk.length; + return bytes; +} + +async function stored() { + const failed = (): Body => new Bun.S3Client({}).file("source").stream(); + const results: Record = {}; + + const readers: Record Promise> = { + "for await": count, + "getReader().read()": async stream => { + const { done, value } = await stream.getReader().read(); + return done ? 0 : value.length; + }, + "pipeTo()": async stream => { + let bytes = 0; + await stream.pipeTo( + new WritableStream({ + write(chunk) { + bytes += chunk.length; + }, + }), + ); + return bytes; + }, + "tee()": stream => count(stream.tee()[0]), + "new Response(stream).bytes()": async stream => (await new Response(stream).bytes()).length, + "new Response(stream).blob()": async stream => (await new Response(stream).blob()).size, + "Bun.write(path, new Response(stream))": stream => Bun.write(path.resolve("out.bin"), new Response(stream)), + }; + for (const [name, read] of Object.entries(readers)) { + results[name] = await outcome(read(failed()), bytes => `resolved ${bytes} bytes`); + } + + // A copy into S3. The write used to resolve and store an empty object under the key. + const puts: string[] = []; + const bucket = Bun.serve({ + port: 0, + hostname: "127.0.0.1", + async fetch(request) { + puts.push(`${request.method} ${new URL(request.url).pathname} ${(await request.bytes()).length} bytes`); + return new Response("", { headers: { ETag: '"etag"' } }); + }, + }); + const client = new Bun.S3Client({ + endpoint: `http://127.0.0.1:${bucket.port}`, + accessKeyId: "id", + secretAccessKey: "secret", + bucket: "bucket", + }); + results["client.write(key, new Response(stream))"] = await outcome( + client.write("copy", new Response(failed())), + bytes => `resolved ${bytes} bytes`, + ); + results["Bun.write(s3file, new Response(stream))"] = await outcome( + Bun.write(client.file("copy"), new Response(failed())), + bytes => `resolved ${bytes} bytes`, + ); + // Waits for requests in flight, so a PUT that went out is counted. + await bucket.stop(); + results["requests the bucket got"] = puts; + + // An upload. The request used to go out as a complete POST with `Content-Length: 0`. + let completeRequests = 0; + const receiver = net.createServer(socket => { + let received = ""; + socket.on("error", () => {}); + socket.on("data", chunk => { + received += chunk.toString("latin1"); + const headEnd = received.indexOf("\r\n\r\n"); + if (headEnd < 0) return; + const length = /^content-length: *(\d+)/im.exec(received.slice(0, headEnd))?.[1]; + const body = received.slice(headEnd + 4); + if (length !== undefined ? body.length >= Number(length) : body.endsWith("0\r\n\r\n")) { + completeRequests++; + socket.end("HTTP/1.1 200 OK\r\nContent-Length: 0\r\nConnection: close\r\n\r\n"); + } + }); + }); + receiver.listen(0, "127.0.0.1"); + await once(receiver, "listening"); + results["fetch(url, { method: 'POST', body: stream })"] = await outcome( + fetch(`http://127.0.0.1:${(receiver.address() as AddressInfo).port}/`, { method: "POST", body: failed() }), + response => `resolved ${response.status}`, + ); + // Waits for the connection to end, so a request that completed is counted. + receiver.close(); + await once(receiver, "close"); + results["complete requests the receiver got"] = completeRequests; + + // A response body. It used to go out as a complete `200` with no body. + await using server = Bun.serve({ + port: 0, + hostname: "127.0.0.1", + error: (error: Error & { code?: string }) => new Response(`error() got ${error.code}`, { status: 500 }), + fetch: () => new Response(failed()), + }); + const served = await fetch(`http://127.0.0.1:${server.port}/`); + results["Bun.serve: return new Response(stream)"] = `${served.status} ${await served.text()}`; + + return results; +} + +async function download() { + const error = (status: string, code: string) => { + const xml = `\n${code}${code}`; + return `HTTP/1.1 ${status}\r\nContent-Type: application/xml\r\nContent-Length: ${xml.length}\r\nConnection: close\r\n\r\n${xml}`; + }; + // The bucket closes the connection behind every answer. + const answers: Record = { + // Announces 100 bytes and sends 50. + cut: `HTTP/1.1 200 OK\r\nContent-Length: 100\r\n\r\n${Buffer.alloc(50, "x").toString()}`, + missing: error("404 Not Found", "NoSuchKey"), + denied: error("403 Forbidden", "AccessDenied"), + ping: "HTTP/1.1 200 OK\r\nContent-Length: 4\r\nConnection: close\r\n\r\npong", + }; + const closed: Record> = {}; + for (const key of Object.keys(answers)) closed[key] = Promise.withResolvers(); + + const bucket = net.createServer(socket => { + socket.on("error", () => {}); + socket.once("data", request => { + const key = /^GET \/bucket\/(\w+)/.exec(request.toString("latin1"))![1]; + socket.on("close", () => closed[key].resolve()); + socket.end(answers[key]); + }); + }); + bucket.listen(0, "127.0.0.1"); + await once(bucket, "listening"); + const endpoint = `http://127.0.0.1:${(bucket.address() as AddressInfo).port}`; + const client = new Bun.S3Client({ endpoint, accessKeyId: "id", secretAccessKey: "secret", bucket: "bucket" }); + + const results: Record = {}; + for (const key of ["cut", "missing", "denied"]) { + const stream: Body = client.file(key).stream(); + // The client closes its end only after it has seen the whole answer, so once the connection + // is closed the failure is on its way to this thread. A round trip through the same HTTP + // thread arrives behind it. + await closed[key].promise; + await (await fetch(`${endpoint}/bucket/ping`)).text(); + results[key] = await outcome(count(stream), bytes => `resolved ${bytes} bytes`); + } + bucket.close(); + return results; +} + +console.log(JSON.stringify(await (process.argv[2] === "stored" ? stored() : download()), null, 2)); diff --git a/test/js/web/fetch/body-mixin-errors.test.ts b/test/js/web/fetch/body-mixin-errors.test.ts index 6287873fb58d..2c9cdf0a8702 100644 --- a/test/js/web/fetch/body-mixin-errors.test.ts +++ b/test/js/web/fetch/body-mixin-errors.test.ts @@ -235,13 +235,21 @@ describe("body-mixin-errors", () => { expect((err as any).code).toBe("ECONNRESET"); } - it.concurrent("fetch: reading a stream that errored while idle rejects instead of ending cleanly", async () => { + type IdleBody = ReadableStream; + it.concurrent.each([ + ["getReader().read()", (body: IdleBody) => body.getReader().read()], + [ + "for await", + async (body: IdleBody) => { + for await (const _ of body); + }, + ], + ["pipeTo()", (body: IdleBody) => body.pipeTo(new WritableStream())], + ["tee()", (body: IdleBody) => body.tee()[0].getReader().read()], + ] as const)("fetch: %s on a stream that errored while idle rejects instead of ending cleanly", async (_, read) => { await withIdleErroredBody(async body => { let err: unknown; - await body - .getReader() - .read() - .catch(e => (err = e)); + await read(body).catch((e: unknown) => (err = e)); expectTruncationError(err); }); }); diff --git a/test/js/workerd/html-rewriter.test.js b/test/js/workerd/html-rewriter.test.js index 1f7ae4c15fbf..dc97be2f1acb 100644 --- a/test/js/workerd/html-rewriter.test.js +++ b/test/js/workerd/html-rewriter.test.js @@ -101,6 +101,28 @@ describe("HTMLRewriter", () => { await expect(res.text()).rejects.toThrow("test"); }); + // `.body` exists but nothing reads it when the handler throws, so the failure + // waits inside the output stream. A read that starts afterwards used to see a + // complete, empty document. + it("error inside element handler rejects a .body that is read only afterwards", async () => { + let input; + const threw = Promise.withResolvers(); + const res = new HTMLRewriter() + .on("div", { + element(element) { + threw.resolve(); + throw new Error("test"); + }, + }) + .transform(new Response(new ReadableStream({ start: controller => void (input = controller) }))); + const body = res.body; + input.enqueue(new TextEncoder().encode("
hello
")); + await threw.promise; + // One turn later the call that ran the handler has returned and failed the output. + await setImmediatePromise(); + await expect(body.getReader().read()).rejects.toThrow("test"); + }); + // Inputs that go through the JS stream pump with data already queued are // drained synchronously inside `transform()`, so the handler throws (and the // pipe fails, detaching its input) while `assign_to_stream` is still on the @@ -1353,6 +1375,27 @@ describe("HTMLRewriter", () => { }); }); + it("a first read on .body after the upstream failed rejects", async () => { + await withPartialBodyServer(async (url, release) => { + const res = await fetch(url); + // `.body` is taken before the upstream fails and read only after it. The + // output stream holds the failure meanwhile; it used to read as a + // complete, empty document. The rewriter is given a wrapper so that + // `res` itself still shows when the failure has landed. + const body = rewriter().transform(new Response(res.body, res)).body; + release(); + // Bun.inspect(res) lists the body stream while the body is pending and + // stops once the body holds the error. By then the error has gone + // through the rewriter into its output. + const deadline = Date.now() + 10_000; + while (Bun.inspect(res).includes("ReadableStream")) { + if (Date.now() > deadline) throw new Error("the upstream failure never reached the body"); + await Bun.sleep(1); + } + expect(await settle(body.getReader().read())).toEqual(rejectedWithConnectionError); + }); + }); + it(".clone() after a failed read throws", async () => { await withPartialBodyServer(async (url, release) => { const res = await fetch(url);