stream: make pipeTo source normalization independent of Writer

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

lib/internal/streams/iter/pull.js

@@ -8,8 +8,6 @@
 
 const {
   ArrayBufferIsView,
-  ArrayFromAsync,
-  ArrayIsArray,
   ArrayPrototypePush,
   ArrayPrototypeSlice,
   FunctionPrototypeCall,
@@ -46,9 +44,7 @@ const {
   fromSync,
   isSyncIterable,
   isAsyncIterable,
-  isPrimitiveChunk,
   isUint8ArrayBatch,
-  normalizeAsyncValue,
 } = require('internal/streams/iter/from');
 
 const {
@@ -62,10 +58,7 @@ const {
 } = require('internal/streams/iter/utils');
 
 const {
-  kValidatedSource,
   kValidatedTransform,
-  toAsyncStreamable,
-  toStreamable,
 } = require('internal/streams/iter/types');
 
 // =============================================================================
@@ -128,22 +121,6 @@ function parsePipeToArgs(args, requiredMethod) {
   };
 }
 
-function canUseSyncIterablePipeToFastPath(source, transforms, signal) {
-  if (signal !== undefined ||
-      transforms.length !== 0 ||
-      isPrimitiveChunk(source) ||
-      ArrayIsArray(source) ||
-      source?.[kValidatedSource] ||
-      !isSyncIterable(source) ||
-      isAsyncIterable(source)) {
-    return false;
-  }
-
-  // Preserve from()'s top-level protocol precedence for custom iterables.
-  return typeof source[toAsyncStreamable] !== 'function' &&
-    typeof source[toStreamable] !== 'function';
-}
-
 // =============================================================================
 // Transform Output Flattening
 // =============================================================================
@@ -1055,9 +1032,7 @@ async function pipeTo(source, ...args) {
   }
 
   const hasWriteSync = typeof writer.writeSync === 'function';
-  const useSyncIterableFastPath =
-    hasWriteSync && canUseSyncIterablePipeToFastPath(source, transforms, signal);
-  const normalized = useSyncIterableFastPath ? undefined : from(source);
+  const normalized = from(source);
 
   let totalBytes = 0;
   const hasWritev = typeof writer.writev === 'function';
@@ -1118,31 +1093,7 @@ async function pipeTo(source, ...args) {
   }
 
   try {
-    if (useSyncIterableFastPath) {
-      // Avoid from()'s async sync-iterable batching path. This keeps writes
-      // incremental for synchronous sources while preserving async
-      // normalization for non-primitive yielded values.
-      for (const value of source) {
-        if (isUint8ArrayBatch(value)) {
-          if (value.length > 0) {
-            const p = writeBatch(value);
-            if (p) await p;
-          }
-          continue;
-        }
-        if (isUint8Array(value)) {
-          const p = writeBatch([value]);
-          if (p) await p;
-          continue;
-        }
-
-        const batch = await ArrayFromAsync(normalizeAsyncValue(value));
-        if (batch.length > 0) {
-          const p = writeBatch(batch);
-          if (p) await p;
-        }
-      }
-    } else if (transforms.length === 0) {
+    if (transforms.length === 0) {
       // Fast path: no transforms - iterate normalized source directly
       if (signal) {
         for await (const batch of yieldAbortable(normalized, signal)) {

test/parallel/test-stream-iter-pipeto.js

@@ -244,7 +244,7 @@ async function testPipeToSyncMinimalWriter() {
   assert.strictEqual(chunks.length > 0, true);
 }
 
-async function testPipeToSyncIterableFastPathWritesIncrementally() {
+async function testPipeToSyncIterableUsesFromBatching() {
   let pulled = 0;
   let firstWritePulled = 0;
   const chunks = [];
@@ -267,15 +267,39 @@ async function testPipeToSyncIterableFastPathWritesIncrementally() {
 
   const totalBytes = await pipeTo(source(), writer);
   assert.strictEqual(totalBytes, 3);
-  assert.strictEqual(firstWritePulled, 1);
+  assert.strictEqual(firstWritePulled, 3);
   assert.deepStrictEqual(chunks, [
     new Uint8Array([0x61]),
     new Uint8Array([0x62]),
     new Uint8Array([0x63]),
   ]);
 }
 
-async function testPipeToSyncIterableFastPathWriteFallback() {
+async function testPipeToSourceNormalizationIndependentOfWriter() {
+  function source() {
+    return {
+      *[Symbol.iterator]() {
+        yield {
+          async *[Symbol.asyncIterator]() {
+            yield 'nested';
+          },
+        };
+      },
+    };
+  }
+
+  for (const hasWriteSync of [false, true]) {
+    const writer = { write: common.mustNotCall() };
+    if (hasWriteSync) writer.writeSync = common.mustNotCall();
+
+    await assert.rejects(
+      pipeTo(source(), writer, { preventClose: true, preventFail: true }),
+      { code: 'ERR_INVALID_ARG_TYPE' },
+    );
+  }
+}
+
+async function testPipeToSyncIterableWriteFallback() {
   const asyncWrites = [];
   const writer = {
     writeSync(chunk) {
@@ -296,7 +320,7 @@ async function testPipeToSyncIterableFastPathWriteFallback() {
   assert.deepStrictEqual(asyncWrites, [new Uint8Array([0x62])]);
 }
 
-async function testPipeToSyncIterableFastPathAsyncValue() {
+async function testPipeToSyncIterableAsyncValue() {
   const chunks = [];
   const writer = {
     write: common.mustNotCall(),
@@ -334,7 +358,8 @@ Promise.all([
   testPipeToSyncPreventClose(),
   testPipeToMinimalWriter(),
   testPipeToSyncMinimalWriter(),
-  testPipeToSyncIterableFastPathWritesIncrementally(),
-  testPipeToSyncIterableFastPathWriteFallback(),
-  testPipeToSyncIterableFastPathAsyncValue(),
+  testPipeToSyncIterableUsesFromBatching(),
+  testPipeToSyncIterableWriteFallback(),
+  testPipeToSyncIterableAsyncValue(),
+  testPipeToSourceNormalizationIndependentOfWriter(),
 ]).then(common.mustCall());