webstreams: make Blob/File/body streams byte streams so BYOB readers attach

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

@@ -8,6 +8,7 @@
 #include "root.h"
 #include "StreamsForward.h"
 
+#include "JSReadableByteStreamController.h"
 #include "JSReadableStreamDefaultController.h"
 #include <JavaScriptCore/JSCast.h>
 #include <JavaScriptCore/JSInternalFieldObjectImpl.h>
@@ -67,13 +68,18 @@ class JSNativeStreamSourceAdapter final : public JSC::JSInternalFieldObjectImpl<
     JSC::JSObject* pendingView() const { return internalField(Field::PendingView).get().getObject(); }
     JSC::JSObject* closer() const { return internalField(Field::Closer).get().getObject(); }
     JSC::JSValue drainValue() const { return internalField(Field::DrainValue).get(); }
-    JSReadableStreamDefaultController* controller() const { return dynamicDowncast<JSReadableStreamDefaultController>(internalField(Field::Controller).get()); }
+    // Exactly one of these is non-null once set: a text-mode adapter installs a default
+    // controller (it enqueues JSStrings), a binary adapter installs a byte controller so
+    // BYOB readers can attach.
+    JSC::JSObject* controller() const { return internalField(Field::Controller).get().getObject(); }
+    JSReadableStreamDefaultController* defaultController() const { return dynamicDowncast<JSReadableStreamDefaultController>(internalField(Field::Controller).get()); }
+    JSReadableByteStreamController* byteController() const { return dynamicDowncast<JSReadableByteStreamController>(internalField(Field::Controller).get()); }
 
     void setHandle(JSC::VM& vm, JSC::JSValue v) { internalField(Field::Handle).set(vm, this, v); }
     void setPendingView(JSC::VM& vm, JSC::JSValue v) { internalField(Field::PendingView).set(vm, this, v); }
     void setCloser(JSC::VM& vm, JSC::JSValue v) { internalField(Field::Closer).set(vm, this, v); }
     void setDrainValue(JSC::VM& vm, JSC::JSValue v) { internalField(Field::DrainValue).set(vm, this, v); }
-    void setController(JSC::VM& vm, JSReadableStreamDefaultController* c) { internalField(Field::Controller).set(vm, this, c); }
+    void setController(JSC::VM& vm, JSC::JSObject* c) { internalField(Field::Controller).set(vm, this, c); }
 
     void clearHandle(JSC::VM& vm) { internalField(Field::Handle).set(vm, this, JSC::jsUndefined()); }
     void clearPendingView(JSC::VM& vm) { internalField(Field::PendingView).set(vm, this, JSC::jsUndefined()); }
@@ -88,6 +94,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 };
+    // the in-flight async pull's PendingView is the head pull-into descriptor's buffer
+    // (respond(n) on fulfilment) rather than an adapter-owned scratch buffer (enqueue()).
+    bool m_pendingIsBYOB : 1 { false };
     Bun::WebStreams::StreamingUTF8DecodeState m_textState;
 
 private:

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

@@ -114,7 +114,7 @@ static JSC::JSPromise* invokePromiseReturningMethod(JSC::VM& vm, JSC::JSGlobalOb
 }
 
 // The [[pullAlgorithm]] dispatch. The reachable kind set on a byte controller is exactly
-// {JavaScript, Nothing, ByteTeeBranch}; the switch is total over SourceKind.
+// {JavaScript, Nothing, ByteTeeBranch, Native}; the switch is total over SourceKind.
 // Returns nullptr with no exception pending when the pull completed synchronously with a
 // non-thenable result: the caller queues the upon-fulfillment handler without a wrapper promise.
 static JSC::JSPromise* performByteControllerPullAlgorithm(JSC::VM& vm, JSC::JSGlobalObject* globalObject, JSReadableByteStreamController* controller)
@@ -156,11 +156,12 @@ static JSC::JSPromise* performByteControllerPullAlgorithm(JSC::VM& vm, JSC::JSGl
         return nullptr;
     case SourceKind::ByteTeeBranch:
         RELEASE_AND_RETURN(scope, byteTeePullAlgorithm(globalObject, uncheckedDowncast<JSStreamTeeState>(controller->m_algorithms.algorithmContext.get()), controller->m_algorithms.teeBranchIndex));
+    case SourceKind::Native:
+        RELEASE_AND_RETURN(scope, nativeSourcePull(globalObject, controller));
     case SourceKind::Transform:
     case SourceKind::TeeBranch:
     case SourceKind::FromIterable:
     case SourceKind::CrossRealm:
-    case SourceKind::Native:
     case SourceKind::TextDecode:
         break;
     }
@@ -190,11 +191,12 @@ static JSC::JSPromise* performByteControllerCancelAlgorithm(JSC::VM& vm, JSC::JS
         RELEASE_AND_RETURN(scope, promiseFulfilledWith(globalObject, JSC::jsUndefined()));
     case SourceKind::ByteTeeBranch:
         RELEASE_AND_RETURN(scope, byteTeeCancelAlgorithm(globalObject, uncheckedDowncast<JSStreamTeeState>(controller->m_algorithms.algorithmContext.get()), controller->m_algorithms.teeBranchIndex, reason));
+    case SourceKind::Native:
+        RELEASE_AND_RETURN(scope, nativeSourceCancel(globalObject, controller, reason));
     case SourceKind::Transform:
     case SourceKind::TeeBranch:
     case SourceKind::FromIterable:
     case SourceKind::CrossRealm:
-    case SourceKind::Native:
     case SourceKind::TextDecode:
         break;
     }

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

@@ -78,9 +78,9 @@ class JSReadableByteStreamController final : public JSC::JSDestructibleObject {
     // stream has NO size algorithm (a byte stream given a size strategy is a RangeError at
     // construction). See SourceAlgorithmSlots (StreamQueue.h).
     // The reachable m_algorithms.kind set on a BYTE controller is EXACTLY
-    // {JavaScript, Nothing, ByteTeeBranch}. CrossRealm is impossible (the cross-realm
+    // {JavaScript, Nothing, ByteTeeBranch, Native}. CrossRealm is impossible (the cross-realm
     // readable endpoint is always a DEFAULT controller — JSCrossRealmTransformState's
-    // back-pointer is exact-typed to one) and Native always uses a DEFAULT controller.
+    // back-pointer is exact-typed to one). A text-mode Native source uses a DEFAULT controller.
     Bun::WebStreams::SourceAlgorithmSlots m_algorithms;
 
     // Internal methods

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

@@ -599,7 +599,14 @@ JSC_DEFINE_HOST_FUNCTION(jsReadableStreamPrototypeFunction_getReader, (JSGlobalO
     }
 
     if (isBYOB) {
-        // A BYOB reader never materializes Bun's lazy modes.
+        // A lazy binary native stream is a byte stream; materialize it so the BYOB reader
+        // attaches. A text-mode native stream and a DirectPending stream can never satisfy
+        // BYOB, so leave them unmaterialized and let SetUpReadableStreamBYOBReader reject
+        // without running user code.
+        if (stream->m_bunMode == BunStreamMode::NativePending && !stream->m_nativeTextMode) {
+            stream->materializeIfNeeded(lexicalGlobalObject);
+            RETURN_IF_EXCEPTION(scope, {});
+        }
         auto* reader = acquireReadableStreamBYOBReader(lexicalGlobalObject, stream);
         RETURN_IF_EXCEPTION(scope, {});
         return JSValue::encode(reader);

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

@@ -244,6 +244,13 @@ template<> JSC::EncodedJSValue JSC_HOST_CALL_ATTRIBUTES JSReadableStreamBYOBRead
     if (!stream)
         return throwVMTypeError(lexicalGlobalObject, scope, "ReadableStreamBYOBReader constructor requires a ReadableStream as its first argument"_s);
 
+    // Same as getReader({mode:"byob"}): a lazy binary native stream materializes into a byte
+    // controller before it is locked.
+    if (stream->m_bunMode == BunStreamMode::NativePending && !stream->m_nativeTextMode) {
+        stream->materializeIfNeeded(lexicalGlobalObject);
+        RETURN_IF_EXCEPTION(scope, {});
+    }
+
     auto* structure = structureForNewTarget(vm, constructor, lexicalGlobalObject, asObject(callFrame->newTarget()));
     RETURN_IF_EXCEPTION(scope, {});
     auto* reader = JSReadableStreamBYOBReader::create(vm, structure);

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

@@ -537,28 +537,20 @@ void readableStreamReaderGenericRelease(JSGlobalObject* globalObject, JSReadable
     case ControllerKind::Default: {
         auto* controller = defaultControllerOf(stream);
         controller->releaseSteps();
-        // Bun: drop the native handle's event-loop ref when its consumer releases the lock.
-        if (stream->m_nativePtr && controller->m_algorithms.kind == SourceKind::Native) {
-            const auto* adapter = uncheckedDowncast<WebCore::JSNativeStreamSourceAdapter>(controller->m_algorithms.algorithmContext.get());
-            if (auto* handle = adapter->handle()) {
-                JSValue updateRef = handle->getIfPropertyExists(globalObject, builtinNames(vm).updateRefPublicName());
-                RETURN_IF_EXCEPTION(scope, void());
-                if (updateRef && updateRef.isCallable()) {
-                    auto callData = JSC::getCallData(updateRef);
-                    MarkedArgumentBuffer args;
-                    args.append(jsBoolean(false));
-                    ASSERT(!args.hasOverflowed());
-                    JSC::call(globalObject, updateRef, callData, handle, args);
-                    RETURN_IF_EXCEPTION(scope, void());
-                }
-            }
-        }
+        if (stream->m_nativePtr && controller->m_algorithms.kind == SourceKind::Native)
+            nativeSourceDropEventLoopRef(globalObject, uncheckedDowncast<WebCore::JSNativeStreamSourceAdapter>(controller->m_algorithms.algorithmContext.get()));
+        RETURN_IF_EXCEPTION(scope, void());
         break;
     }
-    case ControllerKind::Byte:
-        byteControllerOf(stream)->releaseSteps();
+    case ControllerKind::Byte: {
+        auto* controller = byteControllerOf(stream);
+        controller->releaseSteps();
+        if (stream->m_nativePtr && controller->m_algorithms.kind == SourceKind::Native)
+            nativeSourceDropEventLoopRef(globalObject, uncheckedDowncast<WebCore::JSNativeStreamSourceAdapter>(controller->m_algorithms.algorithmContext.get()));
+        RETURN_IF_EXCEPTION(scope, void());
         break;
     }
+    }
     stream->m_reader.clear();
     reader->m_stream.clear();
 }

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

@@ -113,8 +113,9 @@ enum class SourceKind : uint8_t {
     ByteTeeBranch, // a ReadableByteStreamTee branch     (context = the JSStreamTeeState)
     FromIterable, // ReadableStream.from(asyncIterable)  (context = JSStreamFromIterableContext)
     CrossRealm, // receiving end of a postMessage transfer (context = JSCrossRealmTransformState)
-    Native, // Bun: lazily-materialized native source on a DEFAULT controller
-            // (context = JSNativeStreamSourceAdapter)
+    Native, // Bun: lazily-materialized native source — a BYTE controller for binary streams
+            // (so Blob/File/Bytes streams accept BYOB readers), a DEFAULT controller for
+            // text-mode (Body.textStream()) (context = JSNativeStreamSourceAdapter)
     TextDecode, // Body.textStream() reading from an existing byte stream
                 // (algorithmContext = source JSReadableStreamDefaultReader;
                 // decode state inline on m_algorithms.textDecodeState)

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

@@ -212,7 +212,9 @@ extern "C" JSC::EncodedJSValue ReadableStream__empty(Zig::GlobalObject* globalOb
 {
     auto& vm = JSC::getVM(globalObject);
     auto scope = DECLARE_THROW_SCOPE(vm);
-    auto* stream = createReadableStream(globalObject, SourceKind::Nothing, nullptr, jsUndefined());
+    // Every caller is a byte-producing source (an empty Blob/body/subprocess pipe), so a
+    // closed byte stream lets a BYOB reader attach and observe done=true.
+    auto* stream = createReadableByteStream(globalObject, SourceKind::Nothing, nullptr);
     RETURN_IF_EXCEPTION(scope, {});
     readableStreamClose(globalObject, stream);
     RETURN_IF_EXCEPTION(scope, {});

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

@@ -508,17 +508,22 @@ void pipeToReadRequestErrorSteps(JSC::JSGlobalObject*, JSStreamPipeToOperation*,
 
 // BunStreamSource.cpp — the lazy native source and the native-sink pumps.
 
-// lazyLoadStream: installs the Native default controller (or the empty fast path).
+// lazyLoadStream: installs the Native controller (a byte controller, or a default controller
+// for text-mode native streams) or the empty fast path.
 void materializeNativeSource(JSC::JSGlobalObject*, JSReadableStream*); // userJS: yes — BunStreamSource.cpp
 
-// The SourceKind::Native algorithm ARMS. The pull/cancel dispatch is a TOTAL
-// `switch (m_algorithms.kind)` in JSReadableStreamDefaultController.cpp (a Native source is
-// ALWAYS a default controller); these bodies live HERE per BunStreamSource.h's owner rule,
-// so this is the declared bridge between the two files. The controller's algorithmContext is
-// the JSNativeStreamSourceAdapter for all three.
-JSC::JSValue nativeSourceStart(JSC::JSGlobalObject*, JSReadableStreamDefaultController*); // userJS: no (native handle.start; enqueues the drain value) — BunStreamSource.cpp
+// The SourceKind::Native algorithm ARMS. A binary Native source installs a BYTE controller
+// (so Blob/File/Bytes streams accept BYOB readers); a text-mode Native source (Body.textStream())
+// installs a DEFAULT controller. Both controllers' total `switch (m_algorithms.kind)` dispatch
+// into the matching overload here. The controller's algorithmContext is the
+// JSNativeStreamSourceAdapter for all of these.
+JSC::JSValue nativeSourceStart(JSC::JSGlobalObject*, JSNativeStreamSourceAdapter*); // userJS: no (native handle.start; enqueues the drain value) — BunStreamSource.cpp
 JSC::JSPromise* nativeSourcePull(JSC::JSGlobalObject*, JSReadableStreamDefaultController*); // userJS: no (native handle.pull; its promise's reactions are onNativePull*) — BunStreamSource.cpp
+JSC::JSPromise* nativeSourcePull(JSC::JSGlobalObject*, JSReadableByteStreamController*); // userJS: no — BunStreamSource.cpp
 JSC::JSPromise* nativeSourceCancel(JSC::JSGlobalObject*, JSReadableStreamDefaultController*, JSC::JSValue reason); // userJS: no (native handle.cancel + teardown) — BunStreamSource.cpp
+JSC::JSPromise* nativeSourceCancel(JSC::JSGlobalObject*, JSReadableByteStreamController*, JSC::JSValue reason); // userJS: no — BunStreamSource.cpp
+// Bun: drop the native handle's event-loop ref when its consumer releases the lock.
+void nativeSourceDropEventLoopRef(JSC::JSGlobalObject*, const JSNativeStreamSourceAdapter*); // userJS: no — BunStreamSource.cpp
 // readableStreamCancel's ControllerKind::None arm for a still-NativePending stream: calls
 // handle.updateRef(false) + handle.cancel(reason) on m_nativePtr directly, no materialize.
 JSC::JSPromise* cancelPendingNativeSource(JSC::JSGlobalObject*, JSReadableStream*, JSC::JSValue reason); // userJS: no — BunStreamSource.cpp

test/js/web/fetch/blob.test.ts

@@ -560,6 +560,143 @@ test("Blob.slice at an odd byte offset decodes UTF-16LE (BOM) content with text(
   expect(exitCode).toBe(0);
 });
 
+// https://w3c.github.io/FileAPI/#stream-method-algo
+// "return the result of running get stream on this"; get stream creates a byte stream.
+describe("Blob.prototype.stream() is a byte stream (supports BYOB readers)", () => {
+  test("getReader({ mode: 'byob' }) fills the caller's view", async () => {
+    const payload = new Uint8Array(100_000);
+    for (let i = 0; i < payload.length; i++) payload[i] = i & 0xff;
+    const blob = new Blob([payload]);
+
+    const reader = blob.stream().getReader({ mode: "byob" });
+    const first = await reader.read(new Uint8Array(64));
+    expect(first.done).toBe(false);
+    expect(first.value).toBeInstanceOf(Uint8Array);
+    expect(first.value!.byteLength).toBe(64);
+    expect(first.value![0]).toBe(0);
+    expect(first.value![63]).toBe(63);
+
+    // Second read with a different-sized view picks up where the first left off.
+    const second = await reader.read(new Uint8Array(200));
+    expect(second.value!.byteLength).toBe(200);
+    expect(second.value![0]).toBe(64);
+    expect(second.value![199]).toBe((64 + 199) & 0xff);
+
+    await reader.cancel();
+  });
+
+  test("BYOB read loop reassembles the full blob", async () => {
+    const payload = new Uint8Array(10_000);
+    for (let i = 0; i < payload.length; i++) payload[i] = i & 0xff;
+    const blob = new Blob([payload]);
+
+    const reader = blob.stream().getReader({ mode: "byob" });
+    const out = new Uint8Array(payload.length);
+    let offset = 0;
+    let buf = new Uint8Array(777);
+    while (offset < out.length) {
+      const { value, done } = await reader.read(buf);
+      if (done) break;
+      out.set(value, offset);
+      offset += value.byteLength;
+      buf = new Uint8Array(value.buffer, 0, value.buffer.byteLength);
+    }
+    expect(offset).toBe(payload.length);
+    expect(out).toEqual(payload);
+    // The stream closes once the blob is drained.
+    const tail = await reader.read(new Uint8Array(16));
+    expect(tail.done).toBe(true);
+  });
+
+  test("new ReadableStreamBYOBReader(blob.stream()) also works", async () => {
+    const blob = new Blob([new Uint8Array(256).map((_, i) => i)]);
+    const reader = new ReadableStreamBYOBReader(blob.stream());
+    const { value, done } = await reader.read(new Uint8Array(16));
+    expect(done).toBe(false);
+    expect([...value!]).toEqual([...Array(16).keys()]);
+    await reader.cancel();
+  });
+
+  test("Bun.file().stream() supports BYOB readers", async () => {
+    using dir = tempDir("blob-byob", {
+      "data.bin": Buffer.alloc(1024, "abcd").toString("binary"),
+    });
+    const reader = Bun.file(path.join(String(dir), "data.bin"))
+      .stream()
+      .getReader({ mode: "byob" });
+    const { value, done } = await reader.read(new Uint8Array(8));
+    expect(done).toBe(false);
+    expect(Buffer.from(value!).toString()).toBe("abcdabcd");
+    await reader.cancel();
+  });
+
+  test("default reader still works on a Blob byte stream", async () => {
+    const payload = new Uint8Array(5_000).map((_, i) => i & 0xff);
+    const blob = new Blob([payload]);
+    const reader = blob.stream().getReader();
+    const chunks: Uint8Array[] = [];
+    while (true) {
+      const { done, value } = await reader.read();
+      if (done) break;
+      chunks.push(value);
+    }
+    expect(Buffer.concat(chunks)).toEqual(Buffer.from(payload));
+  });
+
+  test("releaseLock after a default reader then attach a BYOB reader", async () => {
+    const payload = new Uint8Array(40_000).map((_, i) => i & 0xff);
+    const stream = new Blob([payload]).stream();
+    const r1 = stream.getReader();
+    const first = await r1.read();
+    expect(first.done).toBe(false);
+    const consumed = first.value!.byteLength;
+    expect(consumed).toBeGreaterThan(0);
+    expect(consumed).toBeLessThan(payload.length);
+    r1.releaseLock();
+
+    // A byte stream permits a BYOB reader after a default reader is released,
+    // and the BYOB read picks up exactly where the default reader left off.
+    const r2 = stream.getReader({ mode: "byob" });
+    const second = await r2.read(new Uint8Array(32));
+    expect(second.done).toBe(false);
+    expect(second.value!.byteLength).toBe(32);
+    expect(Buffer.from(second.value!)).toEqual(Buffer.from(payload.subarray(consumed, consumed + 32)));
+    await r2.cancel();
+  });
+
+  test("empty Blob byte stream closes immediately for a BYOB reader", async () => {
+    const reader = new Blob([]).stream().getReader({ mode: "byob" });
+    const { value, done } = await reader.read(new Uint8Array(8));
+    expect(done).toBe(true);
+    expect(value!.byteLength).toBe(0);
+  });
+
+  test("closing with a partially-filled multi-byte-element BYOB view rejects read() without an uncaught exception", async () => {
+    await using proc = Bun.spawn({
+      cmd: [
+        bunExe(),
+        "-e",
+        `
+          process.on("uncaughtException", e => { console.log("UNCAUGHT"); process.exitCode = 1; });
+          const r = new Blob([new Uint8Array([1, 2, 3])]).stream().getReader({ mode: "byob" });
+          const err = await r.read(new Uint32Array(4)).then(() => null, e => e);
+          console.log(err instanceof TypeError ? "rejected" : "resolved");
+          await r.closed.catch(() => {});
+        `,
+      ],
+      env: bunEnv,
+      stdout: "pipe",
+      stderr: "pipe",
+    });
+    const [stdout, stderr, exitCode] = await Promise.all([proc.stdout.text(), proc.stderr.text(), proc.exited]);
+    expect({ stdout: stdout.trim(), stderr: stderr.trim(), exitCode }).toEqual({
+      stdout: "rejected",
+      stderr: "",
+      exitCode: 0,
+    });
+  });
+});
+
 // structuredClone/postMessage of sliced Blobs and Files is covered by
 // test/js/web/structured-clone-blob-file.test.ts. These tests focus on the
 // consumer paths that go through resolve_size()/resolved_size() rather than