stream: apply more stream/iter cleanups

Author: jasnell

lib/internal/streams/iter/from.js

@@ -555,12 +555,7 @@ function from(input) {
     );
   }
 
-  return {
-    __proto__: null,
-    async *[SymbolAsyncIterator]() {
-      yield* normalizeAsyncSource(input);
-    },
-  };
+  return normalizeAsyncSource(input);
 }
 
 // =============================================================================

lib/internal/streams/iter/pull.js

@@ -10,6 +10,7 @@ const {
   ArrayBufferIsView,
   ArrayPrototypePush,
   ArrayPrototypeSlice,
+  PromisePrototypeThen,
   SymbolAsyncIterator,
   SymbolIterator,
   TypedArrayPrototypeGetByteLength,
@@ -350,7 +351,21 @@ function* applyFusedStatelessSyncTransforms(source, run) {
       current = result;
     }
     if (current === null) continue;
-    yield* processTransformResultSync(current);
+    // Inline normalization with Uint8Array[] batch as the fast path,
+    // matching the async pipeline's check order.
+    if (isUint8ArrayBatch(current)) {
+      if (current.length > 0) yield current;
+    } else if (isUint8Array(current)) {
+      yield [current];
+    } else if (typeof current === 'string') {
+      yield [toUint8Array(current)];
+    } else if (isAnyArrayBuffer(current)) {
+      yield [new Uint8Array(current)];
+    } else if (ArrayBufferIsView(current)) {
+      yield [primitiveToUint8Array(current)];
+    } else {
+      yield* processTransformResultSync(current);
+    }
   }
   // Flush
   let current = null;
@@ -394,8 +409,7 @@ function* applyStatefulSyncTransform(source, transform) {
  * @yields {Uint8Array[]}
  */
 function* createSyncPipeline(source, transforms) {
-  // Normalize source via fromSync()
-  let current = fromSync(source);
+  let current = source;
 
   // Apply transforms - fuse consecutive stateless transforms into a single
   // generator layer to avoid unnecessary generator ticks.
@@ -553,8 +567,7 @@ async function* createAsyncPipeline(source, transforms, signal) {
   // Check for abort
   signal?.throwIfAborted();
 
-  // Normalize source via from()
-  const normalized = from(source);
+  const normalized = source;
 
   // Fast path: no transforms, just yield normalized source directly
   if (transforms.length === 0) {
@@ -661,7 +674,7 @@ function pullSync(source, ...transforms) {
   return {
     __proto__: null,
     *[SymbolIterator]() {
-      yield* createSyncPipeline(source, transforms);
+      yield* createSyncPipeline(fromSync(source), transforms);
     },
   };
 }
@@ -693,7 +706,7 @@ function pull(source, ...args) {
   return {
     __proto__: null,
     async *[SymbolAsyncIterator]() {
-      yield* createAsyncPipeline(source, transforms, signal);
+      yield* createAsyncPipeline(from(source), transforms, signal);
     },
   };
 }
@@ -792,28 +805,50 @@ async function pipeTo(source, ...args) {
   const hasWriteSync = typeof writer.writeSync === 'function';
   const hasWritevSync = typeof writer.writevSync === 'function';
   const hasEndSync = typeof writer.endSync === 'function';
+  // Async fallback for writeBatch when sync write fails partway through.
+  // Continues writing from batch[startIndex] using async write().
+  async function writeBatchAsyncFallback(batch, startIndex) {
+    for (let i = startIndex; i < batch.length; i++) {
+      const chunk = batch[i];
+      if (hasWriteSync && writer.writeSync(chunk)) {
+        // Sync retry succeeded
+      } else {
+        const result = writer.write(
+          chunk, signal ? { __proto__: null, signal } : undefined);
+        if (result !== undefined) {
+          await result;
+        }
+      }
+      totalBytes += TypedArrayPrototypeGetByteLength(chunk);
+    }
+  }
+
   // Write a batch using try-fallback: sync first, async if needed.
-  async function writeBatch(batch) {
+  // Returns undefined on sync success, or a Promise when async fallback
+  // is required. Callers must check: const p = writeBatch(b); if (p) await p;
+  function writeBatch(batch) {
     if (hasWritev && batch.length > 1) {
       if (!hasWritevSync || !writer.writevSync(batch)) {
-        await writer.writev(batch, signal ? { __proto__: null, signal } :
-          undefined);
+        const opts = signal ? { __proto__: null, signal } : undefined;
+        return PromisePrototypeThen(writer.writev(batch, opts), () => {
+          for (let i = 0; i < batch.length; i++) {
+            totalBytes += TypedArrayPrototypeGetByteLength(batch[i]);
+          }
+        });
       }
       for (let i = 0; i < batch.length; i++) {
         totalBytes += TypedArrayPrototypeGetByteLength(batch[i]);
       }
-    } else {
-      for (let i = 0; i < batch.length; i++) {
-        const chunk = batch[i];
-        if (!hasWriteSync || !writer.writeSync(chunk)) {
-          const result = writer.write(
-            chunk, signal ? { __proto__: null, signal } : undefined);
-          if (result !== undefined) {
-            await result;
-          }
-        }
-        totalBytes += TypedArrayPrototypeGetByteLength(chunk);
+      return;
+    }
+    for (let i = 0; i < batch.length; i++) {
+      const chunk = batch[i];
+      if (!hasWriteSync || !writer.writeSync(chunk)) {
+        // Sync path failed at index i - fall back to async for the rest.
+        // Count bytes for chunks already written synchronously (0..i-1).
+        return writeBatchAsyncFallback(batch, i);
       }
+      totalBytes += TypedArrayPrototypeGetByteLength(chunk);
     }
   }
 
@@ -823,11 +858,13 @@ async function pipeTo(source, ...args) {
       if (signal) {
         for await (const batch of normalized) {
           signal.throwIfAborted();
-          await writeBatch(batch);
+          const p = writeBatch(batch);
+          if (p) await p;
         }
       } else {
         for await (const batch of normalized) {
-          await writeBatch(batch);
+          const p = writeBatch(batch);
+          if (p) await p;
         }
       }
     } else {
@@ -836,11 +873,13 @@ async function pipeTo(source, ...args) {
       if (signal) {
         for await (const batch of pipeline) {
           signal.throwIfAborted();
-          await writeBatch(batch);
+          const p = writeBatch(batch);
+          if (p) await p;
         }
       } else {
         for await (const batch of pipeline) {
-          await writeBatch(batch);
+          const p = writeBatch(batch);
+          if (p) await p;
         }
       }
     }