stream: fix merge settlement tagging and falsy error tracking

Signed-off-by: James M Snell <jasnell@gmail.com>
Assisted-by: Opencode
PR-URL: #65652
Reviewed-By: Trivikram Kamat <trivikr.dev@gmail.com>
Reviewed-By: Matteo Collina <matteo.collina@gmail.com>

Author: jasnell

lib/internal/streams/iter/consumers.js

@@ -390,6 +390,8 @@ function ondrain(drainable) {
 // Merge Utility
 // =============================================================================
 
+const kNoMergeError = { __proto__: null };
+
 /**
  * Merge multiple async iterables by yielding values in temporal order.
  * @param {...(AsyncIterable<Uint8Array[]>|object)} args
@@ -440,6 +442,7 @@ function merge(...args) {
       let activeCount = normalized.length;
       let waitResolve = null;
       let onAbort;
+      let stopped = false;
 
       if (signal) {
         onAbort = () => {
@@ -457,11 +460,13 @@ function merge(...args) {
       // Called when a source's .next() settles. Pushes the result into
       // the ready queue and wakes the consumer if it's waiting.
       const onSettled = (iterator, result) => {
+        if (stopped) return;
         if (result.done) {
           activeCount--;
         } else {
           ArrayPrototypePush(ready, {
             __proto__: null,
+            kind: 'value',
             iterator,
             value: result.value,
           });
@@ -472,6 +477,19 @@ function merge(...args) {
         }
       };
 
+      const onRejected = (reason) => {
+        if (stopped) return;
+        ArrayPrototypePush(ready, {
+          __proto__: null,
+          kind: 'error',
+          reason,
+        });
+        if (waitResolve) {
+          waitResolve();
+          waitResolve = null;
+        }
+      };
+
       // Start one .next() per source
       const iterators = [];
       for (let i = 0; i < normalized.length; i++) {
@@ -480,38 +498,27 @@ function merge(...args) {
         PromisePrototypeThen(
           iterator.next(),
           (r) => onSettled(iterator, r),
-          (err) => {
-            ArrayPrototypePush(ready, { __proto__: null, error: err });
-            if (waitResolve) {
-              waitResolve();
-              waitResolve = null;
-            }
-          },
+          onRejected,
         );
       }
 
-      let primaryError;
+      let completed = false;
+      let primaryError = kNoMergeError;
       try {
         while (activeCount > 0 || ready.length > 0) {
           signal?.throwIfAborted();
 
           // Drain ready queue synchronously
           while (ready.length > 0) {
             const item = ArrayPrototypeShift(ready);
-            if (item?.error) {
-              throw item.error;
+            if (item.kind === 'error') {
+              throw item.reason;
             }
             yield item.value;
             PromisePrototypeThen(
               item.iterator.next(),
               (r) => onSettled(item.iterator, r),
-              (err) => {
-                ArrayPrototypePush(ready, { __proto__: null, error: err });
-                if (waitResolve) {
-                  waitResolve();
-                  waitResolve = null;
-                }
-              },
+              onRejected,
             );
           }
 
@@ -526,9 +533,11 @@ function merge(...args) {
             });
           }
         }
+        completed = true;
       } catch (err) {
         primaryError = err;
       } finally {
+        stopped = true;
         if (onAbort !== undefined) {
           signal.removeEventListener('abort', onAbort);
         }
@@ -538,15 +547,15 @@ function merge(...args) {
         await cleanupIterators(
           iterators,
           primaryError,
-          signal?.aborted && primaryError === signal.reason,
+          !completed,
         );
       }
     },
   };
 }
 
 async function cleanupIterators(iterators, primaryError, skipAwaitCleanup) {
-  let cleanupError;
+  let cleanupError = kNoMergeError;
   await SafePromiseAllReturnVoid(iterators, async (iterator) => {
     if (iterator.return) {
       try {
@@ -558,12 +567,12 @@ async function cleanupIterators(iterators, primaryError, skipAwaitCleanup) {
         }
       } catch (err) {
         // Keep the first cleanup error encountered.
-        cleanupError ??= err;
+        if (cleanupError === kNoMergeError) cleanupError = err;
       }
     }
   });
-  if (cleanupError !== undefined) {
-    if (primaryError !== undefined) {
+  if (cleanupError !== kNoMergeError) {
+    if (primaryError !== kNoMergeError) {
       // Both a primary error and a cleanup error occurred.
       // Wrap in SuppressedError so neither is lost:
       // .error = primaryError, .suppressed = cleanupError.
@@ -573,7 +582,7 @@ async function cleanupIterators(iterators, primaryError, skipAwaitCleanup) {
     // No primary error - the cleanup error is the only error.
     throw cleanupError;
   }
-  if (primaryError !== undefined) {
+  if (primaryError !== kNoMergeError) {
     throw primaryError;
   }
 }

test/parallel/test-stream-iter-consumers-merge.js

@@ -108,6 +108,109 @@ async function testMergeSourceError() {
   );
 }
 
+async function testMergeFalsySourceErrors() {
+  const reasons = [undefined, null, false, 0, '', NaN];
+
+  for (const reason of reasons) {
+    const noError = { __proto__: null };
+    let actual = noError;
+    try {
+      await text(merge(rejectedSource(reason), from('other')));
+    } catch (error) {
+      actual = error;
+    }
+    assert.strictEqual(Object.is(actual, reason), true);
+  }
+}
+
+function rejectedSource(reason) {
+  return {
+    __proto__: null,
+    [Symbol.asyncIterator]() {
+      return this;
+    },
+    next() {
+      return Promise.reject(reason);
+    },
+  };
+}
+
+function pendingSource() {
+  return {
+    __proto__: null,
+    [Symbol.asyncIterator]() {
+      return this;
+    },
+    next() {
+      return new Promise(() => {});
+    },
+    return() {
+      return new Promise(() => {});
+    },
+  };
+}
+
+async function testMergeSourceErrorDoesNotAwaitCleanup() {
+  const reason = new Error('source failed');
+
+  const timedOut = { __proto__: null };
+  const outcome = await Promise.race([
+    text(merge(rejectedSource(reason), pendingSource())).then(
+      () => ({ __proto__: null, status: 'fulfilled' }),
+      (error) => ({ __proto__: null, status: 'rejected', error }),
+    ),
+    new Promise((resolve) => setImmediate(resolve, timedOut)),
+  ]);
+
+  assert.notStrictEqual(outcome, timedOut);
+  assert.strictEqual(outcome.status, 'rejected');
+  assert.strictEqual(outcome.error, reason);
+}
+
+async function testMergeBreakDoesNotAwaitCleanup() {
+  async function* readySource() {
+    yield [Uint8Array.of(1)];
+  }
+
+  const timedOut = { __proto__: null };
+  const outcome = await Promise.race([
+    (async () => {
+      for await (const batch of merge(readySource(), pendingSource())) {
+        assert.deepStrictEqual(batch, [Uint8Array.of(1)]);
+        break;
+      }
+      return true;
+    })(),
+    new Promise((resolve) => setImmediate(resolve, timedOut)),
+  ]);
+
+  assert.strictEqual(outcome, true);
+}
+
+async function testMergeNaNAbortDoesNotAwaitCleanup() {
+  const ac = new AbortController();
+  const iterator = merge(pendingSource(), pendingSource(), {
+    __proto__: null,
+    signal: ac.signal,
+  })[Symbol.asyncIterator]();
+  const next = iterator.next();
+  await new Promise(setImmediate);
+  ac.abort(NaN);
+
+  const timedOut = { __proto__: null };
+  const outcome = await Promise.race([
+    next.then(
+      () => ({ __proto__: null, status: 'fulfilled' }),
+      (error) => ({ __proto__: null, status: 'rejected', error }),
+    ),
+    new Promise((resolve) => setImmediate(resolve, timedOut)),
+  ]);
+
+  assert.notStrictEqual(outcome, timedOut);
+  assert.strictEqual(outcome.status, 'rejected');
+  assert.strictEqual(Object.is(outcome.error, NaN), true);
+}
+
 async function testMergeConsumerBreak() {
   let source1Return = false;
   let source2Return = false;
@@ -296,9 +399,8 @@ async function testMergeCleanupErrorOnly() {
   );
 }
 
-// Primary error + cleanup error: a source throws during iteration AND
-// iterator.return() also throws. Should get a SuppressedError.
-async function testMergePrimaryAndCleanupError() {
+// A primary source error must not wait for asynchronous cleanup failures.
+async function testMergePrimaryErrorPrecedesCleanupError() {
   async function* badSource() {
     yield [new TextEncoder().encode('x')];
     throw new Error('primary boom');
@@ -319,15 +421,7 @@ async function testMergePrimaryAndCleanupError() {
         // Consume until error
       }
     },
-    (err) => {
-      assert.ok(
-        err instanceof SuppressedError,
-        `Expected SuppressedError, got ${err.constructor.name}`,
-      );
-      assert.strictEqual(err.error.message, 'primary boom');
-      assert.strictEqual(err.suppressed.message, 'cleanup boom');
-      return true;
-    },
+    { message: 'primary boom' },
   );
 }
 
@@ -360,6 +454,10 @@ Promise.all([
   testMergeWithAbortSignal(),
   testMergeSyncSources(),
   testMergeSourceError(),
+  testMergeFalsySourceErrors(),
+  testMergeSourceErrorDoesNotAwaitCleanup(),
+  testMergeBreakDoesNotAwaitCleanup(),
+  testMergeNaNAbortDoesNotAwaitCleanup(),
   testMergeConsumerBreak(),
   testMergeSignalMidIteration(),
   testMergeSignalDuringPendingMultiSourceRead(),
@@ -368,6 +466,6 @@ Promise.all([
   testMergeStringSources(),
   testMergeObjectLikeSources(),
   testMergeCleanupErrorOnly(),
-  testMergePrimaryAndCleanupError(),
+  testMergePrimaryErrorPrecedesCleanupError(),
   testMergeBreakWithCleanupError(),
 ]).then(common.mustCall());