stream: apply more stream/iter conformance fixes

Author: jasnell

lib/internal/streams/iter/broadcast.js

@@ -151,31 +151,40 @@ class BroadcastImpl {
     const { transforms, options } = parsePushArgs(args);
     const rawConsumer = this.#createRawConsumer();
 
+    // When transforms are present, delegate to pull() which creates its
+    // own internal AbortController that follows the external signal.
+    // When no transforms, return rawConsumer directly (controller elided
+    // per PULL-02 optimization -- no transforms means no signal recipient).
     if (transforms.length > 0) {
+      const pullArgs = [...transforms];
       if (options?.signal) {
-        return pullWithTransforms(
-          rawConsumer, ...transforms, { __proto__: null, signal: options.signal });
+        ArrayPrototypePush(pullArgs,
+                           { __proto__: null, signal: options.signal });
       }
-      return pullWithTransforms(rawConsumer, ...transforms);
+      return pullWithTransforms(rawConsumer, ...pullArgs);
     }
     return rawConsumer;
   }
 
   #createRawConsumer() {
     const state = {
       __proto__: null,
-      cursor: this.#bufferStart + this.#buffer.length,
+      // Start at the oldest buffered entry so late-joining consumers
+      // can read data already in the buffer.
+      cursor: this.#bufferStart,
       resolve: null,
       reject: null,
       detached: false,
     };
 
     this.#consumers.add(state);
-    // New consumer starts at the latest position; min cursor unchanged
-    // unless this is the first consumer.
+    // New consumer starts at buffer start; recalculate min cursor
+    // since this consumer may now be the slowest.
     if (this.#consumers.size === 1) {
       this.#cachedMinCursor = state.cursor;
       this.#minCursorDirty = false;
+    } else {
+      this.#minCursorDirty = true;
     }
     const self = this;
 
@@ -252,7 +261,7 @@ class BroadcastImpl {
     this.#cancelled = true;
     this.#ended = true; // Prevents [kAbort]() from redundantly iterating consumers
 
-    if (reason) {
+    if (reason !== undefined) {
       this.#error = reason;
     }
 
@@ -261,7 +270,7 @@ class BroadcastImpl {
 
     for (const consumer of this.#consumers) {
       if (consumer.resolve) {
-        if (reason) {
+        if (reason !== undefined) {
           consumer.reject?.(reason);
         } else {
           consumer.resolve({ __proto__: null, done: true, value: undefined });

lib/internal/streams/iter/duplex.js

@@ -7,12 +7,14 @@
 
 const {
   SymbolAsyncDispose,
+  SymbolAsyncIterator,
 } = primordials;
 
 const {
   push,
 } = require('internal/streams/iter/push');
 const {
+  validateAbortSignal,
   validateObject,
 } = require('internal/validators');
 
@@ -31,34 +33,54 @@ function duplex(options = { __proto__: null }) {
   if (b !== undefined) {
     validateObject(b, 'options.b');
   }
+  if (signal !== undefined) {
+    validateAbortSignal(signal, 'options.signal');
+  }
 
-  // Channel A writes to B's readable (A->B direction)
+  // Channel A writes to B's readable (A->B direction).
+  // Signal is NOT passed to push() -- we handle abort via close() below.
   const { writer: aWriter, readable: bReadable } = push({
     highWaterMark: a?.highWaterMark ?? highWaterMark,
     backpressure: a?.backpressure ?? backpressure,
-    signal,
   });
 
   // Channel B writes to A's readable (B->A direction)
   const { writer: bWriter, readable: aReadable } = push({
     highWaterMark: b?.highWaterMark ?? highWaterMark,
     backpressure: b?.backpressure ?? backpressure,
-    signal,
   });
 
-  let aWriterRef = aWriter;
-  let bWriterRef = bWriter;
+  let aClosed = false;
+  let bClosed = false;
+  // Track active iterators so close() can call .return() on them
+  let aReadableIterator = null;
+  let bReadableIterator = null;
 
   const channelA = {
     __proto__: null,
     get writer() { return aWriter; },
-    readable: aReadable,
+    // Wrap readable to track the iterator for cleanup on close()
+    get readable() {
+      return {
+        __proto__: null,
+        [SymbolAsyncIterator]() {
+          const iter = aReadable[SymbolAsyncIterator]();
+          aReadableIterator = iter;
+          return iter;
+        },
+      };
+    },
     async close() {
-      if (aWriterRef === null) return;
-      const writer = aWriterRef;
-      aWriterRef = null;
-      if (writer.endSync() < 0) {
-        await writer.end();
+      if (aClosed) return;
+      aClosed = true;
+      // End the writer (signals end-of-stream to B's readable)
+      if (aWriter.endSync() < 0) {
+        await aWriter.end();
+      }
+      // Stop iteration of this channel's readable
+      if (aReadableIterator?.return) {
+        await aReadableIterator.return();
+        aReadableIterator = null;
       }
     },
     [SymbolAsyncDispose]() {
@@ -69,20 +91,48 @@ function duplex(options = { __proto__: null }) {
   const channelB = {
     __proto__: null,
     get writer() { return bWriter; },
-    readable: bReadable,
+    get readable() {
+      return {
+        __proto__: null,
+        [SymbolAsyncIterator]() {
+          const iter = bReadable[SymbolAsyncIterator]();
+          bReadableIterator = iter;
+          return iter;
+        },
+      };
+    },
     async close() {
-      if (bWriterRef === null) return;
-      const writer = bWriterRef;
-      bWriterRef = null;
-      if (writer.endSync() < 0) {
-        await writer.end();
+      if (bClosed) return;
+      bClosed = true;
+      if (bWriter.endSync() < 0) {
+        await bWriter.end();
+      }
+      if (bReadableIterator?.return) {
+        await bReadableIterator.return();
+        bReadableIterator = null;
       }
     },
     [SymbolAsyncDispose]() {
       return this.close();
     },
   };
 
+  // Signal handler: fail both writers with the abort reason so consumers
+  // see the error. This is an error-path shutdown, not a clean close.
+  if (signal) {
+    const abortBoth = () => {
+      const reason = signal.reason;
+      aWriter.fail(reason);
+      bWriter.fail(reason);
+    };
+    if (signal.aborted) {
+      abortBoth();
+    } else {
+      signal.addEventListener('abort', abortBoth,
+                              { __proto__: null, once: true });
+    }
+  }
+
   return [channelA, channelB];
 }
 

lib/internal/streams/iter/share.js

@@ -28,6 +28,8 @@ const {
 } = require('internal/streams/iter/types');
 
 const {
+  from,
+  fromSync,
   isAsyncIterable,
   isSyncIterable,
 } = require('internal/streams/iter/from');
@@ -198,7 +200,7 @@ class ShareImpl {
     if (this.#cancelled) return;
     this.#cancelled = true;
 
-    if (reason) {
+    if (reason !== undefined) {
       this.#sourceError = reason;
     }
 
@@ -208,7 +210,7 @@ class ShareImpl {
 
     for (const consumer of this.#consumers) {
       if (consumer.resolve) {
-        if (reason) {
+        if (reason !== undefined) {
           consumer.reject?.(reason);
         } else {
           consumer.resolve({ __proto__: null, done: true, value: undefined });
@@ -505,7 +507,7 @@ class SyncShareImpl {
     if (this.#cancelled) return;
     this.#cancelled = true;
 
-    if (reason) {
+    if (reason !== undefined) {
       this.#sourceError = reason;
     }
 
@@ -569,10 +571,8 @@ class SyncShareImpl {
 // =============================================================================
 
 function share(source, options = { __proto__: null }) {
-  if (!isAsyncIterable(source) && !isSyncIterable(source)) {
-    throw new ERR_INVALID_ARG_TYPE(
-      'source', ['AsyncIterable', 'Iterable'], source);
-  }
+  // Normalize source via from() - accepts strings, ArrayBuffers, protocols, etc.
+  const normalized = from(source);
   validateObject(options, 'options');
   const {
     highWaterMark = kMultiConsumerDefaultHWM,
@@ -592,7 +592,7 @@ function share(source, options = { __proto__: null }) {
     signal,
   };
 
-  const shareImpl = new ShareImpl(source, opts);
+  const shareImpl = new ShareImpl(normalized, opts);
 
   if (signal) {
     if (signal.aborted) {
@@ -608,9 +608,8 @@ function share(source, options = { __proto__: null }) {
 }
 
 function shareSync(source, options = { __proto__: null }) {
-  if (!isSyncIterable(source)) {
-    throw new ERR_INVALID_ARG_TYPE('source', 'Iterable', source);
-  }
+  // Normalize source via fromSync() - accepts strings, ArrayBuffers, protocols, etc.
+  const normalized = fromSync(source);
   validateObject(options, 'options');
   const {
     highWaterMark = kMultiConsumerDefaultHWM,
@@ -625,7 +624,7 @@ function shareSync(source, options = { __proto__: null }) {
     backpressure,
   };
 
-  return new SyncShareImpl(source, opts);
+  return new SyncShareImpl(normalized, opts);
 }
 
 function isShareable(value) {

test/parallel/test-stream-iter-broadcast-basic.js

@@ -218,6 +218,17 @@ async function testWriterFailIdempotent() {
   }, { message: 'fail!' });
 }
 
+// cancel() with falsy reason (0, "", false) should still treat as error
+async function testCancelWithFalsyReason() {
+  const { broadcast: bc } = broadcast();
+  const consumer = bc.push();
+  const resultPromise = text(consumer).catch((err) => err);
+  await new Promise((resolve) => setImmediate(resolve));
+  bc.cancel(0);
+  const result = await resultPromise;
+  assert.strictEqual(result, 0);
+}
+
 Promise.all([
   testBasicBroadcast(),
   testMultipleWrites(),
@@ -228,6 +239,22 @@ Promise.all([
   testWriterFail(),
   testCancelWithoutReason(),
   testCancelWithReason(),
+  testCancelWithFalsyReason(),
   testFailDetachesConsumers(),
   testWriterFailIdempotent(),
+  testLateJoinerSeesBufferedData(),
 ]).then(common.mustCall());
+
+// Late-joining consumer should read from oldest buffered entry
+async function testLateJoinerSeesBufferedData() {
+  const { writer, broadcast: bc } = broadcast({ highWaterMark: 16 });
+
+  // Write data before any consumer joins
+  writer.writeSync('before-join');
+  writer.endSync();
+
+  // Consumer joins after data is written
+  const consumer = bc.push();
+  const result = await text(consumer);
+  assert.strictEqual(result, 'before-join');
+}

test/parallel/test-stream-iter-share-async.js

@@ -271,4 +271,12 @@ Promise.all([
   testShareLateJoiningConsumer(),
   testShareConsumerBreak(),
   testShareMultipleConsumersConcurrentPull(),
+  testShareStringSource(),
 ]).then(common.mustCall());
+
+// share() accepts string source directly (normalized via from())
+async function testShareStringSource() {
+  const shared = share('hello-share');
+  const result = await text(shared.pull());
+  assert.strictEqual(result, 'hello-share');
+}

test/parallel/test-stream-iter-share-sync.js

@@ -151,4 +151,12 @@ Promise.all([
   testShareSyncCancelMidIteration(),
   testShareSyncCancelWithReason(),
   testShareSyncSourceError(),
+  testShareSyncStringSource(),
 ]).then(common.mustCall());
+
+// shareSync() accepts string source directly (normalized via fromSync())
+function testShareSyncStringSource() {
+  const shared = shareSync('hello-sync-share');
+  const result = textSync(shared.pull());
+  assert.strictEqual(result, 'hello-sync-share');
+}