From e386b6b36e80926e9f9926e905293782562763cc Mon Sep 17 00:00:00 2001
From: robobun <117481402+robobun@users.noreply.github.com>
Date: Tue, 4 Aug 2026 07:31:31 +0000
Subject: [PATCH 1/8] ByteStream: empty buffer before signal_drained in drain()

drain() was calling signal_drained() (producer.ready()) before replacing the
buffer with an empty Vec. A producer whose on_ready inspects the ByteStream's
buffer length to decide whether output backpressure has cleared (RewriterPipe
does this in output_backpressured()) would see the old buffer still present and
bail without waking its own upstream.

In the TextEncoderStream -> HTMLRewriter -> CompressionStream chain this
deadlocked: the rewriter's write() returned Backpressure once one input chunk's
output exceeded the 256-byte high-water mark, parking the encodeIntoSink write
on m_nativeSinkReadyPromise; the only thing that resolves that promise is the
sink's onReady, which RewriterPipe::resume -> drain_pending_input -> src.ready()
drives, and resume() bailed because drain() had signalled while the buffer was
still full. Nothing re-signals once the buffer is actually emptied, so the
encoder stayed parked, the TransformStream's readable never closed, and the
readStreamIntoSink pump's readMany never settled.

on_pull() already empties first and signals second; make drain() match.
---
 src/runtime/webcore/ByteStream.rs     | 12 ++--
 test/js/workerd/html-rewriter.test.js | 80 +++++++++++++++++++++++++++
 2 files changed, 88 insertions(+), 4 deletions(-)

diff --git a/src/runtime/webcore/ByteStream.rs b/src/runtime/webcore/ByteStream.rs
index f873f3b231b0..17c51871051b 100644
--- a/src/runtime/webcore/ByteStream.rs
+++ b/src/runtime/webcore/ByteStream.rs
@@ -673,11 +673,15 @@ impl ByteStream {
     }
 
     pub(crate) fn drain(&self) -> Vec<u8> {
-        if !self.buffer.get().is_empty() {
-            self.signal_drained();
-            return Vec::<u8>::move_from_list(self.buffer.replace(Vec::new()));
+        if self.buffer.get().is_empty() {
+            return Vec::<u8>::default();
         }
-        Vec::<u8>::default()
+        // Empty the buffer BEFORE `signal_drained` (same order as `on_pull`): the
+        // producer's on_ready may inspect `self.buffer.len()` to decide whether
+        // output backpressure has cleared.
+        let drained = Vec::<u8>::move_from_list(self.buffer.replace(Vec::new()));
+        self.signal_drained();
+        drained
     }
 
     /// Take a pre-attach `StreamResult::Err` stashed by [`Self::append`].
diff --git a/test/js/workerd/html-rewriter.test.js b/test/js/workerd/html-rewriter.test.js
index a6380cd1e85f..ead0643fc63d 100644
--- a/test/js/workerd/html-rewriter.test.js
+++ b/test/js/workerd/html-rewriter.test.js
@@ -1741,6 +1741,86 @@ const payloads = [
   },
 ];
 
+// TextEncoderStream → HTMLRewriter → CompressionStream: the rewriter's input is
+// driven by the native-sink bypass (encodeIntoSink), and its output ByteStream
+// is consumed by a JS reader (pipeThrough), so its `drain()` is what wakes the
+// rewriter once a batch of output bytes has been taken. With enough elements
+// that one input chunk's output exceeds the rewriter's output high-water mark,
+// the rewriter's `write` returns Backpressure; this must eventually resolve.
+describe("output consumed via pipeThrough after a native-sink input transform", () => {
+  // ~50 elements → ~800 bytes of rewriter output per input chunk, well above
+  // the default output high-water mark (256).
+  const unit = "<p>abc</p>";
+  const chunk = Buffer.alloc(50 * unit.length, unit).toString();
+  const expected = Buffer.alloc(50 * 16, '<p x="1">abc</p>').toString();
+
+  function makeRewritten() {
+    let i = 0;
+    const body = new ReadableStream({
+      async pull(c) {
+        await Bun.sleep(0);
+        if (i++ === 0) c.enqueue(chunk);
+        else c.close();
+      },
+    });
+    return new HTMLRewriter()
+      .on("p", { element: e => e.setAttribute("x", "1") })
+      .transform(new Response(body.pipeThrough(new TextEncoderStream())));
+  }
+
+  it("completes when read with arrayBuffer()", async () => {
+    const compressed = makeRewritten().body.pipeThrough(new CompressionStream("gzip"));
+    const buf = await new Response(compressed).arrayBuffer();
+    const text = await new Response(
+      new Blob([buf]).stream().pipeThrough(new DecompressionStream("gzip")),
+    ).text();
+    expect(text).toBe(expected);
+  });
+
+  it("completes when served over HTTP", async () => {
+    await using server = Bun.serve({
+      port: 0,
+      fetch: () => new Response(makeRewritten().body.pipeThrough(new CompressionStream("gzip"))),
+    });
+    const buf = await (await fetch(server.url)).arrayBuffer();
+    const text = await new Response(
+      new Blob([buf]).stream().pipeThrough(new DecompressionStream("gzip")),
+    ).text();
+    expect(text).toBe(expected);
+  });
+
+  // CompressionStream is the only native-byte-transform that can sit downstream
+  // of the rewriter's ByteStream output; cover each format so a later per-codec
+  // regression shows up here.
+  it.each(["gzip", "deflate", "deflate-raw", "brotli", "zstd"])(
+    "completes for CompressionStream(%s)",
+    async format => {
+      const compressed = makeRewritten().body.pipeThrough(new CompressionStream(format));
+      const text = await new Response(compressed.pipeThrough(new DecompressionStream(format))).text();
+      expect(text).toBe(expected);
+    },
+  );
+
+  // Two input chunks → the second chunk is what the Backpressure wake must
+  // re-pull from upstream (the single-chunk case only owes the end() call).
+  it("completes across multiple input chunks", async () => {
+    let i = 0;
+    const body = new ReadableStream({
+      async pull(c) {
+        await Bun.sleep(0);
+        if (i++ < 2) c.enqueue(chunk);
+        else c.close();
+      },
+    });
+    const rewritten = new HTMLRewriter()
+      .on("p", { element: e => e.setAttribute("x", "1") })
+      .transform(new Response(body.pipeThrough(new TextEncoderStream())));
+    const compressed = rewritten.body.pipeThrough(new CompressionStream("gzip"));
+    const text = await new Response(compressed.pipeThrough(new DecompressionStream("gzip"))).text();
+    expect(text).toBe(expected + expected);
+  });
+});
+
 payloads.forEach(type => {
   type.test(`works with payload of type ${type.name}`, async () => {
     let calls = 0;

From 5c825b65222a93a50aceed2fa3374aad8a879d7f Mon Sep 17 00:00:00 2001
From: "autofix-ci[bot]" <114827586+autofix-ci[bot]@users.noreply.github.com>
Date: Tue, 4 Aug 2026 07:34:43 +0000
Subject: [PATCH 2/8] [autofix.ci] apply automated fixes

---
 test/js/workerd/html-rewriter.test.js | 21 +++++++--------------
 1 file changed, 7 insertions(+), 14 deletions(-)

diff --git a/test/js/workerd/html-rewriter.test.js b/test/js/workerd/html-rewriter.test.js
index ead0643fc63d..4c1ce8784816 100644
--- a/test/js/workerd/html-rewriter.test.js
+++ b/test/js/workerd/html-rewriter.test.js
@@ -1771,9 +1771,7 @@ describe("output consumed via pipeThrough after a native-sink input transform",
   it("completes when read with arrayBuffer()", async () => {
     const compressed = makeRewritten().body.pipeThrough(new CompressionStream("gzip"));
     const buf = await new Response(compressed).arrayBuffer();
-    const text = await new Response(
-      new Blob([buf]).stream().pipeThrough(new DecompressionStream("gzip")),
-    ).text();
+    const text = await new Response(new Blob([buf]).stream().pipeThrough(new DecompressionStream("gzip"))).text();
     expect(text).toBe(expected);
   });
 
@@ -1783,23 +1781,18 @@ describe("output consumed via pipeThrough after a native-sink input transform",
       fetch: () => new Response(makeRewritten().body.pipeThrough(new CompressionStream("gzip"))),
     });
     const buf = await (await fetch(server.url)).arrayBuffer();
-    const text = await new Response(
-      new Blob([buf]).stream().pipeThrough(new DecompressionStream("gzip")),
-    ).text();
+    const text = await new Response(new Blob([buf]).stream().pipeThrough(new DecompressionStream("gzip"))).text();
     expect(text).toBe(expected);
   });
 
   // CompressionStream is the only native-byte-transform that can sit downstream
   // of the rewriter's ByteStream output; cover each format so a later per-codec
   // regression shows up here.
-  it.each(["gzip", "deflate", "deflate-raw", "brotli", "zstd"])(
-    "completes for CompressionStream(%s)",
-    async format => {
-      const compressed = makeRewritten().body.pipeThrough(new CompressionStream(format));
-      const text = await new Response(compressed.pipeThrough(new DecompressionStream(format))).text();
-      expect(text).toBe(expected);
-    },
-  );
+  it.each(["gzip", "deflate", "deflate-raw", "brotli", "zstd"])("completes for CompressionStream(%s)", async format => {
+    const compressed = makeRewritten().body.pipeThrough(new CompressionStream(format));
+    const text = await new Response(compressed.pipeThrough(new DecompressionStream(format))).text();
+    expect(text).toBe(expected);
+  });
 
   // Two input chunks → the second chunk is what the Backpressure wake must
   // re-pull from upstream (the single-chunk case only owes the end() call).

From 28f333cacf440ba154cea0f4faa46c554e138e10 Mon Sep 17 00:00:00 2001
From: robobun <117481402+robobun@users.noreply.github.com>
Date: Tue, 4 Aug 2026 08:37:15 +0000
Subject: [PATCH 3/8] ByteStream: flush buffered bytes to sink before
 signal_drained in the native-sink wiring paths

The drain() reorder in the previous commit fixed the JS-reader path but opened
a re-entrancy window for the three callers that install self.sink before
calling drain() (wire_native_sink, S3 upload, RequestContext): a producer
whose on_ready feeds synchronously (RewriterPipe) would emit newer bytes to
the sink before the caller wrote the older drained bytes.

Factor resume()'s write-then-signal body into ByteStream::flush_to_sink() and
have those callers use it instead of drain() + manual sink.write. drain() is
now only the JS-reader entry point (native-source adapter / blob hand-off);
a debug_assert enforces it is not called with a sink installed.

The rewriter-output-to-native-sink state was a pre-existing deadlock on main
(resume bailed on output_backpressured and nothing re-signalled); this makes
it complete in order instead.
---
 src/runtime/server/RequestContext.rs  | 13 +++---
 src/runtime/webcore/ByteStream.rs     | 18 +++++++--
 src/runtime/webcore/ReadableStream.rs | 23 ++---------
 src/runtime/webcore/s3/client.rs      | 29 +-------------
 test/js/workerd/html-rewriter.test.js | 57 +++++++++++++++++++++++++++
 5 files changed, 85 insertions(+), 55 deletions(-)

diff --git a/src/runtime/server/RequestContext.rs b/src/runtime/server/RequestContext.rs
index ea47e3d5a730..65a3cbeb9bb9 100644
--- a/src/runtime/server/RequestContext.rs
+++ b/src/runtime/server/RequestContext.rs
@@ -3279,17 +3279,16 @@ where
                                 .set(readable_stream::Strong::init(stream, global_this));
 
                             this.byte_stream.set(Some(byte_stream_nn));
-                            let mut response_buf = byte_stream.drain();
-                            let buffer = response_buf.move_to_list();
-                            let has_body_bytes = !buffer.is_empty();
-                            this.response_buf_owned.set(buffer);
 
                             // we don't set size here because even if we have a hint
                             // uWebSockets won't let us partially write streaming content
                             this.blob.with_mut(|b| b.detach());
 
                             // if we've received metadata and part of the body, send everything we can and drain
-                            if has_body_bytes {
+                            if !byte_stream.buffer.get().is_empty() {
+                                let mut buf = byte_stream.buffer.replace(Vec::new());
+                                byte_stream.offset.set(0);
+                                this.response_buf_owned.set(buf.move_to_list());
                                 resp.run_corked_with_type(
                                     Self::drain_response_buffer_and_metadata_corked,
                                     this.as_ctx_ptr(),
@@ -3308,6 +3307,10 @@ where
                                     this.as_ctx_ptr(),
                                 );
                             }
+                            // Wake the producer now that the buffered bytes are
+                            // written; any output it emits synchronously reaches
+                            // `write_chunk` after them.
+                            byte_stream.flush_to_sink();
                             return;
                         }
                     }
diff --git a/src/runtime/webcore/ByteStream.rs b/src/runtime/webcore/ByteStream.rs
index 17c51871051b..24060f5f3978 100644
--- a/src/runtime/webcore/ByteStream.rs
+++ b/src/runtime/webcore/ByteStream.rs
@@ -167,7 +167,17 @@ impl ByteStream {
             return;
         }
         self.sink_paused.set(false);
+        self.flush_to_sink();
+    }
 
+    /// Write any buffered bytes to `self.sink`, then [`signal_drained`]. The
+    /// native-sink wiring paths call this (instead of [`drain`] + a manual
+    /// `sink.write`) so the buffered bytes reach the sink before the producer
+    /// is woken; a producer whose `on_ready` feeds synchronously (RewriterPipe)
+    /// would otherwise emit newer bytes to the already-installed sink ahead of
+    /// the ones the caller is still holding. Clears `self.sink` when the write
+    /// or the last-chunk flag ends the stream.
+    pub(crate) fn flush_to_sink(&self) {
         let sink = *self.sink.get();
         if sink.is_none() {
             return;
@@ -672,13 +682,15 @@ impl ByteStream {
         // deallocate the storage backing `&mut self` (dangling UAF).
     }
 
+    /// JS-reader drain (native-source adapter's `handle.drain()` and the
+    /// `has_received_last_chunk` blob hand-off). Callers that have already
+    /// installed `self.sink` must use [`flush_to_sink`] instead so the bytes
+    /// are written before the producer is woken.
     pub(crate) fn drain(&self) -> Vec<u8> {
+        debug_assert!(self.sink.get().is_none());
         if self.buffer.get().is_empty() {
             return Vec::<u8>::default();
         }
-        // Empty the buffer BEFORE `signal_drained` (same order as `on_pull`): the
-        // producer's on_ready may inspect `self.buffer.len()` to decide whether
-        // output backpressure has cleared.
         let drained = Vec::<u8>::move_from_list(self.buffer.replace(Vec::new()));
         self.signal_drained();
         drained
diff --git a/src/runtime/webcore/ReadableStream.rs b/src/runtime/webcore/ReadableStream.rs
index 4afa1fed7cb9..fa3888880f8d 100644
--- a/src/runtime/webcore/ReadableStream.rs
+++ b/src/runtime/webcore/ReadableStream.rs
@@ -320,7 +320,7 @@ impl ReadableStream {
         owner_cell: JSValue,
         set_source: impl FnOnce(streams::SourceHandle),
     ) -> NativeWireResult {
-        use streams::{SourceHandle, Start, StreamError, StreamResult, Writable};
+        use streams::{SourceHandle, Start, StreamError, StreamResult};
         use webcore::SinkHandle;
 
         if let Some(byte_stream) = self.ptr.bytes() {
@@ -337,25 +337,8 @@ impl ReadableStream {
                     return NativeWireResult::EndedInline(Some(err));
                 }
 
-                let buffered = byte_stream.drain();
-                let has_last = byte_stream.has_received_last_chunk.get();
-                if !buffered.is_empty() {
-                    let chunk = if has_last {
-                        StreamResult::OwnedAndDone(buffered)
-                    } else {
-                        StreamResult::Owned(buffered)
-                    };
-                    match sink.write(&chunk) {
-                        Writable::Backpressure(_) => byte_stream.sink_paused.set(true),
-                        Writable::Done | Writable::Err(_) => {
-                            byte_stream.sink.set(SinkHandle::None);
-                            return NativeWireResult::EndedInline(None);
-                        }
-                        _ => {}
-                    }
-                }
-                if has_last {
-                    byte_stream.sink.set(SinkHandle::None);
+                byte_stream.flush_to_sink();
+                if byte_stream.sink.get().is_none() {
                     return NativeWireResult::EndedInline(None);
                 }
                 return NativeWireResult::Wired;
diff --git a/src/runtime/webcore/s3/client.rs b/src/runtime/webcore/s3/client.rs
index cf40d5e19b55..c479a08f27ff 100644
--- a/src/runtime/webcore/s3/client.rs
+++ b/src/runtime/webcore/s3/client.rs
@@ -1037,33 +1037,8 @@ pub(crate) fn upload_stream(
                 return Ok(end_promise_value);
             }
 
-            let buffered = byte_stream.drain();
-            let has_last = byte_stream.has_received_last_chunk.get();
-            if !buffered.is_empty() {
-                let chunk = if has_last {
-                    crate::webcore::streams::StreamResult::OwnedAndDone(buffered)
-                } else {
-                    crate::webcore::streams::StreamResult::Owned(buffered)
-                };
-                match sink.write(&chunk) {
-                    crate::webcore::streams::Writable::Backpressure(_) => {
-                        byte_stream.sink_paused.set(true);
-                    }
-                    crate::webcore::streams::Writable::Done
-                    | crate::webcore::streams::Writable::Err(_) => {
-                        byte_stream.sink.set(crate::webcore::SinkHandle::None);
-                        sink.source.clear();
-                        if !sink.ended {
-                            let _ = sink.end(None);
-                        }
-                        ctx.handle_resolve_stream();
-                        return Ok(end_promise_value);
-                    }
-                    _ => {}
-                }
-            }
-            if has_last {
-                byte_stream.sink.set(crate::webcore::SinkHandle::None);
+            byte_stream.flush_to_sink();
+            if byte_stream.sink.get().is_none() {
                 sink.source.clear();
                 if !sink.ended {
                     let _ = sink.end(None);
diff --git a/test/js/workerd/html-rewriter.test.js b/test/js/workerd/html-rewriter.test.js
index 4c1ce8784816..f496a72a72b2 100644
--- a/test/js/workerd/html-rewriter.test.js
+++ b/test/js/workerd/html-rewriter.test.js
@@ -1814,6 +1814,63 @@ describe("output consumed via pipeThrough after a native-sink input transform",
   });
 });
 
+// The same output-backpressure state, consumed by a native sink instead of a
+// JS reader: the sink is installed on the output ByteStream while it still
+// holds more than the rewriter's high-water mark in buffered bytes, and the
+// rewriter's input is already fully consumed (input_ended). flush_to_sink()
+// must write the buffered bytes to the sink before waking the rewriter (which
+// then runs end_rewrite() synchronously).
+describe("output consumed by a native sink after output backpressure", () => {
+  const prefix = "<p>" + Buffer.alloc(100, "A").toString() + "</p>";
+  const suffix = "<q>" + Buffer.alloc(30_000, "B").toString() + "</q>";
+  const html = prefix + "<x></x>" + suffix;
+
+  // Produce a rewriter output ByteStream in the "input_ended + buffer > hwm"
+  // state: .body materialises the ByteStream; resolving the handler gate lets
+  // resume_rewrite() push the 30 KB suffix into that buffer; drain_pending_input
+  // then returns early on output_backpressured(), leaving end_rewrite() owed.
+  async function makeParked() {
+    const gate = Promise.withResolvers();
+    const out = new HTMLRewriter()
+      .on("x", { element: () => gate.promise })
+      .transform(new Response(html));
+    const body = out.body;
+    gate.resolve();
+    await 0;
+    return { out, body };
+  }
+
+  it("chained HTMLRewriter receives the full output in order", async () => {
+    const { body } = await makeParked();
+    const chained = new HTMLRewriter().transform(new Response(body));
+    expect(await chained.text()).toBe(html);
+  });
+
+  it("Bun.serve returning the rewriter Response sends the full output", async () => {
+    await using server = Bun.serve({
+      port: 0,
+      async fetch() {
+        return (await makeParked()).out;
+      },
+    });
+    expect(await (await fetch(server.url)).text()).toBe(html);
+  });
+
+  it("fetch request body wired to the output ByteStream uploads the full output", async () => {
+    const received = Promise.withResolvers();
+    await using server = Bun.serve({
+      port: 0,
+      async fetch(req) {
+        received.resolve(await req.text());
+        return new Response("ok");
+      },
+    });
+    const { body } = await makeParked();
+    await fetch(server.url, { method: "POST", body });
+    expect(await received.promise).toBe(html);
+  });
+});
+
 payloads.forEach(type => {
   type.test(`works with payload of type ${type.name}`, async () => {
     let calls = 0;

From 7542d5f466bab12a55e1b5e47495b68b13615795 Mon Sep 17 00:00:00 2001
From: "autofix-ci[bot]" <114827586+autofix-ci[bot]@users.noreply.github.com>
Date: Tue, 4 Aug 2026 08:49:21 +0000
Subject: [PATCH 4/8] [autofix.ci] apply automated fixes

---
 test/js/workerd/html-rewriter.test.js | 4 +---
 1 file changed, 1 insertion(+), 3 deletions(-)

diff --git a/test/js/workerd/html-rewriter.test.js b/test/js/workerd/html-rewriter.test.js
index f496a72a72b2..2700ee4049aa 100644
--- a/test/js/workerd/html-rewriter.test.js
+++ b/test/js/workerd/html-rewriter.test.js
@@ -1831,9 +1831,7 @@ describe("output consumed by a native sink after output backpressure", () => {
   // then returns early on output_backpressured(), leaving end_rewrite() owed.
   async function makeParked() {
     const gate = Promise.withResolvers();
-    const out = new HTMLRewriter()
-      .on("x", { element: () => gate.promise })
-      .transform(new Response(html));
+    const out = new HTMLRewriter().on("x", { element: () => gate.promise }).transform(new Response(html));
     const body = out.body;
     gate.resolve();
     await 0;

From 10de12c32da19a667068d527458ce73988e94333 Mon Sep 17 00:00:00 2001
From: robobun <117481402+robobun@users.noreply.github.com>
Date: Tue, 4 Aug 2026 08:49:57 +0000
Subject: [PATCH 5/8] tests: note why the pull() macrotask yield is
 load-bearing; dedupe multi-chunk source

---
 test/js/workerd/html-rewriter.test.js | 20 ++++++--------------
 1 file changed, 6 insertions(+), 14 deletions(-)

diff --git a/test/js/workerd/html-rewriter.test.js b/test/js/workerd/html-rewriter.test.js
index 2700ee4049aa..6f3032aed825 100644
--- a/test/js/workerd/html-rewriter.test.js
+++ b/test/js/workerd/html-rewriter.test.js
@@ -1754,12 +1754,15 @@ describe("output consumed via pipeThrough after a native-sink input transform",
   const chunk = Buffer.alloc(50 * unit.length, unit).toString();
   const expected = Buffer.alloc(50 * 16, '<p x="1">abc</p>').toString();
 
-  function makeRewritten() {
+  // The pull() macrotask yield is load-bearing: a synchronous pull lets the
+  // rewrite drain into the pre-stream output_buffer before the downstream
+  // reader attaches, so the ByteStream drain() path under test is never hit.
+  function makeRewritten(chunks = [chunk]) {
     let i = 0;
     const body = new ReadableStream({
       async pull(c) {
         await Bun.sleep(0);
-        if (i++ === 0) c.enqueue(chunk);
+        if (i < chunks.length) c.enqueue(chunks[i++]);
         else c.close();
       },
     });
@@ -1797,18 +1800,7 @@ describe("output consumed via pipeThrough after a native-sink input transform",
   // Two input chunks → the second chunk is what the Backpressure wake must
   // re-pull from upstream (the single-chunk case only owes the end() call).
   it("completes across multiple input chunks", async () => {
-    let i = 0;
-    const body = new ReadableStream({
-      async pull(c) {
-        await Bun.sleep(0);
-        if (i++ < 2) c.enqueue(chunk);
-        else c.close();
-      },
-    });
-    const rewritten = new HTMLRewriter()
-      .on("p", { element: e => e.setAttribute("x", "1") })
-      .transform(new Response(body.pipeThrough(new TextEncoderStream())));
-    const compressed = rewritten.body.pipeThrough(new CompressionStream("gzip"));
+    const compressed = makeRewritten([chunk, chunk]).body.pipeThrough(new CompressionStream("gzip"));
     const text = await new Response(compressed.pipeThrough(new DecompressionStream("gzip"))).text();
     expect(text).toBe(expected + expected);
   });

From 041cd25fd0028eb548b9513395e685113f3378f7 Mon Sep 17 00:00:00 2001
From: robobun <117481402+robobun@users.noreply.github.com>
Date: Tue, 4 Aug 2026 09:13:34 +0000
Subject: [PATCH 6/8] ByteStream::drain: drop the sink-is-none debug_assert

materializeNativeSource can reach drain() with a sink already wired: getReader()
on a natively-locked stream materialises before the lock check throws. Guard the
signal_drained call instead so that path neither panics nor lets a synchronous
producer emit ahead of the drained bytes.
---
 src/runtime/webcore/ByteStream.rs | 10 ++++++++--
 1 file changed, 8 insertions(+), 2 deletions(-)

diff --git a/src/runtime/webcore/ByteStream.rs b/src/runtime/webcore/ByteStream.rs
index 24060f5f3978..ba3a163819e8 100644
--- a/src/runtime/webcore/ByteStream.rs
+++ b/src/runtime/webcore/ByteStream.rs
@@ -687,12 +687,18 @@ impl ByteStream {
     /// installed `self.sink` must use [`flush_to_sink`] instead so the bytes
     /// are written before the producer is woken.
     pub(crate) fn drain(&self) -> Vec<u8> {
-        debug_assert!(self.sink.get().is_none());
         if self.buffer.get().is_empty() {
             return Vec::<u8>::default();
         }
         let drained = Vec::<u8>::move_from_list(self.buffer.replace(Vec::new()));
-        self.signal_drained();
+        // `materializeNativeSource` can reach here with a sink already wired
+        // (getReader() on a natively-locked stream materialises before the
+        // lock check throws); signalling then would let a synchronous producer
+        // emit newer bytes to that sink ahead of `drained`, so only signal on
+        // the intended JS-reader path.
+        if self.sink.get().is_none() {
+            self.signal_drained();
+        }
         drained
     }
 

From 29e33eb50c76796f2b38b8ff8fe03b7926a8e816 Mon Sep 17 00:00:00 2001
From: robobun <117481402+robobun@users.noreply.github.com>
Date: Tue, 4 Aug 2026 09:22:42 +0000
Subject: [PATCH 7/8] Revert the native-sink caller refactor; guard drain()'s
 signal only for the sync producer behind a sink

flush_to_sink() changed the EndedInline contract: it called sink.end() on every
terminal path, and every wire_native_sink caller then ran its own end again
(FetchTasklet::write_end_request double-deref, NetworkSink pump-ref double-release).
The state those changes targeted (RewriterPipe output buffer > hwm then wired
to a native sink) is a pre-existing deadlock on main; leave it as-is rather than
trading it for a refcount over-release.

drain() now empties before signalling (the JS-reader fix) and skips the signal
only for the one producer whose on_ready feeds synchronously (RewriterPipe) when
a sink is installed, so async producers (FetchTasklet, FileReader) are still
resumed exactly as on main.
---
 src/runtime/server/RequestContext.rs  | 13 +++----
 src/runtime/webcore/ByteStream.rs     | 31 ++++++---------
 src/runtime/webcore/ReadableStream.rs | 23 +++++++++--
 src/runtime/webcore/s3/client.rs      | 29 +++++++++++++-
 test/js/workerd/html-rewriter.test.js | 55 ---------------------------
 5 files changed, 63 insertions(+), 88 deletions(-)

diff --git a/src/runtime/server/RequestContext.rs b/src/runtime/server/RequestContext.rs
index 65a3cbeb9bb9..ea47e3d5a730 100644
--- a/src/runtime/server/RequestContext.rs
+++ b/src/runtime/server/RequestContext.rs
@@ -3279,16 +3279,17 @@ where
                                 .set(readable_stream::Strong::init(stream, global_this));
 
                             this.byte_stream.set(Some(byte_stream_nn));
+                            let mut response_buf = byte_stream.drain();
+                            let buffer = response_buf.move_to_list();
+                            let has_body_bytes = !buffer.is_empty();
+                            this.response_buf_owned.set(buffer);
 
                             // we don't set size here because even if we have a hint
                             // uWebSockets won't let us partially write streaming content
                             this.blob.with_mut(|b| b.detach());
 
                             // if we've received metadata and part of the body, send everything we can and drain
-                            if !byte_stream.buffer.get().is_empty() {
-                                let mut buf = byte_stream.buffer.replace(Vec::new());
-                                byte_stream.offset.set(0);
-                                this.response_buf_owned.set(buf.move_to_list());
+                            if has_body_bytes {
                                 resp.run_corked_with_type(
                                     Self::drain_response_buffer_and_metadata_corked,
                                     this.as_ctx_ptr(),
@@ -3307,10 +3308,6 @@ where
                                     this.as_ctx_ptr(),
                                 );
                             }
-                            // Wake the producer now that the buffered bytes are
-                            // written; any output it emits synchronously reaches
-                            // `write_chunk` after them.
-                            byte_stream.flush_to_sink();
                             return;
                         }
                     }
diff --git a/src/runtime/webcore/ByteStream.rs b/src/runtime/webcore/ByteStream.rs
index ba3a163819e8..189d49f752ed 100644
--- a/src/runtime/webcore/ByteStream.rs
+++ b/src/runtime/webcore/ByteStream.rs
@@ -167,17 +167,7 @@ impl ByteStream {
             return;
         }
         self.sink_paused.set(false);
-        self.flush_to_sink();
-    }
 
-    /// Write any buffered bytes to `self.sink`, then [`signal_drained`]. The
-    /// native-sink wiring paths call this (instead of [`drain`] + a manual
-    /// `sink.write`) so the buffered bytes reach the sink before the producer
-    /// is woken; a producer whose `on_ready` feeds synchronously (RewriterPipe)
-    /// would otherwise emit newer bytes to the already-installed sink ahead of
-    /// the ones the caller is still holding. Clears `self.sink` when the write
-    /// or the last-chunk flag ends the stream.
-    pub(crate) fn flush_to_sink(&self) {
         let sink = *self.sink.get();
         if sink.is_none() {
             return;
@@ -682,21 +672,22 @@ impl ByteStream {
         // deallocate the storage backing `&mut self` (dangling UAF).
     }
 
-    /// JS-reader drain (native-source adapter's `handle.drain()` and the
-    /// `has_received_last_chunk` blob hand-off). Callers that have already
-    /// installed `self.sink` must use [`flush_to_sink`] instead so the bytes
-    /// are written before the producer is woken.
     pub(crate) fn drain(&self) -> Vec<u8> {
         if self.buffer.get().is_empty() {
             return Vec::<u8>::default();
         }
+        // Empty first so a producer.on_ready that checks `buffer.len()` sees it
+        // drained. RewriterPipe is the one producer whose on_ready feeds
+        // synchronously; when a sink is installed the native-sink wiring caller
+        // writes `drained` to it afterward, so waking the rewriter here would
+        // let it reach the sink first. Async producers only schedule work.
         let drained = Vec::<u8>::move_from_list(self.buffer.replace(Vec::new()));
-        // `materializeNativeSource` can reach here with a sink already wired
-        // (getReader() on a natively-locked stream materialises before the
-        // lock check throws); signalling then would let a synchronous producer
-        // emit newer bytes to that sink ahead of `drained`, so only signal on
-        // the intended JS-reader path.
-        if self.sink.get().is_none() {
+        let sync_behind_sink = self.sink.get().is_some()
+            && matches!(
+                self.parent_const().producer.get(),
+                streams::SourceHandle::HTMLRewriter(_)
+            );
+        if !sync_behind_sink {
             self.signal_drained();
         }
         drained
diff --git a/src/runtime/webcore/ReadableStream.rs b/src/runtime/webcore/ReadableStream.rs
index fa3888880f8d..4afa1fed7cb9 100644
--- a/src/runtime/webcore/ReadableStream.rs
+++ b/src/runtime/webcore/ReadableStream.rs
@@ -320,7 +320,7 @@ impl ReadableStream {
         owner_cell: JSValue,
         set_source: impl FnOnce(streams::SourceHandle),
     ) -> NativeWireResult {
-        use streams::{SourceHandle, Start, StreamError, StreamResult};
+        use streams::{SourceHandle, Start, StreamError, StreamResult, Writable};
         use webcore::SinkHandle;
 
         if let Some(byte_stream) = self.ptr.bytes() {
@@ -337,8 +337,25 @@ impl ReadableStream {
                     return NativeWireResult::EndedInline(Some(err));
                 }
 
-                byte_stream.flush_to_sink();
-                if byte_stream.sink.get().is_none() {
+                let buffered = byte_stream.drain();
+                let has_last = byte_stream.has_received_last_chunk.get();
+                if !buffered.is_empty() {
+                    let chunk = if has_last {
+                        StreamResult::OwnedAndDone(buffered)
+                    } else {
+                        StreamResult::Owned(buffered)
+                    };
+                    match sink.write(&chunk) {
+                        Writable::Backpressure(_) => byte_stream.sink_paused.set(true),
+                        Writable::Done | Writable::Err(_) => {
+                            byte_stream.sink.set(SinkHandle::None);
+                            return NativeWireResult::EndedInline(None);
+                        }
+                        _ => {}
+                    }
+                }
+                if has_last {
+                    byte_stream.sink.set(SinkHandle::None);
                     return NativeWireResult::EndedInline(None);
                 }
                 return NativeWireResult::Wired;
diff --git a/src/runtime/webcore/s3/client.rs b/src/runtime/webcore/s3/client.rs
index c479a08f27ff..cf40d5e19b55 100644
--- a/src/runtime/webcore/s3/client.rs
+++ b/src/runtime/webcore/s3/client.rs
@@ -1037,8 +1037,33 @@ pub(crate) fn upload_stream(
                 return Ok(end_promise_value);
             }
 
-            byte_stream.flush_to_sink();
-            if byte_stream.sink.get().is_none() {
+            let buffered = byte_stream.drain();
+            let has_last = byte_stream.has_received_last_chunk.get();
+            if !buffered.is_empty() {
+                let chunk = if has_last {
+                    crate::webcore::streams::StreamResult::OwnedAndDone(buffered)
+                } else {
+                    crate::webcore::streams::StreamResult::Owned(buffered)
+                };
+                match sink.write(&chunk) {
+                    crate::webcore::streams::Writable::Backpressure(_) => {
+                        byte_stream.sink_paused.set(true);
+                    }
+                    crate::webcore::streams::Writable::Done
+                    | crate::webcore::streams::Writable::Err(_) => {
+                        byte_stream.sink.set(crate::webcore::SinkHandle::None);
+                        sink.source.clear();
+                        if !sink.ended {
+                            let _ = sink.end(None);
+                        }
+                        ctx.handle_resolve_stream();
+                        return Ok(end_promise_value);
+                    }
+                    _ => {}
+                }
+            }
+            if has_last {
+                byte_stream.sink.set(crate::webcore::SinkHandle::None);
                 sink.source.clear();
                 if !sink.ended {
                     let _ = sink.end(None);
diff --git a/test/js/workerd/html-rewriter.test.js b/test/js/workerd/html-rewriter.test.js
index 6f3032aed825..f5edd7771941 100644
--- a/test/js/workerd/html-rewriter.test.js
+++ b/test/js/workerd/html-rewriter.test.js
@@ -1806,61 +1806,6 @@ describe("output consumed via pipeThrough after a native-sink input transform",
   });
 });
 
-// The same output-backpressure state, consumed by a native sink instead of a
-// JS reader: the sink is installed on the output ByteStream while it still
-// holds more than the rewriter's high-water mark in buffered bytes, and the
-// rewriter's input is already fully consumed (input_ended). flush_to_sink()
-// must write the buffered bytes to the sink before waking the rewriter (which
-// then runs end_rewrite() synchronously).
-describe("output consumed by a native sink after output backpressure", () => {
-  const prefix = "<p>" + Buffer.alloc(100, "A").toString() + "</p>";
-  const suffix = "<q>" + Buffer.alloc(30_000, "B").toString() + "</q>";
-  const html = prefix + "<x></x>" + suffix;
-
-  // Produce a rewriter output ByteStream in the "input_ended + buffer > hwm"
-  // state: .body materialises the ByteStream; resolving the handler gate lets
-  // resume_rewrite() push the 30 KB suffix into that buffer; drain_pending_input
-  // then returns early on output_backpressured(), leaving end_rewrite() owed.
-  async function makeParked() {
-    const gate = Promise.withResolvers();
-    const out = new HTMLRewriter().on("x", { element: () => gate.promise }).transform(new Response(html));
-    const body = out.body;
-    gate.resolve();
-    await 0;
-    return { out, body };
-  }
-
-  it("chained HTMLRewriter receives the full output in order", async () => {
-    const { body } = await makeParked();
-    const chained = new HTMLRewriter().transform(new Response(body));
-    expect(await chained.text()).toBe(html);
-  });
-
-  it("Bun.serve returning the rewriter Response sends the full output", async () => {
-    await using server = Bun.serve({
-      port: 0,
-      async fetch() {
-        return (await makeParked()).out;
-      },
-    });
-    expect(await (await fetch(server.url)).text()).toBe(html);
-  });
-
-  it("fetch request body wired to the output ByteStream uploads the full output", async () => {
-    const received = Promise.withResolvers();
-    await using server = Bun.serve({
-      port: 0,
-      async fetch(req) {
-        received.resolve(await req.text());
-        return new Response("ok");
-      },
-    });
-    const { body } = await makeParked();
-    await fetch(server.url, { method: "POST", body });
-    expect(await received.promise).toBe(html);
-  });
-});
-
 payloads.forEach(type => {
   type.test(`works with payload of type ${type.name}`, async () => {
     let calls = 0;

From bcb38610f0d393b11b77bd36a579b6f04bacdbdd Mon Sep 17 00:00:00 2001
From: robobun <117481402+robobun@users.noreply.github.com>
Date: Tue, 4 Aug 2026 11:05:42 +0000
Subject: [PATCH 8/8] ci: retrigger