stream: propagate abort reason in share and broadcast

trivikr · aduh95 · commit 0008d01f9c64 · 2026-05-27T11:35:43.000+02:00
Pass signal.reason to the multi-consumer cancel paths so signal abort is reported as AbortError instead of clean iterator completion. Also make detached share consumers rethrow a stored source error when they resume after cancellation, preserving the abort reason for pending pulls. Fixes: #63357 Signed-off-by: Kamat, Trivikram <16024985+trivikr@users.noreply.github.com> Assisted-by: openai:gpt-5.5 PR-URL: #63358 Fixes: #63357 Reviewed-By: Antoine du Hamel <duhamelantoine1995@gmail.com>
diff --git a/lib/internal/streams/iter/broadcast.js b/lib/internal/streams/iter/broadcast.js
@@ -686,6 +686,10 @@ function wireBroadcastWriteSignal(entry, signal, resolve, reject, self) {
   signal.addEventListener('abort', onAbort, { __proto__: null, once: true });
 }
 
+function onBroadcastCancel(broadcastImpl, signal) {
+  onSignalAbort(signal, () => broadcastImpl.cancel(signal.reason));
+}
+
 // =============================================================================
 // Public API
 // =============================================================================
@@ -720,7 +724,7 @@ function broadcast(options = { __proto__: null }) {
   broadcastImpl.setWriter(writer);
 
   if (signal) {
-    onSignalAbort(signal, () => broadcastImpl.cancel());
+    onBroadcastCancel(broadcastImpl, signal);
   }
 
   return { __proto__: null, writer, broadcast: broadcastImpl };
diff --git a/lib/internal/streams/iter/share.js b/lib/internal/streams/iter/share.js
@@ -144,6 +144,7 @@ class ShareImpl {
             // cursor must re-pull rather than terminating prematurely.
             for (;;) {
               if (state.detached) {
+                if (self.#sourceError) throw self.#sourceError;
                 return { __proto__: null, done: true, value: undefined };
               }
 
@@ -628,6 +629,10 @@ class SyncShareImpl {
   }
 }
 
+function onShareCancel(shareImpl, signal) {
+  onSignalAbort(signal, () => shareImpl.cancel(signal.reason));
+}
+
 // =============================================================================
 // Public API
 // =============================================================================
@@ -657,7 +662,7 @@ function share(source, options = { __proto__: null }) {
   const shareImpl = new ShareImpl(normalized, opts);
 
   if (signal) {
-    onSignalAbort(signal, () => shareImpl.cancel());
+    onShareCancel(shareImpl, signal);
   }
 
   return shareImpl;
diff --git a/test/parallel/test-stream-iter-broadcast-from.js b/test/parallel/test-stream-iter-broadcast-from.js
@@ -70,11 +70,12 @@ async function testAbortSignal() {
 
   ac.abort();
 
-  const batches = [];
-  for await (const batch of consumer) {
-    batches.push(batch);
-  }
-  assert.strictEqual(batches.length, 0);
+  await assert.rejects(async () => {
+    // eslint-disable-next-line no-unused-vars
+    for await (const _ of consumer) {
+      assert.fail('Should not reach here');
+    }
+  }, { name: 'AbortError' });
 }
 
 async function testAlreadyAbortedSignal() {
@@ -84,11 +85,12 @@ async function testAlreadyAbortedSignal() {
   const { broadcast: bc } = broadcast({ signal: ac.signal });
   const consumer = bc.push();
 
-  const batches = [];
-  for await (const batch of consumer) {
-    batches.push(batch);
-  }
-  assert.strictEqual(batches.length, 0);
+  await assert.rejects(async () => {
+    // eslint-disable-next-line no-unused-vars
+    for await (const _ of consumer) {
+      assert.fail('Should not reach here');
+    }
+  }, { name: 'AbortError' });
 }
 
 // =============================================================================
diff --git a/test/parallel/test-stream-iter-share-async.js b/test/parallel/test-stream-iter-share-async.js
@@ -134,16 +134,66 @@ async function testShareCancelWithReason() {
 
 async function testShareAbortSignal() {
   const ac = new AbortController();
-  const shared = share(from('data'), { signal: ac.signal });
-  const consumer = shared.pull();
+  const reason = new Error('share aborted');
+  const enc = new TextEncoder();
+  async function* source() {
+    yield [enc.encode('a')];
+    yield [enc.encode('b')];
+  }
+  const shared = share(source(), {
+    highWaterMark: 1,
+    backpressure: 'block',
+    signal: ac.signal,
+  });
+  const fast = shared.pull()[Symbol.asyncIterator]();
+  shared.pull();
+
+  await fast.next();
+  const read = fast.next();
+  const rejected = assert.rejects(read, (error) => error === reason);
+  ac.abort(reason);
+
+  await rejected;
+}
+
+async function testShareAbortSignalWhileSourcePullPending() {
+  const ac = new AbortController();
+  const {
+    promise: resumePromise,
+    resolve: resume,
+  } = Promise.withResolvers();
+  const {
+    promise: sourceStartedPromise,
+    resolve: sourceStarted,
+  } = Promise.withResolvers();
+
+  const source = {
+    __proto__: null,
+    [Symbol.asyncIterator]() {
+      return {
+        __proto__: null,
+        async next() {
+          sourceStarted();
+          await resumePromise;
+          return { __proto__: null, done: true, value: undefined };
+        },
+      };
+    },
+  };
+
+  const shared = share(source, { signal: ac.signal });
+  const iter1 = shared.pull()[Symbol.asyncIterator]();
+  const iter2 = shared.pull()[Symbol.asyncIterator]();
+  const read1 = iter1.next();
+  const read2 = iter2.next();
+  const rejected1 = assert.rejects(read1, { name: 'AbortError' });
+  const rejected2 = assert.rejects(read2, { name: 'AbortError' });
 
+  await sourceStartedPromise;
   ac.abort();
+  resume();
 
-  const batches = [];
-  for await (const batch of consumer) {
-    batches.push(batch);
-  }
-  assert.strictEqual(batches.length, 0);
+  await Promise.all([rejected1, rejected2]);
 }
 
 async function testShareAlreadyAborted() {
@@ -153,11 +203,12 @@ async function testShareAlreadyAborted() {
   const shared = share(from('data'), { signal: ac.signal });
   const consumer = shared.pull();
 
-  const batches = [];
-  for await (const batch of consumer) {
-    batches.push(batch);
-  }
-  assert.strictEqual(batches.length, 0);
+  await assert.rejects(async () => {
+    // eslint-disable-next-line no-unused-vars
+    for await (const _ of consumer) {
+      assert.fail('Should not reach here');
+    }
+  }, { name: 'AbortError' });
 }
 
 // =============================================================================
@@ -273,6 +324,7 @@ Promise.all([
   testShareCancelMidIteration(),
   testShareCancelWithReason(),
   testShareAbortSignal(),
+  testShareAbortSignalWhileSourcePullPending(),
   testShareAlreadyAborted(),
   testShareSourceError(),
   testShareLateJoiningConsumer(),

Original file line number	Diff line number	Diff line change
`@@ -144,6 +144,7 @@ class ShareImpl {`
`144`	`144`	`// cursor must re-pull rather than terminating prematurely.`
`145`	`145`	`for (;;) {`
`146`	`146`	`if (state.detached) {`
	`147`	`+ if (self.#sourceError) throw self.#sourceError;`
`147`	`148`	`return { __proto__: null, done: true, value: undefined };`
`148`	`149`	`}`
`149`	`150`
`@@ -628,6 +629,10 @@ class SyncShareImpl {`
`628`	`629`	`}`
`629`	`630`	`}`
`630`	`631`
	`632`	`+function onShareCancel(shareImpl, signal) {`
	`633`	`+ onSignalAbort(signal, () => shareImpl.cancel(signal.reason));`
	`634`	`+}`
	`635`	`+`
`631`	`636`	`// =============================================================================`
`632`	`637`	`// Public API`
`633`	`638`	`// =============================================================================`
`@@ -657,7 +662,7 @@ function share(source, options = { __proto__: null }) {`
`657`	`662`	`const shareImpl = new ShareImpl(normalized, opts);`
`658`	`663`
`659`	`664`	`if (signal) {`
`660`		`- onSignalAbort(signal, () => shareImpl.cancel());`
	`665`	`+ onShareCancel(shareImpl, signal);`
`661`	`666`	`}`
`662`	`667`
`663`	`668`	`return shareImpl;`