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

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 = (_?) => (
@@ -36,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;
   };
 
@@ -72,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;
@@ -142,6 +146,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) {
+      this[kHasResized] = true;
+      this[kSourceOwnsChunks] = true;
     }
     if ($isTypedArrayView(result) && result.byteLength > 0) {
       pushAndCheck(this, result);
@@ -153,7 +160,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(

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

@@ -417,6 +417,8 @@
 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 @@
         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 @@
     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 +631,8 @@
     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 +641,8 @@
         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 +653,7 @@
     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/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())
                 {

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