nodejs
diff --git a/‎lib/internal/streams/iter/push.js‎
Lines changed: 9 additions & 0 deletions b/‎lib/internal/streams/iter/push.js‎
Lines changed: 9 additions & 0 deletions
diff --git a/‎test/parallel/test-stream-iter-push-writer.js‎
Lines changed: 40 additions & 0 deletions b/‎test/parallel/test-stream-iter-push-writer.js‎
Lines changed: 40 additions & 0 deletions
@@ -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;
       }
 
@@ -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(),