stream: avoid draining merged iter sources

Defer reading the next value from each source in merge() until the
merged consumer resumes after receiving the previous value.

Signed-off-by: Kamat, Trivikram <16024985+trivikr@users.noreply.github.com>
Assisted-by: openai:gpt-5.5
PR-URL: #64293
Fixes: #63566
Reviewed-By: James M Snell <jasnell@gmail.com>
Reviewed-By: Gürgün Dayıoğlu <hey@gurgun.day>

Author: trivikr

lib/internal/streams/iter/consumers.js

@@ -463,20 +463,11 @@ function merge(...args) {
         if (result.done) {
           activeCount--;
         } else {
-          ArrayPrototypePush(ready, result.value);
-          // Immediately request the next value from this source
-          // (at most one pending .next() per source)
-          PromisePrototypeThen(
-            iterator.next(),
-            (r) => onSettled(iterator, r),
-            (err) => {
-              ArrayPrototypePush(ready, { __proto__: null, error: err });
-              if (waitResolve) {
-                waitResolve();
-                waitResolve = null;
-              }
-            },
-          );
+          ArrayPrototypePush(ready, {
+            __proto__: null,
+            iterator,
+            value: result.value,
+          });
         }
         if (waitResolve) {
           waitResolve();
@@ -513,7 +504,18 @@ function merge(...args) {
             if (item?.error) {
               throw item.error;
             }
-            yield item;
+            yield item.value;
+            PromisePrototypeThen(
+              item.iterator.next(),
+              (r) => onSettled(item.iterator, r),
+              (err) => {
+                ArrayPrototypePush(ready, { __proto__: null, error: err });
+                if (waitResolve) {
+                  waitResolve();
+                  waitResolve = null;
+                }
+              },
+            );
           }
 
           // If sources are still active, wait for the next settlement

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

@@ -170,6 +170,32 @@ async function testMergeSignalDuringPendingMultiSourceRead() {
   await assert.rejects(next, { name: 'AbortError' });
 }
 
+async function testMergeDoesNotDrainSourcesWhileIdle() {
+  function source(n) {
+    return {
+      __proto__: null,
+      pulls: 0,
+      async *[Symbol.asyncIterator]() {
+        while (this.pulls < n) {
+          yield [Buffer.from(`${++this.pulls}`)];
+        }
+      },
+    };
+  }
+
+  const a = source(5);
+  const b = source(5);
+  const iterator = merge(a, b)[Symbol.asyncIterator]();
+
+  await iterator.next();
+  await new Promise(setImmediate);
+
+  assert.strictEqual(a.pulls, 1);
+  assert.strictEqual(b.pulls, 1);
+
+  await iterator.return?.();
+}
+
 // merge() accepts string sources (normalized via from())
 async function testMergeStringSources() {
   const batches = [];
@@ -306,6 +332,7 @@ Promise.all([
   testMergeConsumerBreak(),
   testMergeSignalMidIteration(),
   testMergeSignalDuringPendingMultiSourceRead(),
+  testMergeDoesNotDrainSourcesWhileIdle(),
   testMergeStringSources(),
   testMergeObjectLikeSources(),
   testMergeCleanupErrorOnly(),