From 00f33d71bfb4cb24b26f2f229c4c427ba506ccde Mon Sep 17 00:00:00 2001 From: robobun <117481402+robobun@users.noreply.github.com> Date: Tue, 10 Nov 2026 20:18:51 +0000 Subject: [PATCH 1/2] streams: reject a direct source without pull() in the ReadableStream constructor A `type: "direct"` stream does not run start(), so a source without pull() can never write or close. The constructor accepted it, and each consumer gave it a meaning of its own: readers rejected with a TypeError, and a native sink took it as a complete empty body and left the stream readable and unlocked, so a later read() never settled. The constructor now throws the TypeError that it already throws for a pull that is not a function. The two guards for a missing pull in readDirectStream and oneShotCallPull are unreachable and are deleted. --- docs/runtime/streams.mdx | 2 ++ packages/bun-types/bun.d.ts | 3 +++ .../webcore/streams/BunStreamConsumers.cpp | 4 ---- .../webcore/streams/BunStreamSource.cpp | 5 ----- .../streams/JSDirectStreamController.cpp | 3 ++- .../webcore/streams/JSDirectStreamSource.h | 3 ++- .../webcore/streams/JSReadableStream.cpp | 5 ++++- test/js/web/streams/streams.test.js | 19 +++++++++++++++++++ 8 files changed, 32 insertions(+), 12 deletions(-) diff --git a/docs/runtime/streams.mdx b/docs/runtime/streams.mdx index cc100dd402d6..81d6b635fc6e 100644 --- a/docs/runtime/streams.mdx +++ b/docs/runtime/streams.mdx @@ -71,6 +71,8 @@ const stream = new ReadableStream({ When using a direct `ReadableStream`, the destination handles all chunk queueing. The destination receives the bytes you pass to `controller.write()`. When the stream is read from JavaScript, Bun buffers the writes and delivers them as `Uint8Array` chunks (strings are UTF-8 encoded). +A direct `ReadableStream` needs a `pull()` function. Bun does not call `start()` on a direct source, so `pull()` is where you write. Without `pull()`, the `ReadableStream` constructor throws a `TypeError`. + ### Ending the stream Call `controller.close()` (or `controller.end()`) once everything is written. Every destination flushes what it has buffered and finishes at that point. `controller.close(error)` fails the stream instead: nothing more is flushed and the consumer rejects with `error`. diff --git a/packages/bun-types/bun.d.ts b/packages/bun-types/bun.d.ts index 7ff2db50d2b1..c8147a37eeae 100644 --- a/packages/bun-types/bun.d.ts +++ b/packages/bun-types/bun.d.ts @@ -603,6 +603,9 @@ declare module "bun" { * * A reader (`getReader()`, `for await`, `pipeTo()`) calls `pull()` again * for a later read, once the previous call has settled. + * + * `pull` is required: a direct stream does not call `start()`. Without + * `pull`, the `ReadableStream` constructor throws a `TypeError`. */ pull: (controller: ReadableStreamDirectController) => void | PromiseLike; type: "direct"; diff --git a/src/jsc/bindings/webcore/streams/BunStreamConsumers.cpp b/src/jsc/bindings/webcore/streams/BunStreamConsumers.cpp index bacc19d0f128..728827dda1ce 100644 --- a/src/jsc/bindings/webcore/streams/BunStreamConsumers.cpp +++ b/src/jsc/bindings/webcore/streams/BunStreamConsumers.cpp @@ -1075,10 +1075,6 @@ static JSValue oneShotCallPull(JSC::VM& vm, JSGlobalObject* globalObject, JSDire { auto scope = DECLARE_THROW_SCOPE(vm); JSObject* pullFunction = source->pullFunction(); - if (!pullFunction) [[unlikely]] { - throwTypeError(globalObject, scope, "The 'pull' method of a direct ReadableStream's underlying source is not a function"_s); - return {}; - } MarkedArgumentBuffer arguments; arguments.append(sink); RELEASE_AND_RETURN(scope, JSC::call(globalObject, pullFunction, JSC::getCallData(pullFunction), source->thisValue(), arguments)); diff --git a/src/jsc/bindings/webcore/streams/BunStreamSource.cpp b/src/jsc/bindings/webcore/streams/BunStreamSource.cpp index f4867711b82c..7b9117bbe56c 100644 --- a/src/jsc/bindings/webcore/streams/BunStreamSource.cpp +++ b/src/jsc/bindings/webcore/streams/BunStreamSource.cpp @@ -731,11 +731,6 @@ static JSValue readDirectStream(JSGlobalObject* globalObject, JSReadableStream* stream->m_bunMode = BunStreamMode::Default; JSObject* pull = source->pullFunction(); - if (!pull) { - sinkController->end(globalObject); - RETURN_IF_EXCEPTION(scope, {}); - return jsUndefined(); - } stream->m_controller.set(vm, stream, sinkController); stream->m_controllerKind = ControllerKind::NativeSink; diff --git a/src/jsc/bindings/webcore/streams/JSDirectStreamController.cpp b/src/jsc/bindings/webcore/streams/JSDirectStreamController.cpp index a8ec19ecf04f..f86c4143ccd3 100644 --- a/src/jsc/bindings/webcore/streams/JSDirectStreamController.cpp +++ b/src/jsc/bindings/webcore/streams/JSDirectStreamController.cpp @@ -51,9 +51,10 @@ void JSDirectStreamSource::finishCreation(VM& vm, JSValue underlyingSource, JSOb { Base::finishCreation(vm); ASSERT(inherits(info())); + ASSERT(pull); auto valueOrNull = [](JSObject* object) { return object ? JSValue(object) : jsNull(); }; Base::internalField(static_cast(Field::UnderlyingSource)).set(vm, this, underlyingSource); - Base::internalField(static_cast(Field::Pull)).set(vm, this, valueOrNull(pull)); + Base::internalField(static_cast(Field::Pull)).set(vm, this, pull); Base::internalField(static_cast(Field::Cancel)).set(vm, this, valueOrNull(cancel)); Base::internalField(static_cast(Field::Close)).set(vm, this, valueOrNull(close)); } diff --git a/src/jsc/bindings/webcore/streams/JSDirectStreamSource.h b/src/jsc/bindings/webcore/streams/JSDirectStreamSource.h index 6c70ad9ab098..c0b20789f908 100644 --- a/src/jsc/bindings/webcore/streams/JSDirectStreamSource.h +++ b/src/jsc/bindings/webcore/streams/JSDirectStreamSource.h @@ -22,8 +22,9 @@ class JSDirectStreamSource final : public JSC::JSInternalFieldObjectImpl<4> { enum class Field : uint32_t { // The user's underlyingSource object, or an internal source's context cell. UnderlyingSource = 0, - // Each is a callable or null. + // A callable: the ReadableStream constructor rejects a direct source without pull(). Pull, + // Each is a callable or null. Cancel, // Bun's close(reason) lifecycle hook: the stream ended or errored. Close, diff --git a/src/jsc/bindings/webcore/streams/JSReadableStream.cpp b/src/jsc/bindings/webcore/streams/JSReadableStream.cpp index 1222dfd927c5..dfa5cd52818d 100644 --- a/src/jsc/bindings/webcore/streams/JSReadableStream.cpp +++ b/src/jsc/bindings/webcore/streams/JSReadableStream.cpp @@ -335,13 +335,16 @@ template<> JSC::EncodedJSValue JSC_HOST_CALL_ATTRIBUTES JSReadableStreamConstruc switch (source.type) { case BunUnderlyingSourceType::Direct: { + // A direct stream does not run start(), so pull() is the only way for its source to write or close. + if (!source.dict.pull) + return throwVMTypeError(lexicalGlobalObject, scope, "The underlying source's 'pull' property must be a function"_s); // Bun's close(reason) hook is not a dictionary member; it is converted like one. JSValue close = asObject(underlyingSource)->get(lexicalGlobalObject, builtinNames(vm).closePublicName()); RETURN_IF_EXCEPTION(scope, {}); if (!close.isUndefined() && !close.isCallable()) return throwVMTypeError(lexicalGlobalObject, scope, "The underlying source's 'close' property must be a function"_s); auto* directSource = JSDirectStreamSource::create(vm, JSStreamsRuntime::from(lexicalGlobalObject)->directStreamSourceStructure(defaultGlobalObject(lexicalGlobalObject)), underlyingSource, - source.dict.pull ? source.dict.pull.getObject() : nullptr, source.dict.cancel ? source.dict.cancel.getObject() : nullptr, close.isUndefined() ? nullptr : close.getObject()); + source.dict.pull.getObject(), source.dict.cancel ? source.dict.cancel.getObject() : nullptr, close.isUndefined() ? nullptr : close.getObject()); // A direct stream has no controller yet; materializeIfNeeded() builds it on first use. stream->m_bunMode = BunStreamMode::DirectPending; stream->m_directSource.set(vm, stream, directSource); diff --git a/test/js/web/streams/streams.test.js b/test/js/web/streams/streams.test.js index 8372f338ca3f..7958454ba9cf 100644 --- a/test/js/web/streams/streams.test.js +++ b/test/js/web/streams/streams.test.js @@ -1416,6 +1416,25 @@ describe("multi-chunk consumers produce exactly the concatenated bytes", () => { expect(reads.toSorted()).toEqual(["cancel", "close", "pull"]); expect(() => new ReadableStream({ type: "direct", pull() {}, close: 1 })).toThrow(TypeError); }); + + // A direct stream does not run start(), so a source without pull() can never write or close. + it.each([ + ["no methods", { type: "direct" }], + ["pull: undefined", { type: "direct", pull: undefined }], + ["start() only", { type: "direct", start() {} }], + ["cancel() only", { type: "direct", cancel() {} }], + ])("a source without pull() does not construct: %s", (_label, source) => { + let thrown; + try { + new ReadableStream(source); + } catch (error) { + thrown = error; + } + expect({ name: thrown?.name, message: thrown?.message }).toEqual({ + name: "TypeError", + message: "The underlying source's 'pull' property must be a function", + }); + }); }); // A type:"direct" pull() is re-invoked per read as a demand signal, but never while a From 5a381f0169e5139dffed10c46033aa864d0b9685 Mon Sep 17 00:00:00 2001 From: robobun <117481402+robobun@users.noreply.github.com> Date: Tue, 10 Nov 2026 23:31:08 +0000 Subject: [PATCH 2/2] streams: say why a direct source needs pull() when it is missing A direct source without pull() gets its own TypeError text. The text names the condition (type is "direct") and the reason (a direct stream does not call start()). A pull that is present and not a function keeps the existing text. --- src/jsc/bindings/webcore/streams/JSReadableStream.cpp | 3 +-- test/js/web/streams/streams.test.js | 2 +- 2 files changed, 2 insertions(+), 3 deletions(-) diff --git a/src/jsc/bindings/webcore/streams/JSReadableStream.cpp b/src/jsc/bindings/webcore/streams/JSReadableStream.cpp index dfa5cd52818d..b2cd1e887a53 100644 --- a/src/jsc/bindings/webcore/streams/JSReadableStream.cpp +++ b/src/jsc/bindings/webcore/streams/JSReadableStream.cpp @@ -335,9 +335,8 @@ template<> JSC::EncodedJSValue JSC_HOST_CALL_ATTRIBUTES JSReadableStreamConstruc switch (source.type) { case BunUnderlyingSourceType::Direct: { - // A direct stream does not run start(), so pull() is the only way for its source to write or close. if (!source.dict.pull) - return throwVMTypeError(lexicalGlobalObject, scope, "The underlying source's 'pull' property must be a function"_s); + return throwVMTypeError(lexicalGlobalObject, scope, "The underlying source's 'pull' property is required when 'type' is \"direct\": a direct stream does not call start()"_s); // Bun's close(reason) hook is not a dictionary member; it is converted like one. JSValue close = asObject(underlyingSource)->get(lexicalGlobalObject, builtinNames(vm).closePublicName()); RETURN_IF_EXCEPTION(scope, {}); diff --git a/test/js/web/streams/streams.test.js b/test/js/web/streams/streams.test.js index 7958454ba9cf..2a994ad33499 100644 --- a/test/js/web/streams/streams.test.js +++ b/test/js/web/streams/streams.test.js @@ -1432,7 +1432,7 @@ describe("multi-chunk consumers produce exactly the concatenated bytes", () => { } expect({ name: thrown?.name, message: thrown?.message }).toEqual({ name: "TypeError", - message: "The underlying source's 'pull' property must be a function", + message: `The underlying source's 'pull' property is required when 'type' is "direct": a direct stream does not call start()`, }); }); });