stream: keep overlapping broadcast reads pending

Broadcast consumers may receive overlapping next() calls on the same
iterator. Queue those reads so chunks satisfy them in call order.
A single written chunk should resolve the earliest pending next();
later next() calls remain pending until more data is written or the
broadcast completes.
Fixes: #63499
Signed-off-by: Kamat, Trivikram <16024985+trivikr@users.noreply.github.com>
Assisted-by: openai:gpt-5.5
PR-URL: #63500
Backport-PR-URL: #64675Fixes: #63499
Reviewed-By: James M Snell <jasnell@gmail.com>

Author: trivikr

lib/internal/streams/iter/broadcast.js

@@ -9,6 +9,7 @@
 const{
  ArrayIsArray,
  ArrayPrototypePush,
+ ArrayPrototypeShift,
  MathMax,
  PromisePrototypeThen,
  PromiseReject,
@@ -146,6 +147,7 @@ class BroadcastImpl {
 cursor: this.#bufferStart,
 resolve: null,
 reject: null,
+pending: [],
 detached: false,
 };
 
@@ -165,9 +167,10 @@ class BroadcastImpl {
 
 functiondetach(){
 state.detached=true;
-state.resolve?.({__proto__: null,done: true,value: undefined});
-state.resolve=null;
-state.reject=null;
+if(state.resolve){
+state.resolve({__proto__: null,done: true,value: undefined});
+}
+self.#resolvePendingDone(state);
 if(self.#deleteConsumer(state)){
 self.#tryTrimBuffer();
 }
@@ -208,6 +211,13 @@ class BroadcastImpl {
 returnkDone;
 }
 
+if(state.resolve){
+const{ promise, resolve, reject }=PromiseWithResolvers();
+ArrayPrototypePush(state.pending,
+{__proto__: null, resolve, reject });
+returnpromise;
+}
+
 const{ promise, resolve, reject }=PromiseWithResolvers();
 state.resolve=resolve;
 state.reject=reject;
@@ -251,6 +261,11 @@ class BroadcastImpl {
 consumer.resolve=null;
 consumer.reject=null;
 }
+if(reason!==undefined){
+this.#rejectPending(consumer,reason);
+}else{
+this.#resolvePendingDone(consumer);
+}
 consumer.detached=true;
 }
 this.#consumers.clear();
@@ -297,7 +312,7 @@ class BroadcastImpl {
 this.#ended =true;
 
 for(constconsumerofthis.#consumers){
-if(consumer.resolve){
+while(consumer.resolve){
 constbufferIndex=consumer.cursor-this.#bufferStart;
 if(bufferIndex<this.#buffer.length){
 constchunk=this.#buffer.get(bufferIndex);
@@ -310,9 +325,15 @@ class BroadcastImpl {
 consumer.resolve({__proto__: null,done: false,value: chunk});
 }else{
 consumer.resolve({__proto__: null,done: true,value: undefined});
+this.#resolvePendingDone(consumer);
+consumer.detached=true;
 }
 consumer.resolve=null;
 consumer.reject=null;
+if(consumer.detached&&this.#deleteConsumer(consumer)){
+this.#tryTrimBuffer();
+break;
+}
 }
 }
 }
@@ -329,6 +350,7 @@ class BroadcastImpl {
 consumer.resolve=null;
 consumer.reject=null;
 }
+this.#rejectPending(consumer,reason);
 consumer.detached=true;
 }
 this.#consumers.clear();
@@ -397,6 +419,11 @@ class BroadcastImpl {
 consumer.resolve=null;
 consumer.reject=null;
 resolve({__proto__: null,done: false,value: chunk});
+if(consumer.detached&&this.#deleteConsumer(consumer)){
+this.#tryTrimBuffer();
+}elseif(this.#promotePending(consumer)){
+ArrayPrototypePush(this.#waiters,consumer);
+}
 }else{
 // Still waiting -- put back
 ArrayPrototypePush(this.#waiters,consumer);
@@ -419,6 +446,31 @@ class BroadcastImpl {
 }
 returnfalse;
 }
+
+ #promotePending(consumer){
+constnext=ArrayPrototypeShift(consumer.pending);
+if(next===undefined)returnfalse;
+consumer.resolve=next.resolve;
+consumer.reject=next.reject;
+returntrue;
+}
+
+ #resolvePendingDone(consumer){
+if(consumer.resolve){
+consumer.resolve=null;
+consumer.reject=null;
+}
+while(consumer.pending.length>0){
+ArrayPrototypeShift(consumer.pending).resolve(
+{__proto__: null,done: true,value: undefined});
+}
+}
+
+ #rejectPending(consumer,reason){
+while(consumer.pending.length>0){
+ArrayPrototypeShift(consumer.pending).reject(reason);
+}
+}
 }
 
 // =============================================================================

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

@@ -3,6 +3,7 @@
 
 constcommon=require('../common');
 constassert=require('assert');
+const{ setTimeout }=require('timers/promises');
 const{ broadcast, text }=require('stream/iter');
 
 // =============================================================================
@@ -255,6 +256,38 @@ async function testLateJoinerSeesBufferedData() {
 assert.strictEqual(result,'before-join');
 }
 
+asyncfunctiontestOverlappingNextKeepsEarlierRead(){
+const{ writer,broadcast: bc}=broadcast();
+constit=bc.push()[Symbol.asyncIterator]();
+
+constfirst=it.next();
+constsecond=it.next();
+
+awaitwriter.write('x');
+
+constsecondResult=awaitPromise.race([
+second.then((value)=>({__proto__: null,settled: true, value })),
+setTimeout(common.platformTimeout(50),
+{__proto__: null,settled: false}),
+]);
+assert.deepStrictEqual(secondResult,{
+__proto__: null,
+settled: false,
+});
+
+constresult=awaitfirst;
+assert.strictEqual(result.done,false);
+assert.strictEqual(Buffer.concat(result.value).toString(),'x');
+
+writer.endSync();
+assert.deepStrictEqual(awaitsecond,{
+__proto__: null,
+done: true,
+value: undefined,
+});
+assert.strictEqual(bc.consumerCount,0);
+}
+
 Promise.all([
 testBasicBroadcast(),
 testMultipleWrites(),
@@ -270,4 +303,5 @@ Promise.all([
 testFailDetachesConsumers(),
 testWriterFailIdempotent(),
 testLateJoinerSeesBufferedData(),
+testOverlappingNextKeepsEarlierRead(),
 ]).then(common.mustCall());

lib/internal/streams/iter/broadcast.js

@@ -9,6 +9,7 @@
 const{
  ArrayIsArray,
  ArrayPrototypePush,
+ ArrayPrototypeShift,
  MathMax,
  PromisePrototypeThen,
  PromiseReject,
@@ -146,6 +147,7 @@ class BroadcastImpl {
 cursor: this.#bufferStart,
 resolve: null,
 reject: null,
+pending: [],
 detached: false,
 };
 
@@ -165,9 +167,10 @@ class BroadcastImpl {
 
 functiondetach(){
 state.detached=true;
-state.resolve?.({__proto__: null,done: true,value: undefined});
-state.resolve=null;
-state.reject=null;
+if(state.resolve){
+state.resolve({__proto__: null,done: true,value: undefined});
+}
+self.#resolvePendingDone(state);
 if(self.#deleteConsumer(state)){
 self.#tryTrimBuffer();
 }
@@ -208,6 +211,13 @@ class BroadcastImpl {
 returnkDone;
 }
 
+if(state.resolve){
+const{ promise, resolve, reject }=PromiseWithResolvers();
+ArrayPrototypePush(state.pending,
+{__proto__: null, resolve, reject });
+returnpromise;
+}
+
 const{ promise, resolve, reject }=PromiseWithResolvers();
 state.resolve=resolve;
 state.reject=reject;
@@ -251,6 +261,11 @@ class BroadcastImpl {
 consumer.resolve=null;
 consumer.reject=null;
 }
+if(reason!==undefined){
+this.#rejectPending(consumer,reason);
+}else{
+this.#resolvePendingDone(consumer);
+}
 consumer.detached=true;
 }
 this.#consumers.clear();
@@ -297,7 +312,7 @@ class BroadcastImpl {
 this.#ended =true;
 
 for(constconsumerofthis.#consumers){
-if(consumer.resolve){
+while(consumer.resolve){
 constbufferIndex=consumer.cursor-this.#bufferStart;
 if(bufferIndex<this.#buffer.length){
 constchunk=this.#buffer.get(bufferIndex);
@@ -310,9 +325,15 @@ class BroadcastImpl {
 consumer.resolve({__proto__: null,done: false,value: chunk});
 }else{
 consumer.resolve({__proto__: null,done: true,value: undefined});
+this.#resolvePendingDone(consumer);
+consumer.detached=true;
 }
 consumer.resolve=null;
 consumer.reject=null;
+if(consumer.detached&&this.#deleteConsumer(consumer)){
+this.#tryTrimBuffer();
+break;
+}
 }
 }
 }
@@ -329,6 +350,7 @@ class BroadcastImpl {
 consumer.resolve=null;
 consumer.reject=null;
 }
+this.#rejectPending(consumer,reason);
 consumer.detached=true;
 }
 this.#consumers.clear();
@@ -397,6 +419,11 @@ class BroadcastImpl {
 consumer.resolve=null;
 consumer.reject=null;
 resolve({__proto__: null,done: false,value: chunk});
+if(consumer.detached&&this.#deleteConsumer(consumer)){
+this.#tryTrimBuffer();
+}elseif(this.#promotePending(consumer)){
+ArrayPrototypePush(this.#waiters,consumer);
+}
 }else{
 // Still waiting -- put back
 ArrayPrototypePush(this.#waiters,consumer);
@@ -419,6 +446,31 @@ class BroadcastImpl {
 }
 returnfalse;
 }
+
+ #promotePending(consumer){
+constnext=ArrayPrototypeShift(consumer.pending);
+if(next===undefined)returnfalse;
+consumer.resolve=next.resolve;
+consumer.reject=next.reject;
+returntrue;
+}
+
+ #resolvePendingDone(consumer){
+if(consumer.resolve){
+consumer.resolve=null;
+consumer.reject=null;
+}
+while(consumer.pending.length>0){
+ArrayPrototypeShift(consumer.pending).resolve(
+{__proto__: null,done: true,value: undefined});
+}
+}
+
+ #rejectPending(consumer,reason){
+while(consumer.pending.length>0){
+ArrayPrototypeShift(consumer.pending).reject(reason);
+}
+}
 }
 
 // =============================================================================

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

@@ -3,6 +3,7 @@
 
 constcommon=require('../common');
 constassert=require('assert');
+const{ setTimeout }=require('timers/promises');
 const{ broadcast, text }=require('stream/iter');
 
 // =============================================================================
@@ -255,6 +256,38 @@ async function testLateJoinerSeesBufferedData() {
 assert.strictEqual(result,'before-join');
 }
 
+asyncfunctiontestOverlappingNextKeepsEarlierRead(){
+const{ writer,broadcast: bc}=broadcast();
+constit=bc.push()[Symbol.asyncIterator]();
+
+constfirst=it.next();
+constsecond=it.next();
+
+awaitwriter.write('x');
+
+constsecondResult=awaitPromise.race([
+second.then((value)=>({__proto__: null,settled: true, value })),
+setTimeout(common.platformTimeout(50),
+{__proto__: null,settled: false}),
+]);
+assert.deepStrictEqual(secondResult,{
+__proto__: null,
+settled: false,
+});
+
+constresult=awaitfirst;
+assert.strictEqual(result.done,false);
+assert.strictEqual(Buffer.concat(result.value).toString(),'x');
+
+writer.endSync();
+assert.deepStrictEqual(awaitsecond,{
+__proto__: null,
+done: true,
+value: undefined,
+});
+assert.strictEqual(bc.consumerCount,0);
+}
+
 Promise.all([
 testBasicBroadcast(),
 testMultipleWrites(),
@@ -270,4 +303,5 @@ Promise.all([
 testFailDetachesConsumers(),
 testWriterFailIdempotent(),
 testLateJoinerSeesBufferedData(),
+testOverlappingNextKeepsEarlierRead(),
 ]).then(common.mustCall());

lib/internal/streams/iter/broadcast.js

@@ -9,6 +9,7 @@
 const{
  ArrayIsArray,
  ArrayPrototypePush,
+ ArrayPrototypeShift,
  MathMax,
  PromisePrototypeThen,
  PromiseReject,
@@ -146,6 +147,7 @@ class BroadcastImpl {
 cursor: this.#bufferStart,
 resolve: null,
 reject: null,
+pending: [],
 detached: false,
 };
 
@@ -165,9 +167,10 @@ class BroadcastImpl {
 
 functiondetach(){
 state.detached=true;
-state.resolve?.({__proto__: null,done: true,value: undefined});
-state.resolve=null;
-state.reject=null;
+if(state.resolve){
+state.resolve({__proto__: null,done: true,value: undefined});
+}
+self.#resolvePendingDone(state);
 if(self.#deleteConsumer(state)){
 self.#tryTrimBuffer();
 }
@@ -208,6 +211,13 @@ class BroadcastImpl {
 returnkDone;
 }
 
+if(state.resolve){
+const{ promise, resolve, reject }=PromiseWithResolvers();
+ArrayPrototypePush(state.pending,
+{__proto__: null, resolve, reject });
+returnpromise;
+}
+
 const{ promise, resolve, reject }=PromiseWithResolvers();
 state.resolve=resolve;
 state.reject=reject;
@@ -251,6 +261,11 @@ class BroadcastImpl {
 consumer.resolve=null;
 consumer.reject=null;
 }
+if(reason!==undefined){
+this.#rejectPending(consumer,reason);
+}else{
+this.#resolvePendingDone(consumer);
+}
 consumer.detached=true;
 }
 this.#consumers.clear();
@@ -297,7 +312,7 @@ class BroadcastImpl {
 this.#ended =true;
 
 for(constconsumerofthis.#consumers){
-if(consumer.resolve){
+while(consumer.resolve){
 constbufferIndex=consumer.cursor-this.#bufferStart;
 if(bufferIndex<this.#buffer.length){
 constchunk=this.#buffer.get(bufferIndex);
@@ -310,9 +325,15 @@ class BroadcastImpl {
 consumer.resolve({__proto__: null,done: false,value: chunk});
 }else{
 consumer.resolve({__proto__: null,done: true,value: undefined});
+this.#resolvePendingDone(consumer);
+consumer.detached=true;
 }
 consumer.resolve=null;
 consumer.reject=null;
+if(consumer.detached&&this.#deleteConsumer(consumer)){
+this.#tryTrimBuffer();
+break;
+}
 }
 }
 }
@@ -329,6 +350,7 @@ class BroadcastImpl {
 consumer.resolve=null;
 consumer.reject=null;
 }
+this.#rejectPending(consumer,reason);
 consumer.detached=true;
 }
 this.#consumers.clear();
@@ -397,6 +419,11 @@ class BroadcastImpl {
 consumer.resolve=null;
 consumer.reject=null;
 resolve({__proto__: null,done: false,value: chunk});
+if(consumer.detached&&this.#deleteConsumer(consumer)){
+this.#tryTrimBuffer();
+}elseif(this.#promotePending(consumer)){
+ArrayPrototypePush(this.#waiters,consumer);
+}
 }else{
 // Still waiting -- put back
 ArrayPrototypePush(this.#waiters,consumer);
@@ -419,6 +446,31 @@ class BroadcastImpl {
 }
 returnfalse;
 }
+
+ #promotePending(consumer){
+constnext=ArrayPrototypeShift(consumer.pending);
+if(next===undefined)returnfalse;
+consumer.resolve=next.resolve;
+consumer.reject=next.reject;
+returntrue;
+}
+
+ #resolvePendingDone(consumer){
+if(consumer.resolve){
+consumer.resolve=null;
+consumer.reject=null;
+}
+while(consumer.pending.length>0){
+ArrayPrototypeShift(consumer.pending).resolve(
+{__proto__: null,done: true,value: undefined});
+}
+}
+
+ #rejectPending(consumer,reason){
+while(consumer.pending.length>0){
+ArrayPrototypeShift(consumer.pending).reject(reason);
+}
+}
 }
 
 // =============================================================================

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

@@ -3,6 +3,7 @@
 
 constcommon=require('../common');
 constassert=require('assert');
+const{ setTimeout }=require('timers/promises');
 const{ broadcast, text }=require('stream/iter');
 
 // =============================================================================
@@ -255,6 +256,38 @@ async function testLateJoinerSeesBufferedData() {
 assert.strictEqual(result,'before-join');
 }
 
+asyncfunctiontestOverlappingNextKeepsEarlierRead(){
+const{ writer,broadcast: bc}=broadcast();
+constit=bc.push()[Symbol.asyncIterator]();
+
+constfirst=it.next();
+constsecond=it.next();
+
+awaitwriter.write('x');
+
+constsecondResult=awaitPromise.race([
+second.then((value)=>({__proto__: null,settled: true, value })),
+setTimeout(common.platformTimeout(50),
+{__proto__: null,settled: false}),
+]);
+assert.deepStrictEqual(secondResult,{
+__proto__: null,
+settled: false,
+});
+
+constresult=awaitfirst;
+assert.strictEqual(result.done,false);
+assert.strictEqual(Buffer.concat(result.value).toString(),'x');
+
+writer.endSync();
+assert.deepStrictEqual(awaitsecond,{
+__proto__: null,
+done: true,
+value: undefined,
+});
+assert.strictEqual(bc.consumerCount,0);
+}
+
 Promise.all([
 testBasicBroadcast(),
 testMultipleWrites(),
@@ -270,4 +303,5 @@ Promise.all([
 testFailDetachesConsumers(),
 testWriterFailIdempotent(),
 testLateJoinerSeesBufferedData(),
+testOverlappingNextKeepsEarlierRead(),
 ]).then(common.mustCall());

lib/internal/streams/iter/broadcast.js

@@ -9,6 +9,7 @@
 const{
  ArrayIsArray,
  ArrayPrototypePush,
+ ArrayPrototypeShift,
  MathMax,
  PromisePrototypeThen,
  PromiseReject,
@@ -146,6 +147,7 @@ class BroadcastImpl {
 cursor: this.#bufferStart,
 resolve: null,
 reject: null,
+pending: [],
 detached: false,
 };
 
@@ -165,9 +167,10 @@ class BroadcastImpl {
 
 functiondetach(){
 state.detached=true;
-state.resolve?.({__proto__: null,done: true,value: undefined});
-state.resolve=null;
-state.reject=null;
+if(state.resolve){
+state.resolve({__proto__: null,done: true,value: undefined});
+}
+self.#resolvePendingDone(state);
 if(self.#deleteConsumer(state)){
 self.#tryTrimBuffer();
 }
@@ -208,6 +211,13 @@ class BroadcastImpl {
 returnkDone;
 }
 
+if(state.resolve){
+const{ promise, resolve, reject }=PromiseWithResolvers();
+ArrayPrototypePush(state.pending,
+{__proto__: null, resolve, reject });
+returnpromise;
+}
+
 const{ promise, resolve, reject }=PromiseWithResolvers();
 state.resolve=resolve;
 state.reject=reject;
@@ -251,6 +261,11 @@ class BroadcastImpl {
 consumer.resolve=null;
 consumer.reject=null;
 }
+if(reason!==undefined){
+this.#rejectPending(consumer,reason);
+}else{
+this.#resolvePendingDone(consumer);
+}
 consumer.detached=true;
 }
 this.#consumers.clear();
@@ -297,7 +312,7 @@ class BroadcastImpl {
 this.#ended =true;
 
 for(constconsumerofthis.#consumers){
-if(consumer.resolve){
+while(consumer.resolve){
 constbufferIndex=consumer.cursor-this.#bufferStart;
 if(bufferIndex<this.#buffer.length){
 constchunk=this.#buffer.get(bufferIndex);
@@ -310,9 +325,15 @@ class BroadcastImpl {
 consumer.resolve({__proto__: null,done: false,value: chunk});
 }else{
 consumer.resolve({__proto__: null,done: true,value: undefined});
+this.#resolvePendingDone(consumer);
+consumer.detached=true;
 }
 consumer.resolve=null;
 consumer.reject=null;
+if(consumer.detached&&this.#deleteConsumer(consumer)){
+this.#tryTrimBuffer();
+break;
+}
 }
 }
 }
@@ -329,6 +350,7 @@ class BroadcastImpl {
 consumer.resolve=null;
 consumer.reject=null;
 }
+this.#rejectPending(consumer,reason);
 consumer.detached=true;
 }
 this.#consumers.clear();
@@ -397,6 +419,11 @@ class BroadcastImpl {
 consumer.resolve=null;
 consumer.reject=null;
 resolve({__proto__: null,done: false,value: chunk});
+if(consumer.detached&&this.#deleteConsumer(consumer)){
+this.#tryTrimBuffer();
+}elseif(this.#promotePending(consumer)){
+ArrayPrototypePush(this.#waiters,consumer);
+}
 }else{
 // Still waiting -- put back
 ArrayPrototypePush(this.#waiters,consumer);
@@ -419,6 +446,31 @@ class BroadcastImpl {
 }
 returnfalse;
 }
+
+ #promotePending(consumer){
+constnext=ArrayPrototypeShift(consumer.pending);
+if(next===undefined)returnfalse;
+consumer.resolve=next.resolve;
+consumer.reject=next.reject;
+returntrue;
+}
+
+ #resolvePendingDone(consumer){
+if(consumer.resolve){
+consumer.resolve=null;
+consumer.reject=null;
+}
+while(consumer.pending.length>0){
+ArrayPrototypeShift(consumer.pending).resolve(
+{__proto__: null,done: true,value: undefined});
+}
+}
+
+ #rejectPending(consumer,reason){
+while(consumer.pending.length>0){
+ArrayPrototypeShift(consumer.pending).reject(reason);
+}
+}
 }
 
 // =============================================================================

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

@@ -3,6 +3,7 @@
 
 constcommon=require('../common');
 constassert=require('assert');
+const{ setTimeout }=require('timers/promises');
 const{ broadcast, text }=require('stream/iter');
 
 // =============================================================================
@@ -255,6 +256,38 @@ async function testLateJoinerSeesBufferedData() {
 assert.strictEqual(result,'before-join');
 }
 
+asyncfunctiontestOverlappingNextKeepsEarlierRead(){
+const{ writer,broadcast: bc}=broadcast();
+constit=bc.push()[Symbol.asyncIterator]();
+
+constfirst=it.next();
+constsecond=it.next();
+
+awaitwriter.write('x');
+
+constsecondResult=awaitPromise.race([
+second.then((value)=>({__proto__: null,settled: true, value })),
+setTimeout(common.platformTimeout(50),
+{__proto__: null,settled: false}),
+]);
+assert.deepStrictEqual(secondResult,{
+__proto__: null,
+settled: false,
+});
+
+constresult=awaitfirst;
+assert.strictEqual(result.done,false);
+assert.strictEqual(Buffer.concat(result.value).toString(),'x');
+
+writer.endSync();
+assert.deepStrictEqual(awaitsecond,{
+__proto__: null,
+done: true,
+value: undefined,
+});
+assert.strictEqual(bc.consumerCount,0);
+}
+
 Promise.all([
 testBasicBroadcast(),
 testMultipleWrites(),
@@ -270,4 +303,5 @@ Promise.all([
 testFailDetachesConsumers(),
 testWriterFailIdempotent(),
 testLateJoinerSeesBufferedData(),
+testOverlappingNextKeepsEarlierRead(),
 ]).then(common.mustCall());

lib/internal/streams/iter/broadcast.js

@@ -9,6 +9,7 @@
 const{
  ArrayIsArray,
  ArrayPrototypePush,
+ ArrayPrototypeShift,
  MathMax,
  PromisePrototypeThen,
  PromiseReject,
@@ -146,6 +147,7 @@ class BroadcastImpl {
 cursor: this.#bufferStart,
 resolve: null,
 reject: null,
+pending: [],
 detached: false,
 };
 
@@ -165,9 +167,10 @@ class BroadcastImpl {
 
 functiondetach(){
 state.detached=true;
-state.resolve?.({__proto__: null,done: true,value: undefined});
-state.resolve=null;
-state.reject=null;
+if(state.resolve){
+state.resolve({__proto__: null,done: true,value: undefined});
+}
+self.#resolvePendingDone(state);
 if(self.#deleteConsumer(state)){
 self.#tryTrimBuffer();
 }
@@ -208,6 +211,13 @@ class BroadcastImpl {
 returnkDone;
 }
 
+if(state.resolve){
+const{ promise, resolve, reject }=PromiseWithResolvers();
+ArrayPrototypePush(state.pending,
+{__proto__: null, resolve, reject });
+returnpromise;
+}
+
 const{ promise, resolve, reject }=PromiseWithResolvers();
 state.resolve=resolve;
 state.reject=reject;
@@ -251,6 +261,11 @@ class BroadcastImpl {
 consumer.resolve=null;
 consumer.reject=null;
 }
+if(reason!==undefined){
+this.#rejectPending(consumer,reason);
+}else{
+this.#resolvePendingDone(consumer);
+}
 consumer.detached=true;
 }
 this.#consumers.clear();
@@ -297,7 +312,7 @@ class BroadcastImpl {
 this.#ended =true;
 
 for(constconsumerofthis.#consumers){
-if(consumer.resolve){
+while(consumer.resolve){
 constbufferIndex=consumer.cursor-this.#bufferStart;
 if(bufferIndex<this.#buffer.length){
 constchunk=this.#buffer.get(bufferIndex);
@@ -310,9 +325,15 @@ class BroadcastImpl {
 consumer.resolve({__proto__: null,done: false,value: chunk});
 }else{
 consumer.resolve({__proto__: null,done: true,value: undefined});
+this.#resolvePendingDone(consumer);
+consumer.detached=true;
 }
 consumer.resolve=null;
 consumer.reject=null;
+if(consumer.detached&&this.#deleteConsumer(consumer)){
+this.#tryTrimBuffer();
+break;
+}
 }
 }
 }
@@ -329,6 +350,7 @@ class BroadcastImpl {
 consumer.resolve=null;
 consumer.reject=null;
 }
+this.#rejectPending(consumer,reason);
 consumer.detached=true;
 }
 this.#consumers.clear();
@@ -397,6 +419,11 @@ class BroadcastImpl {
 consumer.resolve=null;
 consumer.reject=null;
 resolve({__proto__: null,done: false,value: chunk});
+if(consumer.detached&&this.#deleteConsumer(consumer)){
+this.#tryTrimBuffer();
+}elseif(this.#promotePending(consumer)){
+ArrayPrototypePush(this.#waiters,consumer);
+}
 }else{
 // Still waiting -- put back
 ArrayPrototypePush(this.#waiters,consumer);
@@ -419,6 +446,31 @@ class BroadcastImpl {
 }
 returnfalse;
 }
+
+ #promotePending(consumer){
+constnext=ArrayPrototypeShift(consumer.pending);
+if(next===undefined)returnfalse;
+consumer.resolve=next.resolve;
+consumer.reject=next.reject;
+returntrue;
+}
+
+ #resolvePendingDone(consumer){
+if(consumer.resolve){
+consumer.resolve=null;
+consumer.reject=null;
+}
+while(consumer.pending.length>0){
+ArrayPrototypeShift(consumer.pending).resolve(
+{__proto__: null,done: true,value: undefined});
+}
+}
+
+ #rejectPending(consumer,reason){
+while(consumer.pending.length>0){
+ArrayPrototypeShift(consumer.pending).reject(reason);
+}
+}
 }
 
 // =============================================================================

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

@@ -3,6 +3,7 @@
 
 constcommon=require('../common');
 constassert=require('assert');
+const{ setTimeout }=require('timers/promises');
 const{ broadcast, text }=require('stream/iter');
 
 // =============================================================================
@@ -255,6 +256,38 @@ async function testLateJoinerSeesBufferedData() {
 assert.strictEqual(result,'before-join');
 }
 
+asyncfunctiontestOverlappingNextKeepsEarlierRead(){
+const{ writer,broadcast: bc}=broadcast();
+constit=bc.push()[Symbol.asyncIterator]();
+
+constfirst=it.next();
+constsecond=it.next();
+
+awaitwriter.write('x');
+
+constsecondResult=awaitPromise.race([
+second.then((value)=>({__proto__: null,settled: true, value })),
+setTimeout(common.platformTimeout(50),
+{__proto__: null,settled: false}),
+]);
+assert.deepStrictEqual(secondResult,{
+__proto__: null,
+settled: false,
+});
+
+constresult=awaitfirst;
+assert.strictEqual(result.done,false);
+assert.strictEqual(Buffer.concat(result.value).toString(),'x');
+
+writer.endSync();
+assert.deepStrictEqual(awaitsecond,{
+__proto__: null,
+done: true,
+value: undefined,
+});
+assert.strictEqual(bc.consumerCount,0);
+}
+
 Promise.all([
 testBasicBroadcast(),
 testMultipleWrites(),
@@ -270,4 +303,5 @@ Promise.all([
 testFailDetachesConsumers(),
 testWriterFailIdempotent(),
 testLateJoinerSeesBufferedData(),
+testOverlappingNextKeepsEarlierRead(),
 ]).then(common.mustCall());

lib/internal/streams/iter/broadcast.js

@@ -9,6 +9,7 @@
 const{
  ArrayIsArray,
  ArrayPrototypePush,
+ ArrayPrototypeShift,
  MathMax,
  PromisePrototypeThen,
  PromiseReject,
@@ -146,6 +147,7 @@ class BroadcastImpl {
 cursor: this.#bufferStart,
 resolve: null,
 reject: null,
+pending: [],
 detached: false,
 };
 
@@ -165,9 +167,10 @@ class BroadcastImpl {
 
 functiondetach(){
 state.detached=true;
-state.resolve?.({__proto__: null,done: true,value: undefined});
-state.resolve=null;
-state.reject=null;
+if(state.resolve){
+state.resolve({__proto__: null,done: true,value: undefined});
+}
+self.#resolvePendingDone(state);
 if(self.#deleteConsumer(state)){
 self.#tryTrimBuffer();
 }
@@ -208,6 +211,13 @@ class BroadcastImpl {
 returnkDone;
 }
 
+if(state.resolve){
+const{ promise, resolve, reject }=PromiseWithResolvers();
+ArrayPrototypePush(state.pending,
+{__proto__: null, resolve, reject });
+returnpromise;
+}
+
 const{ promise, resolve, reject }=PromiseWithResolvers();
 state.resolve=resolve;
 state.reject=reject;
@@ -251,6 +261,11 @@ class BroadcastImpl {
 consumer.resolve=null;
 consumer.reject=null;
 }
+if(reason!==undefined){
+this.#rejectPending(consumer,reason);
+}else{
+this.#resolvePendingDone(consumer);
+}
 consumer.detached=true;
 }
 this.#consumers.clear();
@@ -297,7 +312,7 @@ class BroadcastImpl {
 this.#ended =true;
 
 for(constconsumerofthis.#consumers){
-if(consumer.resolve){
+while(consumer.resolve){
 constbufferIndex=consumer.cursor-this.#bufferStart;
 if(bufferIndex<this.#buffer.length){
 constchunk=this.#buffer.get(bufferIndex);
@@ -310,9 +325,15 @@ class BroadcastImpl {
 consumer.resolve({__proto__: null,done: false,value: chunk});
 }else{
 consumer.resolve({__proto__: null,done: true,value: undefined});
+this.#resolvePendingDone(consumer);
+consumer.detached=true;
 }
 consumer.resolve=null;
 consumer.reject=null;
+if(consumer.detached&&this.#deleteConsumer(consumer)){
+this.#tryTrimBuffer();
+break;
+}
 }
 }
 }
@@ -329,6 +350,7 @@ class BroadcastImpl {
 consumer.resolve=null;
 consumer.reject=null;
 }
+this.#rejectPending(consumer,reason);
 consumer.detached=true;
 }
 this.#consumers.clear();
@@ -397,6 +419,11 @@ class BroadcastImpl {
 consumer.resolve=null;
 consumer.reject=null;
 resolve({__proto__: null,done: false,value: chunk});
+if(consumer.detached&&this.#deleteConsumer(consumer)){
+this.#tryTrimBuffer();
+}elseif(this.#promotePending(consumer)){
+ArrayPrototypePush(this.#waiters,consumer);
+}
 }else{
 // Still waiting -- put back
 ArrayPrototypePush(this.#waiters,consumer);
@@ -419,6 +446,31 @@ class BroadcastImpl {
 }
 returnfalse;
 }
+
+ #promotePending(consumer){
+constnext=ArrayPrototypeShift(consumer.pending);
+if(next===undefined)returnfalse;
+consumer.resolve=next.resolve;
+consumer.reject=next.reject;
+returntrue;
+}
+
+ #resolvePendingDone(consumer){
+if(consumer.resolve){
+consumer.resolve=null;
+consumer.reject=null;
+}
+while(consumer.pending.length>0){
+ArrayPrototypeShift(consumer.pending).resolve(
+{__proto__: null,done: true,value: undefined});
+}
+}
+
+ #rejectPending(consumer,reason){
+while(consumer.pending.length>0){
+ArrayPrototypeShift(consumer.pending).reject(reason);
+}
+}
 }
 
 // =============================================================================

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

@@ -3,6 +3,7 @@
 
 constcommon=require('../common');
 constassert=require('assert');
+const{ setTimeout }=require('timers/promises');
 const{ broadcast, text }=require('stream/iter');
 
 // =============================================================================
@@ -255,6 +256,38 @@ async function testLateJoinerSeesBufferedData() {
 assert.strictEqual(result,'before-join');
 }
 
+asyncfunctiontestOverlappingNextKeepsEarlierRead(){
+const{ writer,broadcast: bc}=broadcast();
+constit=bc.push()[Symbol.asyncIterator]();
+
+constfirst=it.next();
+constsecond=it.next();
+
+awaitwriter.write('x');
+
+constsecondResult=awaitPromise.race([
+second.then((value)=>({__proto__: null,settled: true, value })),
+setTimeout(common.platformTimeout(50),
+{__proto__: null,settled: false}),
+]);
+assert.deepStrictEqual(secondResult,{
+__proto__: null,
+settled: false,
+});
+
+constresult=awaitfirst;
+assert.strictEqual(result.done,false);
+assert.strictEqual(Buffer.concat(result.value).toString(),'x');
+
+writer.endSync();
+assert.deepStrictEqual(awaitsecond,{
+__proto__: null,
+done: true,
+value: undefined,
+});
+assert.strictEqual(bc.consumerCount,0);
+}
+
 Promise.all([
 testBasicBroadcast(),
 testMultipleWrites(),
@@ -270,4 +303,5 @@ Promise.all([
 testFailDetachesConsumers(),
 testWriterFailIdempotent(),
 testLateJoinerSeesBufferedData(),
+testOverlappingNextKeepsEarlierRead(),
 ]).then(common.mustCall());

lib/internal/streams/iter/broadcast.js

@@ -9,6 +9,7 @@
 const{
  ArrayIsArray,
  ArrayPrototypePush,
+ ArrayPrototypeShift,
  MathMax,
  PromisePrototypeThen,
  PromiseReject,
@@ -146,6 +147,7 @@ class BroadcastImpl {
 cursor: this.#bufferStart,
 resolve: null,
 reject: null,
+pending: [],
 detached: false,
 };
 
@@ -165,9 +167,10 @@ class BroadcastImpl {
 
 functiondetach(){
 state.detached=true;
-state.resolve?.({__proto__: null,done: true,value: undefined});
-state.resolve=null;
-state.reject=null;
+if(state.resolve){
+state.resolve({__proto__: null,done: true,value: undefined});
+}
+self.#resolvePendingDone(state);
 if(self.#deleteConsumer(state)){
 self.#tryTrimBuffer();
 }
@@ -208,6 +211,13 @@ class BroadcastImpl {
 returnkDone;
 }
 
+if(state.resolve){
+const{ promise, resolve, reject }=PromiseWithResolvers();
+ArrayPrototypePush(state.pending,
+{__proto__: null, resolve, reject });
+returnpromise;
+}
+
 const{ promise, resolve, reject }=PromiseWithResolvers();
 state.resolve=resolve;
 state.reject=reject;
@@ -251,6 +261,11 @@ class BroadcastImpl {
 consumer.resolve=null;
 consumer.reject=null;
 }
+if(reason!==undefined){
+this.#rejectPending(consumer,reason);
+}else{
+this.#resolvePendingDone(consumer);
+}
 consumer.detached=true;
 }
 this.#consumers.clear();
@@ -297,7 +312,7 @@ class BroadcastImpl {
 this.#ended =true;
 
 for(constconsumerofthis.#consumers){
-if(consumer.resolve){
+while(consumer.resolve){
 constbufferIndex=consumer.cursor-this.#bufferStart;
 if(bufferIndex<this.#buffer.length){
 constchunk=this.#buffer.get(bufferIndex);
@@ -310,9 +325,15 @@ class BroadcastImpl {
 consumer.resolve({__proto__: null,done: false,value: chunk});
 }else{
 consumer.resolve({__proto__: null,done: true,value: undefined});
+this.#resolvePendingDone(consumer);
+consumer.detached=true;
 }
 consumer.resolve=null;
 consumer.reject=null;
+if(consumer.detached&&this.#deleteConsumer(consumer)){
+this.#tryTrimBuffer();
+break;
+}
 }
 }
 }
@@ -329,6 +350,7 @@ class BroadcastImpl {
 consumer.resolve=null;
 consumer.reject=null;
 }
+this.#rejectPending(consumer,reason);
 consumer.detached=true;
 }
 this.#consumers.clear();
@@ -397,6 +419,11 @@ class BroadcastImpl {
 consumer.resolve=null;
 consumer.reject=null;
 resolve({__proto__: null,done: false,value: chunk});
+if(consumer.detached&&this.#deleteConsumer(consumer)){
+this.#tryTrimBuffer();
+}elseif(this.#promotePending(consumer)){
+ArrayPrototypePush(this.#waiters,consumer);
+}
 }else{
 // Still waiting -- put back
 ArrayPrototypePush(this.#waiters,consumer);
@@ -419,6 +446,31 @@ class BroadcastImpl {
 }
 returnfalse;
 }
+
+ #promotePending(consumer){
+constnext=ArrayPrototypeShift(consumer.pending);
+if(next===undefined)returnfalse;
+consumer.resolve=next.resolve;
+consumer.reject=next.reject;
+returntrue;
+}
+
+ #resolvePendingDone(consumer){
+if(consumer.resolve){
+consumer.resolve=null;
+consumer.reject=null;
+}
+while(consumer.pending.length>0){
+ArrayPrototypeShift(consumer.pending).resolve(
+{__proto__: null,done: true,value: undefined});
+}
+}
+
+ #rejectPending(consumer,reason){
+while(consumer.pending.length>0){
+ArrayPrototypeShift(consumer.pending).reject(reason);
+}
+}
 }
 
 // =============================================================================

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

@@ -3,6 +3,7 @@
 
 constcommon=require('../common');
 constassert=require('assert');
+const{ setTimeout }=require('timers/promises');
 const{ broadcast, text }=require('stream/iter');
 
 // =============================================================================
@@ -255,6 +256,38 @@ async function testLateJoinerSeesBufferedData() {
 assert.strictEqual(result,'before-join');
 }
 
+asyncfunctiontestOverlappingNextKeepsEarlierRead(){
+const{ writer,broadcast: bc}=broadcast();
+constit=bc.push()[Symbol.asyncIterator]();
+
+constfirst=it.next();
+constsecond=it.next();
+
+awaitwriter.write('x');
+
+constsecondResult=awaitPromise.race([
+second.then((value)=>({__proto__: null,settled: true, value })),
+setTimeout(common.platformTimeout(50),
+{__proto__: null,settled: false}),
+]);
+assert.deepStrictEqual(secondResult,{
+__proto__: null,
+settled: false,
+});
+
+constresult=awaitfirst;
+assert.strictEqual(result.done,false);
+assert.strictEqual(Buffer.concat(result.value).toString(),'x');
+
+writer.endSync();
+assert.deepStrictEqual(awaitsecond,{
+__proto__: null,
+done: true,
+value: undefined,
+});
+assert.strictEqual(bc.consumerCount,0);
+}
+
 Promise.all([
 testBasicBroadcast(),
 testMultipleWrites(),
@@ -270,4 +303,5 @@ Promise.all([
 testFailDetachesConsumers(),
 testWriterFailIdempotent(),
 testLateJoinerSeesBufferedData(),
+testOverlappingNextKeepsEarlierRead(),
 ]).then(common.mustCall());

lib/internal/streams/iter/broadcast.js

@@ -9,6 +9,7 @@
 const{
  ArrayIsArray,
  ArrayPrototypePush,
+ ArrayPrototypeShift,
  MathMax,
  PromisePrototypeThen,
  PromiseReject,
@@ -146,6 +147,7 @@ class BroadcastImpl {
 cursor: this.#bufferStart,
 resolve: null,
 reject: null,
+pending: [],
 detached: false,
 };
 
@@ -165,9 +167,10 @@ class BroadcastImpl {
 
 functiondetach(){
 state.detached=true;
-state.resolve?.({__proto__: null,done: true,value: undefined});
-state.resolve=null;
-state.reject=null;
+if(state.resolve){
+state.resolve({__proto__: null,done: true,value: undefined});
+}
+self.#resolvePendingDone(state);
 if(self.#deleteConsumer(state)){
 self.#tryTrimBuffer();
 }
@@ -208,6 +211,13 @@ class BroadcastImpl {
 returnkDone;
 }
 
+if(state.resolve){
+const{ promise, resolve, reject }=PromiseWithResolvers();
+ArrayPrototypePush(state.pending,
+{__proto__: null, resolve, reject });
+returnpromise;
+}
+
 const{ promise, resolve, reject }=PromiseWithResolvers();
 state.resolve=resolve;
 state.reject=reject;
@@ -251,6 +261,11 @@ class BroadcastImpl {
 consumer.resolve=null;
 consumer.reject=null;
 }
+if(reason!==undefined){
+this.#rejectPending(consumer,reason);
+}else{
+this.#resolvePendingDone(consumer);
+}
 consumer.detached=true;
 }
 this.#consumers.clear();
@@ -297,7 +312,7 @@ class BroadcastImpl {
 this.#ended =true;
 
 for(constconsumerofthis.#consumers){
-if(consumer.resolve){
+while(consumer.resolve){
 constbufferIndex=consumer.cursor-this.#bufferStart;
 if(bufferIndex<this.#buffer.length){
 constchunk=this.#buffer.get(bufferIndex);
@@ -310,9 +325,15 @@ class BroadcastImpl {
 consumer.resolve({__proto__: null,done: false,value: chunk});
 }else{
 consumer.resolve({__proto__: null,done: true,value: undefined});
+this.#resolvePendingDone(consumer);
+consumer.detached=true;
 }
 consumer.resolve=null;
 consumer.reject=null;
+if(consumer.detached&&this.#deleteConsumer(consumer)){
+this.#tryTrimBuffer();
+break;
+}
 }
 }
 }
@@ -329,6 +350,7 @@ class BroadcastImpl {
 consumer.resolve=null;
 consumer.reject=null;
 }
+this.#rejectPending(consumer,reason);
 consumer.detached=true;
 }
 this.#consumers.clear();
@@ -397,6 +419,11 @@ class BroadcastImpl {
 consumer.resolve=null;
 consumer.reject=null;
 resolve({__proto__: null,done: false,value: chunk});
+if(consumer.detached&&this.#deleteConsumer(consumer)){
+this.#tryTrimBuffer();
+}elseif(this.#promotePending(consumer)){
+ArrayPrototypePush(this.#waiters,consumer);
+}
 }else{
 // Still waiting -- put back
 ArrayPrototypePush(this.#waiters,consumer);
@@ -419,6 +446,31 @@ class BroadcastImpl {
 }
 returnfalse;
 }
+
+ #promotePending(consumer){
+constnext=ArrayPrototypeShift(consumer.pending);
+if(next===undefined)returnfalse;
+consumer.resolve=next.resolve;
+consumer.reject=next.reject;
+returntrue;
+}
+
+ #resolvePendingDone(consumer){
+if(consumer.resolve){
+consumer.resolve=null;
+consumer.reject=null;
+}
+while(consumer.pending.length>0){
+ArrayPrototypeShift(consumer.pending).resolve(
+{__proto__: null,done: true,value: undefined});
+}
+}
+
+ #rejectPending(consumer,reason){
+while(consumer.pending.length>0){
+ArrayPrototypeShift(consumer.pending).reject(reason);
+}
+}
 }
 
 // =============================================================================

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

@@ -3,6 +3,7 @@
 
 constcommon=require('../common');
 constassert=require('assert');
+const{ setTimeout }=require('timers/promises');
 const{ broadcast, text }=require('stream/iter');
 
 // =============================================================================
@@ -255,6 +256,38 @@ async function testLateJoinerSeesBufferedData() {
 assert.strictEqual(result,'before-join');
 }
 
+asyncfunctiontestOverlappingNextKeepsEarlierRead(){
+const{ writer,broadcast: bc}=broadcast();
+constit=bc.push()[Symbol.asyncIterator]();
+
+constfirst=it.next();
+constsecond=it.next();
+
+awaitwriter.write('x');
+
+constsecondResult=awaitPromise.race([
+second.then((value)=>({__proto__: null,settled: true, value })),
+setTimeout(common.platformTimeout(50),
+{__proto__: null,settled: false}),
+]);
+assert.deepStrictEqual(secondResult,{
+__proto__: null,
+settled: false,
+});
+
+constresult=awaitfirst;
+assert.strictEqual(result.done,false);
+assert.strictEqual(Buffer.concat(result.value).toString(),'x');
+
+writer.endSync();
+assert.deepStrictEqual(awaitsecond,{
+__proto__: null,
+done: true,
+value: undefined,
+});
+assert.strictEqual(bc.consumerCount,0);
+}
+
 Promise.all([
 testBasicBroadcast(),
 testMultipleWrites(),
@@ -270,4 +303,5 @@ Promise.all([
 testFailDetachesConsumers(),
 testWriterFailIdempotent(),
 testLateJoinerSeesBufferedData(),
+testOverlappingNextKeepsEarlierRead(),
 ]).then(common.mustCall());