node:stream: make isReadable/isWritable/isErrored/isDisturbed work with WHATWG web streams

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<Fields extends any[], N extends keyof Fields>(
   base: InternalFieldObject<Fields>,
   number: N,

src/js/builtins/BunBuiltinNames.h

@@ -202,6 +202,7 @@ using namespace JSC;
     macro(warning) \
     macro(webStreamClosedPromise) \
     macro(webStreamControllerError) \
+    macro(webStreamState) \
     macro(writable) \
     macro(writableType) \
     macro(write) \

src/js/internal/streams/utils.ts

@@ -146,13 +146,17 @@ 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);
 }
 
 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 ??

src/jsc/bindings/ZigGlobalObject.cpp

@@ -1715,6 +1715,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))
 {
@@ -1846,6 +1847,21 @@ 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 and read the underlying [[state]] here instead.
+JSC_DEFINE_HOST_FUNCTION(jsWebStreamState, (JSGlobalObject * globalObject, CallFrame* callFrame))
+{
+    JSValue streamValue = callFrame->argument(0);
+    if (auto* readable = dynamicDowncast<WebCore::JSReadableStream>(streamValue))
+        return JSValue::encode(jsNumber(static_cast<int32_t>(readable->m_state)));
+    if (auto* writable = dynamicDowncast<WebCore::JSWritableStream>(streamValue))
+        return JSValue::encode(jsNumber(static_cast<int32_t>(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.
@@ -3062,6 +3078,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),

test/js/node/stream/node-stream.test.js

@@ -3,7 +3,20 @@ 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, duplexPair, finished, PassThrough, Readable, Stream, Transform, Writable } from "node:stream";
+import {
+  Duplex,
+  duplexPair,
+  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";
 
@@ -1666,6 +1679,136 @@ 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 });
+    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", () => {
+    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();
+  });
+});
+
 describe("duplexPair teardown (test-duplex-error.js)", () => {
   const once = (emitter, event) => new Promise(resolve => emitter.once(event, resolve));