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: #63555
Backport-PR-URL: #64675
Fixes: #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(
       new ERR_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 {
       } else if (this.#writerState === 'errored' && this.#error) {
         const pending = this.#pendingReads.shift();
         pending.reject(this.#error);
+      } else if (this.#consumerState === 'returned') {
+        const pending = this.#pendingReads.shift();
+        pending.resolve({ __proto__: null, value: undefined, done: true });
       } else {
         break;
       }

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

@@ -326,6 +326,44 @@ async function testFailRejectsPendingRead() {
   );
 }
 
+// iterator.return() resolves a pending read with done:true
+async function testConsumerReturnResolvesPendingRead() {
+  const { readable } = push();
+
+  const iter = readable[Symbol.asyncIterator]();
+  const readPromise = iter.next();
+
+  await new Promise(setImmediate);
+
+  const returnResult = await iter.return();
+  assert.strictEqual(returnResult.value, undefined);
+  assert.strictEqual(returnResult.done, true);
+
+  const readResult = await readPromise;
+  assert.strictEqual(readResult.value, undefined);
+  assert.strictEqual(readResult.done, true);
+}
+
+// iterator.throw() rejects a pending read with the thrown error
+async function testConsumerThrowRejectsPendingRead() {
+  const { readable } = push();
+
+  const iter = readable[Symbol.asyncIterator]();
+  const readPromise = iter.next();
+
+  await new Promise(setImmediate);
+
+  const err = new Error('consumer read boom');
+  const throwResult = await iter.throw(err);
+  assert.strictEqual(throwResult.value, undefined);
+  assert.strictEqual(throwResult.done, true);
+
+  await assert.rejects(
+    () => readPromise,
+    (e) => e === err,
+  );
+}
+
 // end() while writes are pending rejects those writes
 async function testEndRejectsPendingWrites() {
   const { writer, readable } = push({ highWaterMark: 1, backpressure: 'block' });
@@ -438,6 +476,8 @@ Promise.all([
   testConsumerThrowRejectsWrites(),
   testEndResolvesPendingRead(),
   testFailRejectsPendingRead(),
+  testConsumerReturnResolvesPendingRead(),
+  testConsumerThrowRejectsPendingRead(),
   testEndRejectsPendingWrites(),
   testEndIdempotentWhenClosed(),
   testEndRejectsWhenErrored(),