stream: fix merge abort for pending sources

trivikr · aduh95 · commit d3fa77c5e29f · 2026-07-06T10:03:33.000+02:00
Wake the multi-source merge loop when its abort signal fires so a pending read rejects instead of waiting for another source to settle. When the primary error is the abort reason, do not await iterator cleanup because sources may already be stuck in pending next() calls. Fixes: #64012 Signed-off-by: Kamat, Trivikram <16024985+trivikr@users.noreply.github.com> Assisted-by: openai:gpt-5.5 PR-URL: #64013 Fixes: #64012 Reviewed-By: James M Snell <jasnell@gmail.com>
diff --git a/lib/internal/streams/iter/consumers.js b/lib/internal/streams/iter/consumers.js
@@ -39,6 +39,10 @@ const {
 validateObject,
 } = require('internal/validators');
 
+const {
+ markPromiseAsHandled,
+} = internalBinding('util');
+
 const {
 from,
 fromSync,
@@ -434,6 +438,20 @@ function merge(...args) {
 const ready = [];
 let activeCount = normalized.length;
 let waitResolve = null;
+ let onAbort;
+
+ if (signal) {
+ onAbort = () => {
+ if (waitResolve) {
+ waitResolve();
+ waitResolve = null;
+ }
+ };
+ signal.addEventListener('abort', onAbort, {
+ __proto__: null,
+ once: true,
+ });
+ }
 
 // Called when a source's .next() settles. Pushes the result into
 // the ready queue and wakes the consumer if it's waiting.
@@ -498,27 +516,43 @@ function merge(...args) {
 if (activeCount > 0) {
 await new Promise((resolve) => {
 waitResolve = resolve;
+ if (signal?.aborted) {
+ waitResolve = null;
+ resolve();
+ }
 });
 }
 }
 } catch (err) {
 primaryError = err;
 } finally {
+ if (onAbort !== undefined) {
+ signal.removeEventListener('abort', onAbort);
+ }
 // Clean up: return all iterators. Cleanup errors are not
 // swallowed - a broken iterator.return() (e.g., failing to
 // release a resource) should be visible to the caller.
- await cleanupIterators(iterators, primaryError);
+ await cleanupIterators(
+ iterators,
+ primaryError,
+ signal?.aborted && primaryError === signal.reason,
+ );
 }
 },
 };
 }
 
-async function cleanupIterators(iterators, primaryError) {
+async function cleanupIterators(iterators, primaryError, skipAwaitCleanup) {
 let cleanupError;
 await SafePromiseAllReturnVoid(iterators, async (iterator) => {
 if (iterator.return) {
 try {
- await iterator.return();
+ const result = iterator.return();
+ if (skipAwaitCleanup) {
+ markPromiseAsHandled(result);
+ } else {
+ await result;
+ }
 } catch (err) {
 // Keep the first cleanup error encountered.
 cleanupError ??= err;
diff --git a/test/parallel/test-stream-iter-consumers-merge.js b/test/parallel/test-stream-iter-consumers-merge.js
@@ -151,6 +151,25 @@ async function testMergeSignalMidIteration() {
 await assert.rejects(() => iter.next(), { name: 'AbortError' });
 }
 
+async function testMergeSignalDuringPendingMultiSourceRead() {
+ const ac = new AbortController();
+
+ async function* pending() {
+ await new Promise(() => {});
+ yield [];
+ }
+
+ const iter = merge(pending(), pending(), {
+ __proto__: null,
+ signal: ac.signal,
+ })[Symbol.asyncIterator]();
+
+ const next = iter.next();
+ ac.abort();
+
+ await assert.rejects(next, { name: 'AbortError' });
+}
+
 // merge() accepts string sources (normalized via from())
 async function testMergeStringSources() {
 const batches = [];
@@ -286,6 +305,7 @@ Promise.all([
 testMergeSourceError(),
 testMergeConsumerBreak(),
 testMergeSignalMidIteration(),
+ testMergeSignalDuringPendingMultiSourceRead(),
 testMergeStringSources(),
 testMergeObjectLikeSources(),
 testMergeCleanupErrorOnly(),

-Original file line number
+Diff line change
  validateObject,
 }=require('internal/validators');
 +const{
 + markPromiseAsHandled,
 +}=internalBinding('util');
++
 const{
  from,
  fromSync,
 constready=[];
 letactiveCount=normalized.length;
 letwaitResolve=null;
 +letonAbort;
++
 +if(signal){
 +onAbort=()=>{
 +if(waitResolve){
 +waitResolve();
 +waitResolve=null;
 +}
 +};
 +signal.addEventListener('abort',onAbort,{
 +__proto__: null,
 +once: true,
 +});
 +}
 // Called when a source's .next() settles. Pushes the result into
 // the ready queue and wakes the consumer if it's waiting.
 if(activeCount>0){
 awaitnewPromise((resolve)=>{
 waitResolve=resolve;
 +if(signal?.aborted){
 +waitResolve=null;
 +resolve();
 +}
 });
+}
+}
 }catch(err){
 primaryError=err;
 }finally{
 +if(onAbort!==undefined){
 +signal.removeEventListener('abort',onAbort);
 +}
 // Clean up: return all iterators. Cleanup errors are not
 // swallowed - a broken iterator.return() (e.g., failing to
 // release a resource) should be visible to the caller.
 -awaitcleanupIterators(iterators,primaryError);
 +awaitcleanupIterators(
 +iterators,
 +primaryError,
 +signal?.aborted&&primaryError===signal.reason,
 +);
+}
 },
 };
+}
 -asyncfunctioncleanupIterators(iterators,primaryError){
 +asyncfunctioncleanupIterators(iterators,primaryError,skipAwaitCleanup){
 letcleanupError;
 awaitSafePromiseAllReturnVoid(iterators,async(iterator)=>{
 if(iterator.return){
 try{
 -awaititerator.return();
 +constresult=iterator.return();
 +if(skipAwaitCleanup){
 +markPromiseAsHandled(result);
 +}else{
 +awaitresult;
 +}
 }catch(err){
 // Keep the first cleanup error encountered.
 cleanupError??=err;