stream: abort pending single-source merge reads

Make the single-source merge path abort-aware so signal cancellation
rejects a pending read and closes the active source iterator.

Handle synchronous iterator return values when marking aborted cleanup
as handled.

Signed-off-by: Kamat, Trivikram <16024985+trivikr@users.noreply.github.com>
PR-URL: #64445
Fixes: #64444
Reviewed-By: James M Snell <jasnell@gmail.com>
Reviewed-By: Matteo Collina <matteo.collina@gmail.com>
Reviewed-By: Ethan Arrowood <ethan@arrowood.dev>

Author: trivikr

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;

lib/internal/streams/iter/utils.js

@@ -133,7 +133,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;
           }

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(),