stream: preserve falsy cancellation reasons

Use undefined as the no-error sentinel when cancelling broadcast and
share consumers. This ensures that 0, an empty string, false, and null
are propagated instead of being converted to clean completion.

Make sync share surface cancellation reasons before handling detached
consumers, and add regression coverage for async and sync consumers.

Signed-off-by: Kamat, Trivikram <16024985+trivikr@users.noreply.github.com>
Assisted-by: codex:gpt-5.6-sol

Author: trivikr

lib/internal/streams/iter/broadcast.js

@@ -87,7 +87,7 @@ class BroadcastImpl {
   #consumers = new SafeSet();
   #waiters = [];  // Consumers with pending resolve (subset of #consumers)
   #ended = false;
-  #error = null;
+  #error;
   #cancelled = false;
   #options;
   #writer = null;
@@ -176,7 +176,9 @@ class BroadcastImpl {
           __proto__: null,
           next() {
             if (state.detached) {
-              if (self.#error) return PromiseReject(self.#error);
+              if (self.#error !== undefined) {
+                return PromiseReject(self.#error);
+              }
               return kDone;
             }
 
@@ -193,7 +195,7 @@ class BroadcastImpl {
                 { __proto__: null, done: false, value: chunk });
             }
 
-            if (self.#error) {
+            if (self.#error !== undefined) {
               state.detached = true;
               self.#deleteConsumer(state);
               return PromiseReject(self.#error);
@@ -343,7 +345,7 @@ class BroadcastImpl {
   }
 
   [kAbort](reason) {
-    if (this.#ended || this.#error) return;
+    if (this.#ended || this.#error !== undefined) return;
     this.#error = reason;
     this.#ended = true;
 

lib/internal/streams/iter/share.js

@@ -73,7 +73,7 @@ class ShareImpl {
   #consumers = new SafeSet();
   #sourceIterator = null;
   #sourceExhausted = false;
-  #sourceError = null;
+  #sourceError;
   #cancelled = false;
   #pulling = false;
   #pullWaiters = [];
@@ -129,7 +129,7 @@ class ShareImpl {
       __proto__: null,
       [SymbolAsyncIterator]() {
         const getNext = async () => {
-          if (self.#sourceError) {
+          if (self.#sourceError !== undefined) {
             state.detached = true;
             self.#consumers.delete(state);
             throw self.#sourceError;
@@ -141,7 +141,7 @@ class ShareImpl {
           // cursor must re-pull rather than terminating prematurely.
           for (;;) {
             if (state.detached) {
-              if (self.#sourceError) throw self.#sourceError;
+              if (self.#sourceError !== undefined) throw self.#sourceError;
               return { __proto__: null, done: true, value: undefined };
             }
 
@@ -167,7 +167,7 @@ class ShareImpl {
             if (self.#sourceExhausted) {
               state.detached = true;
               self.#deleteConsumer(state);
-              if (self.#sourceError) throw self.#sourceError;
+              if (self.#sourceError !== undefined) throw self.#sourceError;
               return { __proto__: null, done: true, value: undefined };
             }
 
@@ -176,7 +176,7 @@ class ShareImpl {
             if (shouldBuffer === null) {
               state.detached = true;
               self.#deleteConsumer(state);
-              if (self.#sourceError) throw self.#sourceError;
+              if (self.#sourceError !== undefined) throw self.#sourceError;
               return { __proto__: null, done: true, value: undefined };
             }
 
@@ -260,7 +260,9 @@ class ShareImpl {
 
   async #waitForBufferSpace() {
     while (this.#bufferedBytes >= this.#options.budget) {
-      if (this.#cancelled || this.#sourceError || this.#sourceExhausted) {
+      if (this.#cancelled ||
+          this.#sourceError !== undefined ||
+          this.#sourceExhausted) {
         return this.#cancelled ? null : true;
       }
 
@@ -418,7 +420,7 @@ class SyncShareImpl {
   #consumers = new SafeSet();
   #sourceIterator = null;
   #sourceExhausted = false;
-  #sourceError = null;
+  #sourceError;
   #cancelled = false;
   #cachedMinCursor = 0;
   #cachedMinCursorConsumers = 0;
@@ -467,14 +469,14 @@ class SyncShareImpl {
         return {
           __proto__: null,
           next() {
-            if (state.detached) {
-              return { __proto__: null, done: true, value: undefined };
-            }
-            if (self.#sourceError) {
+            if (self.#sourceError !== undefined) {
               state.detached = true;
               self.#deleteConsumer(state);
               throw self.#sourceError;
             }
+            if (state.detached) {
+              return { __proto__: null, done: true, value: undefined };
+            }
             if (self.#cancelled) {
               state.detached = true;
               self.#deleteConsumer(state);
@@ -535,7 +537,7 @@ class SyncShareImpl {
 
             self.#pullFromSource();
 
-            if (self.#sourceError) {
+            if (self.#sourceError !== undefined) {
               state.detached = true;
               self.#deleteConsumer(state);
               throw self.#sourceError;

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

@@ -260,15 +260,15 @@ 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);
+  for (const reason of [0, '', false, null]) {
+    const { broadcast: bc } = broadcast();
+    const iterator = bc.push()[Symbol.asyncIterator]();
+
+    bc.cancel(reason);
+
+    await assert.rejects(iterator.next(), (error) => error === reason);
+  }
 }
 
 // Late-joining consumer should read from oldest buffered entry

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

@@ -132,6 +132,17 @@ async function testShareCancelWithReason() {
   );
 }
 
+async function testShareCancelWithFalsyReason() {
+  for (const reason of [0, '', false, null]) {
+    const shared = share(from('data'));
+    const iterator = shared.pull()[Symbol.asyncIterator]();
+
+    shared.cancel(reason);
+
+    await assert.rejects(iterator.next(), (error) => error === reason);
+  }
+}
+
 async function testShareAbortSignal() {
   const ac = new AbortController();
   const reason = new Error('share aborted');
@@ -357,6 +368,7 @@ Promise.all([
   testShareCancel(),
   testShareCancelMidIteration(),
   testShareCancelWithReason(),
+  testShareCancelWithFalsyReason(),
   testShareAbortSignal(),
   testShareAbortSignalWhileSourcePullPending(),
   testSharePullAbortSignalRejectsPendingNext(),

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

@@ -87,37 +87,32 @@ function testShareSyncCancelMidIteration() {
 }
 
 function testShareSyncCancelWithReason() {
-  // When cancel(reason) is called, a consumer that hasn't started
-  // iterating is already detached, so it sees done:true (not the error).
-  // But a consumer that is mid-iteration when another consumer cancels
-  // with a reason will see the error on the next pull after cancel.
   const enc = new TextEncoder();
   function* gen() {
     yield [enc.encode('a')];
     yield [enc.encode('b')];
-    yield [enc.encode('c')];
   }
   const shared = shareSync(gen(), { budget: 16384 });
-  const c1 = shared.pull();
-  const c2 = shared.pull();
+  const iterator1 = shared.pull()[Symbol.iterator]();
+  const iterator2 = shared.pull()[Symbol.iterator]();
+  const reason = new Error('sync cancel reason');
+
+  iterator1.next();
+  shared.cancel(reason);
 
-  // c1 reads one item, then c2 cancels with a reason
-  const iter1 = c1[Symbol.iterator]();
-  const first = iter1.next();
-  assert.strictEqual(first.done, false);
+  assert.throws(() => iterator1.next(), (error) => error === reason);
+  assert.throws(() => iterator2.next(), (error) => error === reason);
+}
 
-  shared.cancel(new Error('sync cancel reason'));
+function testShareSyncCancelWithFalsyReason() {
+  for (const reason of [0, '', false, null]) {
+    const shared = shareSync(fromSync('data'));
+    const iterator = shared.pull()[Symbol.iterator]();
 
-  // c1 was already iterating, it's now detached → done
-  const next = iter1.next();
-  assert.strictEqual(next.done, true);
+    shared.cancel(reason);
 
-  // c2 never started, also detached → done (not error)
-  const batches = [];
-  for (const batch of c2) {
-    batches.push(batch);
+    assert.throws(() => iterator.next(), (error) => error === reason);
   }
-  assert.strictEqual(batches.length, 0);
 }
 
 // =============================================================================
@@ -157,6 +152,7 @@ Promise.all([
   testShareSyncCancel(),
   testShareSyncCancelMidIteration(),
   testShareSyncCancelWithReason(),
+  testShareSyncCancelWithFalsyReason(),
   testShareSyncSourceError(),
   testShareSyncStringSource(),
 ]).then(common.mustCall());