stream: respect iter consumer abort signals

Delegate broadcast.push() and share.pull() calls with per-consumer
AbortSignals through pull(), so pending next() calls reject when the
signal aborts.

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

Author: trivikr

lib/internal/streams/iter/broadcast.js

@@ -128,7 +128,7 @@ class BroadcastImpl {
     // own internal AbortController that follows the external signal.
     // When no transforms, return rawConsumer directly (controller elided
     // per PULL-02 optimization -- no transforms means no signal recipient).
-    if (transforms.length > 0) {
+    if (transforms.length > 0 || options?.signal) {
       const pullArgs = [...transforms];
       if (options?.signal) {
         ArrayPrototypePush(pullArgs,

lib/internal/streams/iter/share.js

@@ -97,7 +97,7 @@ class ShareImpl {
     const { transforms, options } = parsePullArgs(args);
     const rawConsumer = this.#createRawConsumer();
 
-    if (transforms.length > 0) {
+    if (transforms.length > 0 || options?.signal) {
       if (options) {
         return pullWithTransforms(rawConsumer, ...transforms, options);
       }

test/parallel/test-stream-iter-broadcast-basic.js

@@ -174,6 +174,19 @@ async function testPendingNextSettlesAfterReturn() {
   assert.strictEqual(result.value, undefined);
 }
 
+async function testPushAbortSignalRejectsPendingNext() {
+  const ac = new AbortController();
+  const reason = new Error('push aborted');
+  const { broadcast: bc } = broadcast();
+  const iter = bc.push({ signal: ac.signal })[Symbol.asyncIterator]();
+
+  const pendingNext = iter.next();
+  const rejected = assert.rejects(pendingNext, (error) => error === reason);
+  ac.abort(reason);
+
+  await rejected;
+}
+
 // =============================================================================
 // Writer fail detaches consumers
 // =============================================================================
@@ -300,6 +313,7 @@ Promise.all([
   testCancelWithReason(),
   testCancelWithFalsyReason(),
   testPendingNextSettlesAfterReturn(),
+  testPushAbortSignalRejectsPendingNext(),
   testFailDetachesConsumers(),
   testWriterFailIdempotent(),
   testLateJoinerSeesBufferedData(),

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

@@ -196,6 +196,25 @@ async function testShareAbortSignalWhileSourcePullPending() {
   await Promise.all([rejected1, rejected2]);
 }
 
+async function testSharePullAbortSignalRejectsPendingNext() {
+  const ac = new AbortController();
+  const reason = new Error('pull aborted');
+  const shared = share(
+    // eslint-disable-next-line require-yield
+    (async function* never() {
+      await new Promise(() => {});
+    })(),
+  );
+  const iter = shared.pull({ signal: ac.signal })[Symbol.asyncIterator]();
+
+  const pendingNext = iter.next();
+  const rejected = assert.rejects(pendingNext, (error) => error === reason);
+  ac.abort(reason);
+
+  await rejected;
+  shared.cancel();
+}
+
 async function testShareAlreadyAborted() {
   const shared = share(from('data'), { signal: AbortSignal.abort() });
   const consumer = shared.pull();
@@ -340,6 +359,7 @@ Promise.all([
   testShareCancelWithReason(),
   testShareAbortSignal(),
   testShareAbortSignalWhileSourcePullPending(),
+  testSharePullAbortSignalRejectsPendingNext(),
   testShareAlreadyAborted(),
   testShareSourceError(),
   testShareLateJoiningConsumer(),