stream: reject pending reads on iterator throw

Settle reads that are already waiting when a push stream iterator is
returned or thrown. This prevents next() promises from hanging after
consumer cancellation.
Fixes: #63554
Signed-off-by: Kamat, Trivikram <16024985+trivikr@users.noreply.github.com>
Assisted-by: openai:gpt-5.5
PR-URL: #63555Fixes: #63554
Reviewed-By: James M Snell <jasnell@gmail.com>
Reviewed-By: Antoine du Hamel <duhamelantoine1995@gmail.com>

Author: trivikr

lib/internal/streams/iter/push.js

@@ -426,6 +426,7 @@ class PushQueue {
 if(this.#consumerState !=='active')return;
 this.#consumerState ='returned';
 this.#cleanup();
+this.#resolvePendingReads();
 this.#rejectPendingWrites(
 newERR_INVALID_STATE.TypeError('Stream closed by consumer'));
 // If closing, reject the pending end promise
@@ -443,7 +444,12 @@ class PushQueue {
 this.#consumerState ='thrown';
 this.#error =error;
 this.#cleanup();
+this.#rejectPendingReads(error);
 this.#rejectPendingWrites(error);
+if(this.#writerState ==='closing'&&this.#pendingEnd){
+this.#pendingEnd.reject(error);
+this.#pendingEnd =null;
+}
 // Reject pending drains - the consumer errored
 this.#rejectPendingDrains(error);
 }
@@ -485,6 +491,9 @@ class PushQueue {
 }elseif(this.#writerState ==='errored'&&this.#error){
 constpending=this.#pendingReads.shift();
 pending.reject(this.#error);
+}elseif(this.#consumerState ==='returned'){
+constpending=this.#pendingReads.shift();
+pending.resolve({__proto__: null,value: undefined,done: true});
 }else{
 break;
 }

test/parallel/test-stream-iter-push-writer.js

@@ -323,6 +323,44 @@ async function testFailRejectsPendingRead() {
 );
 }
 
+// iterator.return() resolves a pending read with done:true
+asyncfunctiontestConsumerReturnResolvesPendingRead(){
+const{ readable }=push();
+
+constiter=readable[Symbol.asyncIterator]();
+constreadPromise=iter.next();
+
+awaitnewPromise(setImmediate);
+
+constreturnResult=awaititer.return();
+assert.strictEqual(returnResult.value,undefined);
+assert.strictEqual(returnResult.done,true);
+
+constreadResult=awaitreadPromise;
+assert.strictEqual(readResult.value,undefined);
+assert.strictEqual(readResult.done,true);
+}
+
+// iterator.throw() rejects a pending read with the thrown error
+asyncfunctiontestConsumerThrowRejectsPendingRead(){
+const{ readable }=push();
+
+constiter=readable[Symbol.asyncIterator]();
+constreadPromise=iter.next();
+
+awaitnewPromise(setImmediate);
+
+consterr=newError('consumer read boom');
+constthrowResult=awaititer.throw(err);
+assert.strictEqual(throwResult.value,undefined);
+assert.strictEqual(throwResult.done,true);
+
+awaitassert.rejects(
+()=>readPromise,
+(e)=>e===err,
+);
+}
+
 // end() while writes are pending rejects those writes
 asyncfunctiontestEndRejectsPendingWrites(){
 const{ writer, readable }=push({highWaterMark: 1,backpressure: 'block'});
@@ -435,6 +473,8 @@ Promise.all([
 testConsumerThrowRejectsWrites(),
 testEndResolvesPendingRead(),
 testFailRejectsPendingRead(),
+testConsumerReturnResolvesPendingRead(),
+testConsumerThrowRejectsPendingRead(),
 testEndRejectsPendingWrites(),
 testEndIdempotentWhenClosed(),
 testEndRejectsWhenErrored(),