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
PR-URL: #64705Fixes: #64704
Reviewed-By: James M Snell <jasnell@gmail.com>

Author: trivikr

lib/internal/streams/iter/broadcast.js

@@ -88,7 +88,7 @@ class BroadcastImpl {
  #consumers =newSafeSet();
  #waiters =[];// Consumers with pending resolve (subset of #consumers)
  #ended =false;
- #error=null;
+ #error;
  #cancelled =false;
  #options;
  #writer =null;
@@ -177,7 +177,9 @@ class BroadcastImpl {
 __proto__: null,
 next(){
 if(state.detached){
-if(self.#error)returnPromiseReject(self.#error);
+if(self.#error !==undefined){
+returnPromiseReject(self.#error);
+}
 returnkDone;
 }
 
@@ -194,7 +196,7 @@ class BroadcastImpl {
 {__proto__: null,done: false,value: chunk});
 }
 
-if(self.#error){
+if(self.#error!==undefined){
 state.detached=true;
 self.#deleteConsumer(state);
 returnPromiseReject(self.#error);
@@ -344,7 +346,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 =newSafeSet();
  #sourceIterator =null;
  #sourceExhausted =false;
- #sourceError=null;
+ #sourceError;
  #cancelled =false;
  #pulling =false;
  #pullWaiters =[];
@@ -129,7 +129,7 @@ class ShareImpl {
 __proto__: null,
 [SymbolAsyncIterator](){
 constgetNext=async()=>{
-if(self.#sourceError){
+if(self.#sourceError!==undefined){
 state.detached=true;
 self.#consumers.delete(state);
 throwself.#sourceError;
@@ -141,7 +141,7 @@ class ShareImpl {
 // cursor must re-pull rather than terminating prematurely.
 for(;;){
 if(state.detached){
-if(self.#sourceError)throwself.#sourceError;
+if(self.#sourceError!==undefined)throwself.#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)throwself.#sourceError;
+if(self.#sourceError!==undefined)throwself.#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)throwself.#sourceError;
+if(self.#sourceError!==undefined)throwself.#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){
 returnthis.#cancelled ? null : true;
 }
 
@@ -418,7 +420,7 @@ class SyncShareImpl {
  #consumers =newSafeSet();
  #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);
 throwself.#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);
 throwself.#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
 asyncfunctiontestCancelWithFalsyReason(){
-const{broadcast: bc}=broadcast();
-constconsumer=bc.push();
-constresultPromise=text(consumer).catch((err)=>err);
-awaitnewPromise((resolve)=>setImmediate(resolve));
-bc.cancel(0);
-constresult=awaitresultPromise;
-assert.strictEqual(result,0);
+for(constreasonof[0,'',false,null]){
+const{broadcast: bc}=broadcast();
+constiterator=bc.push()[Symbol.asyncIterator]();
+
+bc.cancel(reason);
+
+awaitassert.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() {
 );
 }
 
+asyncfunctiontestShareCancelWithFalsyReason(){
+for(constreasonof[0,'',false,null]){
+constshared=share(from('data'));
+constiterator=shared.pull()[Symbol.asyncIterator]();
+
+shared.cancel(reason);
+
+awaitassert.rejects(iterator.next(),(error)=>error===reason);
+}
+}
+
 asyncfunctiontestShareAbortSignal(){
 constac=newAbortController();
 constreason=newError('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() {
 }
 
 functiontestShareSyncCancelWithReason(){
-// 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.
 constenc=newTextEncoder();
 function*gen(){
 yield[enc.encode('a')];
 yield[enc.encode('b')];
-yield[enc.encode('c')];
 }
 constshared=shareSync(gen(),{budget: 16384});
-constc1=shared.pull();
-constc2=shared.pull();
+constiterator1=shared.pull()[Symbol.iterator]();
+constiterator2=shared.pull()[Symbol.iterator]();
+constreason=newError('sync cancel reason');
+
+iterator1.next();
+shared.cancel(reason);
 
-// c1 reads one item, then c2 cancels with a reason
-constiter1=c1[Symbol.iterator]();
-constfirst=iter1.next();
-assert.strictEqual(first.done,false);
+assert.throws(()=>iterator1.next(),(error)=>error===reason);
+assert.throws(()=>iterator2.next(),(error)=>error===reason);
+}
 
-shared.cancel(newError('sync cancel reason'));
+functiontestShareSyncCancelWithFalsyReason(){
+for(constreasonof[0,'',false,null]){
+constshared=shareSync(fromSync('data'));
+constiterator=shared.pull()[Symbol.iterator]();
 
-// c1 was already iterating, it's now detached → done
-constnext=iter1.next();
-assert.strictEqual(next.done,true);
+shared.cancel(reason);
 
-// c2 never started, also detached → done (not error)
-constbatches=[];
-for(constbatchofc2){
-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());

lib/internal/streams/iter/broadcast.js

@@ -88,7 +88,7 @@ class BroadcastImpl {
  #consumers =newSafeSet();
  #waiters =[];// Consumers with pending resolve (subset of #consumers)
  #ended =false;
- #error=null;
+ #error;
  #cancelled =false;
  #options;
  #writer =null;
@@ -177,7 +177,9 @@ class BroadcastImpl {
 __proto__: null,
 next(){
 if(state.detached){
-if(self.#error)returnPromiseReject(self.#error);
+if(self.#error !==undefined){
+returnPromiseReject(self.#error);
+}
 returnkDone;
 }
 
@@ -194,7 +196,7 @@ class BroadcastImpl {
 {__proto__: null,done: false,value: chunk});
 }
 
-if(self.#error){
+if(self.#error!==undefined){
 state.detached=true;
 self.#deleteConsumer(state);
 returnPromiseReject(self.#error);
@@ -344,7 +346,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 =newSafeSet();
  #sourceIterator =null;
  #sourceExhausted =false;
- #sourceError=null;
+ #sourceError;
  #cancelled =false;
  #pulling =false;
  #pullWaiters =[];
@@ -129,7 +129,7 @@ class ShareImpl {
 __proto__: null,
 [SymbolAsyncIterator](){
 constgetNext=async()=>{
-if(self.#sourceError){
+if(self.#sourceError!==undefined){
 state.detached=true;
 self.#consumers.delete(state);
 throwself.#sourceError;
@@ -141,7 +141,7 @@ class ShareImpl {
 // cursor must re-pull rather than terminating prematurely.
 for(;;){
 if(state.detached){
-if(self.#sourceError)throwself.#sourceError;
+if(self.#sourceError!==undefined)throwself.#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)throwself.#sourceError;
+if(self.#sourceError!==undefined)throwself.#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)throwself.#sourceError;
+if(self.#sourceError!==undefined)throwself.#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){
 returnthis.#cancelled ? null : true;
 }
 
@@ -418,7 +420,7 @@ class SyncShareImpl {
  #consumers =newSafeSet();
  #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);
 throwself.#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);
 throwself.#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
 asyncfunctiontestCancelWithFalsyReason(){
-const{broadcast: bc}=broadcast();
-constconsumer=bc.push();
-constresultPromise=text(consumer).catch((err)=>err);
-awaitnewPromise((resolve)=>setImmediate(resolve));
-bc.cancel(0);
-constresult=awaitresultPromise;
-assert.strictEqual(result,0);
+for(constreasonof[0,'',false,null]){
+const{broadcast: bc}=broadcast();
+constiterator=bc.push()[Symbol.asyncIterator]();
+
+bc.cancel(reason);
+
+awaitassert.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() {
 );
 }
 
+asyncfunctiontestShareCancelWithFalsyReason(){
+for(constreasonof[0,'',false,null]){
+constshared=share(from('data'));
+constiterator=shared.pull()[Symbol.asyncIterator]();
+
+shared.cancel(reason);
+
+awaitassert.rejects(iterator.next(),(error)=>error===reason);
+}
+}
+
 asyncfunctiontestShareAbortSignal(){
 constac=newAbortController();
 constreason=newError('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() {
 }
 
 functiontestShareSyncCancelWithReason(){
-// 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.
 constenc=newTextEncoder();
 function*gen(){
 yield[enc.encode('a')];
 yield[enc.encode('b')];
-yield[enc.encode('c')];
 }
 constshared=shareSync(gen(),{budget: 16384});
-constc1=shared.pull();
-constc2=shared.pull();
+constiterator1=shared.pull()[Symbol.iterator]();
+constiterator2=shared.pull()[Symbol.iterator]();
+constreason=newError('sync cancel reason');
+
+iterator1.next();
+shared.cancel(reason);
 
-// c1 reads one item, then c2 cancels with a reason
-constiter1=c1[Symbol.iterator]();
-constfirst=iter1.next();
-assert.strictEqual(first.done,false);
+assert.throws(()=>iterator1.next(),(error)=>error===reason);
+assert.throws(()=>iterator2.next(),(error)=>error===reason);
+}
 
-shared.cancel(newError('sync cancel reason'));
+functiontestShareSyncCancelWithFalsyReason(){
+for(constreasonof[0,'',false,null]){
+constshared=shareSync(fromSync('data'));
+constiterator=shared.pull()[Symbol.iterator]();
 
-// c1 was already iterating, it's now detached → done
-constnext=iter1.next();
-assert.strictEqual(next.done,true);
+shared.cancel(reason);
 
-// c2 never started, also detached → done (not error)
-constbatches=[];
-for(constbatchofc2){
-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());

lib/internal/streams/iter/broadcast.js

@@ -88,7 +88,7 @@ class BroadcastImpl {
  #consumers =newSafeSet();
  #waiters =[];// Consumers with pending resolve (subset of #consumers)
  #ended =false;
- #error=null;
+ #error;
  #cancelled =false;
  #options;
  #writer =null;
@@ -177,7 +177,9 @@ class BroadcastImpl {
 __proto__: null,
 next(){
 if(state.detached){
-if(self.#error)returnPromiseReject(self.#error);
+if(self.#error !==undefined){
+returnPromiseReject(self.#error);
+}
 returnkDone;
 }
 
@@ -194,7 +196,7 @@ class BroadcastImpl {
 {__proto__: null,done: false,value: chunk});
 }
 
-if(self.#error){
+if(self.#error!==undefined){
 state.detached=true;
 self.#deleteConsumer(state);
 returnPromiseReject(self.#error);
@@ -344,7 +346,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 =newSafeSet();
  #sourceIterator =null;
  #sourceExhausted =false;
- #sourceError=null;
+ #sourceError;
  #cancelled =false;
  #pulling =false;
  #pullWaiters =[];
@@ -129,7 +129,7 @@ class ShareImpl {
 __proto__: null,
 [SymbolAsyncIterator](){
 constgetNext=async()=>{
-if(self.#sourceError){
+if(self.#sourceError!==undefined){
 state.detached=true;
 self.#consumers.delete(state);
 throwself.#sourceError;
@@ -141,7 +141,7 @@ class ShareImpl {
 // cursor must re-pull rather than terminating prematurely.
 for(;;){
 if(state.detached){
-if(self.#sourceError)throwself.#sourceError;
+if(self.#sourceError!==undefined)throwself.#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)throwself.#sourceError;
+if(self.#sourceError!==undefined)throwself.#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)throwself.#sourceError;
+if(self.#sourceError!==undefined)throwself.#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){
 returnthis.#cancelled ? null : true;
 }
 
@@ -418,7 +420,7 @@ class SyncShareImpl {
  #consumers =newSafeSet();
  #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);
 throwself.#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);
 throwself.#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
 asyncfunctiontestCancelWithFalsyReason(){
-const{broadcast: bc}=broadcast();
-constconsumer=bc.push();
-constresultPromise=text(consumer).catch((err)=>err);
-awaitnewPromise((resolve)=>setImmediate(resolve));
-bc.cancel(0);
-constresult=awaitresultPromise;
-assert.strictEqual(result,0);
+for(constreasonof[0,'',false,null]){
+const{broadcast: bc}=broadcast();
+constiterator=bc.push()[Symbol.asyncIterator]();
+
+bc.cancel(reason);
+
+awaitassert.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() {
 );
 }
 
+asyncfunctiontestShareCancelWithFalsyReason(){
+for(constreasonof[0,'',false,null]){
+constshared=share(from('data'));
+constiterator=shared.pull()[Symbol.asyncIterator]();
+
+shared.cancel(reason);
+
+awaitassert.rejects(iterator.next(),(error)=>error===reason);
+}
+}
+
 asyncfunctiontestShareAbortSignal(){
 constac=newAbortController();
 constreason=newError('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() {
 }
 
 functiontestShareSyncCancelWithReason(){
-// 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.
 constenc=newTextEncoder();
 function*gen(){
 yield[enc.encode('a')];
 yield[enc.encode('b')];
-yield[enc.encode('c')];
 }
 constshared=shareSync(gen(),{budget: 16384});
-constc1=shared.pull();
-constc2=shared.pull();
+constiterator1=shared.pull()[Symbol.iterator]();
+constiterator2=shared.pull()[Symbol.iterator]();
+constreason=newError('sync cancel reason');
+
+iterator1.next();
+shared.cancel(reason);
 
-// c1 reads one item, then c2 cancels with a reason
-constiter1=c1[Symbol.iterator]();
-constfirst=iter1.next();
-assert.strictEqual(first.done,false);
+assert.throws(()=>iterator1.next(),(error)=>error===reason);
+assert.throws(()=>iterator2.next(),(error)=>error===reason);
+}
 
-shared.cancel(newError('sync cancel reason'));
+functiontestShareSyncCancelWithFalsyReason(){
+for(constreasonof[0,'',false,null]){
+constshared=shareSync(fromSync('data'));
+constiterator=shared.pull()[Symbol.iterator]();
 
-// c1 was already iterating, it's now detached → done
-constnext=iter1.next();
-assert.strictEqual(next.done,true);
+shared.cancel(reason);
 
-// c2 never started, also detached → done (not error)
-constbatches=[];
-for(constbatchofc2){
-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());

lib/internal/streams/iter/broadcast.js

@@ -88,7 +88,7 @@ class BroadcastImpl {
  #consumers =newSafeSet();
  #waiters =[];// Consumers with pending resolve (subset of #consumers)
  #ended =false;
- #error=null;
+ #error;
  #cancelled =false;
  #options;
  #writer =null;
@@ -177,7 +177,9 @@ class BroadcastImpl {
 __proto__: null,
 next(){
 if(state.detached){
-if(self.#error)returnPromiseReject(self.#error);
+if(self.#error !==undefined){
+returnPromiseReject(self.#error);
+}
 returnkDone;
 }
 
@@ -194,7 +196,7 @@ class BroadcastImpl {
 {__proto__: null,done: false,value: chunk});
 }
 
-if(self.#error){
+if(self.#error!==undefined){
 state.detached=true;
 self.#deleteConsumer(state);
 returnPromiseReject(self.#error);
@@ -344,7 +346,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 =newSafeSet();
  #sourceIterator =null;
  #sourceExhausted =false;
- #sourceError=null;
+ #sourceError;
  #cancelled =false;
  #pulling =false;
  #pullWaiters =[];
@@ -129,7 +129,7 @@ class ShareImpl {
 __proto__: null,
 [SymbolAsyncIterator](){
 constgetNext=async()=>{
-if(self.#sourceError){
+if(self.#sourceError!==undefined){
 state.detached=true;
 self.#consumers.delete(state);
 throwself.#sourceError;
@@ -141,7 +141,7 @@ class ShareImpl {
 // cursor must re-pull rather than terminating prematurely.
 for(;;){
 if(state.detached){
-if(self.#sourceError)throwself.#sourceError;
+if(self.#sourceError!==undefined)throwself.#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)throwself.#sourceError;
+if(self.#sourceError!==undefined)throwself.#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)throwself.#sourceError;
+if(self.#sourceError!==undefined)throwself.#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){
 returnthis.#cancelled ? null : true;
 }
 
@@ -418,7 +420,7 @@ class SyncShareImpl {
  #consumers =newSafeSet();
  #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);
 throwself.#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);
 throwself.#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
 asyncfunctiontestCancelWithFalsyReason(){
-const{broadcast: bc}=broadcast();
-constconsumer=bc.push();
-constresultPromise=text(consumer).catch((err)=>err);
-awaitnewPromise((resolve)=>setImmediate(resolve));
-bc.cancel(0);
-constresult=awaitresultPromise;
-assert.strictEqual(result,0);
+for(constreasonof[0,'',false,null]){
+const{broadcast: bc}=broadcast();
+constiterator=bc.push()[Symbol.asyncIterator]();
+
+bc.cancel(reason);
+
+awaitassert.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() {
 );
 }
 
+asyncfunctiontestShareCancelWithFalsyReason(){
+for(constreasonof[0,'',false,null]){
+constshared=share(from('data'));
+constiterator=shared.pull()[Symbol.asyncIterator]();
+
+shared.cancel(reason);
+
+awaitassert.rejects(iterator.next(),(error)=>error===reason);
+}
+}
+
 asyncfunctiontestShareAbortSignal(){
 constac=newAbortController();
 constreason=newError('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() {
 }
 
 functiontestShareSyncCancelWithReason(){
-// 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.
 constenc=newTextEncoder();
 function*gen(){
 yield[enc.encode('a')];
 yield[enc.encode('b')];
-yield[enc.encode('c')];
 }
 constshared=shareSync(gen(),{budget: 16384});
-constc1=shared.pull();
-constc2=shared.pull();
+constiterator1=shared.pull()[Symbol.iterator]();
+constiterator2=shared.pull()[Symbol.iterator]();
+constreason=newError('sync cancel reason');
+
+iterator1.next();
+shared.cancel(reason);
 
-// c1 reads one item, then c2 cancels with a reason
-constiter1=c1[Symbol.iterator]();
-constfirst=iter1.next();
-assert.strictEqual(first.done,false);
+assert.throws(()=>iterator1.next(),(error)=>error===reason);
+assert.throws(()=>iterator2.next(),(error)=>error===reason);
+}
 
-shared.cancel(newError('sync cancel reason'));
+functiontestShareSyncCancelWithFalsyReason(){
+for(constreasonof[0,'',false,null]){
+constshared=shareSync(fromSync('data'));
+constiterator=shared.pull()[Symbol.iterator]();
 
-// c1 was already iterating, it's now detached → done
-constnext=iter1.next();
-assert.strictEqual(next.done,true);
+shared.cancel(reason);
 
-// c2 never started, also detached → done (not error)
-constbatches=[];
-for(constbatchofc2){
-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());

lib/internal/streams/iter/broadcast.js

@@ -88,7 +88,7 @@ class BroadcastImpl {
  #consumers =newSafeSet();
  #waiters =[];// Consumers with pending resolve (subset of #consumers)
  #ended =false;
- #error=null;
+ #error;
  #cancelled =false;
  #options;
  #writer =null;
@@ -177,7 +177,9 @@ class BroadcastImpl {
 __proto__: null,
 next(){
 if(state.detached){
-if(self.#error)returnPromiseReject(self.#error);
+if(self.#error !==undefined){
+returnPromiseReject(self.#error);
+}
 returnkDone;
 }
 
@@ -194,7 +196,7 @@ class BroadcastImpl {
 {__proto__: null,done: false,value: chunk});
 }
 
-if(self.#error){
+if(self.#error!==undefined){
 state.detached=true;
 self.#deleteConsumer(state);
 returnPromiseReject(self.#error);
@@ -344,7 +346,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 =newSafeSet();
  #sourceIterator =null;
  #sourceExhausted =false;
- #sourceError=null;
+ #sourceError;
  #cancelled =false;
  #pulling =false;
  #pullWaiters =[];
@@ -129,7 +129,7 @@ class ShareImpl {
 __proto__: null,
 [SymbolAsyncIterator](){
 constgetNext=async()=>{
-if(self.#sourceError){
+if(self.#sourceError!==undefined){
 state.detached=true;
 self.#consumers.delete(state);
 throwself.#sourceError;
@@ -141,7 +141,7 @@ class ShareImpl {
 // cursor must re-pull rather than terminating prematurely.
 for(;;){
 if(state.detached){
-if(self.#sourceError)throwself.#sourceError;
+if(self.#sourceError!==undefined)throwself.#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)throwself.#sourceError;
+if(self.#sourceError!==undefined)throwself.#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)throwself.#sourceError;
+if(self.#sourceError!==undefined)throwself.#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){
 returnthis.#cancelled ? null : true;
 }
 
@@ -418,7 +420,7 @@ class SyncShareImpl {
  #consumers =newSafeSet();
  #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);
 throwself.#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);
 throwself.#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
 asyncfunctiontestCancelWithFalsyReason(){
-const{broadcast: bc}=broadcast();
-constconsumer=bc.push();
-constresultPromise=text(consumer).catch((err)=>err);
-awaitnewPromise((resolve)=>setImmediate(resolve));
-bc.cancel(0);
-constresult=awaitresultPromise;
-assert.strictEqual(result,0);
+for(constreasonof[0,'',false,null]){
+const{broadcast: bc}=broadcast();
+constiterator=bc.push()[Symbol.asyncIterator]();
+
+bc.cancel(reason);
+
+awaitassert.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() {
 );
 }
 
+asyncfunctiontestShareCancelWithFalsyReason(){
+for(constreasonof[0,'',false,null]){
+constshared=share(from('data'));
+constiterator=shared.pull()[Symbol.asyncIterator]();
+
+shared.cancel(reason);
+
+awaitassert.rejects(iterator.next(),(error)=>error===reason);
+}
+}
+
 asyncfunctiontestShareAbortSignal(){
 constac=newAbortController();
 constreason=newError('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() {
 }
 
 functiontestShareSyncCancelWithReason(){
-// 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.
 constenc=newTextEncoder();
 function*gen(){
 yield[enc.encode('a')];
 yield[enc.encode('b')];
-yield[enc.encode('c')];
 }
 constshared=shareSync(gen(),{budget: 16384});
-constc1=shared.pull();
-constc2=shared.pull();
+constiterator1=shared.pull()[Symbol.iterator]();
+constiterator2=shared.pull()[Symbol.iterator]();
+constreason=newError('sync cancel reason');
+
+iterator1.next();
+shared.cancel(reason);
 
-// c1 reads one item, then c2 cancels with a reason
-constiter1=c1[Symbol.iterator]();
-constfirst=iter1.next();
-assert.strictEqual(first.done,false);
+assert.throws(()=>iterator1.next(),(error)=>error===reason);
+assert.throws(()=>iterator2.next(),(error)=>error===reason);
+}
 
-shared.cancel(newError('sync cancel reason'));
+functiontestShareSyncCancelWithFalsyReason(){
+for(constreasonof[0,'',false,null]){
+constshared=shareSync(fromSync('data'));
+constiterator=shared.pull()[Symbol.iterator]();
 
-// c1 was already iterating, it's now detached → done
-constnext=iter1.next();
-assert.strictEqual(next.done,true);
+shared.cancel(reason);
 
-// c2 never started, also detached → done (not error)
-constbatches=[];
-for(constbatchofc2){
-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());

lib/internal/streams/iter/broadcast.js

@@ -88,7 +88,7 @@ class BroadcastImpl {
  #consumers =newSafeSet();
  #waiters =[];// Consumers with pending resolve (subset of #consumers)
  #ended =false;
- #error=null;
+ #error;
  #cancelled =false;
  #options;
  #writer =null;
@@ -177,7 +177,9 @@ class BroadcastImpl {
 __proto__: null,
 next(){
 if(state.detached){
-if(self.#error)returnPromiseReject(self.#error);
+if(self.#error !==undefined){
+returnPromiseReject(self.#error);
+}
 returnkDone;
 }
 
@@ -194,7 +196,7 @@ class BroadcastImpl {
 {__proto__: null,done: false,value: chunk});
 }
 
-if(self.#error){
+if(self.#error!==undefined){
 state.detached=true;
 self.#deleteConsumer(state);
 returnPromiseReject(self.#error);
@@ -344,7 +346,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 =newSafeSet();
  #sourceIterator =null;
  #sourceExhausted =false;
- #sourceError=null;
+ #sourceError;
  #cancelled =false;
  #pulling =false;
  #pullWaiters =[];
@@ -129,7 +129,7 @@ class ShareImpl {
 __proto__: null,
 [SymbolAsyncIterator](){
 constgetNext=async()=>{
-if(self.#sourceError){
+if(self.#sourceError!==undefined){
 state.detached=true;
 self.#consumers.delete(state);
 throwself.#sourceError;
@@ -141,7 +141,7 @@ class ShareImpl {
 // cursor must re-pull rather than terminating prematurely.
 for(;;){
 if(state.detached){
-if(self.#sourceError)throwself.#sourceError;
+if(self.#sourceError!==undefined)throwself.#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)throwself.#sourceError;
+if(self.#sourceError!==undefined)throwself.#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)throwself.#sourceError;
+if(self.#sourceError!==undefined)throwself.#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){
 returnthis.#cancelled ? null : true;
 }
 
@@ -418,7 +420,7 @@ class SyncShareImpl {
  #consumers =newSafeSet();
  #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);
 throwself.#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);
 throwself.#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
 asyncfunctiontestCancelWithFalsyReason(){
-const{broadcast: bc}=broadcast();
-constconsumer=bc.push();
-constresultPromise=text(consumer).catch((err)=>err);
-awaitnewPromise((resolve)=>setImmediate(resolve));
-bc.cancel(0);
-constresult=awaitresultPromise;
-assert.strictEqual(result,0);
+for(constreasonof[0,'',false,null]){
+const{broadcast: bc}=broadcast();
+constiterator=bc.push()[Symbol.asyncIterator]();
+
+bc.cancel(reason);
+
+awaitassert.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() {
 );
 }
 
+asyncfunctiontestShareCancelWithFalsyReason(){
+for(constreasonof[0,'',false,null]){
+constshared=share(from('data'));
+constiterator=shared.pull()[Symbol.asyncIterator]();
+
+shared.cancel(reason);
+
+awaitassert.rejects(iterator.next(),(error)=>error===reason);
+}
+}
+
 asyncfunctiontestShareAbortSignal(){
 constac=newAbortController();
 constreason=newError('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() {
 }
 
 functiontestShareSyncCancelWithReason(){
-// 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.
 constenc=newTextEncoder();
 function*gen(){
 yield[enc.encode('a')];
 yield[enc.encode('b')];
-yield[enc.encode('c')];
 }
 constshared=shareSync(gen(),{budget: 16384});
-constc1=shared.pull();
-constc2=shared.pull();
+constiterator1=shared.pull()[Symbol.iterator]();
+constiterator2=shared.pull()[Symbol.iterator]();
+constreason=newError('sync cancel reason');
+
+iterator1.next();
+shared.cancel(reason);
 
-// c1 reads one item, then c2 cancels with a reason
-constiter1=c1[Symbol.iterator]();
-constfirst=iter1.next();
-assert.strictEqual(first.done,false);
+assert.throws(()=>iterator1.next(),(error)=>error===reason);
+assert.throws(()=>iterator2.next(),(error)=>error===reason);
+}
 
-shared.cancel(newError('sync cancel reason'));
+functiontestShareSyncCancelWithFalsyReason(){
+for(constreasonof[0,'',false,null]){
+constshared=shareSync(fromSync('data'));
+constiterator=shared.pull()[Symbol.iterator]();
 
-// c1 was already iterating, it's now detached → done
-constnext=iter1.next();
-assert.strictEqual(next.done,true);
+shared.cancel(reason);
 
-// c2 never started, also detached → done (not error)
-constbatches=[];
-for(constbatchofc2){
-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());

lib/internal/streams/iter/broadcast.js

@@ -88,7 +88,7 @@ class BroadcastImpl {
  #consumers =newSafeSet();
  #waiters =[];// Consumers with pending resolve (subset of #consumers)
  #ended =false;
- #error=null;
+ #error;
  #cancelled =false;
  #options;
  #writer =null;
@@ -177,7 +177,9 @@ class BroadcastImpl {
 __proto__: null,
 next(){
 if(state.detached){
-if(self.#error)returnPromiseReject(self.#error);
+if(self.#error !==undefined){
+returnPromiseReject(self.#error);
+}
 returnkDone;
 }
 
@@ -194,7 +196,7 @@ class BroadcastImpl {
 {__proto__: null,done: false,value: chunk});
 }
 
-if(self.#error){
+if(self.#error!==undefined){
 state.detached=true;
 self.#deleteConsumer(state);
 returnPromiseReject(self.#error);
@@ -344,7 +346,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 =newSafeSet();
  #sourceIterator =null;
  #sourceExhausted =false;
- #sourceError=null;
+ #sourceError;
  #cancelled =false;
  #pulling =false;
  #pullWaiters =[];
@@ -129,7 +129,7 @@ class ShareImpl {
 __proto__: null,
 [SymbolAsyncIterator](){
 constgetNext=async()=>{
-if(self.#sourceError){
+if(self.#sourceError!==undefined){
 state.detached=true;
 self.#consumers.delete(state);
 throwself.#sourceError;
@@ -141,7 +141,7 @@ class ShareImpl {
 // cursor must re-pull rather than terminating prematurely.
 for(;;){
 if(state.detached){
-if(self.#sourceError)throwself.#sourceError;
+if(self.#sourceError!==undefined)throwself.#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)throwself.#sourceError;
+if(self.#sourceError!==undefined)throwself.#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)throwself.#sourceError;
+if(self.#sourceError!==undefined)throwself.#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){
 returnthis.#cancelled ? null : true;
 }
 
@@ -418,7 +420,7 @@ class SyncShareImpl {
  #consumers =newSafeSet();
  #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);
 throwself.#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);
 throwself.#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
 asyncfunctiontestCancelWithFalsyReason(){
-const{broadcast: bc}=broadcast();
-constconsumer=bc.push();
-constresultPromise=text(consumer).catch((err)=>err);
-awaitnewPromise((resolve)=>setImmediate(resolve));
-bc.cancel(0);
-constresult=awaitresultPromise;
-assert.strictEqual(result,0);
+for(constreasonof[0,'',false,null]){
+const{broadcast: bc}=broadcast();
+constiterator=bc.push()[Symbol.asyncIterator]();
+
+bc.cancel(reason);
+
+awaitassert.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() {
 );
 }
 
+asyncfunctiontestShareCancelWithFalsyReason(){
+for(constreasonof[0,'',false,null]){
+constshared=share(from('data'));
+constiterator=shared.pull()[Symbol.asyncIterator]();
+
+shared.cancel(reason);
+
+awaitassert.rejects(iterator.next(),(error)=>error===reason);
+}
+}
+
 asyncfunctiontestShareAbortSignal(){
 constac=newAbortController();
 constreason=newError('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() {
 }
 
 functiontestShareSyncCancelWithReason(){
-// 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.
 constenc=newTextEncoder();
 function*gen(){
 yield[enc.encode('a')];
 yield[enc.encode('b')];
-yield[enc.encode('c')];
 }
 constshared=shareSync(gen(),{budget: 16384});
-constc1=shared.pull();
-constc2=shared.pull();
+constiterator1=shared.pull()[Symbol.iterator]();
+constiterator2=shared.pull()[Symbol.iterator]();
+constreason=newError('sync cancel reason');
+
+iterator1.next();
+shared.cancel(reason);
 
-// c1 reads one item, then c2 cancels with a reason
-constiter1=c1[Symbol.iterator]();
-constfirst=iter1.next();
-assert.strictEqual(first.done,false);
+assert.throws(()=>iterator1.next(),(error)=>error===reason);
+assert.throws(()=>iterator2.next(),(error)=>error===reason);
+}
 
-shared.cancel(newError('sync cancel reason'));
+functiontestShareSyncCancelWithFalsyReason(){
+for(constreasonof[0,'',false,null]){
+constshared=shareSync(fromSync('data'));
+constiterator=shared.pull()[Symbol.iterator]();
 
-// c1 was already iterating, it's now detached → done
-constnext=iter1.next();
-assert.strictEqual(next.done,true);
+shared.cancel(reason);
 
-// c2 never started, also detached → done (not error)
-constbatches=[];
-for(constbatchofc2){
-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());

lib/internal/streams/iter/broadcast.js

@@ -88,7 +88,7 @@ class BroadcastImpl {
  #consumers =newSafeSet();
  #waiters =[];// Consumers with pending resolve (subset of #consumers)
  #ended =false;
- #error=null;
+ #error;
  #cancelled =false;
  #options;
  #writer =null;
@@ -177,7 +177,9 @@ class BroadcastImpl {
 __proto__: null,
 next(){
 if(state.detached){
-if(self.#error)returnPromiseReject(self.#error);
+if(self.#error !==undefined){
+returnPromiseReject(self.#error);
+}
 returnkDone;
 }
 
@@ -194,7 +196,7 @@ class BroadcastImpl {
 {__proto__: null,done: false,value: chunk});
 }
 
-if(self.#error){
+if(self.#error!==undefined){
 state.detached=true;
 self.#deleteConsumer(state);
 returnPromiseReject(self.#error);
@@ -344,7 +346,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 =newSafeSet();
  #sourceIterator =null;
  #sourceExhausted =false;
- #sourceError=null;
+ #sourceError;
  #cancelled =false;
  #pulling =false;
  #pullWaiters =[];
@@ -129,7 +129,7 @@ class ShareImpl {
 __proto__: null,
 [SymbolAsyncIterator](){
 constgetNext=async()=>{
-if(self.#sourceError){
+if(self.#sourceError!==undefined){
 state.detached=true;
 self.#consumers.delete(state);
 throwself.#sourceError;
@@ -141,7 +141,7 @@ class ShareImpl {
 // cursor must re-pull rather than terminating prematurely.
 for(;;){
 if(state.detached){
-if(self.#sourceError)throwself.#sourceError;
+if(self.#sourceError!==undefined)throwself.#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)throwself.#sourceError;
+if(self.#sourceError!==undefined)throwself.#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)throwself.#sourceError;
+if(self.#sourceError!==undefined)throwself.#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){
 returnthis.#cancelled ? null : true;
 }
 
@@ -418,7 +420,7 @@ class SyncShareImpl {
  #consumers =newSafeSet();
  #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);
 throwself.#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);
 throwself.#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
 asyncfunctiontestCancelWithFalsyReason(){
-const{broadcast: bc}=broadcast();
-constconsumer=bc.push();
-constresultPromise=text(consumer).catch((err)=>err);
-awaitnewPromise((resolve)=>setImmediate(resolve));
-bc.cancel(0);
-constresult=awaitresultPromise;
-assert.strictEqual(result,0);
+for(constreasonof[0,'',false,null]){
+const{broadcast: bc}=broadcast();
+constiterator=bc.push()[Symbol.asyncIterator]();
+
+bc.cancel(reason);
+
+awaitassert.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() {
 );
 }
 
+asyncfunctiontestShareCancelWithFalsyReason(){
+for(constreasonof[0,'',false,null]){
+constshared=share(from('data'));
+constiterator=shared.pull()[Symbol.asyncIterator]();
+
+shared.cancel(reason);
+
+awaitassert.rejects(iterator.next(),(error)=>error===reason);
+}
+}
+
 asyncfunctiontestShareAbortSignal(){
 constac=newAbortController();
 constreason=newError('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() {
 }
 
 functiontestShareSyncCancelWithReason(){
-// 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.
 constenc=newTextEncoder();
 function*gen(){
 yield[enc.encode('a')];
 yield[enc.encode('b')];
-yield[enc.encode('c')];
 }
 constshared=shareSync(gen(),{budget: 16384});
-constc1=shared.pull();
-constc2=shared.pull();
+constiterator1=shared.pull()[Symbol.iterator]();
+constiterator2=shared.pull()[Symbol.iterator]();
+constreason=newError('sync cancel reason');
+
+iterator1.next();
+shared.cancel(reason);
 
-// c1 reads one item, then c2 cancels with a reason
-constiter1=c1[Symbol.iterator]();
-constfirst=iter1.next();
-assert.strictEqual(first.done,false);
+assert.throws(()=>iterator1.next(),(error)=>error===reason);
+assert.throws(()=>iterator2.next(),(error)=>error===reason);
+}
 
-shared.cancel(newError('sync cancel reason'));
+functiontestShareSyncCancelWithFalsyReason(){
+for(constreasonof[0,'',false,null]){
+constshared=shareSync(fromSync('data'));
+constiterator=shared.pull()[Symbol.iterator]();
 
-// c1 was already iterating, it's now detached → done
-constnext=iter1.next();
-assert.strictEqual(next.done,true);
+shared.cancel(reason);
 
-// c2 never started, also detached → done (not error)
-constbatches=[];
-for(constbatchofc2){
-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());