Response.clone(): copy a chunk when a body-clone branch hands it to script

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

@@ -859,13 +859,13 @@ static JSValue intoArrayLoop(JSC::VM& vm, JSGlobalObject* globalObject, WebCore:
     }
 }
 
-JSValue readableStreamIntoArray(JSGlobalObject* globalObject, WebCore::JSReadableStream* stream)
+JSValue readableStreamIntoArray(JSGlobalObject* globalObject, WebCore::JSReadableStream* stream, ReaderOwner owner)
 {
     auto& vm = getVM(globalObject);
     auto scope = DECLARE_THROW_SCOPE(vm);
     stream->materializeIfNeeded(globalObject);
     RETURN_IF_EXCEPTION(scope, {});
-    auto* reader = acquireReadableStreamDefaultReader(globalObject, stream);
+    auto* reader = acquireReadableStreamDefaultReader(globalObject, stream, owner);
     RETURN_IF_EXCEPTION(scope, {});
     auto* chunks = constructEmptyArray(globalObject, nullptr);
     RETURN_IF_EXCEPTION(scope, {});
@@ -914,7 +914,7 @@ JSValue readableStreamIntoText(JSGlobalObject* globalObject, WebCore::JSReadable
 {
     auto& vm = getVM(globalObject);
     auto scope = DECLARE_THROW_SCOPE(vm);
-    JSValue arrayResult = readableStreamIntoArray(globalObject, stream);
+    JSValue arrayResult = readableStreamIntoArray(globalObject, stream, ReaderOwner::Native);
     RETURN_IF_EXCEPTION(scope, {});
     RELEASE_AND_RETURN(scope, convertChunkArrayPromise(vm, globalObject, arrayResult, ChunkArrayConversion::Text));
 }
@@ -1001,7 +1001,7 @@ static JSValue consumeDirectStreamBody(JSC::VM& vm, JSGlobalObject* globalObject
         controller->m_closeOnPullSettled = true;
     stream->materializeIfNeeded(globalObject);
     RETURN_IF_EXCEPTION(scope, {});
-    auto* reader = acquireReadableStreamDefaultReader(globalObject, stream);
+    auto* reader = acquireReadableStreamDefaultReader(globalObject, stream, ReaderOwner::Native);
     RETURN_IF_EXCEPTION(scope, {});
     auto* context = InternalFieldTuple::create(vm, defaultGlobalObject(globalObject)->internalFieldTupleStructure(), stream, reader);
     RELEASE_AND_RETURN(scope, directConsumeLoopStep(vm, globalObject, context));
@@ -1170,15 +1170,15 @@ JSValue readableStreamToText(JSGlobalObject* globalObject, WebCore::JSReadableSt
     RELEASE_AND_RETURN(scope, readableStreamIntoText(globalObject, stream));
 }
 
-JSValue readableStreamToArray(JSGlobalObject* globalObject, WebCore::JSReadableStream* stream)
+JSValue readableStreamToArray(JSGlobalObject* globalObject, WebCore::JSReadableStream* stream, ReaderOwner owner)
 {
     auto& vm = getVM(globalObject);
     auto scope = DECLARE_THROW_SCOPE(vm);
     if (stream->m_bunMode == BunStreamMode::DirectPending)
         RELEASE_AND_RETURN(scope, readableStreamToArrayDirect(globalObject, stream));
     if (auto* error = unusableStreamError(globalObject, stream))
         RELEASE_AND_RETURN(scope, promiseRejectedWith(globalObject, error));
-    RELEASE_AND_RETURN(scope, readableStreamIntoArray(globalObject, stream));
+    RELEASE_AND_RETURN(scope, readableStreamIntoArray(globalObject, stream, owner));
 }
 
 // The chunk arrays these conversions consume are built by the array pump and never escape
@@ -1257,7 +1257,7 @@ JSValue readableStreamToArrayBuffer(JSGlobalObject* globalObject, WebCore::JSRea
     RETURN_IF_EXCEPTION(scope, {});
     if (fastPath)
         return fastPath;
-    JSValue arrayResult = readableStreamToArray(globalObject, stream);
+    JSValue arrayResult = readableStreamToArray(globalObject, stream, ReaderOwner::Native);
     RETURN_IF_EXCEPTION(scope, {});
     RELEASE_AND_RETURN(scope, convertChunkArrayPromise(vm, globalObject, arrayResult, ChunkArrayConversion::ArrayBuffer));
 }
@@ -1274,7 +1274,7 @@ JSValue readableStreamToBytes(JSGlobalObject* globalObject, WebCore::JSReadableS
     RETURN_IF_EXCEPTION(scope, {});
     if (fastPath)
         return fastPath;
-    JSValue arrayResult = readableStreamToArray(globalObject, stream);
+    JSValue arrayResult = readableStreamToArray(globalObject, stream, ReaderOwner::Native);
     RETURN_IF_EXCEPTION(scope, {});
     RELEASE_AND_RETURN(scope, convertChunkArrayPromise(vm, globalObject, arrayResult, ChunkArrayConversion::Bytes));
 }
@@ -1323,7 +1323,7 @@ JSValue readableStreamToBlob(JSGlobalObject* globalObject, WebCore::JSReadableSt
     RETURN_IF_EXCEPTION(scope, {});
     if (fastPath)
         return fastPath;
-    JSValue arrayResult = readableStreamToArray(globalObject, stream);
+    JSValue arrayResult = readableStreamToArray(globalObject, stream, ReaderOwner::Native);
     RETURN_IF_EXCEPTION(scope, {});
     auto* arrayPromise = dynamicDowncast<JSPromise>(arrayResult);
     if (!arrayPromise) [[unlikely]] {
@@ -1384,7 +1384,7 @@ JSC_DEFINE_HOST_FUNCTION(jsFunctionReadableStreamToArray, (JSGlobalObject * glob
     auto* stream = dynamicDowncast<JSReadableStream>(streamValue);
     if (!stream) [[unlikely]]
         return Bun::ERR::INVALID_ARG_TYPE(scope, globalObject, "stream"_s, "ReadableStream"_s, streamValue);
-    RELEASE_AND_RETURN(scope, JSValue::encode(Bun::WebStreams::readableStreamToArray(globalObject, stream)));
+    RELEASE_AND_RETURN(scope, JSValue::encode(Bun::WebStreams::readableStreamToArray(globalObject, stream, ReaderOwner::Script)));
 }
 
 JSC_DEFINE_HOST_FUNCTION(jsFunctionReadableStreamToArrayBuffer, (JSGlobalObject * globalObject, CallFrame* callFrame))

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

@@ -1139,7 +1139,7 @@ static void rsisBegin(JSC::VM& vm, JSGlobalObject* globalObject, JSReadableStrea
     auto scope = DECLARE_THROW_SCOPE(vm);
     stream->materializeIfNeeded(globalObject);
     RETURN_IF_EXCEPTION(scope, );
-    auto* reader = acquireReadableStreamDefaultReader(globalObject, stream);
+    auto* reader = acquireReadableStreamDefaultReader(globalObject, stream, ReaderOwner::Native);
     RETURN_IF_EXCEPTION(scope, );
     // The sink consumes the stream as a body: it stays locked after rsisFinally releases the pump's reader.
     stream->markConsumedAsBody();

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

@@ -638,7 +638,7 @@ JSC_DEFINE_HOST_FUNCTION(jsReadableStreamPrototypeFunction_getReader, (JSGlobalO
 
     stream->materializeIfNeeded(lexicalGlobalObject);
     RETURN_IF_EXCEPTION(scope, {});
-    auto* reader = acquireReadableStreamDefaultReader(lexicalGlobalObject, stream);
+    auto* reader = acquireReadableStreamDefaultReader(lexicalGlobalObject, stream, ReaderOwner::Script);
     RETURN_IF_EXCEPTION(scope, {});
     return JSValue::encode(reader);
 }
@@ -712,7 +712,7 @@ JSC_DEFINE_HOST_FUNCTION(jsReadableStreamPrototypeFunction_tee, (JSGlobalObject
     auto* stream = dynamicDowncast<JSReadableStream>(callFrame->thisValue());
     if (!stream) [[unlikely]]
         return Bun::ERR::INVALID_THIS(scope, lexicalGlobalObject, "ReadableStream"_s);
-    auto branches = readableStreamTee(lexicalGlobalObject, stream);
+    auto branches = readableStreamTee(lexicalGlobalObject, stream, /* cloneForBranch2 */ false);
     RETURN_IF_EXCEPTION(scope, {});
     auto* array = constructEmptyArray(lexicalGlobalObject, nullptr, 2);
     RETURN_IF_EXCEPTION(scope, {});
@@ -748,7 +748,7 @@ JSC_DEFINE_HOST_FUNCTION(jsReadableStreamPrototypeFunction_values, (JSGlobalObje
 
     auto* domGlobalObject = defaultGlobalObject(lexicalGlobalObject);
     auto* iterator = JSReadableStreamAsyncIterator::create(vm, getDOMStructure<JSReadableStreamAsyncIterator>(vm, *domGlobalObject));
-    auto* reader = acquireReadableStreamDefaultReader(lexicalGlobalObject, stream);
+    auto* reader = acquireReadableStreamDefaultReader(lexicalGlobalObject, stream, ReaderOwner::Script);
     RETURN_IF_EXCEPTION(scope, {});
     iterator->m_reader.set(vm, iterator, reader);
     iterator->m_preventCancel = preventCancel;

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

@@ -66,6 +66,8 @@ class JSReadableStream final : public JSC::JSNonFinalObject {
     // Body.textStream(): the native source adapter decodes each chunk as UTF-8
     // text before enqueue.
     bool m_nativeTextMode : 1 { false };
+    // Bun: the default reader was acquired with ReaderOwner::Native (StreamsForward.h). Never cleared.
+    bool m_nonScriptConsumer : 1 { false };
 
     // [[reader]] — a default reader, a BYOB reader, or null (undefined).
     JSC::WriteBarrier<JSReadableStreamReaderBase> m_reader;

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

@@ -306,11 +306,22 @@ JSValue JSReadableStreamDefaultController::dequeueChunkForRead(JSGlobalObject* g
 {
     auto scope = DECLARE_THROW_SCOPE(getVM(globalObject));
     ASSERT(!m_queue.isEmpty());
+    // Read before the close step below clears the algorithm slots.
+    auto* isolatingTee = teeBranchIsolatingChunks(this);
     JSValue chunk;
     {
         WTF::Locker locker { cellLock() };
         chunk = m_queue.dequeueValue(locker);
     }
+    if (isolatingTee) [[unlikely]] {
+        chunk = isolateTeeBranchChunk(globalObject, isolatingTee, chunk);
+        RETURN_IF_EXCEPTION(scope, {});
+        if (!chunk)
+            return {};
+        // The copy runs user code (userJS: yes) that can close or cancel this branch.
+        if (m_stream->m_state != ReadableStreamState::Readable)
+            return chunk;
+    }
     if (m_closeRequested && m_queue.isEmpty()) {
         readableStreamDefaultControllerClearAlgorithms(this);
         readableStreamClose(globalObject, m_stream.get());
@@ -331,6 +342,13 @@ void JSReadableStreamDefaultController::pullSteps(JSGlobalObject* globalObject,
     if (!m_queue.isEmpty()) {
         JSValue chunk = dequeueChunkForRead(globalObject);
         RETURN_IF_EXCEPTION(scope, void());
+        if (!chunk) {
+            if (stream->m_state == ReadableStreamState::Closed)
+                RELEASE_AND_RETURN(scope, readRequest->closeSteps(globalObject));
+            ASSERT(stream->m_state == ReadableStreamState::Errored);
+            JSValue storedError = stream->m_storedError.get();
+            RELEASE_AND_RETURN(scope, readRequest->errorSteps(globalObject, storedError ? storedError : jsUndefined()));
+        }
         RELEASE_AND_RETURN(scope, readRequest->chunkSteps(globalObject, chunk));
     }
     readableStreamAddReadRequest(globalObject, stream, readRequest);

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

@@ -76,7 +76,8 @@ class JSReadableStreamDefaultController final : public JSC::JSDestructibleObject
     // [[PullSteps]](readRequest) — userJS: YES (may run the user pull algorithm).
     void pullSteps(JSC::JSGlobalObject*, JSReadRequest*);
     // The queue-hit half of [[PullSteps]]: dequeue + the close-or-pull bookkeeping.
-    // Caller checks !m_queue.isEmpty(). Returns the chunk (empty on exception).
+    // Caller checks !m_queue.isEmpty(). Returns the chunk (a copy on a body-clone tee branch),
+    // or empty: on exception, or with none when the copy failed (the stream is errored or closed).
     JSC::JSValue dequeueChunkForRead(JSC::JSGlobalObject*);
     // [[ReleaseSteps]]() — spec: "Return." (no-op). userJS: no.
     void releaseSteps();

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

@@ -192,6 +192,8 @@ static double drainQueueEntriesInto(JSC::VM& vm, JSGlobalObject* globalObject, J
     double size = isByte ? byteController->m_queue.totalSize() : defaultController->m_queue.totalSize();
     size_t queueLength = isByte ? byteController->m_queue.size() : defaultController->m_queue.size();
     // [[queueTotalSize]] is deliberately NOT decremented while draining (see above).
+    unsigned written = 0;
+    double writtenSize = 0;
     for (unsigned i = 0; i < queueLength; ++i) {
         JSValue chunk;
         if (isByte) {
@@ -210,13 +212,44 @@ static double drainQueueEntriesInto(JSC::VM& vm, JSGlobalObject* globalObject, J
             chunk = JSUint8Array::create(globalObject, globalObject->typedArrayStructure(TypeUint8, resizable), WTF::move(buffer), byteOffset, byteLength);
             RETURN_IF_EXCEPTION(scope, size);
         } else {
-            WTF::Locker locker { defaultController->cellLock() };
-            auto& entry = defaultController->m_queue.first();
-            chunk = entry.value.get();
-            defaultController->m_queue.removeFirst(locker);
+            // The copy below runs user code (userJS: yes) that can read or cancel this branch.
+            if (stream->m_state == ReadableStreamState::Errored) [[unlikely]] {
+                JSValue storedError = stream->m_storedError.get();
+                throwException(globalObject, scope, storedError ? storedError : jsUndefined());
+                return size;
+            }
+            if (defaultController->m_queue.isEmpty() || stream->m_state != ReadableStreamState::Readable) [[unlikely]]
+                break;
+            double entrySize = 0;
+            {
+                WTF::Locker locker { defaultController->cellLock() };
+                auto& entry = defaultController->m_queue.first();
+                chunk = entry.value.get();
+                entrySize = entry.size;
+                defaultController->m_queue.removeFirst(locker);
+            }
+            if (auto* isolatingTee = teeBranchIsolatingChunks(defaultController)) [[unlikely]] {
+                chunk = isolateTeeBranchChunk(globalObject, isolatingTee, chunk);
+                RETURN_IF_EXCEPTION(scope, size);
+                if (!chunk) {
+                    if (stream->m_state == ReadableStreamState::Closed)
+                        break;
+                    ASSERT(stream->m_state == ReadableStreamState::Errored);
+                    JSValue storedError = stream->m_storedError.get();
+                    throwException(globalObject, scope, storedError ? storedError : jsUndefined());
+                    return size;
+                }
+            }
+            writtenSize += entrySize;
         }
         into->putDirectIndex(globalObject, base + i, chunk);
         RETURN_IF_EXCEPTION(scope, size);
+        written = i + 1;
+    }
+    if (written < queueLength) [[unlikely]] {
+        size = writtenSize;
+        into->setLength(globalObject, base + written);
+        RETURN_IF_EXCEPTION(scope, size);
     }
 
     if (isByte) {
@@ -728,6 +761,9 @@ JSC_DEFINE_HOST_FUNCTION(jsReadableStreamDefaultReaderPrototypeFunction_read, (J
         RETURN_IF_EXCEPTION(scope, {});
         RELEASE_AND_RETURN(scope, JSValue::encode(promiseResolvedWith(lexicalGlobalObject, result)));
     }
+    // The copy a body-clone branch makes in that fast path can run user code that released this reader.
+    if (!reader->m_stream)
+        RELEASE_AND_RETURN(scope, JSValue::encode(promiseRejectedWith(lexicalGlobalObject, Bun::createError(lexicalGlobalObject, Bun::ErrorCode::ERR_INVALID_STATE_TypeError, "Invalid state: The reader is not attached to a stream"_s))));
     auto* domGlobalObject = defaultGlobalObject(lexicalGlobalObject);
     auto* runtime = JSStreamsRuntime::from(lexicalGlobalObject);
     auto* promise = JSPromise::create(vm, lexicalGlobalObject->promiseStructure());

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

@@ -80,6 +80,10 @@ class JSStreamTeeState final : public JSC::JSInternalFieldObjectImpl<7> {
     // `canceled1` / `canceled2`
     bool m_canceled1 : 1 { false };
     bool m_canceled2 : 1 { false };
+    // Bun's form of the spec's `cloneForBranch2` (body clone passes true, the public tee()
+    // false). Both branch queues hold the one shared chunk; a branch hands a copy to a script
+    // consumer when it dequeues (isolateTeeBranchChunk). Default tee only.
+    bool m_isolateScriptDeliveries : 1 { false };
 
 private:
     JSStreamTeeState(JSC::VM&, JSC::Structure*);

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

@@ -75,13 +75,15 @@ namespace WebCore {
 //     the JSStreamFromIterableContext).
 //   Tee: context = the JSStreamTeeState, except onByteTeeReaderClosedRejected whose context
 //     is an InternalFieldTuple{teeState, thisReader}.
-//   The two *Microtask entries are the tee chunk-steps "queue a microtask" jobs.
+//   The *ReadChunkMicrotask entries are the tee chunk-steps "queue a microtask" jobs;
+//   onDefaultTeeAbortMicrotask cancels the source after a failed chunk copy (value = reason).
 //   onTextDecodeCancelDeferred: context = an InternalFieldTuple{source reader, result promise}.
 #define FOR_EACH_WEB_STREAMS_REACTION_HANDLER_RS_OPERATIONS(V) \
     V(onFromIterablePullFulfilled)                             \
     V(onFromIterableCancelFulfilled)                           \
     V(onDefaultTeeReadChunkMicrotask)                          \
     V(onDefaultTeeReaderClosedRejected)                        \
+    V(onDefaultTeeAbortMicrotask)                              \
     V(onByteTeeReadChunkMicrotask)                             \
     V(onByteTeeReadIntoChunkMicrotask)                         \
     V(onByteTeeReaderClosedRejected)                           \