stream: serialize concurrent share consumer reads

Ensure overlapping next() calls on a single share() consumer resolve
in the same order they were requested.
Fixes: #63477
Signed-off-by: Kamat, Trivikram <16024985+trivikr@users.noreply.github.com>
Assisted-by: openai:gpt-5.5
PR-URL: #63478Fixes: #63477
Reviewed-By: James M Snell <jasnell@gmail.com>

Author: trivikr

lib/internal/streams/iter/share.js

@@ -113,6 +113,7 @@ class ShareImpl {
 resolve: null,
 reject: null,
 detached: false,
+pendingNext: PromiseResolve(),
 };
 
 this.#consumers.add(state);
@@ -129,62 +130,72 @@ class ShareImpl {
 return{
 __proto__: null,
 [SymbolAsyncIterator](){
-return{
-__proto__: null,
-asyncnext(){
-if(self.#sourceError){
-state.detached=true;
-self.#consumers.delete(state);
-throwself.#sourceError;
+constgetNext=async()=>{
+if(self.#sourceError){
+state.detached=true;
+self.#consumers.delete(state);
+throwself.#sourceError;
+}
+
+// Loop until we get data, source is exhausted, or
+// consumer is detached. Multiple consumers may be woken
+// after a single pull - those that find no data at their
+// cursor must re-pull rather than terminating prematurely.
+for(;;){
+if(state.detached){
+if(self.#sourceError)throwself.#sourceError;
+return{__proto__: null,done: true,value: undefined};
 }
 
-// Loop until we get data, source is exhausted, or
-// consumer is detached. Multiple consumers may be woken
-// after a single pull - those that find no data at their
-// cursor must re-pull rather than terminating prematurely.
-for(;;){
-if(state.detached){
-if(self.#sourceError)throwself.#sourceError;
-return{__proto__: null,done: true,value: undefined};
-}
+if(self.#cancelled){
+state.detached=true;
+self.#deleteConsumer(state);
+return{__proto__: null,done: true,value: undefined};
+}
 
-if(self.#cancelled){
-state.detached=true;
-self.#deleteConsumer(state);
-return{__proto__: null,done: true,value: undefined};
+// Check if data is available in buffer
+constbufferIndex=state.cursor-self.#bufferStart;
+if(bufferIndex<self.#buffer.length){
+constchunk=self.#buffer.get(bufferIndex);
+constcursor=state.cursor;
+state.cursor++;
+if(cursor===self.#cachedMinCursor &&
+--self.#cachedMinCursorConsumers ===0){
+self.#tryTrimBuffer();
 }
+return{__proto__: null,done: false,value: chunk};
+}
 
-// Check if data is available in buffer
-constbufferIndex=state.cursor-self.#bufferStart;
-if(bufferIndex<self.#buffer.length){
-constchunk=self.#buffer.get(bufferIndex);
-constcursor=state.cursor;
-state.cursor++;
-if(cursor===self.#cachedMinCursor &&
---self.#cachedMinCursorConsumers ===0){
-self.#tryTrimBuffer();
-}
-return{__proto__: null,done: false,value: chunk};
-}
+if(self.#sourceExhausted){
+state.detached=true;
+self.#deleteConsumer(state);
+if(self.#sourceError)throwself.#sourceError;
+return{__proto__: null,done: true,value: undefined};
+}
 
-if(self.#sourceExhausted){
-state.detached=true;
-self.#deleteConsumer(state);
-if(self.#sourceError)throwself.#sourceError;
-return{__proto__: null,done: true,value: undefined};
-}
+// Need to pull from source - check buffer limit
+constcanPull=awaitself.#waitForBufferSpace();
+if(!canPull){
+state.detached=true;
+self.#deleteConsumer(state);
+if(self.#sourceError)throwself.#sourceError;
+return{__proto__: null,done: true,value: undefined};
+}
 
-// Need to pull from source - check buffer limit
-constcanPull=awaitself.#waitForBufferSpace();
-if(!canPull){
-state.detached=true;
-self.#deleteConsumer(state);
-if(self.#sourceError)throwself.#sourceError;
-return{__proto__: null,done: true,value: undefined};
-}
+awaitself.#pullFromSource();
+}
+};
 
-awaitself.#pullFromSource();
-}
+return{
+__proto__: null,
+next(){
+constnext=PromisePrototypeThen(
+state.pendingNext,
+getNext,
+getNext);
+state.pendingNext=
+PromisePrototypeThen(next,undefined,()=>{});
+returnnext;
 },
 
 asyncreturn(){

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

@@ -306,6 +306,24 @@ async function testShareMultipleConsumersConcurrentPull() {
 assert.strictEqual(t3,expected);
 }
 
+asyncfunctiontestShareConsumerConcurrentNextCalls(){
+asyncfunction*source(){
+constenc=newTextEncoder();
+yield[enc.encode('first')];
+yield[enc.encode('second')];
+}
+
+constshared=share(source());
+constit=shared.pull()[Symbol.asyncIterator]();
+constfirst=it.next();
+constsecond=it.next();
+
+const[r1,r2]=awaitPromise.all([first,second]);
+constdec=newTextDecoder();
+assert.strictEqual(dec.decode(r1.value[0]),'first');
+assert.strictEqual(dec.decode(r2.value[0]),'second');
+}
+
 // share() accepts string source directly (normalized via from())
 asyncfunctiontestShareStringSource(){
 constshared=share('hello-share');
@@ -327,5 +345,6 @@ Promise.all([
 testShareLateJoiningConsumer(),
 testShareConsumerBreak(),
 testShareMultipleConsumersConcurrentPull(),
+testShareConsumerConcurrentNextCalls(),
 testShareStringSource(),
 ]).then(common.mustCall());