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 change
 // 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){
 constpullArgs=[...transforms];
 if(options?.signal){
 ArrayPrototypePush(pullArgs,
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`	`constrawConsumer=this.#createRawConsumer();`
`99`	`99`
`100`		`-if(transforms.length>0){`
	`100`	`+if(transforms.length>0\|\|options?.signal){`
`101`	`101`	`if(options){`
`102`	`102`	`returnpullWithTransforms(rawConsumer, ...transforms,options);`
`103`	`103`	`}`
-Original file line number
+Diff line change
 awaitPromise.all([rejected1,rejected2]);
+}
 +asyncfunctiontestSharePullAbortSignalRejectsPendingNext(){
 +constac=newAbortController();
 +constreason=newError('pull aborted');
 +constshared=share(
 +// eslint-disable-next-line require-yield
 +(asyncfunction*never(){
 +awaitnewPromise(()=>{});
 +})(),
 +);
 +constiter=shared.pull({signal: ac.signal})[Symbol.asyncIterator]();
++
 +constpendingNext=iter.next();
 +constrejected=assert.rejects(pendingNext,(error)=>error===reason);
 +ac.abort(reason);
++
 +awaitrejected;
 +shared.cancel();
 +}
++
 asyncfunctiontestShareAlreadyAborted(){
 constshared=share(from('data'),{signal: AbortSignal.abort()});
 constconsumer=shared.pull();
 testShareCancelWithReason(),
 testShareAbortSignal(),
 testShareAbortSignalWhileSourcePullPending(),
 +testSharePullAbortSignalRejectsPendingNext(),
 testShareAlreadyAborted(),
 testShareSourceError(),
 testShareLateJoiningConsumer(),