From 7762b0316ee4433d149ab701eedf41f71b51cee1 Mon Sep 17 00:00:00 2001 From: robobun <117481402+robobun@users.noreply.github.com> Date: Tue, 4 Aug 2026 06:33:57 +0000 Subject: [PATCH 1/2] TextEncoderStream native-sink: hand string bytes to sink write_latin1/write_utf16 When a native JSSink is attached, TextEncoderStreamEncoder__encodeIntoSink now passes the chunk's WTFStringImpl bytes straight to the sink via the sink's own write_latin1/write_utf16 instead of first transcoding into a per-encoder scratch Vec and then calling write_bytes. Every sink's write_latin1 already has an is_all_ascii fast path that forwards the input slice verbatim, so an all-ASCII 8-bit chunk is now zero-copy from the JSString all the way to the sink buffer (or the socket, for HTTPServerWritable with no buffered data). For 16-bit chunks the encoder trims a trailing lone lead surrogate into pending_lead_surrogate and hands the remainder to write_utf16; every sink's write_utf16 emits U+FFFD for mid-chunk lone surrogates exactly as the encoder did. Only a chunk that follows a carried lead surrogate still goes through the scratch buffer, so the prefix and body produce one Writable. SinkHandle grows write_latin1/write_utf16 that dispatch to each sink's implementation (RewriterPipe gains &self inherent methods so the shared-BackRef HTMLRewriter arm can reach them). The coerced JSString is now rooted across the sink write (to_js_string + EnsureStillAlive, as in JSSink::js_write), since HTMLRewriter content handlers and ArrayBufferSink's onReady are GC points. Tests: native-sink coverage for 8-bit (ASCII + Latin-1 non-ASCII) and 16-bit (mid-chunk + cross-chunk surrogate) chunks, an HTMLRewriter sink with ToString-coerced chunks and Bun.gc(true) inside the handler, and readable-side + native-sink backpressure tests for TextEncoderStream and readable-side backpressure tests for TextDecoderStream. --- src/runtime/api/html_rewriter.rs | 28 +-- src/runtime/webcore.rs | 62 +++++++ .../webcore/TextEncoderStreamEncoder.rs | 86 +++++++--- .../web/encoding/text-decoder-stream.test.ts | 25 +++ .../web/encoding/text-encoder-stream.test.ts | 160 ++++++++++++++++++ 5 files changed, 330 insertions(+), 31 deletions(-) diff --git a/src/runtime/api/html_rewriter.rs b/src/runtime/api/html_rewriter.rs index 0b70f34d3585..9a143e99ae3e 100644 --- a/src/runtime/api/html_rewriter.rs +++ b/src/runtime/api/html_rewriter.rs @@ -1252,6 +1252,22 @@ impl RewriterPipe { Writable::Owned(len) } + pub fn write_latin1(&self, data: &StreamResult) -> Writable { + let bytes = data.slice(); + if bun_core::strings::is_all_ascii(bytes) { + return self.write(data); + } + let mut buf = Vec::new(); + let _ = buf.write_latin1(bytes); + self.write(&StreamResult::Temporary(RawSlice::new(&buf))) + } + + pub fn write_utf16(&self, data: &StreamResult) -> Writable { + let mut buf = Vec::new(); + let _ = buf.write_utf16(data.slice16()); + self.write(&StreamResult::Temporary(RawSlice::new(&buf))) + } + /// `SinkHandle::end` entry — input EOF or terminal upstream error. pub fn end_from_stream(&self, err: Option) { // Detach via `detach_input_source` (not a bare `.set(None)`) so a @@ -1632,18 +1648,10 @@ impl crate::webcore::sink::JsSinkType for RewriterPipe { RewriterPipe::write(self, data) } fn write_utf16(&mut self, data: &StreamResult) -> Writable { - let mut buf = Vec::new(); - let _ = buf.write_utf16(data.slice16()); - RewriterPipe::write(self, &StreamResult::Temporary(RawSlice::new(&buf))) + RewriterPipe::write_utf16(self, data) } fn write_latin1(&mut self, data: &StreamResult) -> Writable { - let bytes = data.slice(); - if bun_core::strings::is_all_ascii(bytes) { - return RewriterPipe::write(self, data); - } - let mut buf = Vec::new(); - let _ = buf.write_latin1(bytes); - RewriterPipe::write(self, &StreamResult::Temporary(RawSlice::new(&buf))) + RewriterPipe::write_latin1(self, data) } fn end(&mut self, err: Option) -> bun_sys::Result<()> { self.end_from_stream(err.map(StreamError::Error)); diff --git a/src/runtime/webcore.rs b/src/runtime/webcore.rs index 9aff312bc5b3..5013dfaf289f 100644 --- a/src/runtime/webcore.rs +++ b/src/runtime/webcore.rs @@ -406,6 +406,68 @@ impl SinkHandle { } } + /// Route a Latin-1 chunk to the sink's own Latin-1→UTF-8 writer so the + /// conversion lands directly in the sink's buffer. Every sink's + /// `write_latin1` has an `is_all_ascii` fast path that forwards the input + /// slice verbatim, so an all-ASCII `Temporary` here is zero-copy end to end. + /// + /// `ServerResponse` is unreachable: only `sink::sink_handle_from_id` (a + /// native transform attached to a JSSink) produces callers of this path. + /// + /// SAFETY: same pointee-liveness invariant as [`Self::write`]. + pub fn write_latin1(&self, data: &streams::Result) -> streams::Writable { + match *self { + SinkHandle::None => streams::Writable::Done, + SinkHandle::ServerResponse(_) => { + debug_assert!(false, "ServerResponse is never a JSSink SinkID"); + streams::Writable::Done + } + // SAFETY: live backref; ByteStream clears sink before free. + SinkHandle::FetchRequestBody(mut p) => unsafe { p.get_mut() }.write_latin1(data), + // SAFETY: live backref; ByteStream clears sink before free. + SinkHandle::S3Upload(mut p) => unsafe { p.get_mut() }.write_latin1(data), + SinkHandle::FileSink(p) => p.write_latin1(data), + SinkHandle::HTMLRewriter(p) => p.write_latin1(data), + // SAFETY: live backref; transform detaches before the JSSink is finalized. + SinkHandle::HttpResponse(mut p) => unsafe { p.get_mut() }.write_latin1(data), + // SAFETY: live backref; transform detaches before the JSSink is finalized. + SinkHandle::HttpsResponse(mut p) => unsafe { p.get_mut() }.write_latin1(data), + // SAFETY: live backref; transform detaches before the JSSink is finalized. + SinkHandle::H3Response(mut p) => unsafe { p.get_mut() }.write_latin1(data), + // SAFETY: live backref; transform detaches before the JSSink is finalized. + SinkHandle::ArrayBuffer(mut p) => unsafe { p.get_mut() }.write_latin1(data), + } + } + + /// As [`Self::write_latin1`] but for a UTF-16 chunk (`data.slice16()`). + /// Every sink's `write_utf16` emits U+FFFD for unpaired surrogates, so the + /// caller is responsible for any cross-chunk surrogate carry. + /// + /// SAFETY: same pointee-liveness invariant as [`Self::write`]. + pub fn write_utf16(&self, data: &streams::Result) -> streams::Writable { + match *self { + SinkHandle::None => streams::Writable::Done, + SinkHandle::ServerResponse(_) => { + debug_assert!(false, "ServerResponse is never a JSSink SinkID"); + streams::Writable::Done + } + // SAFETY: live backref; ByteStream clears sink before free. + SinkHandle::FetchRequestBody(mut p) => unsafe { p.get_mut() }.write_utf16(data), + // SAFETY: live backref; ByteStream clears sink before free. + SinkHandle::S3Upload(mut p) => unsafe { p.get_mut() }.write_utf16(data), + SinkHandle::FileSink(p) => p.write_utf16(data), + SinkHandle::HTMLRewriter(p) => p.write_utf16(data), + // SAFETY: live backref; transform detaches before the JSSink is finalized. + SinkHandle::HttpResponse(mut p) => unsafe { p.get_mut() }.write_utf16(data), + // SAFETY: live backref; transform detaches before the JSSink is finalized. + SinkHandle::HttpsResponse(mut p) => unsafe { p.get_mut() }.write_utf16(data), + // SAFETY: live backref; transform detaches before the JSSink is finalized. + SinkHandle::H3Response(mut p) => unsafe { p.get_mut() }.write_utf16(data), + // SAFETY: live backref; transform detaches before the JSSink is finalized. + SinkHandle::ArrayBuffer(mut p) => unsafe { p.get_mut() }.write_utf16(data), + } + } + /// Signal end-of-stream (or terminal error) to the attached sink. /// /// SAFETY: same pointee-liveness invariant as [`Self::write`]. diff --git a/src/runtime/webcore/TextEncoderStreamEncoder.rs b/src/runtime/webcore/TextEncoderStreamEncoder.rs index b19cc103dc23..0ffdc38ee532 100644 --- a/src/runtime/webcore/TextEncoderStreamEncoder.rs +++ b/src/runtime/webcore/TextEncoderStreamEncoder.rs @@ -263,13 +263,19 @@ pub extern "C" fn TextEncoderStreamEncoder__flushForStream( /// that much memory for the life of the encoder. const SCRATCH_CAP: usize = 64 * 1024; -/// Native-sink transform step: encodes `chunk` into the encoder's reusable -/// scratch buffer and writes it straight to the sink, so a -/// `ByteStream → TextEncoderStream → JSSink` chain allocates no -/// `JSUint8Array` per chunk. Returns the sink's `write_bytes` result (see -/// nativeSinkWriteIsBackpressure for the backpressure-signal shapes), -/// `undefined` for an empty output, or `JSValue::zero` with the exception -/// pending on `global`. +/// Native-sink transform step: hands the chunk's `WTFStringImpl` bytes +/// straight to the sink via [`SinkHandle::write_latin1`] / +/// [`SinkHandle::write_utf16`], so a `ByteStream → TextEncoderStream → JSSink` +/// chain does the Latin-1/UTF-16 → UTF-8 conversion directly into the sink's +/// own buffer with no per-chunk scratch allocation (and zero-copy for an +/// all-ASCII 8-bit chunk — every sink's `write_latin1` has that fast path). +/// A carried lead surrogate from the previous chunk is the only case that +/// still needs a scratch buffer (the replacement / combined astral prefix +/// must be part of the same sink write so only one `Writable` is produced). +/// +/// Returns the sink's write result (see `nativeSinkWriteIsBackpressure` for +/// the backpressure-signal shapes), `undefined` for an empty output, or +/// `JSValue::zero` with the exception pending on `global`. #[unsafe(no_mangle)] #[allow(clippy::not_unsafe_ptr_arg_deref)] pub extern "C" fn TextEncoderStreamEncoder__encodeIntoSink( @@ -279,14 +285,63 @@ pub extern "C" fn TextEncoderStreamEncoder__encodeIntoSink( sink_id: u8, sink_ptr: *mut core::ffi::c_void, ) -> JSValue { - let Ok(str) = chunk.get_zig_string(global) else { + // WebIDL DOMString coercion runs user JS; hold the resulting JSString cell + // so its `WTFStringImpl` survives the sink write even if the sink itself + // runs user JS (HTMLRewriter content handlers) or reaches a GC point + // (ArrayBufferSink's onReady) — same pattern as `JSSink::js_write`. + let Ok(js_str) = chunk.to_js_string(global) else { return JSValue::ZERO; }; + let str = js_str.view(global); + let _keep = bun_jsc::EnsureStillAlive(js_str.to_js()); // SAFETY: `this` is the live encoder owned by the calling JS cell; taken // after the coercion so no user JS runs while the borrow is live. let this = unsafe { &*this }; - // Move the Vec out of the RefCell for the duration of the sink write so a - // (theoretical) re-entrant encode-into-sink call cannot BorrowMut-panic. + let Some(ptr) = NonNull::new(sink_ptr) else { + return JSValue::UNDEFINED; + }; + // SAFETY: `sink_ptr` is a live JSSink of type `sink_id` (the C++ caller + // null-checks it before attaching); the sink copies what it needs. + let handle = unsafe { sink_handle_from_id(sink_id, ptr) }; + if handle.is_none() { + return JSValue::UNDEFINED; + } + + if this.pending_lead_surrogate.get().is_none() { + if !str.is_16bit() { + let bytes = str.slice(); + if bytes.is_empty() { + return JSValue::UNDEFINED; + } + return handle + .write_latin1(&streams::Result::Temporary(RawSlice::new(bytes))) + .to_js(global); + } + + let mut utf16 = str.utf16_slice_aligned(); + // The sink's `write_utf16` emits U+FFFD for every unpaired surrogate + // (matching this encoder for mid-chunk ones); a trailing lone lead is + // the one position where the encoder differs — it carries it to the + // next chunk instead. + if let Some(&last) = utf16.last() { + if strings::u16_is_lead(last) { + this.pending_lead_surrogate.set(Some(last)); + utf16 = &utf16[..utf16.len() - 1]; + } + } + if utf16.is_empty() { + return JSValue::UNDEFINED; + } + let bytes: &[u8] = bytemuck::cast_slice(utf16); + return handle + .write_utf16(&streams::Result::Temporary(RawSlice::new(bytes))) + .to_js(global); + } + + // A lead surrogate was carried from the previous chunk: the prefix and + // body go out as one write so only one `Writable` is produced. Move the + // Vec out of the RefCell for the duration so a (theoretical) re-entrant + // encode-into-sink call cannot BorrowMut-panic. let mut buf = this.scratch.take(); buf.clear(); if str.is_16bit() { @@ -303,17 +358,6 @@ pub extern "C" fn TextEncoderStreamEncoder__encodeIntoSink( this.scratch.replace(buf); return JSValue::UNDEFINED; } - let Some(ptr) = NonNull::new(sink_ptr) else { - this.scratch.replace(buf); - return JSValue::UNDEFINED; - }; - // SAFETY: `sink_ptr` is a live JSSink of type `sink_id` (the C++ caller - // null-checks it before attaching); the sink copies what it needs. - let handle = unsafe { sink_handle_from_id(sink_id, ptr) }; - if handle.is_none() { - this.scratch.replace(buf); - return JSValue::UNDEFINED; - } let wrote = handle .write(&streams::Result::Temporary(RawSlice::new(&buf))) .to_js(global); diff --git a/test/js/web/encoding/text-decoder-stream.test.ts b/test/js/web/encoding/text-decoder-stream.test.ts index cf603bfc031d..23e0b910ca18 100644 --- a/test/js/web/encoding/text-decoder-stream.test.ts +++ b/test/js/web/encoding/text-decoder-stream.test.ts @@ -264,6 +264,31 @@ test("utf-8 fatal: truncated sequence rejects at flush", async () => { await expect(Bun.readableStreamToArray(out)).rejects.toBeInstanceOf(TypeError); }); +// Readable-side backpressure: the readable queue's HWM is 1, so a single write +// completes without a reader, and a second write parks until the readable side +// is drained. Exercises both the utf-8 fast path and the Rust-decoder path. +for (const opts of [{}, { fatal: true }] as const) { + test(`TextDecoderStream readable-side backpressure (fatal=${!!opts.fatal}): second write stays pending until drained`, async () => { + const tds = new TextDecoderStream("utf-8", opts); + const writer = tds.writable.getWriter(); + const reader = tds.readable.getReader(); + + await writer.write(new TextEncoder().encode("first")); + const second = writer.write(new TextEncoder().encode("second")); + const raced = await Promise.race([second.then(() => "done"), Bun.sleep(0).then(() => "pending")]); + expect(raced).toBe("pending"); + + const { value } = await reader.read(); + expect(value).toBe("first"); + await second; + + void writer.close(); + const rest: string[] = []; + for (let r; !(r = await reader.read()).done; ) rest.push(r.value); + expect(rest.join("")).toBe("second"); + }); +} + // The transform/flush arm runs the native decoder directly, so monkeypatching // TextDecoder.prototype.decode no longer reaches it. test("TextDecoderStream does not call a patched TextDecoder.prototype.decode", async () => { diff --git a/test/js/web/encoding/text-encoder-stream.test.ts b/test/js/web/encoding/text-encoder-stream.test.ts index 5c2578ec9479..f878cb908245 100644 --- a/test/js/web/encoding/text-encoder-stream.test.ts +++ b/test/js/web/encoding/text-encoder-stream.test.ts @@ -267,3 +267,163 @@ test("TextEncoderStream -> TextDecoderStream -> TextEncoderStream -> native HTTP expect(out.byteLength).toBe(expected.byteLength); expect(Buffer.compare(out, Buffer.from(expected))).toBe(0); }); + +// Native-sink path: 8-bit (Latin-1) chunks go straight to the sink's own +// write_latin1. An all-ASCII chunk is zero-copy; a chunk with bytes 128-255 +// is widened to 2-byte UTF-8 inside the sink. +test("TextEncoderStream -> native HTTP sink: 8-bit chunks (ASCII + Latin-1 non-ASCII)", async () => { + const ascii = Buffer.alloc(300_000, "plain ascii run.").toString(); + const latin1 = "caf\xe9 \xffna\xefve \xbd"; // é ÿ ï ½ + const chunks = [ascii, latin1, "mid", latin1, ascii]; + const expected = Buffer.from(new TextEncoder().encode(chunks.join(""))); + await using server = Bun.serve({ + port: 0, + fetch() { + const body = new ReadableStream({ + start(c) { + for (const s of chunks) c.enqueue(s); + c.close(); + }, + }); + return new Response(body.pipeThrough(new TextEncoderStream())); + }, + }); + const out = Buffer.from(await (await fetch(server.url)).arrayBuffer()); + expect(out.byteLength).toBe(expected.byteLength); + expect(Buffer.compare(out, expected)).toBe(0); +}); + +// Native-sink path: 16-bit chunks go straight to the sink's own write_utf16. +// The encoder still owns cross-chunk surrogate carry (trailing lone lead), and +// mid-chunk lone surrogates become U+FFFD via the sink. +test("TextEncoderStream -> native HTTP sink: 16-bit chunks with surrogate edge cases", async () => { + const chunks = [ + "A" + leading + leading, // 8-bit "A" roped with a 16-bit tail: mid-chunk lone lead + trailing lone lead + trailing + "B", // completes the carried pair + trailing, // lone trail, no carry -> FFFD + leading + leading + trailing, // mid-chunk lone lead + valid pair + "\u{1F499}".repeat(2000), // long 16-bit run + leading + trailing + leading, // valid pair then trailing lone lead + trailing, // completes + leading, // dangling -> FFFD at flush + ]; + // Interpret lone surrogates the spec way via a per-chunk TextEncoder (each + // chunk's trailing lone lead carries, each chunk's leading lone trail is + // replaced): equivalent to pushing the chunks through a second, known-good + // TextEncoderStream. + const ref = Buffer.from( + await new Response(readableStreamFromArray(chunks).pipeThrough(new TextEncoderStream())).arrayBuffer(), + ); + await using server = Bun.serve({ + port: 0, + fetch() { + const body = new ReadableStream({ + start(c) { + for (const s of chunks) c.enqueue(s); + c.close(); + }, + }); + return new Response(body.pipeThrough(new TextEncoderStream())); + }, + }); + const out = Buffer.from(await (await fetch(server.url)).arrayBuffer()); + expect(out.toString("hex")).toBe(ref.toString("hex")); +}); + +// HTMLRewriter's sink runs user JS (content handlers) while the input slice is +// still being parsed; a non-string chunk exercises the ToString-created JSString +// being kept rooted across that call. +test("TextEncoderStream -> HTMLRewriter native sink: non-string chunks + handler that allocates", async () => { + const ascii = Buffer.alloc(4096, "a").toString(); + const body = new ReadableStream({ + start(c) { + c.enqueue("
"); + c.enqueue({ toString: () => "

" + ascii + "

" }); // 8-bit, ToString-coerced + c.enqueue({ toString: () => "

\u{1F499}

" }); // 16-bit, ToString-coerced + c.enqueue("
"); + c.close(); + }, + }); + let seen = 0; + const out = await new HTMLRewriter() + .on("p", { + element() { + seen++; + Bun.gc(true); + }, + }) + .transform(new Response(body.pipeThrough(new TextEncoderStream()))) + .text(); + expect(seen).toBe(2); + expect(out).toBe("

" + ascii + "

\u{1F499}

"); +}); + +// Poll `get()` once per tick until it stays unchanged for 5 consecutive +// samples (parked on backpressure) or reaches `total` (ran away). +async function waitUntilStable(get: () => number, total: number) { + let last = get(); + let stable = 0; + while (stable < 5 && get() < total) { + await Bun.sleep(1); + const now = get(); + if (now === last) stable++; + else { + stable = 0; + last = now; + } + } +} + +// Native-sink backpressure: when the HTTP response sink's socket buffer fills +// (slow client), the sink's write_latin1 returns Backpressure and the writable +// side parks on m_nativeSinkReadyPromise. Without that, a fast source with a +// stalled client fills the sink buffer unboundedly. +test("TextEncoderStream -> native HTTP sink applies backpressure to a stalled client", async () => { + let pulls = 0; + const chunk = Buffer.alloc(64 * 1024, "x").toString(); + const TOTAL = 200; + await using server = Bun.serve({ + port: 0, + fetch() { + const body = new ReadableStream({ + pull(c) { + pulls++; + c.enqueue(chunk); + if (pulls >= TOTAL) c.close(); + }, + }); + return new Response(body.pipeThrough(new TextEncoderStream())); + }, + }); + const res = await fetch(server.url); + const reader = res.body!.getReader(); + await reader.read(); + await waitUntilStable(() => pulls, TOTAL); + const pullsWhileStalled = pulls; + while (!(await reader.read()).done) {} + expect(pullsWhileStalled).toBeLessThan(TOTAL); + expect(pulls).toBe(TOTAL); +}); + +// Readable-side backpressure (no native sink): the readable queue's HWM is 1, +// so a single write completes without a reader, and a second write parks until +// the readable side is drained. +test("TextEncoderStream readable-side backpressure: second write stays pending until drained", async () => { + const tes = new TextEncoderStream(); + const writer = tes.writable.getWriter(); + const reader = tes.readable.getReader(); + + await writer.write("first"); + const second = writer.write("second"); + const raced = await Promise.race([second.then(() => "done"), Bun.sleep(0).then(() => "pending")]); + expect(raced).toBe("pending"); + + const { value } = await reader.read(); + expect(new TextDecoder().decode(value)).toBe("first"); + await second; + + void writer.close(); + const rest: Uint8Array[] = []; + for (let r; !(r = await reader.read()).done; ) rest.push(r.value); + expect(new TextDecoder().decode(Buffer.concat(rest))).toBe("second"); +}); From 744749673507ab58ca472ec5db1c55442af680cd Mon Sep 17 00:00:00 2001 From: robobun <117481402+robobun@users.noreply.github.com> Date: Tue, 4 Aug 2026 07:01:27 +0000 Subject: [PATCH 2/2] address review: async-pull source variants, Bun.peek.status for backpressure, fix SAFETY comment - Parameterize the native-sink round-trip tests (8-bit, 16-bit, HTMLRewriter) over two source shapes: start() enqueues everything up front, and async pull with a macrotask yield so each chunk goes through encodeIntoSink while the response is streaming. - Replace the time-polled native-sink backpressure test with a writer-driven one: write 256 KiB chunks until a write stays pending across a macrotask (Bun.peek.status), then drain the client and assert the parked write resolves and the byte count matches. The previous version also had a writer<->fetch deadlock because fetch() waits for the first body bytes. - Readable-side backpressure tests (encoder + decoder) now use Bun.peek.status(second) instead of racing against Bun.sleep(0). - Await writer.close() after draining so a close rejection fails the test. - Add a non-ASCII Latin-1 chunk to the HTMLRewriter rooting test so the RewriterPipe write_latin1 Vec-copy branch is also covered. - Reword the SAFETY comment on the encoder &*this borrow: the sink write can run user JS; the soundness argument is shared-borrow + Cell/RefCell, not absence of user JS. --- .../webcore/TextEncoderStreamEncoder.rs | 6 +- .../web/encoding/text-decoder-stream.test.ts | 7 +- .../web/encoding/text-encoder-stream.test.ts | 259 +++++++++--------- 3 files changed, 141 insertions(+), 131 deletions(-) diff --git a/src/runtime/webcore/TextEncoderStreamEncoder.rs b/src/runtime/webcore/TextEncoderStreamEncoder.rs index 0ffdc38ee532..1ad84ce67c0c 100644 --- a/src/runtime/webcore/TextEncoderStreamEncoder.rs +++ b/src/runtime/webcore/TextEncoderStreamEncoder.rs @@ -294,8 +294,10 @@ pub extern "C" fn TextEncoderStreamEncoder__encodeIntoSink( }; let str = js_str.view(global); let _keep = bun_jsc::EnsureStillAlive(js_str.to_js()); - // SAFETY: `this` is the live encoder owned by the calling JS cell; taken - // after the coercion so no user JS runs while the borrow is live. + // SAFETY: `this` is the live encoder owned by the calling JS cell (on the + // C++ caller's stack, so it cannot be finalized mid-call). The sink write + // below can run user JS, but re-entry only forms another shared `&*this`; + // all state is `Cell`/`RefCell` and `scratch` is moved out before the write. let this = unsafe { &*this }; let Some(ptr) = NonNull::new(sink_ptr) else { return JSValue::UNDEFINED; diff --git a/test/js/web/encoding/text-decoder-stream.test.ts b/test/js/web/encoding/text-decoder-stream.test.ts index 23e0b910ca18..b056c2eef8d2 100644 --- a/test/js/web/encoding/text-decoder-stream.test.ts +++ b/test/js/web/encoding/text-decoder-stream.test.ts @@ -275,16 +275,17 @@ for (const opts of [{}, { fatal: true }] as const) { await writer.write(new TextEncoder().encode("first")); const second = writer.write(new TextEncoder().encode("second")); - const raced = await Promise.race([second.then(() => "done"), Bun.sleep(0).then(() => "pending")]); - expect(raced).toBe("pending"); + expect(Bun.peek.status(second)).toBe("pending"); const { value } = await reader.read(); expect(value).toBe("first"); + expect(Bun.peek.status(second)).toBe("pending"); await second; - void writer.close(); + const closed = writer.close(); const rest: string[] = []; for (let r; !(r = await reader.read()).done; ) rest.push(r.value); + await closed; expect(rest.join("")).toBe("second"); }); } diff --git a/test/js/web/encoding/text-encoder-stream.test.ts b/test/js/web/encoding/text-encoder-stream.test.ts index f878cb908245..a65536ca2a46 100644 --- a/test/js/web/encoding/text-encoder-stream.test.ts +++ b/test/js/web/encoding/text-encoder-stream.test.ts @@ -1,4 +1,4 @@ -import { expect, test } from "bun:test"; +import { describe, expect, test } from "bun:test"; import { readableStreamFromArray } from "harness"; // META: global=window,worker @@ -268,141 +268,147 @@ test("TextEncoderStream -> TextDecoderStream -> TextEncoderStream -> native HTTP expect(Buffer.compare(out, Buffer.from(expected))).toBe(0); }); -// Native-sink path: 8-bit (Latin-1) chunks go straight to the sink's own -// write_latin1. An all-ASCII chunk is zero-copy; a chunk with bytes 128-255 -// is widened to 2-byte UTF-8 inside the sink. -test("TextEncoderStream -> native HTTP sink: 8-bit chunks (ASCII + Latin-1 non-ASCII)", async () => { - const ascii = Buffer.alloc(300_000, "plain ascii run.").toString(); - const latin1 = "caf\xe9 \xffna\xefve \xbd"; // é ÿ ï ½ - const chunks = [ascii, latin1, "mid", latin1, ascii]; - const expected = Buffer.from(new TextEncoder().encode(chunks.join(""))); - await using server = Bun.serve({ - port: 0, - fetch() { - const body = new ReadableStream({ - start(c) { - for (const s of chunks) c.enqueue(s); - c.close(); - }, - }); - return new Response(body.pipeThrough(new TextEncoderStream())); - }, +// Two source shapes: `start` enqueues everything before the HTTP sink attaches +// (readMany drains a pre-filled queue); `async pull` with a macrotask yield +// keeps the source incomplete, so each chunk goes through encodeIntoSink while +// the response is actively streaming. +const sourceShapes = { + start(chunks: readonly unknown[]) { + return new ReadableStream({ + start(c) { + for (const s of chunks) c.enqueue(s); + c.close(); + }, + }); + }, + "async pull"(chunks: readonly unknown[]) { + let i = 0; + return new ReadableStream({ + async pull(c) { + await Bun.sleep(0); + if (i < chunks.length) c.enqueue(chunks[i++]); + else c.close(); + }, + }); + }, +}; + +describe.each(Object.entries(sourceShapes))("TextEncoderStream -> native sink (%s source)", (_, source) => { + // 8-bit (Latin-1) chunks go straight to the sink's own write_latin1. An + // all-ASCII chunk is zero-copy; a chunk with bytes 128-255 is widened to + // 2-byte UTF-8 inside the sink. + test("HTTP sink: 8-bit chunks (ASCII + Latin-1 non-ASCII)", async () => { + const ascii = Buffer.alloc(300_000, "plain ascii run.").toString(); + const latin1 = "caf\xe9 \xffna\xefve \xbd"; // é ÿ ï ½ + const chunks = [ascii, latin1, "mid", latin1, ascii]; + const expected = Buffer.from(new TextEncoder().encode(chunks.join(""))); + await using server = Bun.serve({ + port: 0, + fetch: () => new Response(source(chunks).pipeThrough(new TextEncoderStream())), + }); + const out = Buffer.from(await (await fetch(server.url)).arrayBuffer()); + expect(out.byteLength).toBe(expected.byteLength); + expect(Buffer.compare(out, expected)).toBe(0); }); - const out = Buffer.from(await (await fetch(server.url)).arrayBuffer()); - expect(out.byteLength).toBe(expected.byteLength); - expect(Buffer.compare(out, expected)).toBe(0); -}); -// Native-sink path: 16-bit chunks go straight to the sink's own write_utf16. -// The encoder still owns cross-chunk surrogate carry (trailing lone lead), and -// mid-chunk lone surrogates become U+FFFD via the sink. -test("TextEncoderStream -> native HTTP sink: 16-bit chunks with surrogate edge cases", async () => { - const chunks = [ - "A" + leading + leading, // 8-bit "A" roped with a 16-bit tail: mid-chunk lone lead + trailing lone lead - trailing + "B", // completes the carried pair - trailing, // lone trail, no carry -> FFFD - leading + leading + trailing, // mid-chunk lone lead + valid pair - "\u{1F499}".repeat(2000), // long 16-bit run - leading + trailing + leading, // valid pair then trailing lone lead - trailing, // completes - leading, // dangling -> FFFD at flush - ]; - // Interpret lone surrogates the spec way via a per-chunk TextEncoder (each - // chunk's trailing lone lead carries, each chunk's leading lone trail is - // replaced): equivalent to pushing the chunks through a second, known-good - // TextEncoderStream. - const ref = Buffer.from( - await new Response(readableStreamFromArray(chunks).pipeThrough(new TextEncoderStream())).arrayBuffer(), - ); - await using server = Bun.serve({ - port: 0, - fetch() { - const body = new ReadableStream({ - start(c) { - for (const s of chunks) c.enqueue(s); - c.close(); - }, - }); - return new Response(body.pipeThrough(new TextEncoderStream())); - }, + // 16-bit chunks go straight to the sink's own write_utf16. The encoder still + // owns cross-chunk surrogate carry (trailing lone lead), and mid-chunk lone + // surrogates become U+FFFD via the sink. + test("HTTP sink: 16-bit chunks with surrogate edge cases", async () => { + const chunks = [ + "A" + leading + leading, // 8-bit "A" roped with a 16-bit tail: mid-chunk lone lead + trailing lone lead + trailing + "B", // completes the carried pair + trailing, // lone trail, no carry -> FFFD + leading + leading + trailing, // mid-chunk lone lead + valid pair + Buffer.alloc(32_000, "\u{1F499}").toString(), // long 16-bit run + leading + trailing + leading, // valid pair then trailing lone lead + trailing, // completes + leading, // dangling -> FFFD at flush + ]; + // Reference via the non-sink encodeForStream path (JSUint8Array per chunk). + const ref = Buffer.from( + await new Response(readableStreamFromArray(chunks).pipeThrough(new TextEncoderStream())).arrayBuffer(), + ); + await using server = Bun.serve({ + port: 0, + fetch: () => new Response(source(chunks).pipeThrough(new TextEncoderStream())), + }); + const out = Buffer.from(await (await fetch(server.url)).arrayBuffer()); + expect(out.toString("hex")).toBe(ref.toString("hex")); }); - const out = Buffer.from(await (await fetch(server.url)).arrayBuffer()); - expect(out.toString("hex")).toBe(ref.toString("hex")); -}); -// HTMLRewriter's sink runs user JS (content handlers) while the input slice is -// still being parsed; a non-string chunk exercises the ToString-created JSString -// being kept rooted across that call. -test("TextEncoderStream -> HTMLRewriter native sink: non-string chunks + handler that allocates", async () => { - const ascii = Buffer.alloc(4096, "a").toString(); - const body = new ReadableStream({ - start(c) { - c.enqueue("
"); - c.enqueue({ toString: () => "

" + ascii + "

" }); // 8-bit, ToString-coerced - c.enqueue({ toString: () => "

\u{1F499}

" }); // 16-bit, ToString-coerced - c.enqueue("
"); - c.close(); - }, + // HTMLRewriter's sink runs user JS (content handlers) while the input slice + // is still being parsed; a non-string chunk exercises the ToString-created + // JSString being kept rooted across that call. + test("HTMLRewriter sink: non-string chunks + handler that allocates", async () => { + const ascii = Buffer.alloc(4096, "a").toString(); + const chunks = [ + "
", + { toString: () => "

" + ascii + "

" }, // 8-bit all-ASCII, ToString-coerced (RewriterPipe zero-copy branch) + { toString: () => "

caf\xe9

" }, // 8-bit non-ASCII, ToString-coerced (RewriterPipe Vec-copy branch) + { toString: () => "

\u{1F499}

" }, // 16-bit, ToString-coerced + "
", + ]; + let seen = 0; + const out = await new HTMLRewriter() + .on("p", { + element() { + seen++; + Bun.gc(true); + }, + }) + .transform(new Response(source(chunks).pipeThrough(new TextEncoderStream()))) + .text(); + expect(seen).toBe(3); + expect(out).toBe("

" + ascii + "

caf\xe9

\u{1F499}

"); }); - let seen = 0; - const out = await new HTMLRewriter() - .on("p", { - element() { - seen++; - Bun.gc(true); - }, - }) - .transform(new Response(body.pipeThrough(new TextEncoderStream()))) - .text(); - expect(seen).toBe(2); - expect(out).toBe("

" + ascii + "

\u{1F499}

"); }); -// Poll `get()` once per tick until it stays unchanged for 5 consecutive -// samples (parked on backpressure) or reaches `total` (ran away). -async function waitUntilStable(get: () => number, total: number) { - let last = get(); - let stable = 0; - while (stable < 5 && get() < total) { - await Bun.sleep(1); - const now = get(); - if (now === last) stable++; - else { - stable = 0; - last = now; - } - } -} - // Native-sink backpressure: when the HTTP response sink's socket buffer fills -// (slow client), the sink's write_latin1 returns Backpressure and the writable -// side parks on m_nativeSinkReadyPromise. Without that, a fast source with a -// stalled client fills the sink buffer unboundedly. -test("TextEncoderStream -> native HTTP sink applies backpressure to a stalled client", async () => { - let pulls = 0; - const chunk = Buffer.alloc(64 * 1024, "x").toString(); - const TOTAL = 200; +// (slow client), the sink's write_latin1 returns Backpressure, the transform +// arm parks on m_nativeSinkReadyPromise, and the next writer.write() stays +// pending until the client drains. Drive the writable side directly so the +// parked write is observable via Bun.peek.status (no time-based polling). +test("TextEncoderStream -> native HTTP sink applies backpressure to the writer", async () => { + const chunk = Buffer.alloc(256 * 1024, "x").toString(); + const MAX = 128; // upper bound; backpressure engages well before this (tens of 256 KiB chunks) + const gotWriter = Promise.withResolvers>(); await using server = Bun.serve({ port: 0, fetch() { - const body = new ReadableStream({ - pull(c) { - pulls++; - c.enqueue(chunk); - if (pulls >= TOTAL) c.close(); - }, - }); - return new Response(body.pipeThrough(new TextEncoderStream())); + const tes = new TextEncoderStream(); + gotWriter.resolve(tes.writable.getWriter()); + return new Response(tes.readable); }, }); - const res = await fetch(server.url); - const reader = res.body!.getReader(); - await reader.read(); - await waitUntilStable(() => pulls, TOTAL); - const pullsWhileStalled = pulls; - while (!(await reader.read()).done) {} - expect(pullsWhileStalled).toBeLessThan(TOTAL); - expect(pulls).toBe(TOTAL); + // The response headers flow once the body produces its first bytes, so write + // before awaiting fetch() to avoid a writer↔fetch deadlock. + const resP = fetch(server.url); + const writer = await gotWriter.promise; + await writer.write(chunk); + const reader = (await resP).body!.getReader(); + + let parked: Promise | undefined; + let written = 1; + for (; written < MAX; written++) { + const w = writer.write(chunk); + // One macrotask: the transform + native-sink write has run and either + // resolved w (sink accepted) or left it pending on m_nativeSinkReadyPromise. + await new Promise(r => setImmediate(r)); + if (Bun.peek.status(w) === "pending") { + parked = w; + break; + } + } + expect(written).toBeLessThan(MAX); + expect(parked && Bun.peek.status(parked)).toBe("pending"); + + const closed = writer.close(); + let received = 0; + for (let r; !(r = await reader.read()).done; ) received += r.value!.byteLength; + await parked; + await closed; + expect(received).toBe((written + 1) * chunk.length); }); // Readable-side backpressure (no native sink): the readable queue's HWM is 1, @@ -415,15 +421,16 @@ test("TextEncoderStream readable-side backpressure: second write stays pending u await writer.write("first"); const second = writer.write("second"); - const raced = await Promise.race([second.then(() => "done"), Bun.sleep(0).then(() => "pending")]); - expect(raced).toBe("pending"); + expect(Bun.peek.status(second)).toBe("pending"); const { value } = await reader.read(); expect(new TextDecoder().decode(value)).toBe("first"); + expect(Bun.peek.status(second)).toBe("pending"); await second; - void writer.close(); + const closed = writer.close(); const rest: Uint8Array[] = []; for (let r; !(r = await reader.read()).done; ) rest.push(r.value); + await closed; expect(new TextDecoder().decode(Buffer.concat(rest))).toBe("second"); });