ByteStream: hand off owned buffers from on_pull/on_data and skip the adapter pull view

src/js/internal/streams/native-readable.ts

@@ -142,6 +142,9 @@ 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) {
+      // Start::ReadyOwned: don't grow the pull view past Readable's hwm.
+      this[kHasResized] = true;
     }
     if ($isTypedArrayView(result) && result.byteLength > 0) {
       pushAndCheck(this, result);

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,6 @@ static JSValue nativeDecodePullResult(JSC::VM& vm, JSGlobalObject* globalObject,
         return jsUndefined();
     }
     if (auto* chunk = dynamicDowncast<JSC::JSArrayBufferView>(result)) {
-        if (!isClosed)
-            nativeAdjustChunkSize(adapter, chunk->byteLength());
         if (chunk->byteLength() > 0) {
             if (adapter->m_textMode) {
                 nativeEnqueueTextChunk(globalObject, controller, adapter->m_textState, chunk->span(), /* flush */ false);
@@ -556,7 +556,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));
@@ -645,7 +651,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);

src/jsc/bindings/webcore/streams/BunStreamSource.h

@@ -88,6 +88,8 @@ 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 (Start::ReadyOwned): never allocate PendingView.
+    bool m_sourceOwnsChunks : 1 { false };
     Bun::WebStreams::StreamingUTF8DecodeState m_textState;
 
 private:

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

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
@@ -32,7 +30,6 @@ pub struct ByteStream {
     pub(crate) pending_buffer: Cell<*mut [u8]>,
     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`].
@@ -54,7 +51,6 @@ impl Default for ByteStream {
             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),
@@ -133,17 +129,7 @@ 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)))
+        streams::Start::ReadyOwned
     }
 
     fn value(&self) -> JSValue {
@@ -376,10 +362,55 @@ impl ByteStream {
             return Ok(());
         }
 
-        let chunk = stream.slice();
-
         if self.pending.get().state == streams::PendingState::Pending {
             debug_assert!(self.buffer.get().is_empty());
+
+            // 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 {
+                    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 {
+                            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
+                            } else {
+                                streams::Result::OwnedAndDone(owned)
+                            }
+                        } else {
+                            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
@@ -525,9 +556,32 @@ impl ByteStream {
 
     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());
 
+        // 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.pending_value.get().get().is_none());
+                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;
+            }
+            return streams::Result::Pending(self.pending.as_ptr());
+        }
+
         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`
@@ -592,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();
@@ -602,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();

src/runtime/webcore/ReadableStream.rs

@@ -1117,10 +1117,11 @@ 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 {
+            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 +1134,7 @@ 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)),
+            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),

src/runtime/webcore/streams.rs

@@ -73,6 +73,8 @@ pub enum Start {
     },
     FileSink(FileSinkOptions),
     Ready,
+    /// `on_pull` hands back `Owned` chunks; the adapter skips its pull view.
+    ReadyOwned,
     OwnedAndDone(Vec<u8>),
 }
 
@@ -97,7 +99,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) => {

test/js/bun/http/serve-request-body-pipeline-memory.test.ts

@@ -0,0 +1,114 @@
+// `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 { expect, test } from "bun:test";
+import { connect } from "node:net";
+
+type Seen = { len: number; backing: number; off: number };
+
+async function runUpload(
+  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>());
+  // `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) => {
+    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)), onChunk);
+        handlerDone(seen);
+      } catch (e) {
+        fail(e);
+      }
+      return new Response("ok");
+    },
+  });
+
+  const sock = connect({ port: server.port, host: "127.0.0.1" });
+  sock.on("error", fail);
+  await new Promise<void>((res, rej) => {
+    sock.once("connect", () => res());
+    sock.once("error", rej);
+  });
+
+  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 [i, n] of writes.entries()) {
+      sock.write(Buffer.alloc(n, 0x61));
+      await acks[i].promise;
+    }
+    sock.end();
+    return await handlerP;
+  } finally {
+    sock.destroy();
+  }
+}
+
+function checkRightSized(seen: Seen[], expectedTotal: number, minChunks: number) {
+  let total = 0;
+  for (const { len } of seen) total += len;
+  expect(total).toBe(expectedTotal);
+  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.
+  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 writes = [8 * 1024, 8 * 1024, BODY - 16 * 1024];
+  const seen = await runUpload(
+    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,
+    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.