From 05e3902e2bdc6de81e823eaa1b990089651eb533 Mon Sep 17 00:00:00 2001 From: robobun <117481402+robobun@users.noreply.github.com> Date: Sat, 18 Jul 2026 10:28:36 +0000 Subject: [PATCH 1/6] node:stream: make isReadable/isWritable/isErrored/isDisturbed work with WHATWG web streams Node brands its web stream classes with Symbol.for("nodejs.stream.*") getters, and stream.isReadable/isWritable/isErrored/isDisturbed check those symbols first. Bun's ReadableStream/WritableStream are native JSC cells with no such getters, so all four helpers fell through to the node-stream shape probe and returned null/false for every web-stream state. Add a private $webStreamState() intrinsic that reads the native [[state]] slot directly, and duck-type ReadableStream/WritableStream operands in the four helpers (after the symbol check, so user brands still take precedence). isDisturbed reuses the existing $disturbed private accessor on ReadableStream. --- src/js/builtins.d.ts | 8 +++ src/js/builtins/BunBuiltinNames.h | 1 + src/js/internal/streams/utils.ts | 11 +++- src/jsc/bindings/ZigGlobalObject.cpp | 17 +++++ test/js/node/stream/node-stream.test.js | 87 ++++++++++++++++++++++++- 5 files changed, 122 insertions(+), 2 deletions(-) diff --git a/src/js/builtins.d.ts b/src/js/builtins.d.ts index 60cfb900918a..b8aeea901fe1 100644 --- a/src/js/builtins.d.ts +++ b/src/js/builtins.d.ts @@ -120,6 +120,14 @@ declare function $webStreamClosedPromise(stream: ReadableStream | WritableStream * no-op once the stream is no longer readable/writable. Throws for any other value. */ declare function $webStreamControllerError(stream: ReadableStream | WritableStream, error: unknown): void; + +/** + * Read a WHATWG ReadableStream/WritableStream's [[state]] internal slot as an integer. + * ReadableStream: 0 = readable, 1 = closed, 2 = errored. + * WritableStream: 0 = writable, 1 = erroring, 2 = errored, 3 = closed. + * Throws for any other value. + */ +declare function $webStreamState(stream: ReadableStream | WritableStream): number; declare function $getInternalField( base: InternalFieldObject, number: N, diff --git a/src/js/builtins/BunBuiltinNames.h b/src/js/builtins/BunBuiltinNames.h index e7cce800edda..69a4fc2f88bd 100644 --- a/src/js/builtins/BunBuiltinNames.h +++ b/src/js/builtins/BunBuiltinNames.h @@ -201,6 +201,7 @@ using namespace JSC; macro(warning) \ macro(webStreamClosedPromise) \ macro(webStreamControllerError) \ + macro(webStreamState) \ macro(writable) \ macro(writableType) \ macro(write) \ diff --git a/src/js/internal/streams/utils.ts b/src/js/internal/streams/utils.ts index e4bc3915bce0..ff4fb85986d9 100644 --- a/src/js/internal/streams/utils.ts +++ b/src/js/internal/streams/utils.ts @@ -146,6 +146,9 @@ function isReadableFinished(stream, strict?) { function isReadable(stream) { if (stream && stream[kIsReadable] != null) return stream[kIsReadable]; + // Bun's WHATWG streams are native and don't carry node's kIsReadable brand; read [[state]] + // directly to match node's ReadableStream.prototype[kIsReadable] getter. + if (isReadableStream(stream)) return $webStreamState(stream) === 0; // "readable" if (typeof stream?.readable !== "boolean") return null; if (isDestroyed(stream)) return false; return isReadableNodeStream(stream) && stream.readable && !isReadableFinished(stream); @@ -153,6 +156,7 @@ function isReadable(stream) { function isWritable(stream) { if (stream && stream[kIsWritable] != null) return stream[kIsWritable]; + if (isWritableStream(stream)) return $webStreamState(stream) === 0; // "writable" if (typeof stream?.writable !== "boolean") return null; if (isDestroyed(stream)) return false; return isWritableNodeStream(stream) && stream.writable && !isWritableEnded(stream); @@ -261,13 +265,18 @@ function willEmitClose(stream) { } function isDisturbed(stream) { - return !!(stream && (stream[kIsDisturbed] ?? (stream.readableDidRead || stream.readableAborted))); + return !!( + stream && + (stream[kIsDisturbed] ?? + (isReadableStream(stream) ? stream.$disturbed : stream.readableDidRead || stream.readableAborted)) + ); } function isErrored(stream) { return !!( stream && (stream[kIsErrored] ?? + (isReadableStream(stream) || isWritableStream(stream) ? $webStreamState(stream) === 2 : undefined) ?? stream.readableErrored ?? stream.writableErrored ?? stream._readableState?.errorEmitted ?? diff --git a/src/jsc/bindings/ZigGlobalObject.cpp b/src/jsc/bindings/ZigGlobalObject.cpp index 42dd3fa08aac..f9804a38bc80 100644 --- a/src/jsc/bindings/ZigGlobalObject.cpp +++ b/src/jsc/bindings/ZigGlobalObject.cpp @@ -1842,6 +1842,22 @@ JSC_DEFINE_HOST_FUNCTION(jsWebStreamClosedPromise, (JSGlobalObject * globalObjec return JSValue::encode(throwTypeError(globalObject, scope, "Expected a ReadableStream or WritableStream"_s)); } +// node:stream's isReadable/isWritable/isErrored/isDisturbed expect WHATWG streams to carry the +// Symbol.for("nodejs.stream.*") brand getters that node's own web stream impl defines. Bun's +// native streams don't, so those helpers duck-type the argument and read the underlying +// [[state]] here instead. Callers gate on isReadableStream()/isWritableStream() first. +JSC_DEFINE_HOST_FUNCTION(jsWebStreamState, (JSGlobalObject * globalObject, CallFrame* callFrame)) +{ + JSValue streamValue = callFrame->argument(0); + if (auto* readable = dynamicDowncast(streamValue)) + return JSValue::encode(jsNumber(static_cast(readable->m_state))); + if (auto* writable = dynamicDowncast(streamValue)) + return JSValue::encode(jsNumber(static_cast(writable->m_state))); + + auto scope = DECLARE_THROW_SCOPE(getVM(globalObject)); + return JSValue::encode(throwTypeError(globalObject, scope, "Expected a ReadableStream or WritableStream"_s)); +} + // node:stream's addAbortSignal() errors a WHATWG stream when the signal fires. Its isWebStream() // gate also admits TransformStream, which has no controller to error — node throws there too (it // never sets kControllerErrorFunction on one), so the throw below is reachable, not dead code. @@ -3058,6 +3074,7 @@ void GlobalObject::addBuiltinGlobals(JSC::VM& vm) GlobalPropertyInfo(builtinNames.pokePromiseAsHandledPrivateName(), JSFunction::create(vm, this, 1, String(), jsBunPokePromiseAsHandled, ImplementationVisibility::Public), PropertyAttribute::DontDelete | PropertyAttribute::ReadOnly), GlobalPropertyInfo(builtinNames.webStreamClosedPromisePrivateName(), JSFunction::create(vm, this, 1, String(), jsWebStreamClosedPromise, ImplementationVisibility::Public), PropertyAttribute::DontDelete | PropertyAttribute::ReadOnly), GlobalPropertyInfo(builtinNames.webStreamControllerErrorPrivateName(), JSFunction::create(vm, this, 2, String(), jsWebStreamControllerError, ImplementationVisibility::Public), PropertyAttribute::DontDelete | PropertyAttribute::ReadOnly), + GlobalPropertyInfo(builtinNames.webStreamStatePrivateName(), JSFunction::create(vm, this, 1, String(), jsWebStreamState, ImplementationVisibility::Public), PropertyAttribute::DontDelete | PropertyAttribute::ReadOnly), GlobalPropertyInfo(builtinNames.fulfillModuleSyncPrivateName(), JSFunction::create(vm, this, 1, String(), functionFulfillModuleSync, ImplementationVisibility::Public), PropertyAttribute::DontDelete | PropertyAttribute::ReadOnly), GlobalPropertyInfo(builtinNames.esmNamespaceForCjsPrivateName(), JSFunction::create(vm, this, 1, String(), functionEsmNamespaceForCjs, ImplementationVisibility::Public), PropertyAttribute::DontDelete | PropertyAttribute::ReadOnly), GlobalPropertyInfo(builtinNames.esmRegistryDeletePrivateName(), JSFunction::create(vm, this, 1, String(), functionEsmRegistryDelete, ImplementationVisibility::Public), PropertyAttribute::DontDelete | PropertyAttribute::ReadOnly), diff --git a/test/js/node/stream/node-stream.test.js b/test/js/node/stream/node-stream.test.js index 0489f2dc1df4..22ad69e4e75f 100644 --- a/test/js/node/stream/node-stream.test.js +++ b/test/js/node/stream/node-stream.test.js @@ -2,7 +2,7 @@ import { describe, expect, it, jest } from "bun:test"; import { bunEnv, bunExe, isGlibcVersionAtLeast, isMacOS, tmpdirSync } from "harness"; import { createReadStream, mkdirSync, writeFileSync } from "node:fs"; import { tmpdir } from "node:os"; -import { Duplex, finished, PassThrough, Readable, Stream, Transform, Writable } from "node:stream"; +import { Duplex, finished, isDisturbed, isErrored, isReadable, isWritable, PassThrough, Readable, Stream, Transform, Writable } from "node:stream"; import { finished as finishedP } from "node:stream/promises"; import { join } from "path"; @@ -1601,3 +1601,88 @@ describe("stream operators argument validation (nodejs/node#59529)", () => { } }); }); + +// Node documents WHATWG web streams as valid operands for these helpers and brands its own +// web stream classes with Symbol.for("nodejs.stream.*") getters. Bun's streams are native and +// don't carry those symbols, so the helpers read the internal [[state]]/[[disturbed]] directly. +describe("isReadable/isWritable/isErrored/isDisturbed on WHATWG web streams", () => { + const probe = stream => ({ + isReadable: isReadable(stream), + isWritable: isWritable(stream), + isErrored: isErrored(stream), + isDisturbed: isDisturbed(stream), + }); + + it("ReadableStream: readable", () => { + const rs = new ReadableStream({ start(c) { c.enqueue("x"); } }); + expect(probe(rs)).toEqual({ isReadable: true, isWritable: null, isErrored: false, isDisturbed: false }); + }); + + it("ReadableStream: closed", () => { + const rs = new ReadableStream({ start(c) { c.close(); } }); + expect(probe(rs)).toEqual({ isReadable: false, isWritable: null, isErrored: false, isDisturbed: false }); + }); + + it("ReadableStream: errored + disturbed after read()", async () => { + const rs = new ReadableStream({ start(c) { c.error(new Error("boom")); } }); + await rs.getReader().read().catch(() => {}); + expect(probe(rs)).toEqual({ isReadable: false, isWritable: null, isErrored: true, isDisturbed: true }); + }); + + it("ReadableStream: disturbed after cancel()", async () => { + const rs = new ReadableStream({ start(c) { c.enqueue("x"); } }); + await rs.cancel(); + expect(probe(rs)).toEqual({ isReadable: false, isWritable: null, isErrored: false, isDisturbed: true }); + }); + + it("ReadableStream: disturbed after successful read(), still readable", async () => { + const rs = new ReadableStream({ start(c) { c.enqueue("x"); } }); + const reader = rs.getReader(); + await reader.read(); + reader.releaseLock(); + expect(probe(rs)).toEqual({ isReadable: true, isWritable: null, isErrored: false, isDisturbed: true }); + }); + + it("WritableStream: writable", () => { + const ws = new WritableStream({ write() {} }); + expect(probe(ws)).toEqual({ isReadable: null, isWritable: true, isErrored: false, isDisturbed: false }); + }); + + it("WritableStream: closed", async () => { + const ws = new WritableStream({ write() {} }); + const writer = ws.getWriter(); + await writer.close(); + writer.releaseLock(); + expect(probe(ws)).toEqual({ isReadable: null, isWritable: false, isErrored: false, isDisturbed: false }); + }); + + it("WritableStream: errored", async () => { + const ws = new WritableStream({ start(c) { c.error(new Error("boom")); } }); + await ws.getWriter().closed.catch(() => {}); + expect(probe(ws)).toEqual({ isReadable: null, isWritable: false, isErrored: true, isDisturbed: false }); + }); + + it("TransformStream falls through (no web-stream brand)", () => { + const ts = new TransformStream(); + expect(probe(ts)).toEqual({ isReadable: null, isWritable: null, isErrored: false, isDisturbed: false }); + }); + + it("Symbol.for('nodejs.stream.*') overrides still take precedence", () => { + const rs = new ReadableStream({ start(c) { c.enqueue("x"); } }); + Object.defineProperty(rs, Symbol.for("nodejs.stream.readable"), { value: false }); + expect(isReadable(rs)).toBe(false); + }); + + it("node Readable/Writable operands are unaffected", () => { + const r = new Readable({ read() {} }); + expect(isReadable(r)).toBe(true); + expect(isErrored(r)).toBe(false); + expect(isDisturbed(r)).toBe(false); + r.destroy(); + + const w = new Writable({ write(chunk, enc, cb) { cb(); } }); + expect(isWritable(w)).toBe(true); + expect(isErrored(w)).toBe(false); + w.destroy(); + }); +}); From 96e06a6e8e9615cce74fe2e8b10f3f51ed66f6e7 Mon Sep 17 00:00:00 2001 From: "autofix-ci[bot]" <114827586+autofix-ci[bot]@users.noreply.github.com> Date: Sat, 18 Jul 2026 10:31:16 +0000 Subject: [PATCH 2/6] [autofix.ci] apply automated fixes --- test/js/node/stream/node-stream.test.js | 67 +++++++++++++++++++++---- 1 file changed, 57 insertions(+), 10 deletions(-) diff --git a/test/js/node/stream/node-stream.test.js b/test/js/node/stream/node-stream.test.js index 22ad69e4e75f..a4d233472dd6 100644 --- a/test/js/node/stream/node-stream.test.js +++ b/test/js/node/stream/node-stream.test.js @@ -2,7 +2,19 @@ import { describe, expect, it, jest } from "bun:test"; import { bunEnv, bunExe, isGlibcVersionAtLeast, isMacOS, tmpdirSync } from "harness"; import { createReadStream, mkdirSync, writeFileSync } from "node:fs"; import { tmpdir } from "node:os"; -import { Duplex, finished, isDisturbed, isErrored, isReadable, isWritable, PassThrough, Readable, Stream, Transform, Writable } from "node:stream"; +import { + Duplex, + finished, + isDisturbed, + isErrored, + isReadable, + isWritable, + PassThrough, + Readable, + Stream, + Transform, + Writable, +} from "node:stream"; import { finished as finishedP } from "node:stream/promises"; import { join } from "path"; @@ -1614,29 +1626,52 @@ describe("isReadable/isWritable/isErrored/isDisturbed on WHATWG web streams", () }); it("ReadableStream: readable", () => { - const rs = new ReadableStream({ start(c) { c.enqueue("x"); } }); + const rs = new ReadableStream({ + start(c) { + c.enqueue("x"); + }, + }); expect(probe(rs)).toEqual({ isReadable: true, isWritable: null, isErrored: false, isDisturbed: false }); }); it("ReadableStream: closed", () => { - const rs = new ReadableStream({ start(c) { c.close(); } }); + const rs = new ReadableStream({ + start(c) { + c.close(); + }, + }); expect(probe(rs)).toEqual({ isReadable: false, isWritable: null, isErrored: false, isDisturbed: false }); }); it("ReadableStream: errored + disturbed after read()", async () => { - const rs = new ReadableStream({ start(c) { c.error(new Error("boom")); } }); - await rs.getReader().read().catch(() => {}); + const rs = new ReadableStream({ + start(c) { + c.error(new Error("boom")); + }, + }); + await rs + .getReader() + .read() + .catch(() => {}); expect(probe(rs)).toEqual({ isReadable: false, isWritable: null, isErrored: true, isDisturbed: true }); }); it("ReadableStream: disturbed after cancel()", async () => { - const rs = new ReadableStream({ start(c) { c.enqueue("x"); } }); + const rs = new ReadableStream({ + start(c) { + c.enqueue("x"); + }, + }); await rs.cancel(); expect(probe(rs)).toEqual({ isReadable: false, isWritable: null, isErrored: false, isDisturbed: true }); }); it("ReadableStream: disturbed after successful read(), still readable", async () => { - const rs = new ReadableStream({ start(c) { c.enqueue("x"); } }); + const rs = new ReadableStream({ + start(c) { + c.enqueue("x"); + }, + }); const reader = rs.getReader(); await reader.read(); reader.releaseLock(); @@ -1657,7 +1692,11 @@ describe("isReadable/isWritable/isErrored/isDisturbed on WHATWG web streams", () }); it("WritableStream: errored", async () => { - const ws = new WritableStream({ start(c) { c.error(new Error("boom")); } }); + const ws = new WritableStream({ + start(c) { + c.error(new Error("boom")); + }, + }); await ws.getWriter().closed.catch(() => {}); expect(probe(ws)).toEqual({ isReadable: null, isWritable: false, isErrored: true, isDisturbed: false }); }); @@ -1668,7 +1707,11 @@ describe("isReadable/isWritable/isErrored/isDisturbed on WHATWG web streams", () }); it("Symbol.for('nodejs.stream.*') overrides still take precedence", () => { - const rs = new ReadableStream({ start(c) { c.enqueue("x"); } }); + const rs = new ReadableStream({ + start(c) { + c.enqueue("x"); + }, + }); Object.defineProperty(rs, Symbol.for("nodejs.stream.readable"), { value: false }); expect(isReadable(rs)).toBe(false); }); @@ -1680,7 +1723,11 @@ describe("isReadable/isWritable/isErrored/isDisturbed on WHATWG web streams", () expect(isDisturbed(r)).toBe(false); r.destroy(); - const w = new Writable({ write(chunk, enc, cb) { cb(); } }); + const w = new Writable({ + write(chunk, enc, cb) { + cb(); + }, + }); expect(isWritable(w)).toBe(true); expect(isErrored(w)).toBe(false); w.destroy(); From f65ff0f343c18fedbaa5bba48239d4323ec68988 Mon Sep 17 00:00:00 2001 From: robobun <117481402+robobun@users.noreply.github.com> Date: Sat, 18 Jul 2026 10:37:40 +0000 Subject: [PATCH 3/6] test: cover Symbol.for('nodejs.stream.*') override precedence for all four predicates --- test/js/node/stream/node-stream.test.js | 10 ++++++++++ 1 file changed, 10 insertions(+) diff --git a/test/js/node/stream/node-stream.test.js b/test/js/node/stream/node-stream.test.js index a4d233472dd6..b69f9c17e05a 100644 --- a/test/js/node/stream/node-stream.test.js +++ b/test/js/node/stream/node-stream.test.js @@ -1713,7 +1713,17 @@ describe("isReadable/isWritable/isErrored/isDisturbed on WHATWG web streams", () }, }); Object.defineProperty(rs, Symbol.for("nodejs.stream.readable"), { value: false }); + Object.defineProperty(rs, Symbol.for("nodejs.stream.errored"), { value: true }); + Object.defineProperty(rs, Symbol.for("nodejs.stream.disturbed"), { value: true }); expect(isReadable(rs)).toBe(false); + expect(isErrored(rs)).toBe(true); + expect(isDisturbed(rs)).toBe(true); + + const ws = new WritableStream({ write() {} }); + Object.defineProperty(ws, Symbol.for("nodejs.stream.writable"), { value: false }); + Object.defineProperty(ws, Symbol.for("nodejs.stream.errored"), { value: true }); + expect(isWritable(ws)).toBe(false); + expect(isErrored(ws)).toBe(true); }); it("node Readable/Writable operands are unaffected", () => { From 51c40da8cecb46e742053789c6b677af1334a7d1 Mon Sep 17 00:00:00 2001 From: robobun <117481402+robobun@users.noreply.github.com> Date: Sat, 18 Jul 2026 10:46:12 +0000 Subject: [PATCH 4/6] style: trim jsWebStreamState comment to 3 lines --- src/jsc/bindings/ZigGlobalObject.cpp | 3 +-- 1 file changed, 1 insertion(+), 2 deletions(-) diff --git a/src/jsc/bindings/ZigGlobalObject.cpp b/src/jsc/bindings/ZigGlobalObject.cpp index f9804a38bc80..47d30805aa3d 100644 --- a/src/jsc/bindings/ZigGlobalObject.cpp +++ b/src/jsc/bindings/ZigGlobalObject.cpp @@ -1844,8 +1844,7 @@ JSC_DEFINE_HOST_FUNCTION(jsWebStreamClosedPromise, (JSGlobalObject * globalObjec // node:stream's isReadable/isWritable/isErrored/isDisturbed expect WHATWG streams to carry the // Symbol.for("nodejs.stream.*") brand getters that node's own web stream impl defines. Bun's -// native streams don't, so those helpers duck-type the argument and read the underlying -// [[state]] here instead. Callers gate on isReadableStream()/isWritableStream() first. +// native streams don't, so those helpers duck-type and read the underlying [[state]] here instead. JSC_DEFINE_HOST_FUNCTION(jsWebStreamState, (JSGlobalObject * globalObject, CallFrame* callFrame)) { JSValue streamValue = callFrame->argument(0); From fb866999abbfdb383014d77ec6eb6a2f1dfda6bc Mon Sep 17 00:00:00 2001 From: robobun <117481402+robobun@users.noreply.github.com> Date: Sat, 18 Jul 2026 11:04:33 +0000 Subject: [PATCH 5/6] style: add JSC_DECLARE_HOST_FUNCTION forward decl for jsWebStreamState --- src/jsc/bindings/ZigGlobalObject.cpp | 1 + 1 file changed, 1 insertion(+) diff --git a/src/jsc/bindings/ZigGlobalObject.cpp b/src/jsc/bindings/ZigGlobalObject.cpp index 47d30805aa3d..36ee480e93d3 100644 --- a/src/jsc/bindings/ZigGlobalObject.cpp +++ b/src/jsc/bindings/ZigGlobalObject.cpp @@ -1711,6 +1711,7 @@ JSC_DECLARE_HOST_FUNCTION(jsBunPeekPromiseSettledValue); JSC_DECLARE_HOST_FUNCTION(jsBunPokePromiseAsHandled); JSC_DECLARE_HOST_FUNCTION(jsWebStreamClosedPromise); JSC_DECLARE_HOST_FUNCTION(jsWebStreamControllerError); +JSC_DECLARE_HOST_FUNCTION(jsWebStreamState); JSC_DEFINE_HOST_FUNCTION(makeGetterTypeErrorForBuiltins, (JSGlobalObject * globalObject, CallFrame* callFrame)) { From a42ed8a5ed4c301052ec6271912100d6d7a2cf02 Mon Sep 17 00:00:00 2001 From: robobun <117481402+robobun@users.noreply.github.com> Date: Wed, 22 Jul 2026 10:06:46 +0000 Subject: [PATCH 6/6] ci: retrigger