stream: ensure pipeToSync requires synchronous close

Signed-off-by: James M Snell <jasnell@gmail.com>
Assisted-by: Opencode
PR-URL: https://github.com/nodejs/node/pull/65652
Reviewed-By: Trivikram Kamat <trivikr.dev@gmail.com>
Reviewed-By: Matteo Collina <matteo.collina@gmail.com>

lib/internal/streams/iter/pull.js

@@ -24,6 +24,7 @@ const {
   codes: {
     ERR_INVALID_ARG_TYPE,
     ERR_INVALID_ARG_VALUE,
+    ERR_INVALID_STATE,
     ERR_OUT_OF_RANGE,
   },
 } = require('internal/errors');
@@ -981,6 +982,13 @@ function pullWithConsumerCleanup(source, transforms, signal) {
  */
 function pipeToSync(source, ...args) {
   const { transforms, writer, options } = parsePipeToArgs(args, 'writeSync');
+  const hasWritevSync = typeof writer.writevSync === 'function';
+  const endSync = writer.endSync;
+
+  if (!options?.preventClose && typeof endSync !== 'function') {
+    throw new ERR_INVALID_ARG_TYPE(
+      'writer.endSync', 'Function', endSync);
+  }
 
   // Normalize source and create pipeline
   const normalized = fromSync(source);
@@ -989,8 +997,6 @@ function pipeToSync(source, ...args) {
     normalized;
 
   let totalBytes = 0;
-  const hasWritevSync = typeof writer.writevSync === 'function';
-  const hasEndSync = typeof writer.endSync === 'function';
 
   try {
     for (const batch of pipeline) {
@@ -1019,8 +1025,9 @@ function pipeToSync(source, ...args) {
     }
 
     if (!options?.preventClose) {
-      if (!hasEndSync || writer.endSync() < 0) {
-        writer.end?.();
+      if (FunctionPrototypeCall(endSync, writer) < 0) {
+        throw new ERR_INVALID_STATE(
+          'Writer could not be closed synchronously');
       }
     }
   } catch (error) {

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

@@ -1,33 +1,41 @@
 // Flags: --experimental-stream-iter
 'use strict';
 
-// Edge case tests for pipeToSync: endSync fallback, preventFail.
+// Edge case tests for pipeToSync close and failure behavior.
 
 const common = require('../common');
 const assert = require('assert');
 const { pipeToSync, fromSync } = require('stream/iter');
 
-// pipeToSync endSync returns negative → falls back to end()
-async function testPipeToSyncEndSyncFallback() {
+// pipeToSync cannot complete when endSync() requires async fallback.
+async function testPipeToSyncEndSyncFailure() {
   let endCalled = false;
   const writer = {
     writeSync() { return true; },
-    endSync() { return -1; }, // Negative → triggers end() fallback
+    endSync() { return -1; },
     end() { endCalled = true; },
   };
-  pipeToSync(fromSync('data'), writer);
-  assert.strictEqual(endCalled, true);
+  assert.throws(
+    () => pipeToSync(fromSync('data'), writer, { preventFail: true }),
+    { code: 'ERR_INVALID_STATE' },
+  );
+  assert.strictEqual(endCalled, false);
 }
 
-// pipeToSync endSync missing → falls back to end()
+// pipeToSync requires endSync() when closing is enabled.
 async function testPipeToSyncNoEndSync() {
+  let writeCalled = false;
   let endCalled = false;
   const writer = {
-    writeSync() { return true; },
+    writeSync() { writeCalled = true; return true; },
     end() { endCalled = true; },
   };
-  pipeToSync(fromSync('data'), writer);
-  assert.strictEqual(endCalled, true);
+  assert.throws(
+    () => pipeToSync(fromSync('data'), writer),
+    { code: 'ERR_INVALID_ARG_TYPE' },
+  );
+  assert.strictEqual(writeCalled, false);
+  assert.strictEqual(endCalled, false);
 }
 
 // pipeToSync with preventFail: true — source error does NOT call fail()
@@ -61,7 +69,7 @@ async function testPipeToSyncPreventClose() {
 }
 
 Promise.all([
-  testPipeToSyncEndSyncFallback(),
+  testPipeToSyncEndSyncFailure(),
   testPipeToSyncNoEndSync(),
   testPipeToSyncPreventFail(),
   testPipeToSyncPreventClose(),

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

@@ -71,6 +71,7 @@ async function testPipeToSyncSourceError() {
   let failCalled = false;
   const writer = {
     writeSync() { return true; },
+    endSync: common.mustNotCall(),
     fail(reason) { failCalled = true; },
   };
   function* failingSource() {
@@ -127,6 +128,7 @@ async function testPipeToSyncWithTransforms() {
   const chunks = [];
   const writer = {
     writeSync(chunk) { chunks.push(new TextDecoder().decode(chunk)); return true; },
+    endSync() { return 0; },
   };
   const upper = (batch) => {
     if (batch === null) return null;
@@ -160,6 +162,7 @@ async function testPipeToSyncWriterTransformMethodIgnored() {
       chunks.push(new TextDecoder().decode(chunk));
       return true;
     },
+    endSync() { return 0; },
   };
 
   pipeToSync(fromSync('hello'), writer);
@@ -240,7 +243,7 @@ async function testPipeToSyncMinimalWriter() {
     },
   };
 
-  pipeToSync(fromSync('minimal-sync'), minimalWriter);
+  pipeToSync(fromSync('minimal-sync'), minimalWriter, { preventClose: true });
   assert.strictEqual(chunks.length > 0, true);
 }
 

test/parallel/test-stream-iter-resizable-buffers.js

@@ -161,7 +161,8 @@ async function testPipeRejectsWriterResize() {
     fail: common.mustCall(),
   };
   assert.throws(
-    () => pipeToSync([new Uint8Array(syncBuffer)], syncWriter),
+    () => pipeToSync(
+      [new Uint8Array(syncBuffer)], syncWriter, { preventClose: true }),
     kResizeError,
   );
 }