stream: reject push iterator.throw() with error

Keep the existing consumer cancellation side effects, but
reject the iterator.throw() call with the supplied error
instead of resolving with done: true.

Signed-off-by: Kamat, Trivikram <16024985+trivikr@users.noreply.github.com>
Assisted-by: openai:gpt-5.5
PR-URL: #64380
Fixes: #64378
Reviewed-By: James M Snell <jasnell@gmail.com>

Author: trivikr

lib/internal/streams/iter/push.js

@@ -668,7 +668,7 @@ function createReadable(queue) {
         },
         async throw(error) {
           queue.consumerThrow(error);
-          return { __proto__: null, done: true, value: undefined };
+          throw error;
         },
       };
     },

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

@@ -147,10 +147,16 @@ async function testOndrainRejectsOnConsumerThrow() {
   // Consumer throws via iterator.throw() before draining enough
   // to clear backpressure. The drain should reject.
   const iter = readable[Symbol.asyncIterator]();
-  await iter.throw(new Error('consumer error'));
+  const err = new Error('consumer error');
+  const drainRejects = assert.rejects(drainPromise, (e) => e === err);
+  const pendingWriteRejects = pendingWrite.catch(() => {});
+  await assert.rejects(
+    () => iter.throw(err),
+    (e) => e === err,
+  );
 
-  await assert.rejects(drainPromise, /consumer error/);
-  await pendingWrite.catch(() => {}); // Ignore write rejection
+  await drainRejects;
+  await pendingWriteRejects; // Ignore write rejection
 }
 
 async function testWritev() {
@@ -303,7 +309,11 @@ async function testConsumerThrowRejectsWrites() {
   writer.writeSync('a');
 
   const iter = readable[Symbol.asyncIterator]();
-  await iter.throw(new Error('consumer boom'));
+  const err = new Error('consumer boom');
+  await assert.rejects(
+    () => iter.throw(err),
+    (e) => e === err,
+  );
 
   // Subsequent async writes should reject with the consumer's error
   await assert.rejects(
@@ -312,6 +322,18 @@ async function testConsumerThrowRejectsWrites() {
   );
 }
 
+async function testConsumerThrowRejectsWithThrownError() {
+  const { readable } = push();
+
+  const iter = readable[Symbol.asyncIterator]();
+  const err = new Error('boom');
+
+  await assert.rejects(
+    () => iter.throw(err),
+    (e) => e === err,
+  );
+}
+
 // end() resolves a pending read as done:true
 async function testEndResolvesPendingRead() {
   const { writer, readable } = push();
@@ -373,14 +395,16 @@ async function testConsumerThrowRejectsPendingRead() {
   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(
+  const readRejects = assert.rejects(
     () => readPromise,
     (e) => e === err,
   );
+  await assert.rejects(
+    () => iter.throw(err),
+    (e) => e === err,
+  );
+
+  await readRejects;
 }
 
 // end() while writes are pending rejects those writes
@@ -527,6 +551,7 @@ Promise.all([
   testWriteUint8Array(),
   testOndrainWaitsForDrain(),
   testConsumerThrowRejectsWrites(),
+  testConsumerThrowRejectsWithThrownError(),
   testEndResolvesPendingRead(),
   testFailRejectsPendingRead(),
   testFailRejectsFutureReadWithFalsyReason(),