From 3be437a9580a53867fef14c3dc7a52ccd8f2184a Mon Sep 17 00:00:00 2001
From: robobun <117481402+robobun@users.noreply.github.com>
Date: Sat, 1 Aug 2026 13:57:14 +0000
Subject: [PATCH 01/15] ByteStream: hand off owned buffers from on_pull/on_data
 and skip the adapter pull view

ByteStream is a push source: body chunks arrive via on_data and are held in a
native Vec until on_pull takes them. on_pull used to copy those bytes into the
adapter-supplied Uint8Array view and return IntoArray; the adapter allocated a
256-512 KiB scratch view per stream to receive that copy. on_pull now takes the
Vec and returns Owned (the same allocation becomes the Uint8Array backing), and
on_data's pending path resolves with Owned directly instead of copying into the
parked view.

Since the view is never written, on_start now returns Start::ReadyOwned, which
start_from_js encodes as -1. materializeNativeSource treats a negative start
result as m_sourceOwnsChunks: nativeGetInternalBuffer then skips the view
allocation and passes jsUndefined() to pull (pull_from_js passes an empty slice
when the view is missing). nativeDecodePullResult no longer grows m_chunkSize
when the source returns its own ArrayBufferView (the view was not the
bottleneck). native-readable.ts mirrors the negative start result by keeping
its pull buffer at MIN_BUFFER_SIZE.

Measured on the Hono upload-stream route (oha -c100 -n2000, 2 MiB bodies,
release build on Linux x64) this lowers server VmHWM by roughly 10 percent and
post-full-GC RSS by roughly 19 percent; the remaining peak is the per-request
native buffer Vec (REQUEST_BODY_HIGH_WATER_MARK) plus in-flight body chunks.
---
 src/js/internal/streams/native-readable.ts    |   5 +
 .../webcore/streams/BunStreamSource.cpp       |  22 ++-
 .../webcore/streams/BunStreamSource.h         |   3 +
 src/runtime/webcore/ByteStream.rs             | 168 ++++++------------
 src/runtime/webcore/ReadableStream.rs         |  12 +-
 src/runtime/webcore/streams.rs                |   6 +-
 6 files changed, 96 insertions(+), 120 deletions(-)

diff --git a/src/js/internal/streams/native-readable.ts b/src/js/internal/streams/native-readable.ts
index 65307212b821..3ecd84cbc737 100644
--- a/src/js/internal/streams/native-readable.ts
+++ b/src/js/internal/streams/native-readable.ts
@@ -142,6 +142,11 @@ function read(this: NativeReadable, maxToRead: number) {
     if (typeof result === "number" && result > 1) {
       this[kHasResized] = true;
       this[kHighWaterMark] = Math.min(this[kHighWaterMark], result);
+    } else if (typeof result === "number" && result < 0) {
+      // Source allocates its own chunks; the pull view is unused, so keep it
+      // at the minimum size.
+      this[kHasResized] = true;
+      this[kHighWaterMark] = MIN_BUFFER_SIZE;
     }
     if ($isTypedArrayView(result) && result.byteLength > 0) {
       pushAndCheck(this, result);
diff --git a/src/jsc/bindings/webcore/streams/BunStreamSource.cpp b/src/jsc/bindings/webcore/streams/BunStreamSource.cpp
index 5784f4cdc03a..0656e7af228b 100644
--- a/src/jsc/bindings/webcore/streams/BunStreamSource.cpp
+++ b/src/jsc/bindings/webcore/streams/BunStreamSource.cpp
@@ -417,6 +417,8 @@ static JSC::JSUint8Array* uint8Subarray(JSGlobalObject* globalObject, JSC::JSUin
 static JSC::JSUint8Array* nativeGetInternalBuffer(JSC::VM& vm, JSGlobalObject* globalObject, JSNativeStreamSourceAdapter* adapter)
 {
     auto scope = DECLARE_THROW_SCOPE(vm);
+    if (adapter->m_sourceOwnsChunks)
+        return nullptr;
     const size_t chunkSize = adapter->m_chunkSize;
     if (JSObject* pending = adapter->pendingView()) {
         auto* view = uncheckedDowncast<JSC::JSUint8Array>(pending);
@@ -478,8 +480,8 @@ static JSValue nativeDecodePullResult(JSC::VM& vm, JSGlobalObject* globalObject,
         return jsUndefined();
     }
     if (auto* chunk = dynamicDowncast<JSC::JSArrayBufferView>(result)) {
-        if (!isClosed)
-            nativeAdjustChunkSize(adapter, chunk->byteLength());
+        // The source allocated its own buffer; the adapter's pull view was not
+        // the size bottleneck, so don't grow it.
         if (chunk->byteLength() > 0) {
             if (adapter->m_textMode) {
                 nativeEnqueueTextChunk(globalObject, controller, adapter->m_textState, chunk->span(), /* flush */ false);
@@ -556,7 +558,13 @@ void materializeNativeSource(JSGlobalObject* globalObject, JSReadableStream* str
     auto* adapter = WebCore::JSNativeStreamSourceAdapter::create(vm, runtime->nativeStreamSourceAdapterStructure(domGlobalObject));
     adapter->setHandle(vm, handle);
     adapter->m_textMode = stream->m_nativeTextMode;
-    adapter->m_chunkSize = std::max(static_cast<size_t>(chunkSize), autoAllocateChunkSize);
+    if (chunkSize < 0) {
+        adapter->m_sourceOwnsChunks = true;
+        adapter->m_hasResized = true;
+        adapter->m_chunkSize = 0;
+    } else {
+        adapter->m_chunkSize = std::max(static_cast<size_t>(chunkSize), autoAllocateChunkSize);
+    }
     auto* closer = JSC::constructEmptyArray(globalObject, nullptr, 1);
     RETURN_IF_EXCEPTION(scope, );
     closer->putDirectIndex(globalObject, 0, jsBoolean(false));
@@ -625,7 +633,8 @@ static JSPromise* nativeSourcePullImpl(JSC::VM& vm, JSGlobalObject* globalObject
     closer->putDirectIndex(globalObject, 0, jsBoolean(false));
     RETURN_IF_EXCEPTION(scope, nullptr);
 
-    if (JSObject* pendingObject = adapter->pendingView()) {
+    JSObject* pendingObject = adapter->pendingView();
+    if (pendingObject || adapter->m_sourceOwnsChunks) {
         MarkedArgumentBuffer noArgs;
         JSValue drained = invokeMethod(vm, globalObject, handle, builtinNames(vm).drainPublicName(), noArgs);
         RETURN_IF_EXCEPTION(scope, nullptr);
@@ -634,7 +643,8 @@ static JSPromise* nativeSourcePullImpl(JSC::VM& vm, JSGlobalObject* globalObject
         if (isTruthy) {
             bool isClosed = nativeCloserFlag(vm, globalObject, adapter);
             RETURN_IF_EXCEPTION(scope, nullptr);
-            JSValue newView = nativeDecodePullResult(vm, globalObject, adapter, controller, drained, uncheckedDowncast<JSC::JSUint8Array>(pendingObject), isClosed);
+            JSC::JSUint8Array* pendingView = pendingObject ? uncheckedDowncast<JSC::JSUint8Array>(pendingObject) : nullptr;
+            JSValue newView = nativeDecodePullResult(vm, globalObject, adapter, controller, drained, pendingView, isClosed);
             RETURN_IF_EXCEPTION(scope, nullptr);
             nativeStorePendingView(vm, adapter, newView);
             return nullptr;
@@ -645,7 +655,7 @@ static JSPromise* nativeSourcePullImpl(JSC::VM& vm, JSGlobalObject* globalObject
     RETURN_IF_EXCEPTION(scope, nullptr);
 
     MarkedArgumentBuffer pullArgs;
-    pullArgs.append(view);
+    pullArgs.append(view ? JSValue(view) : jsUndefined());
     pullArgs.append(closer);
     ASSERT(!pullArgs.hasOverflowed());
     JSValue result = invokeMethod(vm, globalObject, handle, builtinNames(vm).pullPublicName(), pullArgs);
diff --git a/src/jsc/bindings/webcore/streams/BunStreamSource.h b/src/jsc/bindings/webcore/streams/BunStreamSource.h
index 95dd4bebdeb6..7228b88b8d00 100644
--- a/src/jsc/bindings/webcore/streams/BunStreamSource.h
+++ b/src/jsc/bindings/webcore/streams/BunStreamSource.h
@@ -88,6 +88,9 @@ class JSNativeStreamSourceAdapter final : public JSC::JSInternalFieldObjectImpl<
     bool m_closed : 1 { false };
     // Body.textStream(): each pulled byte span is UTF-8-decoded before enqueue.
     bool m_textMode : 1 { false };
+    // start() returned <0: the source hands back its own allocations instead
+    // of writing into the pull view. The adapter never allocates PendingView.
+    bool m_sourceOwnsChunks : 1 { false };
     Bun::WebStreams::StreamingUTF8DecodeState m_textState;
 
 private:
diff --git a/src/runtime/webcore/ByteStream.rs b/src/runtime/webcore/ByteStream.rs
index f873f3b231b0..08be9cb3b75b 100644
--- a/src/runtime/webcore/ByteStream.rs
+++ b/src/runtime/webcore/ByteStream.rs
@@ -6,7 +6,7 @@ use bun_jsc::{self as jsc, JSGlobalObject, JSValue, JsCell};
 use bun_sys::Error as SysError;
 
 use crate::webcore::SinkHandle;
-use crate::webcore::streams::{self, BufferAction, IntoArray};
+use crate::webcore::streams::{self, BufferAction};
 use crate::webcore::{blob, readable_stream};
 
 bun_output::declare_scope!(ByteStream, visible);
@@ -133,17 +133,9 @@ impl ByteStream {
             return streams::Start::OwnedAndDone(Vec::<u8>::move_from_list(buffer));
         }
 
-        if self.high_water_mark == 0 {
-            return streams::Start::Ready;
-        }
-
-        // For HTTP, the maximum streaming response body size will be 512 KB.
-        // #define LIBUS_RECV_BUFFER_LENGTH 524288
-        // For HTTPS, the size is probably quite a bit lower like 64 KB due to TLS transmission.
-        // We add 1 extra page size so that if there's a little bit of excess buffered data, we avoid extra allocations.
-        let page_size: blob::SizeType =
-            blob::SizeType::try_from(bun_sys::page_size()).expect("int cast");
-        streams::Start::ChunkSize((512 * 1024 + page_size).min(self.high_water_mark.max(page_size)))
+        // `on_pull`/`on_data` hand the reader `Owned` buffers rather than copying
+        // into the adapter's pull view, so the adapter skips allocating one.
+        streams::Start::ReadyOwned
     }
 
     fn value(&self) -> JSValue {
@@ -233,7 +225,7 @@ impl ByteStream {
         self.parent_const().producer.get().start();
     }
 
-    pub(crate) fn on_data(&self, mut stream: streams::Result) -> Result<(), bun_jsc::JsTerminated> {
+    pub(crate) fn on_data(&self, stream: streams::Result) -> Result<(), bun_jsc::JsTerminated> {
         bun_jsc::mark_binding!();
         if self.done.get() {
             // The owned `Vec<u8>`/`Vec`
@@ -376,73 +368,54 @@ impl ByteStream {
             return Ok(());
         }
 
-        let chunk = stream.slice();
-
         if self.pending.get().state == streams::PendingState::Pending {
             debug_assert!(self.buffer.get().is_empty());
-            // Re-derive the destination from the GC-rooted view instead of trusting the
-            // raw pointer captured at pull time: JS can detach or transfer the backing
-            // ArrayBuffer between the pull and the data arriving, leaving
-            // `pending_buffer` dangling. A detached view re-derives to an empty slice.
-            let global = self.parent_const().global_this();
-            let mut pending_view = self
-                .pending_value
-                .get()
-                .get()
-                .and_then(|view| view.as_array_buffer(global))
-                .unwrap_or_default();
-            let pending_buf = pending_view.slice_mut();
-            let to_copy_len = chunk.len().min(pending_buf.len());
-            let pending_buffer_len = pending_buf.len();
-            debug_assert!(pending_buf.as_ptr() != chunk.as_ptr());
-            pending_buf[..to_copy_len].copy_from_slice(&chunk[..to_copy_len]);
-            let has_remaining = chunk.len() > to_copy_len;
             self.pending_buffer.set(Self::empty_pending_buffer());
-
-            let is_really_done =
-                self.has_received_last_chunk.get() && to_copy_len <= pending_buffer_len;
-
-            if is_really_done {
-                self.done.set(true);
-
-                if to_copy_len == 0 {
-                    if matches!(stream, streams::Result::Err(_)) {
-                        let err = core::mem::replace(&mut stream, streams::Result::Done);
-                        self.pending.with_mut(|p| p.result = err);
+            // Drop the rooted pull view; the chunk is handed off as its own
+            // allocation below, so the view is never written into.
+            self.pending_value
+                .with_mut(|pv| pv.clear_without_deallocation());
+
+            let is_done = self.has_received_last_chunk.get();
+            let result = match stream {
+                streams::Result::Err(_) => {
+                    self.done.set(true);
+                    stream
+                }
+                streams::Result::Done => {
+                    self.done.set(true);
+                    streams::Result::Done
+                }
+                streams::Result::Owned(owned) | streams::Result::OwnedAndDone(owned) => {
+                    if is_done {
+                        self.done.set(true);
+                        if owned.is_empty() {
+                            streams::Result::Done
+                        } else {
+                            streams::Result::OwnedAndDone(owned)
+                        }
                     } else {
-                        self.pending.with_mut(|p| p.result = streams::Result::Done);
+                        streams::Result::Owned(owned)
                     }
-                } else {
-                    let v = self.value();
-                    self.pending.with_mut(|p| {
-                        p.result = streams::Result::IntoArrayAndDone(IntoArray {
-                            value: v,
-                            len: to_copy_len as blob::SizeType, // @truncate
-                        });
-                    });
                 }
-            } else {
-                let v = self.value();
-                self.pending.with_mut(|p| {
-                    p.result = streams::Result::IntoArray(IntoArray {
-                        value: v,
-                        len: to_copy_len as blob::SizeType, // @truncate
-                    });
-                });
-            }
+                streams::Result::Temporary(temp) | streams::Result::TemporaryAndDone(temp) => {
+                    let owned = temp.slice().to_vec();
+                    if is_done {
+                        self.done.set(true);
+                        if owned.is_empty() {
+                            streams::Result::Done
+                        } else {
+                            streams::Result::OwnedAndDone(owned)
+                        }
+                    } else {
+                        streams::Result::Owned(owned)
+                    }
+                }
+                _ => unreachable!(),
+            };
 
-            if has_remaining {
-                self.append(stream, to_copy_len)
-                    .unwrap_or_else(|_| panic!("Out of memory while copying request body"));
-            } else {
-                // Only resume the producer when the whole chunk fit the pull
-                // view. When the tail spilled into `buffer` the next `on_pull`
-                // signals once it drains, so resuming now would let another
-                // producer chunk land with no reader to take it (it would go
-                // straight to `append` below), inflating `buffer` and the
-                // producer's own staging buffer by an extra recv each cycle.
-                self.signal_drained();
-            }
+            self.pending.with_mut(|p| p.result = result);
+            self.signal_drained();
 
             bun_output::scoped_log!(ByteStream, "ByteStream.onData pending.run()");
 
@@ -523,51 +496,26 @@ impl ByteStream {
         self.pending_value.with_mut(|pv| pv.set(global, view));
     }
 
-    fn on_pull(&self, buffer: &mut [u8], view: JSValue) -> streams::Result {
+    fn on_pull(&self, _buffer: &mut [u8], view: JSValue) -> streams::Result {
         bun_jsc::mark_binding!();
-        debug_assert!(!buffer.is_empty());
         debug_assert!(self.buffer_action.get().is_none());
 
         if !self.buffer.get().is_empty() {
             debug_assert!(self.value().is_empty()); // == .zero
-            // R-2: confine the `&mut Vec<u8>` to a `with_mut` so no `JsCell`
-            // borrow escapes the copy. The result tuple drives the rest.
-            let (to_write, remaining_in_buffer_len) = self.buffer.with_mut(|b| {
-                let to_write = (b.len() - self.offset.get()).min(buffer.len());
-                let remaining_in_buffer_len = to_write; // length of `this.buffer.items[this.offset..][0..to_write]`
-
-                buffer[..to_write].copy_from_slice(&b[self.offset.get()..][..to_write]);
-
-                if self.offset.get() + to_write == b.len() {
-                    self.offset.set(0);
-                    b.clear();
-                } else {
-                    self.offset.set(self.offset.get() + to_write);
-                }
-                (to_write, remaining_in_buffer_len)
-            });
-
-            if self.buffer.get().is_empty() {
-                self.signal_drained();
+            let offset = self.offset.replace(0);
+            let mut owned = self.buffer.replace(Vec::new());
+            if offset > 0 {
+                owned.drain(..offset);
             }
 
-            if self.has_received_last_chunk.get() && remaining_in_buffer_len == 0 {
-                self.buffer.with_mut(|b| {
-                    b.clear();
-                    b.shrink_to_fit();
-                });
-                self.done.set(true);
+            self.signal_drained();
 
-                return streams::Result::IntoArrayAndDone(IntoArray {
-                    value: view,
-                    len: to_write as blob::SizeType, // @truncate
-                });
+            if self.has_received_last_chunk.get() {
+                self.done.set(true);
+                return streams::Result::OwnedAndDone(owned);
             }
 
-            return streams::Result::IntoArray(IntoArray {
-                value: view,
-                len: to_write as blob::SizeType, // @truncate
-            });
+            return streams::Result::Owned(owned);
         }
 
         if self.has_received_last_chunk.get() {
@@ -581,8 +529,8 @@ impl ByteStream {
             return streams::Result::Done;
         }
 
-        // Raw borrow of a JS-owned buffer; rooted by `set_value`.
-        self.pending_buffer.set(std::ptr::from_mut::<[u8]>(buffer));
+        // Parked until `on_data`. The pull view is never written into; it is
+        // rooted only so `on_cancel` can observe that a pull was outstanding.
         self.set_value(view);
 
         // R-2: `JsCell::as_ptr` yields the stable `*mut Pending` that the
diff --git a/src/runtime/webcore/ReadableStream.rs b/src/runtime/webcore/ReadableStream.rs
index 6ef9d5771301..8143b0eb5f36 100644
--- a/src/runtime/webcore/ReadableStream.rs
+++ b/src/runtime/webcore/ReadableStream.rs
@@ -1117,10 +1117,13 @@ impl<C: SourceContext> NewSource<C> {
         let this_jsvalue = call_frame.this();
         let [view, flags] = call_frame.arguments_as_array::<2>();
         view.ensure_still_alive();
-        let Some(mut buffer) = view.as_array_buffer(global_this) else {
-            return Ok(JSValue::UNDEFINED);
+        let result = if let Some(mut buffer) = view.as_array_buffer(global_this) {
+            self.on_pull_from_js(buffer.slice_mut(), view)
+        } else {
+            // `Start::ReadyOwned`: the adapter passes no pull view and the
+            // source allocates its own chunk.
+            self.on_pull_from_js(&mut [], view)
         };
-        let result = self.on_pull_from_js(buffer.slice_mut(), view);
         Self::process_result(this_jsvalue, global_this, flags, result)
     }
 
@@ -1133,6 +1136,9 @@ impl<C: SourceContext> NewSource<C> {
         match self.on_start_from_js() {
             streams::Start::Empty => Ok(JSValue::js_number(0.0)),
             streams::Start::Ready => Ok(JSValue::js_number(16384.0)),
+            // Negative chunk size tells the native adapter this source hands
+            // back its own allocations; it then skips allocating a pull view.
+            streams::Start::ReadyOwned => Ok(JSValue::js_number(-1.0)),
             streams::Start::ChunkSize(size) => Ok(JSValue::js_number(size as f64)),
             streams::Start::Err(err) => Err(global_this.throw_value(err.to_js(global_this))),
             rc => rc.to_js(global_this),
diff --git a/src/runtime/webcore/streams.rs b/src/runtime/webcore/streams.rs
index a26a8e0305a0..1ffd0f9422e9 100644
--- a/src/runtime/webcore/streams.rs
+++ b/src/runtime/webcore/streams.rs
@@ -73,6 +73,10 @@ pub enum Start {
     },
     FileSink(FileSinkOptions),
     Ready,
+    /// Streaming, and `on_pull` hands back its own allocations (`Owned`/
+    /// `Temporary`) rather than writing into the adapter's pull view. The
+    /// adapter skips allocating that view.
+    ReadyOwned,
     OwnedAndDone(Vec<u8>),
 }
 
@@ -97,7 +101,7 @@ pub enum StartTag {
 impl Start {
     pub fn to_js(self, global_this: &JSGlobalObject) -> JsResult<JSValue> {
         match self {
-            Start::Empty | Start::Ready => Ok(JSValue::UNDEFINED),
+            Start::Empty | Start::Ready | Start::ReadyOwned => Ok(JSValue::UNDEFINED),
             Start::ChunkSize(chunk) => Ok(JSValue::from(chunk)),
             Start::Err(err) => Err(err.throw(global_this)),
             Start::OwnedAndDone(list) => {

From cb20309f83d9e9995d0606dc5dcac598c238e18c Mon Sep 17 00:00:00 2001
From: robobun <117481402+robobun@users.noreply.github.com>
Date: Sat, 1 Aug 2026 14:40:59 +0000
Subject: [PATCH 02/15] Add test asserting request-body chunks are right-sized;
 drop dead ByteStream fields

The test sends a 2 MiB body in small writes after the handler's first pull has
parked and asserts every chunk the reader sees has backing === byteLength and
byteOffset === 0. On main each chunk is a subarray over the adapter's
~516 KiB scratch view.

Dead-field cleanup flagged by review: offset, pending_buffer, and
high_water_mark are no longer written non-trivially or read after the Owned
handoff, so remove them along with append's offset parameter, the
empty_pending_buffer helper, and the two Body.rs writers.
---
 src/runtime/server/RequestContext.rs          |   2 +-
 src/runtime/webcore/Body.rs                   |   2 -
 src/runtime/webcore/ByteStream.rs             |  57 ++------
 ...serve-request-body-pipeline-memory.test.ts | 136 ++++++++++++++++++
 4 files changed, 149 insertions(+), 48 deletions(-)
 create mode 100644 test/js/bun/http/serve-request-body-pipeline-memory.test.ts

diff --git a/src/runtime/server/RequestContext.rs b/src/runtime/server/RequestContext.rs
index 845980f5cb61..041ddb0622cb 100644
--- a/src/runtime/server/RequestContext.rs
+++ b/src/runtime/server/RequestContext.rs
@@ -4063,7 +4063,7 @@ where
                 let _ = bytes.on_data(WebCore::streams::Result::Temporary(borrowed));
 
                 // What `on_data` buffered; `on_stream_drained` resumes once it empties.
-                let buffered = bytes.buffer.get().len().saturating_sub(bytes.offset.get());
+                let buffered = bytes.buffer.get().len();
                 if bytes.buffer_action.get().is_some()
                     || (bytes.sink.get().is_some() && !bytes.sink_paused.get())
                 {
diff --git a/src/runtime/webcore/Body.rs b/src/runtime/webcore/Body.rs
index 201e73b7c338..ff8b78c8736c 100644
--- a/src/runtime/webcore/Body.rs
+++ b/src/runtime/webcore/Body.rs
@@ -884,7 +884,6 @@ impl Value {
 
         match drain_result {
             DrainResult::EstimatedSize(estimated_size) => {
-                reader.context.high_water_mark = estimated_size as blob::SizeType;
                 reader
                     .context
                     .size_hint
@@ -1509,7 +1508,6 @@ impl Value {
 
         match drain_result {
             DrainResult::EstimatedSize(estimated_size) => {
-                reader.context.high_water_mark = estimated_size as blob::SizeType;
                 reader
                     .context
                     .size_hint
diff --git a/src/runtime/webcore/ByteStream.rs b/src/runtime/webcore/ByteStream.rs
index 08be9cb3b75b..88a415a31ed7 100644
--- a/src/runtime/webcore/ByteStream.rs
+++ b/src/runtime/webcore/ByteStream.rs
@@ -14,9 +14,7 @@ bun_output::declare_scope!(ByteStream, visible);
 /// R-2 (`sharedThis`): every JS-reachable inherent method takes `&self` so a
 /// re-entrant JS call (e.g. `pending.run()` → JS → `onPull`) cannot stack two
 /// `&mut ByteStream`. Fields mutated on those paths are wrapped in `Cell`
-/// (Copy scalars / raw ptrs) or [`JsCell`] (non-Copy). `high_water_mark` /
-/// `size_hint` are written only at init time (before the JS wrapper exists)
-/// and stay bare.
+/// (Copy scalars / raw ptrs) or [`JsCell`] (non-Copy).
 ///
 /// The `SourceContext` trait still spells its callbacks `&mut self` (shared
 /// across `ByteBlobLoader` / `FileReader`); the trait impl below auto-derefs
@@ -26,13 +24,8 @@ pub struct ByteStream {
     pub(crate) has_received_last_chunk: Cell<bool>,
     pub(crate) pending: JsCell<streams::Pending>,
     pub(crate) done: Cell<bool>,
-    /// Borrowed view into a JS `Uint8Array` passed from `on_pull`; kept alive by `pending_value`.
-    // Raw fat slice ptr because the backing store is JS-heap-owned and rooted via
-    // `pending_value: Strong`. Never freed by Rust.
-    pub(crate) pending_buffer: Cell<*mut [u8]>,
+    /// Rooted only so `on_cancel` can observe that a pull was outstanding.
     pub(crate) pending_value: JsCell<StrongOptional>, // jsc.Strong.Optional
-    pub offset: Cell<usize>,
-    pub(crate) high_water_mark: blob::SizeType,
     /// Native sink this stream is piped into; `on_data` dispatches and honors `Writable`.
     pub(crate) sink: JsCell<SinkHandle>,
     /// Set on `Writable::Backpressure` (buffer instead of write); cleared by [`Self::resume`].
@@ -51,10 +44,7 @@ impl Default for ByteStream {
                 ..Default::default()
             }),
             done: Cell::new(false),
-            pending_buffer: Cell::new(Self::empty_pending_buffer()),
             pending_value: JsCell::new(StrongOptional::empty()),
-            offset: Cell::new(0),
-            high_water_mark: 0,
             sink: JsCell::new(SinkHandle::None),
             sink_paused: Cell::new(false),
             size_hint: Cell::new(0),
@@ -109,11 +99,6 @@ impl readable_stream::SourceContext for ByteStream {
 bun_core::impl_field_parent! { ByteStream => Source.context; pub fn parent_const; pub fn parent; }
 
 impl ByteStream {
-    #[inline]
-    const fn empty_pending_buffer() -> *mut [u8] {
-        core::ptr::slice_from_raw_parts_mut(core::ptr::NonNull::<u8>::dangling().as_ptr(), 0)
-    }
-
     /// Init-time reset. Runs before the JS
     /// wrapper exists, so `&mut self` is sound here (R-2 exemption).
     pub(crate) fn setup(&mut self) {
@@ -167,7 +152,6 @@ impl ByteStream {
 
         if !self.buffer.get().is_empty() {
             let buffered = self.buffer.replace(Vec::new());
-            self.offset.set(0);
             let result = if self.has_received_last_chunk.get() {
                 streams::Result::OwnedAndDone(buffered)
             } else {
@@ -255,8 +239,7 @@ impl ByteStream {
 
             if self.sink_paused.get() {
                 bun_output::scoped_log!(ByteStream, "ByteStream.onData sink paused → buffer");
-                self.append(stream, 0)
-                    .unwrap_or_else(|_| panic!("Out of memory while copying request body"));
+                self.append(stream);
                 return Ok(());
             }
 
@@ -370,7 +353,6 @@ impl ByteStream {
 
         if self.pending.get().state == streams::PendingState::Pending {
             debug_assert!(self.buffer.get().is_empty());
-            self.pending_buffer.set(Self::empty_pending_buffer());
             // Drop the rooted pull view; the chunk is handed off as its own
             // allocation below, so the view is never written into.
             self.pending_value
@@ -430,24 +412,19 @@ impl ByteStream {
 
         bun_output::scoped_log!(ByteStream, "ByteStream.onData no action just append");
 
-        self.append(stream, 0)
-            .unwrap_or_else(|_| panic!("Out of memory while copying request body"));
+        self.append(stream);
         Ok(())
     }
 
-    fn append(&self, stream: streams::Result, offset: usize) -> Result<(), bun_alloc::AllocError> {
+    fn append(&self, stream: streams::Result) {
         if self.buffer.get().capacity() == 0 {
             match stream {
                 streams::Result::Owned(mut owned) | streams::Result::OwnedAndDone(mut owned) => {
                     // `move_to_list_managed` moves the buffer, no copy.
                     self.buffer.set(owned.move_to_list_managed());
-                    self.offset.set(self.offset.get() + offset);
                 }
                 streams::Result::TemporaryAndDone(temp) | streams::Result::Temporary(temp) => {
-                    let chunk = &temp.slice()[offset..];
-                    let mut buf = Vec::with_capacity(chunk.len());
-                    buf.extend_from_slice(chunk);
-                    self.buffer.set(buf);
+                    self.buffer.set(temp.slice().to_vec());
                 }
                 streams::Result::Err(err) => {
                     self.pending
@@ -456,17 +433,15 @@ impl ByteStream {
                 streams::Result::Done => {}
                 _ => unreachable!(),
             }
-            return Ok(());
+            return;
         }
 
         match stream {
             streams::Result::TemporaryAndDone(temp) | streams::Result::Temporary(temp) => {
-                self.buffer
-                    .with_mut(|b| b.extend_from_slice(&temp.slice()[offset..]));
+                self.buffer.with_mut(|b| b.extend_from_slice(temp.slice()));
             }
             streams::Result::OwnedAndDone(owned) | streams::Result::Owned(owned) => {
-                self.buffer
-                    .with_mut(|b| b.extend_from_slice(&owned.slice()[offset..]));
+                self.buffer.with_mut(|b| b.extend_from_slice(owned.slice()));
                 // `owned: Vec<u8>` drops here.
             }
             streams::Result::Err(err) => {
@@ -486,8 +461,6 @@ impl ByteStream {
             // We don't support the rest of these yet
             _ => unreachable!(),
         }
-
-        Ok(())
     }
 
     fn set_value(&self, view: JSValue) {
@@ -502,11 +475,7 @@ impl ByteStream {
 
         if !self.buffer.get().is_empty() {
             debug_assert!(self.value().is_empty()); // == .zero
-            let offset = self.offset.replace(0);
-            let mut owned = self.buffer.replace(Vec::new());
-            if offset > 0 {
-                owned.drain(..offset);
-            }
+            let owned = self.buffer.replace(Vec::new());
 
             self.signal_drained();
 
@@ -529,8 +498,8 @@ impl ByteStream {
             return streams::Result::Done;
         }
 
-        // Parked until `on_data`. The pull view is never written into; it is
-        // rooted only so `on_cancel` can observe that a pull was outstanding.
+        // Parked until `on_data`. Rooting the (possibly `undefined`) pull view
+        // lets `on_cancel` observe that a pull was outstanding.
         self.set_value(view);
 
         // R-2: `JsCell::as_ptr` yields the stable `*mut Pending` that the
@@ -551,7 +520,6 @@ impl ByteStream {
         self.pending_value.with_mut(|pv| pv.deinit());
 
         if !view.is_empty() {
-            self.pending_buffer.set(Self::empty_pending_buffer());
             self.pending.with_mut(|p| {
                 p.result.release();
                 p.result = streams::Result::Done;
@@ -596,7 +564,6 @@ impl ByteStream {
         if !self.done.get() {
             self.done.set(true);
 
-            self.pending_buffer.set(Self::empty_pending_buffer());
             let is_promise = self.pending.with_mut(|p| {
                 p.result.release();
                 p.result = streams::Result::Done;
diff --git a/test/js/bun/http/serve-request-body-pipeline-memory.test.ts b/test/js/bun/http/serve-request-body-pipeline-memory.test.ts
new file mode 100644
index 000000000000..a482e399a74c
--- /dev/null
+++ b/test/js/bun/http/serve-request-body-pipeline-memory.test.ts
@@ -0,0 +1,136 @@
+// `for await (chunk of req.body)` (what `pipeline(req.body, writable)` runs
+// via `pumpToNode`) should yield right-sized chunks. A `Bun.serve` request
+// body is a push source (`ByteStream`): bytes arrive via `on_data` and are
+// handed to the reader as the owning allocation. Previously `on_pull`/`on_data`
+// copied into the native-source adapter's ~256-512 KiB scratch view and
+// surfaced each chunk as a subarray over that whole backing, so every in-flight
+// request held a ~0.5 MB `ArrayBuffer` regardless of how little data was
+// actually read.
+
+import { test, expect } from "bun:test";
+import { connect } from "node:net";
+import { Readable } from "node:stream";
+
+test("for await (req.body) chunks are backed by right-sized buffers, not the adapter's scratch view", async () => {
+  type Seen = { len: number; backing: number; off: number };
+  let handlerDone!: (v: Seen[]) => void;
+  const handlerP = new Promise<Seen[]>(r => {
+    handlerDone = r;
+  });
+  let pullParked!: () => void;
+  const pullParkedP = new Promise<void>(r => {
+    pullParked = r;
+  });
+
+  await using server = Bun.serve({
+    port: 0,
+    async fetch(req) {
+      const seen: Seen[] = [];
+      // Let the client know the first pull has parked (no body bytes yet) so
+      // the first chunk is resolved from on_data, not from drain().
+      queueMicrotask(() => queueMicrotask(pullParked));
+      for await (const chunk of req.body!) {
+        seen.push({
+          len: chunk.byteLength,
+          backing: chunk.buffer.byteLength,
+          off: chunk.byteOffset,
+        });
+      }
+      handlerDone(seen);
+      return new Response("ok");
+    },
+  });
+
+  const sock = connect({ port: server.port, host: "127.0.0.1" });
+  await new Promise<void>((res, rej) => {
+    sock.once("connect", () => res());
+    sock.once("error", rej);
+  });
+
+  // Content-Length large enough that on_start would have sized the pull view
+  // at its ~512 KiB ceiling on main.
+  const BODY = 2 * 1024 * 1024;
+  sock.write(`POST / HTTP/1.1\r\nHost: x\r\nContent-Length: ${BODY}\r\nConnection: close\r\n\r\n`);
+  await pullParkedP;
+  // A couple of small chunks first so their backing size is unambiguous.
+  sock.write(Buffer.alloc(8 * 1024, 0x61));
+  await Bun.sleep(20);
+  sock.write(Buffer.alloc(8 * 1024, 0x62));
+  await Bun.sleep(20);
+  sock.write(Buffer.alloc(BODY - 16 * 1024, 0x63));
+  sock.end();
+
+  const seen = await handlerP;
+  sock.destroy();
+
+  let total = 0;
+  for (const { len } of seen) total += len;
+  expect(total).toBe(BODY);
+
+  // Every chunk is its own allocation: backing size equals payload size and
+  // the view starts at offset 0. On main each chunk was a subarray into a
+  // single ~516 KiB scratch view (backing >> len, off advancing per chunk).
+  for (const { len, backing, off } of seen) {
+    expect({ len, backing, off }).toEqual({ len, backing: len, off: 0 });
+  }
+});
+
+test("Readable.fromWeb(req.body) chunks are backed by right-sized buffers", async () => {
+  type Seen = { len: number; backing: number; off: number };
+  let handlerDone!: (v: Seen[]) => void;
+  const handlerP = new Promise<Seen[]>(r => {
+    handlerDone = r;
+  });
+  let pullParked!: () => void;
+  const pullParkedP = new Promise<void>(r => {
+    pullParked = r;
+  });
+
+  await using server = Bun.serve({
+    port: 0,
+    async fetch(req) {
+      const r = Readable.fromWeb(req.body as any);
+      const seen: Seen[] = [];
+      r.on("data", (chunk: Buffer) => {
+        seen.push({
+          len: chunk.byteLength,
+          backing: chunk.buffer.byteLength,
+          off: chunk.byteOffset,
+        });
+      });
+      queueMicrotask(() => queueMicrotask(pullParked));
+      await new Promise<void>(res => r.once("end", () => res()));
+      handlerDone(seen);
+      return new Response("ok");
+    },
+  });
+
+  const sock = connect({ port: server.port, host: "127.0.0.1" });
+  await new Promise<void>((res, rej) => {
+    sock.once("connect", () => res());
+    sock.once("error", rej);
+  });
+
+  const BODY = 64 * 1024;
+  sock.write(`POST / HTTP/1.1\r\nHost: x\r\nContent-Length: ${BODY}\r\nConnection: close\r\n\r\n`);
+  await pullParkedP;
+  for (let i = 0; i < 4; i++) {
+    sock.write(Buffer.alloc(BODY / 4, 0x61 + i));
+    await Bun.sleep(20);
+  }
+  sock.end();
+
+  const seen = await handlerP;
+  sock.destroy();
+
+  let total = 0;
+  for (const { len } of seen) total += len;
+  expect(total).toBe(BODY);
+
+  // Same invariant for the node:stream adapter path: the native-readable pull
+  // loop previously pre-allocated a 64-256 KiB Buffer and pushed subarrays
+  // into it; now it receives the source's own allocation.
+  for (const { len, backing, off } of seen) {
+    expect({ len, backing, off }).toEqual({ len, backing: len, off: 0 });
+  }
+});

From 3a36bdb24b5fe47fc6f8a9980fb7ee3434ea4f0f Mon Sep 17 00:00:00 2001
From: "autofix-ci[bot]" <114827586+autofix-ci[bot]@users.noreply.github.com>
Date: Sat, 1 Aug 2026 14:43:47 +0000
Subject: [PATCH 03/15] [autofix.ci] apply automated fixes

---
 test/js/bun/http/serve-request-body-pipeline-memory.test.ts | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/test/js/bun/http/serve-request-body-pipeline-memory.test.ts b/test/js/bun/http/serve-request-body-pipeline-memory.test.ts
index a482e399a74c..b07164add943 100644
--- a/test/js/bun/http/serve-request-body-pipeline-memory.test.ts
+++ b/test/js/bun/http/serve-request-body-pipeline-memory.test.ts
@@ -7,7 +7,7 @@
 // request held a ~0.5 MB `ArrayBuffer` regardless of how little data was
 // actually read.
 
-import { test, expect } from "bun:test";
+import { expect, test } from "bun:test";
 import { connect } from "node:net";
 import { Readable } from "node:stream";
 

From ecb6ac6dea0b6984df0b50915fa4ac91d1562159 Mon Sep 17 00:00:00 2001
From: robobun <117481402+robobun@users.noreply.github.com>
Date: Sat, 1 Aug 2026 14:54:10 +0000
Subject: [PATCH 04/15] Trim new code comments to one line each

---
 src/js/internal/streams/native-readable.ts           | 3 +--
 src/jsc/bindings/webcore/streams/BunStreamSource.cpp | 2 --
 src/jsc/bindings/webcore/streams/BunStreamSource.h   | 3 +--
 src/runtime/webcore/ByteStream.rs                    | 7 +------
 src/runtime/webcore/ReadableStream.rs                | 4 ----
 src/runtime/webcore/streams.rs                       | 4 +---
 6 files changed, 4 insertions(+), 19 deletions(-)

diff --git a/src/js/internal/streams/native-readable.ts b/src/js/internal/streams/native-readable.ts
index 3ecd84cbc737..97b9fd86fa8d 100644
--- a/src/js/internal/streams/native-readable.ts
+++ b/src/js/internal/streams/native-readable.ts
@@ -143,8 +143,7 @@ function read(this: NativeReadable, maxToRead: number) {
       this[kHasResized] = true;
       this[kHighWaterMark] = Math.min(this[kHighWaterMark], result);
     } else if (typeof result === "number" && result < 0) {
-      // Source allocates its own chunks; the pull view is unused, so keep it
-      // at the minimum size.
+      // Start::ReadyOwned: the pull view is unused.
       this[kHasResized] = true;
       this[kHighWaterMark] = MIN_BUFFER_SIZE;
     }
diff --git a/src/jsc/bindings/webcore/streams/BunStreamSource.cpp b/src/jsc/bindings/webcore/streams/BunStreamSource.cpp
index 0656e7af228b..355001ccc298 100644
--- a/src/jsc/bindings/webcore/streams/BunStreamSource.cpp
+++ b/src/jsc/bindings/webcore/streams/BunStreamSource.cpp
@@ -480,8 +480,6 @@ static JSValue nativeDecodePullResult(JSC::VM& vm, JSGlobalObject* globalObject,
         return jsUndefined();
     }
     if (auto* chunk = dynamicDowncast<JSC::JSArrayBufferView>(result)) {
-        // The source allocated its own buffer; the adapter's pull view was not
-        // the size bottleneck, so don't grow it.
         if (chunk->byteLength() > 0) {
             if (adapter->m_textMode) {
                 nativeEnqueueTextChunk(globalObject, controller, adapter->m_textState, chunk->span(), /* flush */ false);
diff --git a/src/jsc/bindings/webcore/streams/BunStreamSource.h b/src/jsc/bindings/webcore/streams/BunStreamSource.h
index 7228b88b8d00..5b201a6c1174 100644
--- a/src/jsc/bindings/webcore/streams/BunStreamSource.h
+++ b/src/jsc/bindings/webcore/streams/BunStreamSource.h
@@ -88,8 +88,7 @@ class JSNativeStreamSourceAdapter final : public JSC::JSInternalFieldObjectImpl<
     bool m_closed : 1 { false };
     // Body.textStream(): each pulled byte span is UTF-8-decoded before enqueue.
     bool m_textMode : 1 { false };
-    // start() returned <0: the source hands back its own allocations instead
-    // of writing into the pull view. The adapter never allocates PendingView.
+    // start() returned <0 (Start::ReadyOwned): never allocate PendingView.
     bool m_sourceOwnsChunks : 1 { false };
     Bun::WebStreams::StreamingUTF8DecodeState m_textState;
 
diff --git a/src/runtime/webcore/ByteStream.rs b/src/runtime/webcore/ByteStream.rs
index 88a415a31ed7..9b914a116c62 100644
--- a/src/runtime/webcore/ByteStream.rs
+++ b/src/runtime/webcore/ByteStream.rs
@@ -118,8 +118,6 @@ impl ByteStream {
             return streams::Start::OwnedAndDone(Vec::<u8>::move_from_list(buffer));
         }
 
-        // `on_pull`/`on_data` hand the reader `Owned` buffers rather than copying
-        // into the adapter's pull view, so the adapter skips allocating one.
         streams::Start::ReadyOwned
     }
 
@@ -353,8 +351,6 @@ impl ByteStream {
 
         if self.pending.get().state == streams::PendingState::Pending {
             debug_assert!(self.buffer.get().is_empty());
-            // Drop the rooted pull view; the chunk is handed off as its own
-            // allocation below, so the view is never written into.
             self.pending_value
                 .with_mut(|pv| pv.clear_without_deallocation());
 
@@ -498,8 +494,7 @@ impl ByteStream {
             return streams::Result::Done;
         }
 
-        // Parked until `on_data`. Rooting the (possibly `undefined`) pull view
-        // lets `on_cancel` observe that a pull was outstanding.
+        // Rooted only so `on_cancel` can observe an outstanding pull.
         self.set_value(view);
 
         // R-2: `JsCell::as_ptr` yields the stable `*mut Pending` that the
diff --git a/src/runtime/webcore/ReadableStream.rs b/src/runtime/webcore/ReadableStream.rs
index 8143b0eb5f36..342c0e0ad7c4 100644
--- a/src/runtime/webcore/ReadableStream.rs
+++ b/src/runtime/webcore/ReadableStream.rs
@@ -1120,8 +1120,6 @@ impl<C: SourceContext> NewSource<C> {
         let result = if let Some(mut buffer) = view.as_array_buffer(global_this) {
             self.on_pull_from_js(buffer.slice_mut(), view)
         } else {
-            // `Start::ReadyOwned`: the adapter passes no pull view and the
-            // source allocates its own chunk.
             self.on_pull_from_js(&mut [], view)
         };
         Self::process_result(this_jsvalue, global_this, flags, result)
@@ -1136,8 +1134,6 @@ impl<C: SourceContext> NewSource<C> {
         match self.on_start_from_js() {
             streams::Start::Empty => Ok(JSValue::js_number(0.0)),
             streams::Start::Ready => Ok(JSValue::js_number(16384.0)),
-            // Negative chunk size tells the native adapter this source hands
-            // back its own allocations; it then skips allocating a pull view.
             streams::Start::ReadyOwned => Ok(JSValue::js_number(-1.0)),
             streams::Start::ChunkSize(size) => Ok(JSValue::js_number(size as f64)),
             streams::Start::Err(err) => Err(global_this.throw_value(err.to_js(global_this))),
diff --git a/src/runtime/webcore/streams.rs b/src/runtime/webcore/streams.rs
index 1ffd0f9422e9..b8138c45c72b 100644
--- a/src/runtime/webcore/streams.rs
+++ b/src/runtime/webcore/streams.rs
@@ -73,9 +73,7 @@ pub enum Start {
     },
     FileSink(FileSinkOptions),
     Ready,
-    /// Streaming, and `on_pull` hands back its own allocations (`Owned`/
-    /// `Temporary`) rather than writing into the adapter's pull view. The
-    /// adapter skips allocating that view.
+    /// `on_pull` hands back `Owned` chunks; the adapter skips its pull view.
     ReadyOwned,
     OwnedAndDone(Vec<u8>),
 }

From 3088550ecaa9b04178daf406724d6794fa530a2b Mon Sep 17 00:00:00 2001
From: robobun <117481402+robobun@users.noreply.github.com>
Date: Sat, 1 Aug 2026 15:09:19 +0000
Subject: [PATCH 05/15] native-readable: skip the pull Buffer.alloc entirely
 for ReadyOwned sources

The kHighWaterMark assignment was dead: _read(n) is always passed Readable's
own state.highWaterMark, so getRemainingChunk's nullish fallback never fires.
Stash the negative start() result on kSourceOwnsChunks instead and pass
undefined straight to ptr.pull (pull_from_js already handles a missing view).

Also wire handler-side errors in the new test to the awaited promise so a
regression surfaces the stream error instead of timing out.
---
 src/js/internal/streams/native-readable.ts    |   6 +-
 ...serve-request-body-pipeline-memory.test.ts | 152 ++++++++----------
 2 files changed, 66 insertions(+), 92 deletions(-)

diff --git a/src/js/internal/streams/native-readable.ts b/src/js/internal/streams/native-readable.ts
index 97b9fd86fa8d..fcd911b48e9e 100644
--- a/src/js/internal/streams/native-readable.ts
+++ b/src/js/internal/streams/native-readable.ts
@@ -20,6 +20,7 @@ const kHighWaterMark = Symbol("highWaterMark");
 const kPendingRead = Symbol("pendingRead");
 const kHasResized = Symbol("hasResized");
 const kRemainingChunk = Symbol("remainingChunk");
+const kSourceOwnsChunks = Symbol("sourceOwnsChunks");
 
 const MIN_BUFFER_SIZE = 512;
 let dynamicallyAdjustChunkSize = (_?) => (
@@ -143,9 +144,8 @@ function read(this: NativeReadable, maxToRead: number) {
       this[kHasResized] = true;
       this[kHighWaterMark] = Math.min(this[kHighWaterMark], result);
     } else if (typeof result === "number" && result < 0) {
-      // Start::ReadyOwned: the pull view is unused.
       this[kHasResized] = true;
-      this[kHighWaterMark] = MIN_BUFFER_SIZE;
+      this[kSourceOwnsChunks] = true;
     }
     if ($isTypedArrayView(result) && result.byteLength > 0) {
       pushAndCheck(this, result);
@@ -157,7 +157,7 @@ function read(this: NativeReadable, maxToRead: number) {
       pushAndCheck(this, drainResult);
     }
   }
-  const chunk = getRemainingChunk(this, maxToRead);
+  const chunk = this[kSourceOwnsChunks] ? undefined : getRemainingChunk(this, maxToRead);
   var result = ptr.pull(chunk, this[kCloseState]);
   $assert(result !== undefined);
   $debug(
diff --git a/test/js/bun/http/serve-request-body-pipeline-memory.test.ts b/test/js/bun/http/serve-request-body-pipeline-memory.test.ts
index b07164add943..964c8f6bbed9 100644
--- a/test/js/bun/http/serve-request-body-pipeline-memory.test.ts
+++ b/test/js/bun/http/serve-request-body-pipeline-memory.test.ts
@@ -11,126 +11,100 @@ import { expect, test } from "bun:test";
 import { connect } from "node:net";
 import { Readable } from "node:stream";
 
-test("for await (req.body) chunks are backed by right-sized buffers, not the adapter's scratch view", async () => {
-  type Seen = { len: number; backing: number; off: number };
-  let handlerDone!: (v: Seen[]) => void;
-  const handlerP = new Promise<Seen[]>(r => {
-    handlerDone = r;
-  });
-  let pullParked!: () => void;
-  const pullParkedP = new Promise<void>(r => {
-    pullParked = r;
-  });
+type Seen = { len: number; backing: number; off: number };
+
+async function runUpload(
+  handler: (req: Request, onParked: () => void) => Promise<Seen[]>,
+  bodyBytes: number,
+  writes: number[],
+): Promise<Seen[]> {
+  const { promise: handlerP, resolve: handlerDone, reject: handlerFail } = Promise.withResolvers<Seen[]>();
+  const { promise: pullParkedP, resolve: pullParked } = Promise.withResolvers<void>();
 
   await using server = Bun.serve({
     port: 0,
     async fetch(req) {
-      const seen: Seen[] = [];
-      // Let the client know the first pull has parked (no body bytes yet) so
-      // the first chunk is resolved from on_data, not from drain().
-      queueMicrotask(() => queueMicrotask(pullParked));
-      for await (const chunk of req.body!) {
-        seen.push({
-          len: chunk.byteLength,
-          backing: chunk.buffer.byteLength,
-          off: chunk.byteOffset,
-        });
+      try {
+        const seen = await handler(req, () => queueMicrotask(() => queueMicrotask(pullParked)));
+        handlerDone(seen);
+      } catch (e) {
+        handlerFail(e);
       }
-      handlerDone(seen);
       return new Response("ok");
     },
   });
 
   const sock = connect({ port: server.port, host: "127.0.0.1" });
+  sock.on("error", handlerFail);
   await new Promise<void>((res, rej) => {
     sock.once("connect", () => res());
     sock.once("error", rej);
   });
 
-  // Content-Length large enough that on_start would have sized the pull view
-  // at its ~512 KiB ceiling on main.
-  const BODY = 2 * 1024 * 1024;
-  sock.write(`POST / HTTP/1.1\r\nHost: x\r\nContent-Length: ${BODY}\r\nConnection: close\r\n\r\n`);
+  sock.write(`POST / HTTP/1.1\r\nHost: x\r\nContent-Length: ${bodyBytes}\r\nConnection: close\r\n\r\n`);
   await pullParkedP;
-  // A couple of small chunks first so their backing size is unambiguous.
-  sock.write(Buffer.alloc(8 * 1024, 0x61));
-  await Bun.sleep(20);
-  sock.write(Buffer.alloc(8 * 1024, 0x62));
-  await Bun.sleep(20);
-  sock.write(Buffer.alloc(BODY - 16 * 1024, 0x63));
+  for (const n of writes) {
+    sock.write(Buffer.alloc(n, 0x61));
+    await Bun.sleep(20);
+  }
   sock.end();
 
   const seen = await handlerP;
   sock.destroy();
+  return seen;
+}
 
+function checkRightSized(seen: Seen[], expectedTotal: number) {
   let total = 0;
   for (const { len } of seen) total += len;
-  expect(total).toBe(BODY);
-
-  // Every chunk is its own allocation: backing size equals payload size and
-  // the view starts at offset 0. On main each chunk was a subarray into a
-  // single ~516 KiB scratch view (backing >> len, off advancing per chunk).
+  expect(total).toBe(expectedTotal);
+  // Every chunk is its own allocation. On main each chunk was a subarray into
+  // a single ~516 KiB (for await) / ~64 KiB (fromWeb) scratch view, so
+  // `backing >> len` and `off` advanced per chunk.
   for (const { len, backing, off } of seen) {
     expect({ len, backing, off }).toEqual({ len, backing: len, off: 0 });
   }
+}
+
+test("for await (req.body) chunks are backed by right-sized buffers, not the adapter's scratch view", async () => {
+  // Content-Length large enough that on_start would have sized the pull view
+  // at its ~512 KiB ceiling on main.
+  const BODY = 2 * 1024 * 1024;
+  const seen = await runUpload(
+    async (req, onParked) => {
+      const out: Seen[] = [];
+      // Let the client know the first pull has parked (no body bytes yet) so
+      // the first chunk is resolved from on_data, not from drain().
+      onParked();
+      for await (const chunk of req.body!) {
+        out.push({ len: chunk.byteLength, backing: chunk.buffer.byteLength, off: chunk.byteOffset });
+      }
+      return out;
+    },
+    BODY,
+    [8 * 1024, 8 * 1024, BODY - 16 * 1024],
+  );
+  checkRightSized(seen, BODY);
 });
 
 test("Readable.fromWeb(req.body) chunks are backed by right-sized buffers", async () => {
-  type Seen = { len: number; backing: number; off: number };
-  let handlerDone!: (v: Seen[]) => void;
-  const handlerP = new Promise<Seen[]>(r => {
-    handlerDone = r;
-  });
-  let pullParked!: () => void;
-  const pullParkedP = new Promise<void>(r => {
-    pullParked = r;
-  });
-
-  await using server = Bun.serve({
-    port: 0,
-    async fetch(req) {
+  const BODY = 64 * 1024;
+  const seen = await runUpload(
+    async (req, onParked) => {
       const r = Readable.fromWeb(req.body as any);
-      const seen: Seen[] = [];
+      const out: Seen[] = [];
       r.on("data", (chunk: Buffer) => {
-        seen.push({
-          len: chunk.byteLength,
-          backing: chunk.buffer.byteLength,
-          off: chunk.byteOffset,
-        });
+        out.push({ len: chunk.byteLength, backing: chunk.buffer.byteLength, off: chunk.byteOffset });
       });
-      queueMicrotask(() => queueMicrotask(pullParked));
-      await new Promise<void>(res => r.once("end", () => res()));
-      handlerDone(seen);
-      return new Response("ok");
+      onParked();
+      await new Promise<void>((res, rej) => {
+        r.once("end", () => res());
+        r.once("error", rej);
+      });
+      return out;
     },
-  });
-
-  const sock = connect({ port: server.port, host: "127.0.0.1" });
-  await new Promise<void>((res, rej) => {
-    sock.once("connect", () => res());
-    sock.once("error", rej);
-  });
-
-  const BODY = 64 * 1024;
-  sock.write(`POST / HTTP/1.1\r\nHost: x\r\nContent-Length: ${BODY}\r\nConnection: close\r\n\r\n`);
-  await pullParkedP;
-  for (let i = 0; i < 4; i++) {
-    sock.write(Buffer.alloc(BODY / 4, 0x61 + i));
-    await Bun.sleep(20);
-  }
-  sock.end();
-
-  const seen = await handlerP;
-  sock.destroy();
-
-  let total = 0;
-  for (const { len } of seen) total += len;
-  expect(total).toBe(BODY);
-
-  // Same invariant for the node:stream adapter path: the native-readable pull
-  // loop previously pre-allocated a 64-256 KiB Buffer and pushed subarrays
-  // into it; now it receives the source's own allocation.
-  for (const { len, backing, off } of seen) {
-    expect({ len, backing, off }).toEqual({ len, backing: len, off: 0 });
-  }
+    BODY,
+    [BODY / 4, BODY / 4, BODY / 4, BODY / 4],
+  );
+  checkRightSized(seen, BODY);
 });

From c306505a04709a72f2b9bde4944b5c25dc57e1ab Mon Sep 17 00:00:00 2001
From: robobun <117481402+robobun@users.noreply.github.com>
Date: Sat, 1 Aug 2026 15:50:20 +0000
Subject: [PATCH 06/15] Drop ByteStream.pending_value and wire pullParkedP
 reject in the test

pending_value now only ever held jsUndefined() and was read solely by
on_cancel's outstanding-pull check, which pending.state == Pending already
answers. Remove the field, set_value()/value(), the clear in on_data, and the
two deinit() sites; on_cancel and the on_pull debug_assert now gate on
pending.state directly.

Test: give pullParkedP a reject arm shared with handlerP so a handler/socket
failure before onParked surfaces instead of hanging, and move the body writes
and handler await into a try/finally that always sock.destroy()s.
---
 src/runtime/webcore/ByteStream.rs             | 34 ++-----------------
 ...serve-request-body-pipeline-memory.test.ts | 31 ++++++++++-------
 2 files changed, 21 insertions(+), 44 deletions(-)

diff --git a/src/runtime/webcore/ByteStream.rs b/src/runtime/webcore/ByteStream.rs
index 9b914a116c62..7f79b7be0f76 100644
--- a/src/runtime/webcore/ByteStream.rs
+++ b/src/runtime/webcore/ByteStream.rs
@@ -1,7 +1,6 @@
 use core::cell::Cell;
 
 use bun_collections::VecExt;
-use bun_jsc::strong::Optional as StrongOptional;
 use bun_jsc::{self as jsc, JSGlobalObject, JSValue, JsCell};
 use bun_sys::Error as SysError;
 
@@ -24,8 +23,6 @@ pub struct ByteStream {
     pub(crate) has_received_last_chunk: Cell<bool>,
     pub(crate) pending: JsCell<streams::Pending>,
     pub(crate) done: Cell<bool>,
-    /// Rooted only so `on_cancel` can observe that a pull was outstanding.
-    pub(crate) pending_value: JsCell<StrongOptional>, // jsc.Strong.Optional
     /// Native sink this stream is piped into; `on_data` dispatches and honors `Writable`.
     pub(crate) sink: JsCell<SinkHandle>,
     /// Set on `Writable::Backpressure` (buffer instead of write); cleared by [`Self::resume`].
@@ -44,7 +41,6 @@ impl Default for ByteStream {
                 ..Default::default()
             }),
             done: Cell::new(false),
-            pending_value: JsCell::new(StrongOptional::empty()),
             sink: JsCell::new(SinkHandle::None),
             sink_paused: Cell::new(false),
             size_hint: Cell::new(0),
@@ -121,16 +117,6 @@ impl ByteStream {
         streams::Start::ReadyOwned
     }
 
-    fn value(&self) -> JSValue {
-        self.pending_value.with_mut(|pv| {
-            let Some(result) = pv.get() else {
-                return JSValue::ZERO;
-            };
-            pv.clear_without_deallocation();
-            result
-        })
-    }
-
     pub(crate) fn unpipe_without_deref(&self) {
         self.sink.set(SinkHandle::None);
         self.sink_paused.set(false);
@@ -351,8 +337,6 @@ impl ByteStream {
 
         if self.pending.get().state == streams::PendingState::Pending {
             debug_assert!(self.buffer.get().is_empty());
-            self.pending_value
-                .with_mut(|pv| pv.clear_without_deallocation());
 
             let is_done = self.has_received_last_chunk.get();
             let result = match stream {
@@ -459,18 +443,12 @@ impl ByteStream {
         }
     }
 
-    fn set_value(&self, view: JSValue) {
-        bun_jsc::mark_binding!();
-        let global = self.parent_const().global_this();
-        self.pending_value.with_mut(|pv| pv.set(global, view));
-    }
-
-    fn on_pull(&self, _buffer: &mut [u8], view: JSValue) -> streams::Result {
+    fn on_pull(&self, _buffer: &mut [u8], _view: JSValue) -> streams::Result {
         bun_jsc::mark_binding!();
         debug_assert!(self.buffer_action.get().is_none());
 
         if !self.buffer.get().is_empty() {
-            debug_assert!(self.value().is_empty()); // == .zero
+            debug_assert!(self.pending.get().state != streams::PendingState::Pending);
             let owned = self.buffer.replace(Vec::new());
 
             self.signal_drained();
@@ -494,9 +472,6 @@ impl ByteStream {
             return streams::Result::Done;
         }
 
-        // Rooted only so `on_cancel` can observe an outstanding pull.
-        self.set_value(view);
-
         // R-2: `JsCell::as_ptr` yields the stable `*mut Pending` that the
         // returned `streams::Result::Pending` raw-backref needs.
         streams::Result::Pending(self.pending.as_ptr())
@@ -504,7 +479,6 @@ impl ByteStream {
 
     fn on_cancel(&self) {
         bun_jsc::mark_binding!();
-        let view = self.value();
         if self.buffer.get().capacity() > 0 {
             self.buffer.with_mut(|b| {
                 b.clear();
@@ -512,9 +486,8 @@ impl ByteStream {
             });
         }
         self.done.set(true);
-        self.pending_value.with_mut(|pv| pv.deinit());
 
-        if !view.is_empty() {
+        if self.pending.get().state == streams::PendingState::Pending {
             self.pending.with_mut(|p| {
                 p.result.release();
                 p.result = streams::Result::Done;
@@ -555,7 +528,6 @@ impl ByteStream {
             });
         }
 
-        self.pending_value.with_mut(|pv| pv.deinit());
         if !self.done.get() {
             self.done.set(true);
 
diff --git a/test/js/bun/http/serve-request-body-pipeline-memory.test.ts b/test/js/bun/http/serve-request-body-pipeline-memory.test.ts
index 964c8f6bbed9..0a65aec6c3c3 100644
--- a/test/js/bun/http/serve-request-body-pipeline-memory.test.ts
+++ b/test/js/bun/http/serve-request-body-pipeline-memory.test.ts
@@ -19,7 +19,11 @@ async function runUpload(
   writes: number[],
 ): Promise<Seen[]> {
   const { promise: handlerP, resolve: handlerDone, reject: handlerFail } = Promise.withResolvers<Seen[]>();
-  const { promise: pullParkedP, resolve: pullParked } = Promise.withResolvers<void>();
+  const { promise: pullParkedP, resolve: pullParked, reject: pullParkedFail } = Promise.withResolvers<void>();
+  const fail = (e: unknown) => {
+    handlerFail(e);
+    pullParkedFail(e);
+  };
 
   await using server = Bun.serve({
     port: 0,
@@ -28,30 +32,31 @@ async function runUpload(
         const seen = await handler(req, () => queueMicrotask(() => queueMicrotask(pullParked)));
         handlerDone(seen);
       } catch (e) {
-        handlerFail(e);
+        fail(e);
       }
       return new Response("ok");
     },
   });
 
   const sock = connect({ port: server.port, host: "127.0.0.1" });
-  sock.on("error", handlerFail);
+  sock.on("error", fail);
   await new Promise<void>((res, rej) => {
     sock.once("connect", () => res());
     sock.once("error", rej);
   });
 
-  sock.write(`POST / HTTP/1.1\r\nHost: x\r\nContent-Length: ${bodyBytes}\r\nConnection: close\r\n\r\n`);
-  await pullParkedP;
-  for (const n of writes) {
-    sock.write(Buffer.alloc(n, 0x61));
-    await Bun.sleep(20);
+  try {
+    sock.write(`POST / HTTP/1.1\r\nHost: x\r\nContent-Length: ${bodyBytes}\r\nConnection: close\r\n\r\n`);
+    await pullParkedP;
+    for (const n of writes) {
+      sock.write(Buffer.alloc(n, 0x61));
+      await Bun.sleep(20);
+    }
+    sock.end();
+    return await handlerP;
+  } finally {
+    sock.destroy();
   }
-  sock.end();
-
-  const seen = await handlerP;
-  sock.destroy();
-  return seen;
 }
 
 function checkRightSized(seen: Seen[], expectedTotal: number) {

From 3bfb59f07636b430ce2204288500b6c8532b152a Mon Sep 17 00:00:00 2001
From: robobun <117481402+robobun@users.noreply.github.com>
Date: Sat, 1 Aug 2026 16:08:07 +0000
Subject: [PATCH 07/15] Initialize kSourceOwnsChunks in
 constructNativeReadable; assert chunk count in test

---
 src/js/internal/streams/native-readable.ts                  | 5 ++++-
 test/js/bun/http/serve-request-body-pipeline-memory.test.ts | 4 ++++
 2 files changed, 8 insertions(+), 1 deletion(-)

diff --git a/src/js/internal/streams/native-readable.ts b/src/js/internal/streams/native-readable.ts
index fcd911b48e9e..cf7dade6d241 100644
--- a/src/js/internal/streams/native-readable.ts
+++ b/src/js/internal/streams/native-readable.ts
@@ -37,7 +37,8 @@ type NativeReadable = typeof import("node:stream").Readable &
     [kPendingRead]: boolean;
     [kHighWaterMark]: number;
     [kHasResized]: boolean;
-    [kRemainingChunk]: Buffer;
+    [kRemainingChunk]: Buffer | undefined;
+    [kSourceOwnsChunks]: boolean;
     debugId: number;
   };
 
@@ -73,6 +74,8 @@ function constructNativeReadable(readableStream: ReadableStream, options): Nativ
   stream[kPendingRead] = false;
   stream[kHasResized] = !dynamicallyAdjustChunkSize();
   stream[kCloseState] = [false];
+  stream[kRemainingChunk] = undefined;
+  stream[kSourceOwnsChunks] = false;
 
   const highWaterMark = options.highWaterMark;
   stream[kHighWaterMark] = typeof highWaterMark === "number" ? highWaterMark : 256 * 1024;
diff --git a/test/js/bun/http/serve-request-body-pipeline-memory.test.ts b/test/js/bun/http/serve-request-body-pipeline-memory.test.ts
index 0a65aec6c3c3..ba83280c804a 100644
--- a/test/js/bun/http/serve-request-body-pipeline-memory.test.ts
+++ b/test/js/bun/http/serve-request-body-pipeline-memory.test.ts
@@ -63,6 +63,10 @@ function checkRightSized(seen: Seen[], expectedTotal: number) {
   let total = 0;
   for (const { len } of seen) total += len;
   expect(total).toBe(expectedTotal);
+  // The writes are paced so on_data fires more than once; if the kernel
+  // coalesced them into a single recv this assertion surfaces it instead of
+  // silently dropping the multi-chunk coverage.
+  expect(seen.length).toBeGreaterThan(1);
   // Every chunk is its own allocation. On main each chunk was a subarray into
   // a single ~516 KiB (for await) / ~64 KiB (fromWeb) scratch view, so
   // `backing >> len` and `off` advanced per chunk.

From 5600b4dd4fb56eb5aee9920fe97ea8f0abb5f972 Mon Sep 17 00:00:00 2001
From: robobun <117481402+robobun@users.noreply.github.com>
Date: Sat, 1 Aug 2026 16:28:39 +0000
Subject: [PATCH 08/15] Test: replace write-pacing sleep with a per-chunk ack
 from the handler

The client now awaits onChunk() after each write, so the count assertion is
deterministic rather than relying on 20 ms of wall time outrunning kernel
coalescing.
---
 ...serve-request-body-pipeline-memory.test.ts | 37 +++++++++++--------
 1 file changed, 22 insertions(+), 15 deletions(-)

diff --git a/test/js/bun/http/serve-request-body-pipeline-memory.test.ts b/test/js/bun/http/serve-request-body-pipeline-memory.test.ts
index ba83280c804a..f2663e85f157 100644
--- a/test/js/bun/http/serve-request-body-pipeline-memory.test.ts
+++ b/test/js/bun/http/serve-request-body-pipeline-memory.test.ts
@@ -14,22 +14,28 @@ import { Readable } from "node:stream";
 type Seen = { len: number; backing: number; off: number };
 
 async function runUpload(
-  handler: (req: Request, onParked: () => void) => Promise<Seen[]>,
+  handler: (req: Request, onParked: () => void, onChunk: () => void) => Promise<Seen[]>,
   bodyBytes: number,
   writes: number[],
 ): Promise<Seen[]> {
   const { promise: handlerP, resolve: handlerDone, reject: handlerFail } = Promise.withResolvers<Seen[]>();
   const { promise: pullParkedP, resolve: pullParked, reject: pullParkedFail } = Promise.withResolvers<void>();
+  // One ack per client write so the next write only leaves after the server
+  // has observed the previous one as a distinct on_data chunk.
+  const acks = writes.map(() => Promise.withResolvers<void>());
+  let ackIndex = 0;
+  const onChunk = () => acks[ackIndex++]?.resolve();
   const fail = (e: unknown) => {
     handlerFail(e);
     pullParkedFail(e);
+    for (const a of acks) a.reject(e);
   };
 
   await using server = Bun.serve({
     port: 0,
     async fetch(req) {
       try {
-        const seen = await handler(req, () => queueMicrotask(() => queueMicrotask(pullParked)));
+        const seen = await handler(req, () => queueMicrotask(() => queueMicrotask(pullParked)), onChunk);
         handlerDone(seen);
       } catch (e) {
         fail(e);
@@ -48,9 +54,9 @@ async function runUpload(
   try {
     sock.write(`POST / HTTP/1.1\r\nHost: x\r\nContent-Length: ${bodyBytes}\r\nConnection: close\r\n\r\n`);
     await pullParkedP;
-    for (const n of writes) {
+    for (const [i, n] of writes.entries()) {
       sock.write(Buffer.alloc(n, 0x61));
-      await Bun.sleep(20);
+      await acks[i].promise;
     }
     sock.end();
     return await handlerP;
@@ -59,14 +65,11 @@ async function runUpload(
   }
 }
 
-function checkRightSized(seen: Seen[], expectedTotal: number) {
+function checkRightSized(seen: Seen[], expectedTotal: number, minChunks: number) {
   let total = 0;
   for (const { len } of seen) total += len;
   expect(total).toBe(expectedTotal);
-  // The writes are paced so on_data fires more than once; if the kernel
-  // coalesced them into a single recv this assertion surfaces it instead of
-  // silently dropping the multi-chunk coverage.
-  expect(seen.length).toBeGreaterThan(1);
+  expect(seen.length).toBeGreaterThanOrEqual(minChunks);
   // Every chunk is its own allocation. On main each chunk was a subarray into
   // a single ~516 KiB (for await) / ~64 KiB (fromWeb) scratch view, so
   // `backing >> len` and `off` advanced per chunk.
@@ -79,31 +82,35 @@ test("for await (req.body) chunks are backed by right-sized buffers, not the ada
   // Content-Length large enough that on_start would have sized the pull view
   // at its ~512 KiB ceiling on main.
   const BODY = 2 * 1024 * 1024;
+  const writes = [8 * 1024, 8 * 1024, BODY - 16 * 1024];
   const seen = await runUpload(
-    async (req, onParked) => {
+    async (req, onParked, onChunk) => {
       const out: Seen[] = [];
       // Let the client know the first pull has parked (no body bytes yet) so
       // the first chunk is resolved from on_data, not from drain().
       onParked();
       for await (const chunk of req.body!) {
         out.push({ len: chunk.byteLength, backing: chunk.buffer.byteLength, off: chunk.byteOffset });
+        onChunk();
       }
       return out;
     },
     BODY,
-    [8 * 1024, 8 * 1024, BODY - 16 * 1024],
+    writes,
   );
-  checkRightSized(seen, BODY);
+  checkRightSized(seen, BODY, writes.length);
 });
 
 test("Readable.fromWeb(req.body) chunks are backed by right-sized buffers", async () => {
   const BODY = 64 * 1024;
+  const writes = [BODY / 4, BODY / 4, BODY / 4, BODY / 4];
   const seen = await runUpload(
-    async (req, onParked) => {
+    async (req, onParked, onChunk) => {
       const r = Readable.fromWeb(req.body as any);
       const out: Seen[] = [];
       r.on("data", (chunk: Buffer) => {
         out.push({ len: chunk.byteLength, backing: chunk.buffer.byteLength, off: chunk.byteOffset });
+        onChunk();
       });
       onParked();
       await new Promise<void>((res, rej) => {
@@ -113,7 +120,7 @@ test("Readable.fromWeb(req.body) chunks are backed by right-sized buffers", asyn
       return out;
     },
     BODY,
-    [BODY / 4, BODY / 4, BODY / 4, BODY / 4],
+    writes,
   );
-  checkRightSized(seen, BODY);
+  checkRightSized(seen, BODY, writes.length);
 });

From 546bf7429ad86b72e41d0b0e110ee028319ab74b Mon Sep 17 00:00:00 2001
From: robobun <117481402+robobun@users.noreply.github.com>
Date: Sat, 1 Aug 2026 17:02:08 +0000
Subject: [PATCH 09/15] Test: mark every ack/handler promise observed so a
 failure surfaces once

---
 test/js/bun/http/serve-request-body-pipeline-memory.test.ts | 6 ++++++
 1 file changed, 6 insertions(+)

diff --git a/test/js/bun/http/serve-request-body-pipeline-memory.test.ts b/test/js/bun/http/serve-request-body-pipeline-memory.test.ts
index f2663e85f157..4c24a8e26836 100644
--- a/test/js/bun/http/serve-request-body-pipeline-memory.test.ts
+++ b/test/js/bun/http/serve-request-body-pipeline-memory.test.ts
@@ -23,6 +23,12 @@ async function runUpload(
   // One ack per client write so the next write only leaves after the server
   // has observed the previous one as a distinct on_data chunk.
   const acks = writes.map(() => Promise.withResolvers<void>());
+  // `fail` rejects every promise, but only the one currently awaited has a
+  // handler at that point; mark the rest observed so a regression surfaces as
+  // one failure instead of one plus N unhandled-rejection noise entries.
+  void handlerP.catch(() => {});
+  void pullParkedP.catch(() => {});
+  for (const a of acks) void a.promise.catch(() => {});
   let ackIndex = 0;
   const onChunk = () => acks[ackIndex++]?.resolve();
   const fail = (e: unknown) => {

From 8537aa0f7ccc1602e749a064cb415e587c03ecf7 Mon Sep 17 00:00:00 2001
From: robobun <117481402+robobun@users.noreply.github.com>
Date: Sat, 1 Aug 2026 17:32:05 +0000
Subject: [PATCH 10/15] Drop the m_sourceOwnsChunks drain-path disjunct

on_pull subsumes drain for a ReadyOwned source and also writes closer[0] on
OwnedAndDone, so the last-chunk close lands in the same pull instead of the
next one. pendingView is provably null under m_sourceOwnsChunks, so the gate
reverts to the original if (pendingView()).
---
 src/jsc/bindings/webcore/streams/BunStreamSource.cpp | 6 ++----
 1 file changed, 2 insertions(+), 4 deletions(-)

diff --git a/src/jsc/bindings/webcore/streams/BunStreamSource.cpp b/src/jsc/bindings/webcore/streams/BunStreamSource.cpp
index 355001ccc298..a567ee5342b7 100644
--- a/src/jsc/bindings/webcore/streams/BunStreamSource.cpp
+++ b/src/jsc/bindings/webcore/streams/BunStreamSource.cpp
@@ -631,8 +631,7 @@ static JSPromise* nativeSourcePullImpl(JSC::VM& vm, JSGlobalObject* globalObject
     closer->putDirectIndex(globalObject, 0, jsBoolean(false));
     RETURN_IF_EXCEPTION(scope, nullptr);
 
-    JSObject* pendingObject = adapter->pendingView();
-    if (pendingObject || adapter->m_sourceOwnsChunks) {
+    if (JSObject* pendingObject = adapter->pendingView()) {
         MarkedArgumentBuffer noArgs;
         JSValue drained = invokeMethod(vm, globalObject, handle, builtinNames(vm).drainPublicName(), noArgs);
         RETURN_IF_EXCEPTION(scope, nullptr);
@@ -641,8 +640,7 @@ static JSPromise* nativeSourcePullImpl(JSC::VM& vm, JSGlobalObject* globalObject
         if (isTruthy) {
             bool isClosed = nativeCloserFlag(vm, globalObject, adapter);
             RETURN_IF_EXCEPTION(scope, nullptr);
-            JSC::JSUint8Array* pendingView = pendingObject ? uncheckedDowncast<JSC::JSUint8Array>(pendingObject) : nullptr;
-            JSValue newView = nativeDecodePullResult(vm, globalObject, adapter, controller, drained, pendingView, isClosed);
+            JSValue newView = nativeDecodePullResult(vm, globalObject, adapter, controller, drained, uncheckedDowncast<JSC::JSUint8Array>(pendingObject), isClosed);
             RETURN_IF_EXCEPTION(scope, nullptr);
             nativeStorePendingView(vm, adapter, newView);
             return nullptr;

From 2adce47e821c2d7d352b8a0034653d3a7aa3f351 Mon Sep 17 00:00:00 2001
From: robobun <117481402+robobun@users.noreply.github.com>
Date: Sat, 1 Aug 2026 19:06:58 +0000
Subject: [PATCH 11/15] ci: retrigger


From fe6bc933b9a087975d840f16127e36f10bcb287a Mon Sep 17 00:00:00 2001
From: robobun <117481402+robobun@users.noreply.github.com>
Date: Sun, 2 Aug 2026 00:16:18 +0000
Subject: [PATCH 12/15] Keep the view-copy path for native-readable.ts; Owned
 handoff only for the C++ adapter

Readable.fromWeb's native-readable.ts calls _read ahead of downstream
consumption (node:stream Readable's maybeReadMore fires while state.length <
hwm regardless of src.pause()), so handing it the whole ByteStream.buffer and
signal_drained()ing on every pull let the fetch socket re-resume one recv ahead
of the writer. fetch-backpressure.test.ts download-proxy window went from
~170 MB back to ~250-280 MB.

on_pull/on_data now dispatch on whether the caller passed a pull view:

- buffer empty (C++ native-source adapter, m_sourceOwnsChunks): Owned handoff,
  right-sized chunks, signal_drained when self.buffer is taken.
- view present (native-readable.ts): main's copy-into-view + offset + IntoArray
  path, which meters chunks to the view's size and only signals once self.buffer
  is fully drained.

native-readable.ts keeps passing a view sized to Readable's hwm; the -1 from
start() just sets kHasResized so the view stays at that size. Readable.fromWeb
test dropped accordingly; streams-leak.test.ts's pull-buffer-reuse assertion
updated to the right-sized invariant it was really guarding.
---
 src/js/internal/streams/native-readable.ts    |  11 +-
 src/runtime/server/RequestContext.rs          |   2 +-
 src/runtime/webcore/ByteStream.rs             | 275 +++++++++++++++---
 ...serve-request-body-pipeline-memory.test.ts |  30 +-
 test/js/web/streams/streams-leak.test.ts      |  15 +-
 5 files changed, 246 insertions(+), 87 deletions(-)

diff --git a/src/js/internal/streams/native-readable.ts b/src/js/internal/streams/native-readable.ts
index cf7dade6d241..4c7652aa9ee3 100644
--- a/src/js/internal/streams/native-readable.ts
+++ b/src/js/internal/streams/native-readable.ts
@@ -20,7 +20,6 @@ const kHighWaterMark = Symbol("highWaterMark");
 const kPendingRead = Symbol("pendingRead");
 const kHasResized = Symbol("hasResized");
 const kRemainingChunk = Symbol("remainingChunk");
-const kSourceOwnsChunks = Symbol("sourceOwnsChunks");
 
 const MIN_BUFFER_SIZE = 512;
 let dynamicallyAdjustChunkSize = (_?) => (
@@ -37,8 +36,7 @@ type NativeReadable = typeof import("node:stream").Readable &
     [kPendingRead]: boolean;
     [kHighWaterMark]: number;
     [kHasResized]: boolean;
-    [kRemainingChunk]: Buffer | undefined;
-    [kSourceOwnsChunks]: boolean;
+    [kRemainingChunk]: Buffer;
     debugId: number;
   };
 
@@ -74,8 +72,6 @@ function constructNativeReadable(readableStream: ReadableStream, options): Nativ
   stream[kPendingRead] = false;
   stream[kHasResized] = !dynamicallyAdjustChunkSize();
   stream[kCloseState] = [false];
-  stream[kRemainingChunk] = undefined;
-  stream[kSourceOwnsChunks] = false;
 
   const highWaterMark = options.highWaterMark;
   stream[kHighWaterMark] = typeof highWaterMark === "number" ? highWaterMark : 256 * 1024;
@@ -147,8 +143,9 @@ function read(this: NativeReadable, maxToRead: number) {
       this[kHasResized] = true;
       this[kHighWaterMark] = Math.min(this[kHighWaterMark], result);
     } else if (typeof result === "number" && result < 0) {
+      // Start::ReadyOwned: the source meters via the pull view's size, so keep
+      // the view at Readable's own hwm and don't grow it.
       this[kHasResized] = true;
-      this[kSourceOwnsChunks] = true;
     }
     if ($isTypedArrayView(result) && result.byteLength > 0) {
       pushAndCheck(this, result);
@@ -160,7 +157,7 @@ function read(this: NativeReadable, maxToRead: number) {
       pushAndCheck(this, drainResult);
     }
   }
-  const chunk = this[kSourceOwnsChunks] ? undefined : getRemainingChunk(this, maxToRead);
+  const chunk = getRemainingChunk(this, maxToRead);
   var result = ptr.pull(chunk, this[kCloseState]);
   $assert(result !== undefined);
   $debug(
diff --git a/src/runtime/server/RequestContext.rs b/src/runtime/server/RequestContext.rs
index 041ddb0622cb..845980f5cb61 100644
--- a/src/runtime/server/RequestContext.rs
+++ b/src/runtime/server/RequestContext.rs
@@ -4063,7 +4063,7 @@ where
                 let _ = bytes.on_data(WebCore::streams::Result::Temporary(borrowed));
 
                 // What `on_data` buffered; `on_stream_drained` resumes once it empties.
-                let buffered = bytes.buffer.get().len();
+                let buffered = bytes.buffer.get().len().saturating_sub(bytes.offset.get());
                 if bytes.buffer_action.get().is_some()
                     || (bytes.sink.get().is_some() && !bytes.sink_paused.get())
                 {
diff --git a/src/runtime/webcore/ByteStream.rs b/src/runtime/webcore/ByteStream.rs
index 7f79b7be0f76..291100d21cab 100644
--- a/src/runtime/webcore/ByteStream.rs
+++ b/src/runtime/webcore/ByteStream.rs
@@ -1,11 +1,12 @@
 use core::cell::Cell;
 
 use bun_collections::VecExt;
+use bun_jsc::strong::Optional as StrongOptional;
 use bun_jsc::{self as jsc, JSGlobalObject, JSValue, JsCell};
 use bun_sys::Error as SysError;
 
 use crate::webcore::SinkHandle;
-use crate::webcore::streams::{self, BufferAction};
+use crate::webcore::streams::{self, BufferAction, IntoArray};
 use crate::webcore::{blob, readable_stream};
 
 bun_output::declare_scope!(ByteStream, visible);
@@ -23,6 +24,12 @@ pub struct ByteStream {
     pub(crate) has_received_last_chunk: Cell<bool>,
     pub(crate) pending: JsCell<streams::Pending>,
     pub(crate) done: Cell<bool>,
+    /// Borrowed view into a JS `Uint8Array` passed from `on_pull`; kept alive by `pending_value`.
+    // Raw fat slice ptr because the backing store is JS-heap-owned and rooted via
+    // `pending_value: Strong`. Never freed by Rust.
+    pub(crate) pending_buffer: Cell<*mut [u8]>,
+    pub(crate) pending_value: JsCell<StrongOptional>, // jsc.Strong.Optional
+    pub offset: Cell<usize>,
     /// Native sink this stream is piped into; `on_data` dispatches and honors `Writable`.
     pub(crate) sink: JsCell<SinkHandle>,
     /// Set on `Writable::Backpressure` (buffer instead of write); cleared by [`Self::resume`].
@@ -41,6 +48,9 @@ impl Default for ByteStream {
                 ..Default::default()
             }),
             done: Cell::new(false),
+            pending_buffer: Cell::new(Self::empty_pending_buffer()),
+            pending_value: JsCell::new(StrongOptional::empty()),
+            offset: Cell::new(0),
             sink: JsCell::new(SinkHandle::None),
             sink_paused: Cell::new(false),
             size_hint: Cell::new(0),
@@ -95,6 +105,11 @@ impl readable_stream::SourceContext for ByteStream {
 bun_core::impl_field_parent! { ByteStream => Source.context; pub fn parent_const; pub fn parent; }
 
 impl ByteStream {
+    #[inline]
+    const fn empty_pending_buffer() -> *mut [u8] {
+        core::ptr::slice_from_raw_parts_mut(core::ptr::NonNull::<u8>::dangling().as_ptr(), 0)
+    }
+
     /// Init-time reset. Runs before the JS
     /// wrapper exists, so `&mut self` is sound here (R-2 exemption).
     pub(crate) fn setup(&mut self) {
@@ -117,6 +132,16 @@ impl ByteStream {
         streams::Start::ReadyOwned
     }
 
+    fn value(&self) -> JSValue {
+        self.pending_value.with_mut(|pv| {
+            let Some(result) = pv.get() else {
+                return JSValue::ZERO;
+            };
+            pv.clear_without_deallocation();
+            result
+        })
+    }
+
     pub(crate) fn unpipe_without_deref(&self) {
         self.sink.set(SinkHandle::None);
         self.sink_paused.set(false);
@@ -136,6 +161,7 @@ impl ByteStream {
 
         if !self.buffer.get().is_empty() {
             let buffered = self.buffer.replace(Vec::new());
+            self.offset.set(0);
             let result = if self.has_received_last_chunk.get() {
                 streams::Result::OwnedAndDone(buffered)
             } else {
@@ -193,7 +219,7 @@ impl ByteStream {
         self.parent_const().producer.get().start();
     }
 
-    pub(crate) fn on_data(&self, stream: streams::Result) -> Result<(), bun_jsc::JsTerminated> {
+    pub(crate) fn on_data(&self, mut stream: streams::Result) -> Result<(), bun_jsc::JsTerminated> {
         bun_jsc::mark_binding!();
         if self.done.get() {
             // The owned `Vec<u8>`/`Vec`
@@ -223,7 +249,8 @@ impl ByteStream {
 
             if self.sink_paused.get() {
                 bun_output::scoped_log!(ByteStream, "ByteStream.onData sink paused → buffer");
-                self.append(stream);
+                self.append(stream, 0)
+                    .unwrap_or_else(|_| panic!("Out of memory while copying request body"));
                 return Ok(());
             }
 
@@ -338,46 +365,119 @@ impl ByteStream {
         if self.pending.get().state == streams::PendingState::Pending {
             debug_assert!(self.buffer.get().is_empty());
 
-            let is_done = self.has_received_last_chunk.get();
-            let result = match stream {
-                streams::Result::Err(_) => {
-                    self.done.set(true);
-                    stream
-                }
-                streams::Result::Done => {
-                    self.done.set(true);
-                    streams::Result::Done
-                }
-                streams::Result::Owned(owned) | streams::Result::OwnedAndDone(owned) => {
-                    if is_done {
+            // The pending pull came without a view (C++ native-source adapter):
+            // hand the chunk off as its own allocation. `signal_drained` stays
+            // gated by the view-copy branch below so the producer is not
+            // resumed ahead of node:stream Readable's buffer (which that path
+            // meters via the view size).
+            if self.pending_value.get().get().is_none() {
+                let is_done = self.has_received_last_chunk.get();
+                let result = match stream {
+                    streams::Result::Err(_) => {
+                        self.done.set(true);
+                        stream
+                    }
+                    streams::Result::Done => {
                         self.done.set(true);
-                        if owned.is_empty() {
-                            streams::Result::Done
+                        streams::Result::Done
+                    }
+                    streams::Result::Owned(owned) | streams::Result::OwnedAndDone(owned) => {
+                        if is_done {
+                            self.done.set(true);
+                            if owned.is_empty() {
+                                streams::Result::Done
+                            } else {
+                                streams::Result::OwnedAndDone(owned)
+                            }
                         } else {
-                            streams::Result::OwnedAndDone(owned)
+                            streams::Result::Owned(owned)
                         }
-                    } else {
-                        streams::Result::Owned(owned)
                     }
-                }
-                streams::Result::Temporary(temp) | streams::Result::TemporaryAndDone(temp) => {
-                    let owned = temp.slice().to_vec();
-                    if is_done {
-                        self.done.set(true);
-                        if owned.is_empty() {
-                            streams::Result::Done
+                    streams::Result::Temporary(temp) | streams::Result::TemporaryAndDone(temp) => {
+                        let owned = temp.slice().to_vec();
+                        if is_done {
+                            self.done.set(true);
+                            if owned.is_empty() {
+                                streams::Result::Done
+                            } else {
+                                streams::Result::OwnedAndDone(owned)
+                            }
                         } else {
-                            streams::Result::OwnedAndDone(owned)
+                            streams::Result::Owned(owned)
                         }
+                    }
+                    _ => unreachable!(),
+                };
+                self.pending.with_mut(|p| p.result = result);
+                self.signal_drained();
+                self.pending.with_mut(|p| p.run());
+                return Ok(());
+            }
+
+            let chunk = stream.slice();
+            // Re-derive the destination from the GC-rooted view instead of trusting the
+            // raw pointer captured at pull time: JS can detach or transfer the backing
+            // ArrayBuffer between the pull and the data arriving, leaving
+            // `pending_buffer` dangling. A detached view re-derives to an empty slice.
+            let global = self.parent_const().global_this();
+            let mut pending_view = self
+                .pending_value
+                .get()
+                .get()
+                .and_then(|view| view.as_array_buffer(global))
+                .unwrap_or_default();
+            let pending_buf = pending_view.slice_mut();
+            let to_copy_len = chunk.len().min(pending_buf.len());
+            let pending_buffer_len = pending_buf.len();
+            debug_assert!(pending_buf.as_ptr() != chunk.as_ptr());
+            pending_buf[..to_copy_len].copy_from_slice(&chunk[..to_copy_len]);
+            let has_remaining = chunk.len() > to_copy_len;
+            self.pending_buffer.set(Self::empty_pending_buffer());
+
+            let is_really_done =
+                self.has_received_last_chunk.get() && to_copy_len <= pending_buffer_len;
+
+            if is_really_done {
+                self.done.set(true);
+
+                if to_copy_len == 0 {
+                    if matches!(stream, streams::Result::Err(_)) {
+                        let err = core::mem::replace(&mut stream, streams::Result::Done);
+                        self.pending.with_mut(|p| p.result = err);
                     } else {
-                        streams::Result::Owned(owned)
+                        self.pending.with_mut(|p| p.result = streams::Result::Done);
                     }
+                } else {
+                    let v = self.value();
+                    self.pending.with_mut(|p| {
+                        p.result = streams::Result::IntoArrayAndDone(IntoArray {
+                            value: v,
+                            len: to_copy_len as blob::SizeType, // @truncate
+                        });
+                    });
                 }
-                _ => unreachable!(),
-            };
+            } else {
+                let v = self.value();
+                self.pending.with_mut(|p| {
+                    p.result = streams::Result::IntoArray(IntoArray {
+                        value: v,
+                        len: to_copy_len as blob::SizeType, // @truncate
+                    });
+                });
+            }
 
-            self.pending.with_mut(|p| p.result = result);
-            self.signal_drained();
+            if has_remaining {
+                self.append(stream, to_copy_len)
+                    .unwrap_or_else(|_| panic!("Out of memory while copying request body"));
+            } else {
+                // Only resume the producer when the whole chunk fit the pull
+                // view. When the tail spilled into `buffer` the next `on_pull`
+                // signals once it drains, so resuming now would let another
+                // producer chunk land with no reader to take it (it would go
+                // straight to `append` below), inflating `buffer` and the
+                // producer's own staging buffer by an extra recv each cycle.
+                self.signal_drained();
+            }
 
             bun_output::scoped_log!(ByteStream, "ByteStream.onData pending.run()");
 
@@ -392,19 +492,24 @@ impl ByteStream {
 
         bun_output::scoped_log!(ByteStream, "ByteStream.onData no action just append");
 
-        self.append(stream);
+        self.append(stream, 0)
+            .unwrap_or_else(|_| panic!("Out of memory while copying request body"));
         Ok(())
     }
 
-    fn append(&self, stream: streams::Result) {
+    fn append(&self, stream: streams::Result, offset: usize) -> Result<(), bun_alloc::AllocError> {
         if self.buffer.get().capacity() == 0 {
             match stream {
                 streams::Result::Owned(mut owned) | streams::Result::OwnedAndDone(mut owned) => {
                     // `move_to_list_managed` moves the buffer, no copy.
                     self.buffer.set(owned.move_to_list_managed());
+                    self.offset.set(self.offset.get() + offset);
                 }
                 streams::Result::TemporaryAndDone(temp) | streams::Result::Temporary(temp) => {
-                    self.buffer.set(temp.slice().to_vec());
+                    let chunk = &temp.slice()[offset..];
+                    let mut buf = Vec::with_capacity(chunk.len());
+                    buf.extend_from_slice(chunk);
+                    self.buffer.set(buf);
                 }
                 streams::Result::Err(err) => {
                     self.pending
@@ -413,15 +518,17 @@ impl ByteStream {
                 streams::Result::Done => {}
                 _ => unreachable!(),
             }
-            return;
+            return Ok(());
         }
 
         match stream {
             streams::Result::TemporaryAndDone(temp) | streams::Result::Temporary(temp) => {
-                self.buffer.with_mut(|b| b.extend_from_slice(temp.slice()));
+                self.buffer
+                    .with_mut(|b| b.extend_from_slice(&temp.slice()[offset..]));
             }
             streams::Result::OwnedAndDone(owned) | streams::Result::Owned(owned) => {
-                self.buffer.with_mut(|b| b.extend_from_slice(owned.slice()));
+                self.buffer
+                    .with_mut(|b| b.extend_from_slice(&owned.slice()[offset..]));
                 // `owned: Vec<u8>` drops here.
             }
             streams::Result::Err(err) => {
@@ -441,24 +548,89 @@ impl ByteStream {
             // We don't support the rest of these yet
             _ => unreachable!(),
         }
+
+        Ok(())
+    }
+
+    fn set_value(&self, view: JSValue) {
+        bun_jsc::mark_binding!();
+        let global = self.parent_const().global_this();
+        self.pending_value.with_mut(|pv| pv.set(global, view));
     }
 
-    fn on_pull(&self, _buffer: &mut [u8], _view: JSValue) -> streams::Result {
+    fn on_pull(&self, buffer: &mut [u8], view: JSValue) -> streams::Result {
         bun_jsc::mark_binding!();
         debug_assert!(self.buffer_action.get().is_none());
 
+        // The C++ native-source adapter passes no pull view (`Start::ReadyOwned`
+        // → `m_sourceOwnsChunks`); hand off `self.buffer` as the chunk directly.
+        // native-readable.ts passes a view and meters via the copy path below so
+        // node:stream Readable's over-eager `_read` stays bounded.
+        if buffer.is_empty() {
+            if !self.buffer.get().is_empty() {
+                debug_assert!(self.value().is_empty());
+                debug_assert_eq!(self.offset.get(), 0);
+                let owned = self.buffer.replace(Vec::new());
+                self.signal_drained();
+                if self.has_received_last_chunk.get() {
+                    self.done.set(true);
+                    return streams::Result::OwnedAndDone(owned);
+                }
+                return streams::Result::Owned(owned);
+            }
+            if self.has_received_last_chunk.get() {
+                if matches!(self.pending.get().result, streams::Result::Err(_)) {
+                    return self
+                        .pending
+                        .with_mut(|p| core::mem::replace(&mut p.result, streams::Result::Done));
+                }
+                return streams::Result::Done;
+            }
+            // No pull view: `on_data` dispatches on `pending_value` being empty
+            // to resolve with `Owned` instead of copying into a view.
+            return streams::Result::Pending(self.pending.as_ptr());
+        }
+
         if !self.buffer.get().is_empty() {
-            debug_assert!(self.pending.get().state != streams::PendingState::Pending);
-            let owned = self.buffer.replace(Vec::new());
+            debug_assert!(self.value().is_empty()); // == .zero
+            // R-2: confine the `&mut Vec<u8>` to a `with_mut` so no `JsCell`
+            // borrow escapes the copy. The result tuple drives the rest.
+            let (to_write, remaining_in_buffer_len) = self.buffer.with_mut(|b| {
+                let to_write = (b.len() - self.offset.get()).min(buffer.len());
+                let remaining_in_buffer_len = to_write; // length of `this.buffer.items[this.offset..][0..to_write]`
 
-            self.signal_drained();
+                buffer[..to_write].copy_from_slice(&b[self.offset.get()..][..to_write]);
 
-            if self.has_received_last_chunk.get() {
+                if self.offset.get() + to_write == b.len() {
+                    self.offset.set(0);
+                    b.clear();
+                } else {
+                    self.offset.set(self.offset.get() + to_write);
+                }
+                (to_write, remaining_in_buffer_len)
+            });
+
+            if self.buffer.get().is_empty() {
+                self.signal_drained();
+            }
+
+            if self.has_received_last_chunk.get() && remaining_in_buffer_len == 0 {
+                self.buffer.with_mut(|b| {
+                    b.clear();
+                    b.shrink_to_fit();
+                });
                 self.done.set(true);
-                return streams::Result::OwnedAndDone(owned);
+
+                return streams::Result::IntoArrayAndDone(IntoArray {
+                    value: view,
+                    len: to_write as blob::SizeType, // @truncate
+                });
             }
 
-            return streams::Result::Owned(owned);
+            return streams::Result::IntoArray(IntoArray {
+                value: view,
+                len: to_write as blob::SizeType, // @truncate
+            });
         }
 
         if self.has_received_last_chunk.get() {
@@ -472,6 +644,10 @@ impl ByteStream {
             return streams::Result::Done;
         }
 
+        // Raw borrow of a JS-owned buffer; rooted by `set_value`.
+        self.pending_buffer.set(std::ptr::from_mut::<[u8]>(buffer));
+        self.set_value(view);
+
         // R-2: `JsCell::as_ptr` yields the stable `*mut Pending` that the
         // returned `streams::Result::Pending` raw-backref needs.
         streams::Result::Pending(self.pending.as_ptr())
@@ -479,6 +655,7 @@ impl ByteStream {
 
     fn on_cancel(&self) {
         bun_jsc::mark_binding!();
+        let view = self.value();
         if self.buffer.get().capacity() > 0 {
             self.buffer.with_mut(|b| {
                 b.clear();
@@ -486,8 +663,10 @@ impl ByteStream {
             });
         }
         self.done.set(true);
+        self.pending_value.with_mut(|pv| pv.deinit());
 
-        if self.pending.get().state == streams::PendingState::Pending {
+        if !view.is_empty() {
+            self.pending_buffer.set(Self::empty_pending_buffer());
             self.pending.with_mut(|p| {
                 p.result.release();
                 p.result = streams::Result::Done;
@@ -528,9 +707,11 @@ impl ByteStream {
             });
         }
 
+        self.pending_value.with_mut(|pv| pv.deinit());
         if !self.done.get() {
             self.done.set(true);
 
+            self.pending_buffer.set(Self::empty_pending_buffer());
             let is_promise = self.pending.with_mut(|p| {
                 p.result.release();
                 p.result = streams::Result::Done;
diff --git a/test/js/bun/http/serve-request-body-pipeline-memory.test.ts b/test/js/bun/http/serve-request-body-pipeline-memory.test.ts
index 4c24a8e26836..792f7fb33f65 100644
--- a/test/js/bun/http/serve-request-body-pipeline-memory.test.ts
+++ b/test/js/bun/http/serve-request-body-pipeline-memory.test.ts
@@ -9,7 +9,6 @@
 
 import { expect, test } from "bun:test";
 import { connect } from "node:net";
-import { Readable } from "node:stream";
 
 type Seen = { len: number; backing: number; off: number };
 
@@ -107,26 +106,9 @@ test("for await (req.body) chunks are backed by right-sized buffers, not the ada
   checkRightSized(seen, BODY, writes.length);
 });
 
-test("Readable.fromWeb(req.body) chunks are backed by right-sized buffers", async () => {
-  const BODY = 64 * 1024;
-  const writes = [BODY / 4, BODY / 4, BODY / 4, BODY / 4];
-  const seen = await runUpload(
-    async (req, onParked, onChunk) => {
-      const r = Readable.fromWeb(req.body as any);
-      const out: Seen[] = [];
-      r.on("data", (chunk: Buffer) => {
-        out.push({ len: chunk.byteLength, backing: chunk.buffer.byteLength, off: chunk.byteOffset });
-        onChunk();
-      });
-      onParked();
-      await new Promise<void>((res, rej) => {
-        r.once("end", () => res());
-        r.once("error", rej);
-      });
-      return out;
-    },
-    BODY,
-    writes,
-  );
-  checkRightSized(seen, BODY, writes.length);
-});
+// `Readable.fromWeb(req.body)` is deliberately left on the copy-into-view path:
+// node:stream Readable calls `_read` ahead of downstream consumption, so
+// metering via the pull view's size is what keeps the producer paused while
+// a chunk is still sitting in the Readable's own buffer. The C++ adapter
+// above pulls once per reader.read(), so it can hand off whole buffers
+// without that hazard.
diff --git a/test/js/web/streams/streams-leak.test.ts b/test/js/web/streams/streams-leak.test.ts
index 0a2862ade9c1..4308d849fc7d 100644
--- a/test/js/web/streams/streams-leak.test.ts
+++ b/test/js/web/streams/streams-leak.test.ts
@@ -1,7 +1,7 @@
 import { expect, test } from "bun:test";
 import { bunEnv, bunExe, isASAN, isWindows, rss, tempDir } from "harness";
 
-test("native ReadableStream reuses the pull buffer across small reads", async () => {
+test("native ReadableStream backs small reads with right-sized buffers", async () => {
   // #getInternalBuffer used to rotate to a fresh autoAllocateChunkSize
   // (256KB) Uint8Array whenever $data.length < chunkSize — true after
   // every nonzero read, since #handleNumberResult stores the tail
@@ -60,14 +60,13 @@ test("native ReadableStream reuses the pull buffer across small reads", async ()
   // through the native pull path.
   expect(chunks.length).toBeGreaterThanOrEqual(CHUNKS_TO_WRITE);
 
-  // Consecutive small reads should land in the same backing buffer (the
-  // tail subarray is reused until a read fills it). 128 bytes of 2-byte
-  // chunks fits well inside one 256KB buffer, so the whole stream should
-  // share a handful at most. Pre-fix every chunk had its own 256KB
-  // buffer, so this was ~chunks.length.
-  const distinctBuffers = new Set(chunks.map(c => c.buffer));
-  expect(distinctBuffers.size).toBeLessThan(8);
+  // Each chunk is handed off as its own right-sized allocation; pre-fix each
+  // was a subarray over a fresh 256 KB Gigacage buffer.
+  for (const c of chunks) {
+    expect(c.buffer.byteLength).toBe(c.byteLength);
+  }
 
+  const distinctBuffers = new Set(chunks.map(c => c.buffer));
   let backingBytes = 0;
   for (const buf of distinctBuffers) backingBytes += buf.byteLength;
   // Pre-fix this was ~chunks.length * 256KB ≈ 16 MB.

From 41ebfb84fc164d382a5bb3251ba95f81021c2777 Mon Sep 17 00:00:00 2001
From: robobun <117481402+robobun@users.noreply.github.com>
Date: Sun, 2 Aug 2026 00:18:41 +0000
Subject: [PATCH 13/15] Trim code comments to satisfy comment-cop

---
 src/js/internal/streams/native-readable.ts |  3 +--
 src/runtime/webcore/ByteStream.rs          | 14 +++-----------
 2 files changed, 4 insertions(+), 13 deletions(-)

diff --git a/src/js/internal/streams/native-readable.ts b/src/js/internal/streams/native-readable.ts
index 4c7652aa9ee3..583d6fc1b949 100644
--- a/src/js/internal/streams/native-readable.ts
+++ b/src/js/internal/streams/native-readable.ts
@@ -143,8 +143,7 @@ function read(this: NativeReadable, maxToRead: number) {
       this[kHasResized] = true;
       this[kHighWaterMark] = Math.min(this[kHighWaterMark], result);
     } else if (typeof result === "number" && result < 0) {
-      // Start::ReadyOwned: the source meters via the pull view's size, so keep
-      // the view at Readable's own hwm and don't grow it.
+      // Start::ReadyOwned: don't grow the pull view past Readable's hwm.
       this[kHasResized] = true;
     }
     if ($isTypedArrayView(result) && result.byteLength > 0) {
diff --git a/src/runtime/webcore/ByteStream.rs b/src/runtime/webcore/ByteStream.rs
index 291100d21cab..82c2cbf8b980 100644
--- a/src/runtime/webcore/ByteStream.rs
+++ b/src/runtime/webcore/ByteStream.rs
@@ -365,11 +365,7 @@ impl ByteStream {
         if self.pending.get().state == streams::PendingState::Pending {
             debug_assert!(self.buffer.get().is_empty());
 
-            // The pending pull came without a view (C++ native-source adapter):
-            // hand the chunk off as its own allocation. `signal_drained` stays
-            // gated by the view-copy branch below so the producer is not
-            // resumed ahead of node:stream Readable's buffer (which that path
-            // meters via the view size).
+            // Pending pull with no view parked (C++ adapter): Owned handoff.
             if self.pending_value.get().get().is_none() {
                 let is_done = self.has_received_last_chunk.get();
                 let result = match stream {
@@ -562,10 +558,8 @@ impl ByteStream {
         bun_jsc::mark_binding!();
         debug_assert!(self.buffer_action.get().is_none());
 
-        // The C++ native-source adapter passes no pull view (`Start::ReadyOwned`
-        // → `m_sourceOwnsChunks`); hand off `self.buffer` as the chunk directly.
-        // native-readable.ts passes a view and meters via the copy path below so
-        // node:stream Readable's over-eager `_read` stays bounded.
+        // No pull view (C++ adapter, `m_sourceOwnsChunks`): Owned handoff.
+        // With a view (native-readable.ts): metered copy-into-view below.
         if buffer.is_empty() {
             if !self.buffer.get().is_empty() {
                 debug_assert!(self.value().is_empty());
@@ -586,8 +580,6 @@ impl ByteStream {
                 }
                 return streams::Result::Done;
             }
-            // No pull view: `on_data` dispatches on `pending_value` being empty
-            // to resolve with `Owned` instead of copying into a view.
             return streams::Result::Pending(self.pending.as_ptr());
         }
 

From 6829d2f82eb54e925e7fd8a71f658057ad29ab5b Mon Sep 17 00:00:00 2001
From: robobun <117481402+robobun@users.noreply.github.com>
Date: Sun, 2 Aug 2026 00:19:25 +0000
Subject: [PATCH 14/15] One-line the on_pull dispatch comment

---
 src/runtime/webcore/ByteStream.rs | 3 +--
 1 file changed, 1 insertion(+), 2 deletions(-)

diff --git a/src/runtime/webcore/ByteStream.rs b/src/runtime/webcore/ByteStream.rs
index 82c2cbf8b980..e8a7813cbbd3 100644
--- a/src/runtime/webcore/ByteStream.rs
+++ b/src/runtime/webcore/ByteStream.rs
@@ -558,8 +558,7 @@ impl ByteStream {
         bun_jsc::mark_binding!();
         debug_assert!(self.buffer_action.get().is_none());
 
-        // No pull view (C++ adapter, `m_sourceOwnsChunks`): Owned handoff.
-        // With a view (native-readable.ts): metered copy-into-view below.
+        // No pull view = C++ adapter Owned handoff; view = native-readable's metered copy below.
         if buffer.is_empty() {
             if !self.buffer.get().is_empty() {
                 debug_assert!(self.value().is_empty());

From e179877bc995d2b3421287954b2d376d80f429d5 Mon Sep 17 00:00:00 2001
From: robobun <bot@oven.sh>
Date: Sun, 2 Aug 2026 00:33:14 +0000
Subject: [PATCH 15/15] on_cancel: gate pending.run on pending.state, not on a
 stored view

The Owned-handoff Pending path parks without setting pending_value, so the
previous !view.is_empty() check left a parked pull unsettled on cancel,
pinning the ReadableStream via the protected promise. Gate on
pending.state == Pending instead, which is the invariant both park paths
establish.

Also make the on_pull Owned-path debug_assert non-consuming.
---
 src/runtime/webcore/ByteStream.rs | 5 ++---
 1 file changed, 2 insertions(+), 3 deletions(-)

diff --git a/src/runtime/webcore/ByteStream.rs b/src/runtime/webcore/ByteStream.rs
index e8a7813cbbd3..ae782c80078b 100644
--- a/src/runtime/webcore/ByteStream.rs
+++ b/src/runtime/webcore/ByteStream.rs
@@ -561,7 +561,7 @@ impl ByteStream {
         // No pull view = C++ adapter Owned handoff; view = native-readable's metered copy below.
         if buffer.is_empty() {
             if !self.buffer.get().is_empty() {
-                debug_assert!(self.value().is_empty());
+                debug_assert!(self.pending_value.get().get().is_none());
                 debug_assert_eq!(self.offset.get(), 0);
                 let owned = self.buffer.replace(Vec::new());
                 self.signal_drained();
@@ -646,7 +646,6 @@ impl ByteStream {
 
     fn on_cancel(&self) {
         bun_jsc::mark_binding!();
-        let view = self.value();
         if self.buffer.get().capacity() > 0 {
             self.buffer.with_mut(|b| {
                 b.clear();
@@ -656,7 +655,7 @@ impl ByteStream {
         self.done.set(true);
         self.pending_value.with_mut(|pv| pv.deinit());
 
-        if !view.is_empty() {
+        if self.pending.get().state == streams::PendingState::Pending {
             self.pending_buffer.set(Self::empty_pending_buffer());
             self.pending.with_mut(|p| {
                 p.result.release();