stream: respect iter consumer abort signals

trivikr · richardlau · commit f7adcd835992 · 2026-07-03T16:03:26.000Z
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>
diff --git a/lib/internal/streams/iter/broadcast.js b/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,
diff --git a/lib/internal/streams/iter/share.js b/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);
       }
diff --git a/test/parallel/test-stream-iter-broadcast-basic.js b/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(),
diff --git a/test/parallel/test-stream-iter-share-async.js b/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(),

Original file line number	Diff line number	Diff line change
`@@ -97,7 +97,7 @@ class ShareImpl {`
`97`	`97`	`const { transforms, options } = parsePullArgs(args);`
`98`	`98`	`const rawConsumer = this.#createRawConsumer();`
`99`	`99`
`100`		`- if (transforms.length > 0) {`
	`100`	`+ if (transforms.length > 0 \|\| options?.signal) {`
`101`	`101`	`if (options) {`
`102`	`102`	`return pullWithTransforms(rawConsumer, ...transforms, options);`
`103`	`103`	`}`