stream: use Writer try-fallback pattern in pipeTo and Broadcast.from

The `Writer` interface sync methods are designed as try-fallback
pairs: attempt the fast sync path first, fall back to async only
when rejected or unavailable. This was not being followed in

`pipeTo`, `pipeToSync`, or `Broadcast.from()`.
`pipeTo` now attempts `writeSync`/`writevSync` before `write`/`writev`,
`endSync` before `end`, and `failSync` before `fail`, with graceful
handling when *Sync methods are not present on the writer.

`pipeToSync` similarly prefers *Sync methods with fallback to
the non-sync variants for writers that lack them.

`Broadcast.from()` applies the same pattern when writing source
data to the internal broadcast writer.

Additionally, writes in p`ipeTo are now sequential rather than
parallel (previously used `Promise.all`), which respects writer
backpressure signals.

Author: jasnell

lib/internal/streams/iter/broadcast.js

@@ -683,6 +683,7 @@ const Broadcast = {
     const signal = options?.signal;
 
     (async () => {
+      const w = result.writer;
       try {
         if (isAsyncIterable(input)) {
           for await (const chunks of input) {
@@ -691,11 +692,11 @@ const Broadcast = {
                 lazyDOMException('Aborted', 'AbortError');
             }
             if (ArrayIsArray(chunks)) {
-              await result.writer.writev(
-                chunks, signal ? { signal } : undefined);
-            } else {
-              await result.writer.write(
-                chunks, signal ? { signal } : undefined);
+              if (!w.writevSync(chunks)) {
+                await w.writev(chunks, signal ? { signal } : undefined);
+              }
+            } else if (!w.writeSync(chunks)) {
+              await w.write(chunks, signal ? { signal } : undefined);
             }
           }
         } else if (isSyncIterable(input)) {
@@ -705,18 +706,23 @@ const Broadcast = {
                 lazyDOMException('Aborted', 'AbortError');
             }
             if (ArrayIsArray(chunks)) {
-              await result.writer.writev(
-                chunks, signal ? { signal } : undefined);
-            } else {
-              await result.writer.write(
-                chunks, signal ? { signal } : undefined);
+              if (!w.writevSync(chunks)) {
+                await w.writev(chunks, signal ? { signal } : undefined);
+              }
+            } else if (!w.writeSync(chunks)) {
+              await w.write(chunks, signal ? { signal } : undefined);
             }
           }
         }
-        await result.writer.end(signal ? { signal } : undefined);
+        if (w.endSync() < 0) {
+          await w.end(signal ? { signal } : undefined);
+        }
       } catch (error) {
-        await result.writer.fail(
-          isError(error) ? error : new ERR_INVALID_ARG_TYPE('error', 'Error', String(error)));
+        const err = isError(error) ? error :
+          new ERR_INVALID_ARG_TYPE('error', 'Error', String(error));
+        if (!w.failSync(err)) {
+          await w.fail(err);
+        }
       }
     })();
 

lib/internal/streams/iter/pull.js

@@ -10,7 +10,6 @@ const {
   ArrayIsArray,
   ArrayPrototypePush,
   ArrayPrototypeSlice,
-  SafePromiseAllReturnVoid,
   String,
   SymbolAsyncIterator,
   SymbolIterator,
@@ -610,22 +609,33 @@ function pipeToSync(source, ...args) {
     source;
 
   let totalBytes = 0;
+  const hasWriteSync = typeof writer.writeSync === 'function';
+  const hasEndSync = typeof writer.endSync === 'function';
+  const hasFailSync = typeof writer.failSync === 'function';
 
   try {
     for (const batch of pipeline) {
       for (let i = 0; i < batch.length; i++) {
         const chunk = batch[i];
-        writer.write(chunk);
+        if (!hasWriteSync || !writer.writeSync(chunk)) {
+          writer.write(chunk);
+        }
         totalBytes += TypedArrayPrototypeGetByteLength(chunk);
       }
     }
 
     if (!options?.preventClose) {
-      writer.end();
+      if (!hasEndSync || writer.endSync() < 0) {
+        writer.end();
+      }
     }
   } catch (error) {
     if (!options?.preventFail) {
-      writer.fail(isError(error) ? error : new ERR_OPERATION_FAILED(String(error)));
+      const err = isError(error) ? error :
+        new ERR_OPERATION_FAILED(String(error));
+      if (!hasFailSync || !writer.failSync(err)) {
+        writer.fail(err);
+      }
     }
     throw error;
   }
@@ -657,27 +667,31 @@ async function pipeTo(source, ...args) {
 
   let totalBytes = 0;
   const hasWritev = typeof writer.writev === 'function';
+  const hasWritevSync = typeof writer.writevSync === 'function';
+  const hasWriteSync = typeof writer.writeSync === 'function';
 
-  // Helper to write a batch efficiently
+  // Helper to write a batch efficiently using try-fallback pattern:
+  // attempt the sync path first, fall back to async if rejected.
   const writeBatch = async (batch) => {
     if (hasWritev && batch.length > 1) {
-      await writer.writev(batch, signal ? { signal } : undefined);
+      if (!hasWritevSync ||
+          !writer.writevSync(batch)) {
+        await writer.writev(batch, signal ? { signal } : undefined);
+      }
       for (let i = 0; i < batch.length; i++) {
         totalBytes += TypedArrayPrototypeGetByteLength(batch[i]);
       }
     } else {
-      const promises = [];
       for (let i = 0; i < batch.length; i++) {
         const chunk = batch[i];
-        const result = writer.write(chunk, signal ? { signal } : undefined);
-        if (result !== undefined) {
-          ArrayPrototypePush(promises, result);
+        if (!hasWriteSync || !writer.writeSync(chunk)) {
+          const result = writer.write(chunk, signal ? { signal } : undefined);
+          if (result !== undefined) {
+            await result;
+          }
         }
         totalBytes += TypedArrayPrototypeGetByteLength(chunk);
       }
-      if (promises.length > 0) {
-        await SafePromiseAllReturnVoid(promises);
-      }
     }
   };
 
@@ -719,12 +733,19 @@ async function pipeTo(source, ...args) {
     }
 
     if (!options?.preventClose) {
-      await writer.end(signal ? { signal } : undefined);
+      if (typeof writer.endSync !== 'function' ||
+          writer.endSync() < 0) {
+        await writer.end(signal ? { signal } : undefined);
+      }
     }
   } catch (error) {
     if (!options?.preventFail) {
-      await writer.fail(
-        isError(error) ? error : new ERR_OPERATION_FAILED(String(error)));
+      const err = isError(error) ? error :
+        new ERR_OPERATION_FAILED(String(error));
+      if (typeof writer.failSync !== 'function' ||
+          !writer.failSync(err)) {
+        await writer.fail(err);
+      }
     }
     throw error;
   }