ByteStream: hold a pending JS error in a visited slot instead of a Strong

src/runtime/api/streams.classes.ts

@@ -89,7 +89,15 @@ function source(name) {
     // (`sink` backref). A chained transform needs both.
     // `closeAdapter`: the C++ JSNativeStreamSourceAdapter to notify from `on_close` (BunStreamSource.cpp sets it).
     // `lockedStream`: the stream `ReadableStream__lockNative` locked, which `NewSource::end_locked_stream` closes or errors. A Blob source never gets a native sink.
-    values: ["pendingPromise", "closeAdapter", "owner", "sinkOwner", ...(name !== "Blob" ? ["lockedStream"] : [])],
+    // `pendingError`: a JS error the producer delivered while nothing could take it (`ByteStream::take_pending_error`). Not a `Strong`: the error can reference the Response that owns the stream, and a cycle through a root is never collected.
+    values: [
+      "pendingPromise",
+      "closeAdapter",
+      "owner",
+      "sinkOwner",
+      ...(name !== "Blob" ? ["lockedStream"] : []),
+      ...(name === "Bytes" ? ["pendingError"] : []),
+    ],
   });
 }
 

src/runtime/webcore/ByteStream.rs

@@ -5,6 +5,7 @@ use bun_jsc::strong::Optional as StrongOptional;
 use bun_jsc::{self as jsc, JSGlobalObject, JSValue, JsCell};
 use bun_sys::Error as SysError;
 
+use crate::generated_classes::js_BytesInternalReadableStreamSource as js_source;
 use crate::webcore::streams::{self, BufferAction, IntoArray};
 use crate::webcore::{DrainResult, SinkHandle, blob, readable_stream};
 
@@ -24,6 +25,7 @@ pub struct ByteStream {
     pub(crate) buffer: JsCell<Vec<u8>>,
     pub(crate) has_received_last_chunk: Cell<bool>,
     pub(crate) pending: JsCell<streams::Pending>,
+    pending_error: JsCell<PendingError>,
     pub(crate) done: Cell<bool>,
     /// Borrowed view into a JS `Uint8Array` passed from `on_pull`; kept alive by `pending_value`.
     // Raw fat slice ptr because the backing store is JS-heap-owned and rooted via
@@ -49,6 +51,7 @@ impl Default for ByteStream {
                 result: streams::Result::Done,
                 ..Default::default()
             }),
+            pending_error: JsCell::new(PendingError::None),
             done: Cell::new(false),
             pending_buffer: Cell::new(Self::empty_pending_buffer()),
             pending_value: JsCell::new(StrongOptional::empty()),
@@ -62,6 +65,15 @@ impl Default for ByteStream {
     }
 }
 
+/// A producer error that no reader has taken yet ([`ByteStream::take_pending_error`]).
+enum PendingError {
+    None,
+    System(Box<SysError>),
+    AbortReason(jsc::CommonAbortReason),
+    /// In the wrapper's visited `pendingError` slot. A `Strong` would leak an error that references its own Response.
+    InWrapperSlot,
+}
+
 /// ReadableStream source backed by a ByteStream.
 pub(crate) type Source = readable_stream::NewSource<ByteStream>;
 
@@ -677,10 +689,7 @@ impl ByteStream {
                     buf.extend_from_slice(chunk);
                     self.buffer.set(buf);
                 }
-                streams::Result::Err(err) => {
-                    self.pending
-                        .with_mut(|p| p.result = streams::Result::Err(err));
-                }
+                streams::Result::Err(err) => self.set_pending_error(err),
                 streams::Result::Done => {}
                 _ => unreachable!(),
             }
@@ -707,8 +716,7 @@ impl ByteStream {
                     b.clear();
                     b.shrink_to_fit();
                 });
-                self.pending
-                    .with_mut(|p| p.result = streams::Result::Err(err));
+                self.set_pending_error(err);
             }
             streams::Result::Done => {}
             // We don't support the rest of these yet
@@ -718,6 +726,46 @@ impl ByteStream {
         Ok(())
     }
 
+    fn set_pending_error(&self, err: streams::StreamError) {
+        self.clear_pending_error();
+        let pending_error = match err {
+            streams::StreamError::Error(err) => PendingError::System(Box::new(err)),
+            streams::StreamError::AbortReason(reason) => PendingError::AbortReason(reason),
+            streams::StreamError::JSValue(value) => {
+                let source = self.parent_const();
+                // Every reader comes through the wrapper: once it is collected, nothing can take the error.
+                match source.this_jsvalue.try_get() {
+                    Some(wrapper) => {
+                        js_source::pending_error_set_cached(
+                            wrapper,
+                            source.global_this(),
+                            value.get().unwrap_or(JSValue::UNDEFINED),
+                        );
+                        PendingError::InWrapperSlot
+                    }
+                    None => PendingError::None,
+                }
+            }
+        };
+        self.pending_error.set(pending_error);
+    }
+
+    fn clear_pending_error(&self) {
+        if matches!(
+            self.pending_error.replace(PendingError::None),
+            PendingError::InWrapperSlot
+        ) {
+            let source = self.parent_const();
+            if let Some(wrapper) = source.this_jsvalue.try_get() {
+                js_source::pending_error_set_cached(
+                    wrapper,
+                    source.global_this(),
+                    JSValue::UNDEFINED,
+                );
+            }
+        }
+    }
+
     fn set_value(&self, view: JSValue) {
         bun_jsc::mark_binding!();
         let global = self.parent_const().global_this();
@@ -774,10 +822,8 @@ 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(_)) {
-                return self
-                    .pending
-                    .with_mut(|p| core::mem::replace(&mut p.result, streams::Result::Done));
+            if let Some(err) = self.take_pending_error() {
+                return streams::Result::Err(err);
             }
             return streams::Result::Done;
         }
@@ -800,7 +846,7 @@ impl ByteStream {
             let global = self.parent_const().global_this();
             streams::StreamError::JSValue(StrongOptional::create(reason, global))
         } else {
-            // Kept for a reader that already holds this source and pulls later. A stored `reason` would be a GC root.
+            // Kept for a reader that already holds this source and pulls later.
             streams::StreamError::AbortReason(jsc::CommonAbortReason::UserAbort)
         }));
     }
@@ -910,16 +956,20 @@ impl ByteStream {
 
     /// Take a pre-attach `StreamResult::Err` stashed by [`Self::append`].
     pub fn take_pending_error(&self) -> Option<streams::StreamError> {
-        self.pending.with_mut(|p| {
-            if matches!(p.result, streams::Result::Err(_)) {
-                match core::mem::replace(&mut p.result, streams::Result::Done) {
-                    streams::Result::Err(e) => Some(e),
-                    _ => None,
-                }
-            } else {
-                None
+        match self.pending_error.replace(PendingError::None) {
+            PendingError::None => None,
+            PendingError::System(err) => Some(streams::StreamError::Error(*err)),
+            PendingError::AbortReason(reason) => Some(streams::StreamError::AbortReason(reason)),
+            PendingError::InWrapperSlot => {
+                let source = self.parent_const();
+                let wrapper = source.this_jsvalue.try_get()?;
+                let global = source.global_this();
+                let err =
+                    StrongOptional::create(js_source::pending_error_get_cached(wrapper)?, global);
+                js_source::pending_error_set_cached(wrapper, global, JSValue::UNDEFINED);
+                Some(streams::StreamError::JSValue(err))
             }
-        })
+        }
     }
 
     pub(crate) fn to_any_blob(&self) -> Option<blob::Any> {
@@ -930,6 +980,7 @@ impl ByteStream {
                 p.result.release();
                 p.result = streams::Result::Done;
             });
+            self.clear_pending_error();
             self.parent_const().is_closed.set(true);
             return Some(blob::Any::InternalBlob(blob::Internal {
                 bytes: buffer,
@@ -949,10 +1000,9 @@ impl ByteStream {
             return Err(cx.global().throw(format_args!("Cannot buffer value twice")));
         }
 
-        if let streams::Result::Err(err) = &self.pending.get().result {
+        if let Some(err) = self.take_pending_error() {
             let err_js = err.to_js(cx.global());
             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();

src/runtime/webcore/s3/client.rs

@@ -840,23 +840,10 @@ pub(crate) fn upload_stream(
         // unrolled here.
         ReadableStreamPtr::Bytes(_) => {
             // BACKREF: see `Source::bytes()` — payload live while the
-            // ReadableStream JS wrapper is rooted. R-2: `pending` is `JsCell`.
+            // ReadableStream JS wrapper is rooted.
             let stream = readable_stream.ptr.bytes().expect("matched Bytes");
-            if matches!(
-                stream.pending.get().result,
-                crate::webcore::streams::StreamResult::Err(_)
-            ) {
+            if let Some(err) = stream.take_pending_error() {
                 // we got an error, fail early
-                let err = match stream.pending.with_mut(|p| {
-                    core::mem::replace(&mut p.result, crate::webcore::streams::StreamResult::Done)
-                }) {
-                    crate::webcore::streams::StreamResult::Err(err) => err,
-                    _ => unreachable!(),
-                };
-                stream.pending.set(crate::webcore::streams::Pending {
-                    result: crate::webcore::streams::StreamResult::Done,
-                    ..Default::default()
-                });
                 let js_err = err.to_js(cx.global());
                 js_err.ensure_still_alive();
                 return Ok(bun_jsc::JSPromise::rejected_promise(cx.global(), js_err).to_js());

test/js/workerd/html-rewriter-leak.test.ts

@@ -1076,6 +1076,169 @@ describe("handlers that nothing else references stay alive", () => {
   });
 });
 
+// A handler's error reaches the output stream while nothing can take it: the
+// stream exists but nothing reads it, or its reader has a backlog. The stream's
+// native source used to keep that error in a Strong until the next read. A
+// Strong is a GC root, so an error that references its own Response closed a
+// cycle through a root: the Response, both streams, the transform and its
+// lol-html parser were never collected. The error now sits in a visited slot
+// of the source's wrapper.
+describe("a handler error that nothing has read yet", () => {
+  const N = 60;
+  const encoder = new TextEncoder();
+  const turn = () => new Promise<void>(resolve => setImmediate(resolve));
+
+  // A rewrite of a streamed input. Its `p` handler fails, by default with an error that references the output Response.
+  const start = (
+    thrown = (response: Response): unknown => Object.assign(new Error("handler failed"), { response }),
+  ) => {
+    let controller!: ReadableStreamDefaultController;
+    let handled = Promise.withResolvers<void>();
+    const holder: { response?: Response } = {};
+    const response = new HTMLRewriter()
+      .on("b", { element: () => handled.resolve() })
+      .on("p", {
+        element() {
+          handled.resolve();
+          throw thrown(holder.response!);
+        },
+      })
+      .transform(new Response(new ReadableStream({ start: c => void (controller = c) })));
+    holder.response = response;
+    return {
+      response,
+      // Resolves once the handler for `html` ran and the native call that ran it has returned:
+      // the chunk's output, or the handler's error, is in the output stream.
+      async send(html: string) {
+        handled = Promise.withResolvers();
+        controller.enqueue(encoder.encode(html));
+        await handled.promise;
+        await turn();
+      },
+    };
+  };
+
+  // How the output is held when the handler fails. In both, the error finds no read to reject.
+  const failures = {
+    "an output stream that nothing reads": async () => {
+      const rewrite = start();
+      const body = rewrite.response.body!;
+      await rewrite.send("<p>fails</p>");
+      return { ...rewrite, body, reader: undefined };
+    },
+    // The first chunk fills the reader's queue, so the stream stops asking its source for more.
+    "an output stream whose reader has a backlog": async () => {
+      const rewrite = start();
+      const body = rewrite.response.body!;
+      const reader = body.getReader();
+      await rewrite.send("<b>first</b>");
+      await rewrite.send("<b>second</b>");
+      await rewrite.send("<p>fails</p>");
+      return { ...rewrite, body, reader };
+    },
+  };
+
+  const counts = async () => {
+    // A collected transform releases its pipe from a task, and the pipe holds more.
+    for (let round = 0; round < 3; round++) {
+      Bun.gc(true);
+      await turn();
+    }
+    const { objectTypeCounts, protectedObjectTypeCounts } = heapStats();
+    return {
+      responses: objectTypeCounts.Response ?? 0,
+      protectedErrors: protectedObjectTypeCounts.Error ?? 0,
+    };
+  };
+
+  test.each(Object.entries(failures))("does not pin its Response: %s", async (_, fail) => {
+    for (let i = 0; i < 10; i++) await fail();
+    const before = await counts();
+    for (let i = 0; i < N; i++) await fail();
+    const after = await counts();
+
+    // Unfixed: N protected errors, with N Responses behind them.
+    expect(after.protectedErrors - before.protectedErrors).toBeLessThan(N / 4);
+    expect(after.responses - before.responses).toBeLessThan(N / 4);
+  });
+
+  // The other half of a visited slot: the error has to stay alive until something takes it.
+  // These pass before the change too (a Strong cannot die): they guard the slot.
+  describe("is still the one its next reader gets, after a collection", () => {
+    // A full collection, then garbage in the cell sizes of what the handlers throw, so that
+    // an error collected by mistake is reused and cannot reach its reader by luck.
+    const churn = () => {
+      Bun.gc(true);
+      const junk: unknown[] = [];
+      for (let i = 0; i < 5_000; i++) {
+        junk.push(Object.assign(new Error("junk"), { response: junk }), ["junk", i].join(" "));
+      }
+    };
+    const describeError = (error: unknown, response: Response) => ({
+      isError: error instanceof Error,
+      message: (error as Error)?.message,
+      referencesItsResponse: (error as { response?: unknown })?.response === response,
+    });
+    const theHandlerError = { isError: true, message: "handler failed", referencesItsResponse: true };
+
+    test("text()", async () => {
+      const { response } = await failures["an output stream that nothing reads"]();
+      churn();
+      expect(describeError(await response.text().catch(error => error), response)).toEqual(theHandlerError);
+    });
+
+    test("Bun.readableStreamToBytes()", async () => {
+      const { response, body } = await failures["an output stream that nothing reads"]();
+      churn();
+      expect(describeError(await Bun.readableStreamToBytes(body).catch(error => error), response)).toEqual(
+        theHandlerError,
+      );
+    });
+
+    test("read(), after the chunk that was already queued", async () => {
+      const { response, reader } = await failures["an output stream whose reader has a backlog"]();
+      churn();
+      expect(new TextDecoder().decode((await reader!.read()).value)).toBe("<b>first</b>");
+      expect(describeError(await reader!.read().catch(error => error), response)).toEqual(theHandlerError);
+    });
+
+    test("a second rewriter that takes the stream", async () => {
+      const { response, body } = await failures["an output stream that nothing reads"]();
+      churn();
+      const second = new HTMLRewriter().on("b", { element() {} }).transform(new Response(body));
+      expect(describeError(await second.text().catch(error => error), response)).toEqual(theHandlerError);
+    });
+
+    // Script can throw anything. A weak handle, which only an object can have, would not do for the slot.
+    test("text(), when the handler threw a string", async () => {
+      const rewrite = start(() => ["a", "thrown", "string"].join(" "));
+      void rewrite.response.body;
+      await rewrite.send("<p>fails</p>");
+      churn();
+      expect(await rewrite.response.text().catch(error => error)).toBe("a thrown string");
+    });
+
+    // The upload looks for the error before it makes a request.
+    test("an S3 upload of the stream", async () => {
+      let requests = 0;
+      await using server = Bun.serve({
+        port: 0,
+        fetch: () => (requests++, new Response(null, { status: 500 })),
+      });
+      const file = new Bun.S3Client({
+        accessKeyId: "test",
+        secretAccessKey: "test",
+        bucket: "test",
+        endpoint: server.url.href,
+      }).file("key");
+      const { response, body } = await failures["an output stream that nothing reads"]();
+      churn();
+      expect(describeError(await file.write(body).catch(error => error), response)).toEqual(theHandlerError);
+      expect(requests).toBe(0);
+    });
+  });
+});
+
 const withoutAsanWarning = (stderr: string) =>
   stderr
     .split("\n")