diff --git a/src/runtime/api/html_rewriter.rs b/src/runtime/api/html_rewriter.rs
index 1baf57fbdb9e..768830469d96 100644
--- a/src/runtime/api/html_rewriter.rs
+++ b/src/runtime/api/html_rewriter.rs
@@ -755,6 +755,9 @@ impl BufferOutputSink {
// ref taken for the in-flight bufferer.
unsafe { BufferOutputSink::deref(sink) };
return Ok(match buffering_error {
+ // The bufferer entered JS and it threw: that exception is
+ // already pending on the VM, so surface it instead of masking it.
+ crate::Error::JSError => return Err(jsc::JsError::Thrown),
crate::Error::StreamAlreadyUsed => {
let err = system_error(
"ERR_STREAM_ALREADY_FINISHED",
diff --git a/src/runtime/error.rs b/src/runtime/error.rs
index 84720687203d..18ef55cf1d27 100644
--- a/src/runtime/error.rs
+++ b/src/runtime/error.rs
@@ -30,8 +30,6 @@ pub enum Error {
StreamAlreadyUsed,
#[error("InvalidStream")]
InvalidStream,
- #[error("UnsupportedStreamType")]
- UnsupportedStreamType,
#[error("JSError")]
JSError,
#[error("ERR_TLS_CERT_ALTNAME_INVALID")]
@@ -609,7 +607,6 @@ impl Error {
Self::FmtError => "FmtError",
Self::StreamAlreadyUsed => "StreamAlreadyUsed",
Self::InvalidStream => "InvalidStream",
- Self::UnsupportedStreamType => "UnsupportedStreamType",
Self::JSError => "JSError",
Self::ERR_TLS_CERT_ALTNAME_INVALID => "ERR_TLS_CERT_ALTNAME_INVALID",
Self::RequestBodyNotReusable => "RequestBodyNotReusable",
diff --git a/src/runtime/webcore/Body.rs b/src/runtime/webcore/Body.rs
index e2e1f0049767..c62f23b74df1 100644
--- a/src/runtime/webcore/Body.rs
+++ b/src/runtime/webcore/Body.rs
@@ -1,6 +1,5 @@
//! https://developer.mozilla.org/en-US/docs/Web/API/Body
-use bun_collections::VecExt;
use core::ffi::c_void;
use core::ptr::NonNull;
@@ -18,7 +17,6 @@ use bun_http_types::MimeType::MimeType;
use crate::jsc::HTTPHeaderName;
pub use crate::webcore::InternalBlob;
use crate::webcore::form_data::AsyncFormDataExt as _;
-use crate::webcore::sink::{self, ArrayBufferSink};
use bun_core::{MutableString, String as BunString, ZigString};
use bun_core::{WTFStringImpl, WTFStringImplExt as _, WTFStringImplStruct};
use bun_jsc::ZigStringJsc as _;
@@ -1635,9 +1633,6 @@ impl Value {
// JSC-integration: extract / BodyMixin (host-fn methods) / ValueBufferer.
// ────────────────────────────────────────────────────────────────────────────
-// `sink::JSSink` is a free generic (inherent associated types are unstable).
-type ArrayBufferJSSink = sink::JSSink;
-
// https://github.com/WebKit/webkit/blob/main/Source/WebCore/Modules/fetch/FetchBody.cpp#L45
pub(crate) fn extract(global_this: &JSGlobalObject, value: JSValue) -> JsResult {
let body_value = Value::from_js(global_this, value)?;
@@ -2235,7 +2230,6 @@ pub struct ValueBufferer<'a> {
pub ctx: *mut c_void,
pub on_finished_buffering: ValueBuffererCallback,
- pub js_sink: Option>,
pub byte_stream: Option>,
// readable stream strong ref to keep byte stream alive
pub readable_stream_ref: webcore::readable_stream::Strong,
@@ -2254,13 +2248,6 @@ impl<'a> Drop for ValueBufferer<'a> {
bun_ptr::BackRef::from(byte_stream).unpipe_without_deref();
}
self.readable_stream_ref.deinit();
-
- if let Some(mut buffer_stream) = self.js_sink.take() {
- buffer_stream.detach_self(self.global);
- // The wrapper is a `Box>`; dropping it
- // frees the box and runs `Vec`'s Drop.
- drop(buffer_stream);
- }
}
}
@@ -2273,7 +2260,6 @@ impl<'a> ValueBufferer<'a> {
Self {
ctx,
on_finished_buffering: on_finish,
- js_sink: None,
byte_stream: None,
readable_stream_ref: Default::default(),
global,
@@ -2415,7 +2401,7 @@ impl<'a> ValueBufferer<'a> {
let Some(sink) = Self::take_ctx(args.ptr[args.len - 1]) else {
return Ok(JSValue::UNDEFINED);
};
- sink.handle_resolve_stream(true);
+ sink.handle_resolve_stream(args.ptr[0], true);
Ok(JSValue::UNDEFINED)
}
@@ -2433,12 +2419,6 @@ impl<'a> ValueBufferer<'a> {
}
fn handle_reject_stream(&mut self, err: JSValue, is_async: bool) {
- if let Some(mut wrapper) = self.js_sink.take() {
- wrapper.detach_self(self.global);
- // see `Drop` impl — dropping the Box frees the wrapper
- // and runs `Vec`'s Drop.
- drop(wrapper);
- }
// `jsc::strong::Optional` owns a GC root; `ptr::read`-duplicating it would
// double-deinit. Transfer the single owner directly to the callback; the callback
// (or its returned `ValueError`'s Drop) is responsible for releasing it.
@@ -2446,15 +2426,81 @@ impl<'a> ValueBufferer<'a> {
(self.on_finished_buffering)(self.ctx, b"", Some(ValueError::JSValue(ref_)), is_async);
}
- fn handle_resolve_stream(&mut self, is_async: bool) {
- if let Some(wrapper) = &self.js_sink {
- let bytes = wrapper.sink.bytes.slice();
- bun_core::scoped_log!(BodyValueBufferer, "handleResolveStream {}", bytes.len());
- (self.on_finished_buffering)(self.ctx, bytes, None, is_async);
- } else {
- bun_core::scoped_log!(BodyValueBufferer, "handleResolveStream no sink");
- (self.on_finished_buffering)(self.ctx, b"", None, is_async);
+ /// `resolved_value` is what `readableStreamToArrayBuffer` fulfilled with: an
+ /// `ArrayBuffer`, or a `Uint8Array` when the stream yielded a single string
+ /// chunk.
+ ///
+ /// The bytes are copied into `stream_buffer` first. The single-chunk fast
+ /// path hands back the *user's own* ArrayBuffer, and the consumer tokenizes
+ /// the slice in place while re-entering user JS, which can detach it
+ /// (`ArrayBuffer.prototype.transfer`) and free the backing store mid-read.
+ fn handle_resolve_stream(&mut self, resolved_value: JSValue, is_async: bool) {
+ // Only the `Source::Bytes` pipe appends to `stream_buffer`, and a bufferer
+ // drives exactly one source, so this is the sole writer on this path.
+ debug_assert!(self.stream_buffer.list.is_empty());
+ if let Some(array_buffer) = resolved_value.as_array_buffer(self.global) {
+ let _ = self.stream_buffer.write(array_buffer.slice());
+ }
+ let bytes = self.stream_buffer.list.as_slice();
+ bun_core::scoped_log!(BodyValueBufferer, "handleResolveStream {}", bytes.len());
+ (self.on_finished_buffering)(self.ctx, bytes, None, is_async);
+ }
+
+ /// Buffer a JS-backed stream (`new ReadableStream({...})` or a `type:
+ /// "direct"` stream) through `readableStreamToArrayBuffer` — the same path
+ /// `new Response(stream).arrayBuffer()` takes. Only the JS runtime knows how
+ /// to drive these sources; `byte_stream`'s native pipe cannot.
+ fn buffer_js_readable_stream(&mut self, stream: ReadableStream) -> crate::Result<()> {
+ let global = self.global;
+
+ // The builtin's C++ wrapper returns under a `ThrowScope`, so its
+ // simulated throw has to be observed here; a bare `is_empty()` check is
+ // invisible to `validateExceptionChecks` and trips the next scope.
+ let promise_value = {
+ bun_jsc::validation_scope!(scope, global);
+ let value = global.readable_stream_to_array_buffer(stream.value);
+ scope.assert_exception_presence_matches(value.is_empty());
+ value
+ };
+ // Release the GC root `buffer_locked_body_value` took. The builtin owns
+ // the stream through its own reader now, and whatever drives the source
+ // keeps the returned promise alive. Rooting it here also roots the
+ // promise chain — and so the `NativePromiseContext` cell — forever, so
+ // an abandoned transform could never be collected. See `set_promise`.
+ self.readable_stream_ref.deinit();
+ if promise_value.is_empty() {
+ // The builtin threw (e.g. the stream yielded a chunk that is neither
+ // a string nor a view); the exception is pending on the VM.
+ return Err(crate::Error::JSError);
+ }
+ promise_value.ensure_still_alive();
+
+ // Unreachable: the C++ wrapper throws a TypeError when the builtin
+ // hands back anything other than a promise, caught by `is_empty` above.
+ let Some(promise) = promise_value.as_any_promise() else {
+ return Err(crate::Error::InvalidStream);
+ };
+ match promise.unwrap(global.vm(), jsc::PromiseUnwrapMode::MarkHandled) {
+ jsc::PromiseResult::Pending => {
+ // The +1 the owner took for this in-flight buffering doubles as
+ // the cell's ref: settling consumes it via `on_finished_buffering`,
+ // and a promise GC'd without settling releases it through
+ // `Bun__NativePromiseContext__destroy`.
+ let cell = crate::api::NativePromiseContext::create(
+ global,
+ std::ptr::from_mut::(self),
+ );
+ promise_value.then_with_value(
+ global,
+ cell,
+ Bun__BodyValueBufferer__onResolveStream,
+ Bun__BodyValueBufferer__onRejectStream,
+ );
+ }
+ jsc::PromiseResult::Fulfilled(value) => self.handle_resolve_stream(value, false),
+ jsc::PromiseResult::Rejected(err) => self.handle_reject_stream(err, false),
}
+ Ok(())
}
fn buffer_locked_body_value(
@@ -2499,9 +2545,7 @@ impl<'a> ValueBufferer<'a> {
| webcore::readable_stream::Source::File(_) => unreachable!(),
webcore::readable_stream::Source::JavaScript
| webcore::readable_stream::Source::Direct => {
- // this is broken right now
- // return self.create_js_sink(stream);
- return Err(crate::Error::UnsupportedStreamType);
+ return self.buffer_js_readable_stream(stream);
}
webcore::readable_stream::Source::Bytes(byte_stream_ptr) => {
// BACKREF: see `Source::bytes()` — payload owned by the
diff --git a/test/js/workerd/html-rewriter.test.js b/test/js/workerd/html-rewriter.test.js
index 6ffc837062ed..5f297d3e482f 100644
--- a/test/js/workerd/html-rewriter.test.js
+++ b/test/js/workerd/html-rewriter.test.js
@@ -1,3 +1,4 @@
+import { heapStats } from "bun:jsc";
import { afterAll, beforeAll, describe, expect, it } from "bun:test";
import { once } from "events";
import fs from "fs";
@@ -303,6 +304,339 @@ describe("HTMLRewriter", () => {
});
});
+ describe("transform() accepts a JavaScript-backed ReadableStream body", () => {
+ // https://github.com/oven-sh/bun/issues/14216
+ // https://github.com/oven-sh/bun/issues/11758
+ const encode = s => new TextEncoder().encode(s);
+
+ function rewriter() {
+ return new HTMLRewriter().on("p", {
+ element(element) {
+ element.setInnerContent("bye");
+ },
+ });
+ }
+
+ function streamOf(...chunks) {
+ return new ReadableStream({
+ start(controller) {
+ for (const chunk of chunks) controller.enqueue(chunk);
+ controller.close();
+ },
+ });
+ }
+
+ it("single Uint8Array chunk", async () => {
+ const transformed = rewriter().transform(new Response(streamOf(encode("hi
"))));
+ expect(await transformed.text()).toBe("bye
");
+ });
+
+ it("single string chunk", async () => {
+ // readableStreamToArrayBuffer hands a single string chunk back as a
+ // Uint8Array view rather than an ArrayBuffer, so this exercises a
+ // different branch than the binary chunk above.
+ const transformed = rewriter().transform(new Response(streamOf("hi
")));
+ expect(await transformed.text()).toBe("bye
");
+ });
+
+ it("an element split across chunk boundaries", async () => {
+ const transformed = rewriter().transform(
+ new Response(streamOf(encode("h"), encode("i
two
"))),
+ );
+ expect(await transformed.text()).toBe("bye
bye
");
+ });
+
+ it("mixed string and binary chunks", async () => {
+ const transformed = rewriter().transform(new Response(streamOf("a
", encode("b
"))));
+ expect(await transformed.text()).toBe("bye
bye
");
+ });
+
+ it("empty stream", async () => {
+ let endCalls = 0;
+ const transformed = new HTMLRewriter()
+ .onDocument({
+ end() {
+ endCalls++;
+ },
+ })
+ .transform(new Response(streamOf()));
+ expect(await transformed.text()).toBe("");
+ expect(endCalls).toBe(1);
+ });
+
+ it("a direct stream", async () => {
+ const body = new ReadableStream({
+ type: "direct",
+ pull(controller) {
+ controller.write("hi
");
+ controller.close();
+ },
+ });
+ expect(await rewriter().transform(new Response(body)).text()).toBe("bye
");
+ });
+
+ it("a stream that only produces chunks after transform() returns", async () => {
+ // start() stays pending across transform(), so the rewriter has to take
+ // the asynchronous path instead of buffering everything up front.
+ const { promise: gate, resolve: openGate } = Promise.withResolvers();
+ const body = new ReadableStream({
+ async start(controller) {
+ await gate;
+ controller.enqueue(encode("hi
"));
+ controller.close();
+ },
+ });
+ const text = rewriter().transform(new Response(body)).text();
+ openGate();
+ expect(await text).toBe("bye
");
+ });
+
+ it("every way of reading the transformed response", async () => {
+ const read = {
+ text: response => response.text(),
+ arrayBuffer: async response => new TextDecoder().decode(await response.arrayBuffer()),
+ bytes: async response => new TextDecoder().decode(await response.bytes()),
+ blob: response => response.blob().then(blob => blob.text()),
+ json: response => response.json().then(value => JSON.stringify(value)),
+ getReader: async response => {
+ const reader = response.body.getReader();
+ const parts = [];
+ for (let chunk = await reader.read(); !chunk.done; chunk = await reader.read()) {
+ parts.push(new TextDecoder().decode(chunk.value));
+ }
+ return parts.join("");
+ },
+ readableStreamToText: response => Bun.readableStreamToText(response.body),
+ };
+
+ const html = 'hi
there
';
+ const expected = 'bye
bye
';
+ for (const [name, consume] of Object.entries(read)) {
+ const transformed = rewriter().transform(new Response(streamOf(encode(html))));
+ if (name === "json") {
+ // Not valid JSON — but it must fail as a JSON parse error, which
+ // still proves the transformed bytes reached the parser.
+ await expect(consume(transformed)).rejects.toThrow(/JSON/i);
+ continue;
+ }
+ expect({ [name]: await consume(transformed) }).toEqual({ [name]: expected });
+ }
+ });
+
+ it("element handlers observe the streamed document", async () => {
+ const tags = [];
+ const transformed = new HTMLRewriter()
+ .on("*", {
+ element(element) {
+ tags.push(element.tagName);
+ },
+ })
+ .transform(new Response(streamOf(encode(""))));
+ expect(await transformed.text()).toBe("");
+ expect(tags).toEqual(["div", "p"]);
+ });
+
+ it("a stream that errors rejects the transformed body", async () => {
+ const body = new ReadableStream({
+ start(controller) {
+ controller.enqueue(encode("hi
"));
+ controller.error(new Error("upstream boom"));
+ },
+ });
+ const transformed = rewriter().transform(new Response(body));
+ // Must reject rather than resolve with the truncated document.
+ await expect(transformed.text()).rejects.toThrow("upstream boom");
+ });
+
+ it("a stream that errors after transform() returns rejects the transformed body", async () => {
+ const { promise: gate, resolve: openGate } = Promise.withResolvers();
+ const body = new ReadableStream({
+ async start(controller) {
+ await gate;
+ controller.error(new Error("late boom"));
+ },
+ });
+ const text = rewriter().transform(new Response(body)).text();
+ openGate();
+ await expect(text).rejects.toThrow("late boom");
+ });
+
+ it("a chunk that is neither a string nor a view throws", async () => {
+ const transform = () => rewriter().transform(new Response(streamOf(42)));
+ // The underlying TypeError must surface, not "Failed to pipe stream".
+ expect(transform).toThrow(TypeError);
+ expect(transform).not.toThrow("Failed to pipe stream");
+ });
+
+ it("reusing the transformed response's source stream throws", async () => {
+ const response = new Response(streamOf(encode("hi
")));
+ expect(await rewriter().transform(response).text()).toBe("bye
");
+ expect(() => rewriter().transform(response)).toThrow("Response body already used");
+ });
+
+ it("does not rewrite out of the source buffer a handler can detach", async () => {
+ // readableStreamToArrayBuffer's single-chunk fast path returns the user's
+ // own ArrayBuffer. lol-html tokenizes its input in place and dispatches
+ // handlers mid-scan, so a handler that mutates (or transfers, then frees)
+ // that buffer must not be able to reach into the bytes still being parsed.
+ let chunk;
+ const body = new ReadableStream({
+ start(controller) {
+ chunk = encode("xy");
+ controller.enqueue(chunk);
+ controller.close();
+ },
+ });
+ const transformed = new HTMLRewriter()
+ .on("a", {
+ element() {
+ // overwrite "" (not yet tokenized) with ""
+ chunk.set(encode("qqq"), 9);
+ // and drop the backing store the rewriter would be reading
+ chunk.buffer.transfer();
+ Bun.gc(true);
+ },
+ })
+ .transform(new Response(body));
+ expect(await transformed.text()).toBe("xy");
+ });
+
+ // The bufferer drops its GC root on the source once the builtin owns it, so
+ // a live transform is kept alive only by whatever can still settle the
+ // stream. That holds because settling needs the controller, and the
+ // controller holds the stream. Each case hides the stream from userland and
+ // collects hard before letting it finish.
+ describe("a source the bufferer no longer roots still completes", () => {
+ const cases = {
+ "controller held only by a timer": () =>
+ new ReadableStream({
+ start(controller) {
+ setTimeout(() => {
+ controller.enqueue(encode("hi
"));
+ controller.close();
+ }, 1);
+ },
+ }),
+ "controller escaping to an outer scope": () => {
+ let escaped;
+ const stream = new ReadableStream({
+ start(controller) {
+ escaped = controller;
+ },
+ });
+ queueMicrotask(() => {
+ escaped.enqueue(encode("hi
"));
+ escaped.close();
+ });
+ return stream;
+ },
+ "controller reachable only from a pending pull": () =>
+ new ReadableStream({
+ type: "direct",
+ async pull(controller) {
+ await Bun.sleep(1);
+ controller.write("hi
");
+ controller.close();
+ },
+ }),
+ };
+
+ for (const [name, makeStream] of Object.entries(cases)) {
+ it(name, async () => {
+ const transformed = rewriter().transform(new Response(makeStream()));
+ // Collect aggressively while the source is still in flight.
+ for (let i = 0; i < 3; i++) {
+ Bun.gc(true);
+ await Bun.sleep(1);
+ }
+ expect(await transformed.text()).toBe("bye
");
+ });
+ }
+ });
+
+ it("does not leak a transform whose source stream never settles", async () => {
+ // The bufferer must not hold a Strong to the source: the stream reaches
+ // the promise chain, which reaches the NativePromiseContext cell holding
+ // the sink's refcount, so rooting it means an abandoned transform can
+ // never be collected.
+ const abandon = () => {
+ const body = new ReadableStream({
+ pull() {
+ return new Promise(() => {});
+ },
+ });
+ rewriter().transform(new Response(body));
+ };
+ // BufferOutputSink's ref is released on the event loop, so drain between
+ // collections rather than measuring straight after Bun.gc().
+ const settle = async () => {
+ for (let i = 0; i < 5; i++) {
+ Bun.gc(true);
+ await Bun.sleep(1);
+ }
+ };
+ const live = () => heapStats().objectTypeCounts.Response ?? 0;
+
+ for (let i = 0; i < 10; i++) abandon(); // warm up lazy structures
+ await settle();
+ const before = live();
+ for (let i = 0; i < 200; i++) abandon();
+ await settle();
+ // Pre-fix this grew by exactly 200 (one leaked sink per transform, each
+ // pinning its output Response).
+ expect(live() - before).toBeLessThan(20);
+ });
+
+ // Resolves with "" instead: the `.body` getter builds a ByteStream the
+ // producer is never told about, so done() closes it empty. Pre-existing and
+ // not specific to JS sources — a fetch body that is still mid-stream when
+ // transform() returns does the same thing on main (#19305, and #6068 for
+ // the Bun.serve shape, which hangs). Un-skip once the output side is fixed.
+ it.todo(".body of a transform whose source is still pending", async () => {
+ const { promise: gate, resolve: openGate } = Promise.withResolvers();
+ const body = new ReadableStream({
+ async start(controller) {
+ await gate;
+ controller.enqueue(encode("hi
"));
+ controller.close();
+ },
+ });
+ const transformed = rewriter().transform(new Response(body));
+ const reader = transformed.body.getReader();
+ openGate();
+ const parts = [];
+ for (let chunk = await reader.read(); !chunk.done; chunk = await reader.read()) {
+ parts.push(new TextDecoder().decode(chunk.value));
+ }
+ expect(parts.join("")).toBe("bye
");
+ });
+
+ it("served over Bun.serve", async () => {
+ using server = Bun.serve({
+ port: 0,
+ fetch() {
+ const body = new ReadableStream({
+ start(controller) {
+ controller.enqueue(encode("hello world"));
+ controller.close();
+ },
+ });
+ return new HTMLRewriter()
+ .on("b", {
+ element(element) {
+ element.before("", { html: true });
+ element.after("
", { html: true });
+ element.removeAndKeepContent();
+ },
+ })
+ .transform(new Response(body, { headers: { "content-type": "text/html" } }));
+ },
+ });
+ const response = await fetch(server.url);
+ expect(await response.text()).toBe("hello world
");
+ });
+ });
+
it("HTMLRewriter: async replacement using fetch + Bun.serve", async () => {
await gcTick();
let content;
@@ -946,12 +1280,12 @@ const payloads = [
{
name: "direct",
data: getStream("direct", "none"),
- test: it.todo,
+ test: it,
},
{
name: "default",
data: getStream("default", "none"),
- test: it.todo,
+ test: it,
},
{
name: "file",