stream: fix nested async flushing with infinite sources

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

Author: jasnell

doc/api/stream_iter.md

@@ -545,7 +545,8 @@ added: v25.9.0
 
 Create an async byte stream from the given input. Strings are UTF-8 encoded.
 `ArrayBuffer` and `ArrayBufferView` values are wrapped as `Uint8Array`. Arrays
-and iterables in `input` are recursively flattened and normalized.
+and iterables in `input` are recursively flattened and normalized. Flattened
+values may be split across implementation-defined bounded batches.
 
 Objects implementing `Symbol.for('Stream.toAsyncStreamable')` or
 `Symbol.for('Stream.toStreamable')` are converted via those protocols. The

lib/internal/streams/iter/from.js

@@ -363,9 +363,13 @@ async function* normalizeAsyncSource(source) {
         continue;
       }
       // Slow path: normalize the value
-      const batch = [];
+      let batch = [];
       for await (const chunk of normalizeAsyncValue(value)) {
         ArrayPrototypePush(batch, chunk);
+        if (batch.length === FROM_BATCH_SIZE) {
+          yield batch;
+          batch = [];
+        }
       }
       if (batch.length > 0) {
         yield batch;

test/parallel/test-stream-iter-from-async.js

@@ -31,6 +31,30 @@ async function testFromAsyncGenerator() {
   assert.deepStrictEqual(batches[1][0], new Uint8Array([30, 40]));
 }
 
+async function testFromBoundsNestedAsyncIterable() {
+  let nestedClosed = false;
+  async function* nested() {
+    try {
+      let value = 0;
+      while (true) yield new Uint8Array([value++]);
+    } finally {
+      nestedClosed = true;
+    }
+  }
+
+  async function* source() {
+    yield nested();
+  }
+
+  const iterator = from(source())[Symbol.asyncIterator]();
+  const first = await iterator.next();
+  assert.strictEqual(first.done, false);
+  assert.strictEqual(first.value.length, 128);
+
+  await iterator.return();
+  assert.strictEqual(nestedClosed, true);
+}
+
 async function testFromSyncIterableAsAsync() {
   // Sync iterable passed to from() should work
   function* gen() {
@@ -274,6 +298,7 @@ function testFromUndefinedThrows() {
 Promise.all([
   testFromString(),
   testFromAsyncGenerator(),
+  testFromBoundsNestedAsyncIterable(),
   testFromSyncIterableAsAsync(),
   testFromSyncIterableAwaitsPromiseValues(),
   testFromSyncIterableRejectsNestedAsyncIterable(),