diff --git a/lib/internal/streams/iter/consumers.js b/lib/internal/streams/iter/consumers.js index 8e18c1b5b930ea..3894b3fa24a8c4 100644 --- a/lib/internal/streams/iter/consumers.js +++ b/lib/internal/streams/iter/consumers.js @@ -426,8 +426,10 @@ function merge(...args) { if (normalized.length === 0) return; if (normalized.length === 1) { - for await (const batch of normalized[0]) { - signal?.throwIfAborted(); + const source = signal !== undefined && isAsyncIterable(sources[0]) ? + from(yieldAbortable(sources[0], signal)) : + yieldAbortable(normalized[0], signal); + for await (const batch of source) { yield batch; } return; diff --git a/lib/internal/streams/iter/utils.js b/lib/internal/streams/iter/utils.js index ec02eca43f3385..c86f317f490015 100644 --- a/lib/internal/streams/iter/utils.js +++ b/lib/internal/streams/iter/utils.js @@ -142,7 +142,10 @@ function yieldAbortable(source, signal) { if (!completed && typeof iterator.return === 'function') { const result = iterator.return(); if (aborted) { - markPromiseAsHandled(result); + // PromiseResolve(result) can reject if result is a thenable that + // rejects, so mark it as handled even though the abort takes + // precedence over the result of iterator.return(). + markPromiseAsHandled(PromiseResolve(result)); } else { await result; } diff --git a/test/parallel/test-stream-iter-consumers-merge.js b/test/parallel/test-stream-iter-consumers-merge.js index 859fcf8c1580fa..84aeb24b615915 100644 --- a/test/parallel/test-stream-iter-consumers-merge.js +++ b/test/parallel/test-stream-iter-consumers-merge.js @@ -170,6 +170,37 @@ async function testMergeSignalDuringPendingMultiSourceRead() { await assert.rejects(next, { name: 'AbortError' }); } +async function testMergeSignalDuringPendingSingleSourceRead() { + const ac = new AbortController(); + let returned = false; + const source = { + __proto__: null, + [Symbol.asyncIterator]() { + return this; + }, + next() { + // Intentionally never settle to verify that aborting interrupts a pending read. + return new Promise(() => {}); + }, + return() { + returned = true; + return { __proto__: null, done: true }; + }, + }; + + const iter = merge(source, { + __proto__: null, + signal: ac.signal, + })[Symbol.asyncIterator](); + + const next = iter.next(); + await new Promise(setImmediate); + ac.abort(); + + await assert.rejects(next, { name: 'AbortError' }); + assert.strictEqual(returned, true); +} + async function testMergeDoesNotDrainSourcesWhileIdle() { function source(n) { return { @@ -332,6 +363,7 @@ Promise.all([ testMergeConsumerBreak(), testMergeSignalMidIteration(), testMergeSignalDuringPendingMultiSourceRead(), + testMergeSignalDuringPendingSingleSourceRead(), testMergeDoesNotDrainSourcesWhileIdle(), testMergeStringSources(), testMergeObjectLikeSources(),