TextEncoderStream native-sink: write straight to the sink's write_latin1/write_utf16

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<StreamError>) {
         // 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<SysError>) -> bun_sys::Result<()> {
         self.end_from_stream(err.map(StreamError::Error));

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`].

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,65 @@ 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;
     };
-    // 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 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 (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 };
-    // 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 +360,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);

test/js/web/encoding/text-decoder-stream.test.ts

@@ -264,6 +264,32 @@ 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"));
+    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;
+
+    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");
+  });
+}
+
 // 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 () => {