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

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

@@ -12,7 +12,7 @@
 #include "root.h"
 #include "StreamsForward.h"
 
-#include "JSReadableStreamDefaultController.h"
+#include "JSReadableByteStreamController.h"
 #include <JavaScriptCore/JSDestructibleObject.h>
 #include <JavaScriptCore/Weak.h>
 
@@ -55,13 +55,16 @@ class JSNativeStreamSourceAdapter final : public JSC::JSDestructibleObject {
     JSC::WriteBarrier<JSC::Unknown> m_drainValue;
     // THE ONE SANCTIONED JSC::Weak in the subsystem. Null-check EVERY read: null ⇒ the JS
     // consumer side was collected ⇒ drop the data / no-op. Assigned lazily — never eagerly.
-    JSC::Weak<JSReadableStreamDefaultController> m_controller;
+    JSC::Weak<JSReadableByteStreamController> m_controller;
     // adaptive chunk size (256 KiB default, doubled once up to 2 MiB).
     size_t m_chunkSize { 0 };
     // #hasResized — the one-shot chunk-size adaptation already happened.
     bool m_hasResized : 1 { false };
     // #closed
     bool m_closed : 1 { false };
+    // the in-flight async pull's m_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 };
 
 private:
     JSNativeStreamSourceAdapter(JSC::VM&, JSC::Structure*);

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

@@ -113,7 +113,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)
@@ -155,11 +155,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:
         break;
     }
     RELEASE_ASSERT_NOT_REACHED();
@@ -188,11 +189,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:
         break;
     }
     RELEASE_ASSERT_NOT_REACHED();

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).
     Bun::WebStreams::SourceAlgorithmSlots m_algorithms;
 
     // Internal methods

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

@@ -599,7 +599,13 @@ JSC_DEFINE_HOST_FUNCTION(jsReadableStreamPrototypeFunction_getReader, (JSGlobalO
     }
 
     if (isBYOB) {
-        // A BYOB reader never materializes Bun's lazy modes.
+        // A lazy native stream is a byte stream; materialize it so the BYOB reader attaches.
+        // A DirectPending stream can never satisfy BYOB, so leave it unmaterialized and let
+        // SetUpReadableStreamBYOBReader reject it without running user code.
+        if (stream->m_bunMode == BunStreamMode::NativePending) {
+            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,14 @@ 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 native stream materializes into a byte
+    // controller before it is locked. A DirectPending stream is left alone so it rejects
+    // without running user code.
+    if (stream->m_bunMode == BunStreamMode::NativePending) {
+        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/JSReadableStreamDefaultController.cpp

@@ -53,10 +53,10 @@ static JSC::JSPromise* invokePromiseReturningMethod(JSC::VM& vm, JSC::JSGlobalOb
     RELEASE_AND_RETURN(scope, promiseResolvedWith(globalObject, result));
 }
 
-// The [[pullAlgorithm]] dispatch. ByteTeeBranch is byte-controller-only and CrossRealm sources
-// are never created (transferable streams are unimplemented); 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.
+// The [[pullAlgorithm]] dispatch. Native and ByteTeeBranch are byte-controller-only and CrossRealm
+// sources are never created (transferable streams are unimplemented); 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* performDefaultControllerPullAlgorithm(JSC::VM& vm, JSC::JSGlobalObject* globalObject, JSReadableStreamDefaultController* controller)
 {
     auto scope = DECLARE_THROW_SCOPE(vm);
@@ -104,7 +104,6 @@ static JSC::JSPromise* performDefaultControllerPullAlgorithm(JSC::VM& vm, JSC::J
     case SourceKind::FromIterable:
         RELEASE_AND_RETURN(scope, fromIterablePullAlgorithm(globalObject, controller));
     case SourceKind::Native:
-        RELEASE_AND_RETURN(scope, nativeSourcePull(globalObject, controller));
     case SourceKind::ByteTeeBranch:
     case SourceKind::CrossRealm:
         break;
@@ -140,7 +139,6 @@ static JSC::JSPromise* performDefaultControllerCancelAlgorithm(JSC::VM& vm, JSC:
     case SourceKind::FromIterable:
         RELEASE_AND_RETURN(scope, fromIterableCancelAlgorithm(globalObject, controller, reason));
     case SourceKind::Native:
-        RELEASE_AND_RETURN(scope, nativeSourceCancel(globalObject, controller, reason));
     case SourceKind::ByteTeeBranch:
     case SourceKind::CrossRealm:
         break;

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

@@ -499,8 +499,11 @@ void readableStreamReaderGenericRelease(JSGlobalObject* globalObject, JSReadable
         }
         break;
     }
-    case ControllerKind::Default: {
-        auto* controller = defaultControllerOf(stream);
+    case ControllerKind::Default:
+        defaultControllerOf(stream)->releaseSteps();
+        break;
+    case ControllerKind::Byte: {
+        auto* controller = byteControllerOf(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) {
@@ -520,9 +523,6 @@ void readableStreamReaderGenericRelease(JSGlobalObject* globalObject, JSReadable
         }
         break;
     }
-    case ControllerKind::Byte:
-        byteControllerOf(stream)->releaseSteps();
-        break;
     }
     stream->m_reader.clear();
     reader->m_stream.clear();

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

@@ -114,7 +114,7 @@ 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
+    Native, // Bun: lazily-materialized native source on a BYTE controller
             // (context = JSNativeStreamSourceAdapter)
 };
 

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

@@ -182,7 +182,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

@@ -488,17 +488,17 @@ 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 byte controller (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
-JSC::JSPromise* nativeSourcePull(JSC::JSGlobalObject*, JSReadableStreamDefaultController*); // userJS: no (native handle.pull; its promise's reactions are onNativePull*) — BunStreamSource.cpp
-JSC::JSPromise* nativeSourceCancel(JSC::JSGlobalObject*, JSReadableStreamDefaultController*, JSC::JSValue reason); // userJS: no (native handle.cancel + teardown) — BunStreamSource.cpp
+// `switch (m_algorithms.kind)` in JSReadableByteStreamController.cpp (a Native source is
+// ALWAYS a byte controller so Blob/File/Bytes streams accept BYOB readers); 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*, JSReadableByteStreamController*); // userJS: no (native handle.start; enqueues the drain value) — BunStreamSource.cpp
+JSC::JSPromise* nativeSourcePull(JSC::JSGlobalObject*, JSReadableByteStreamController*); // userJS: no (native handle.pull; its promise's reactions are onNativePull*) — BunStreamSource.cpp
+JSC::JSPromise* nativeSourceCancel(JSC::JSGlobalObject*, JSReadableByteStreamController*, JSC::JSValue reason); // userJS: no (native handle.cancel + teardown) — BunStreamSource.cpp
 // The JSSink entry point (GlobalObject::assignToStream's body). Returns undefined or
 // a JSPromise (the Signal protocol's value).
 JSC::JSValue assignToStream(JSC::JSGlobalObject*, JSReadableStream*, JSC::JSValue jsSinkController); // userJS: yes — BunStreamSource.cpp

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

@@ -559,6 +559,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