stream: do not fail the writer when pipeToSync() cannot close

After writing every chunk, pipeToSync() treated endSync() returning -1
like any other error: it threw ERR_INVALID_STATE and, unless
preventFail was set, called writer.fail() with it. -1 only means that
the writer cannot close synchronously, e.g. a push() writer whose
consumer has not drained it yet. All of the data had been accepted,
but failing the writer discarded it, so the consumer saw an error
instead of the end of the stream, and the caller could not recover.

pipeToSync() still throws ERR_INVALID_STATE in that case, since it
never falls back to the async end(), but it no longer fails the
writer. The caller can still close it, e.g. with `await writer.end()`.

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

Author: jasnell

doc/api/stream_iter.md

@@ -718,6 +718,13 @@ Synchronous version of [`pipeTo()`][]. The `source`, all transforms, and the
 The `writer` must have the `*Sync` methods (`writeSync`, `writevSync`,
 `endSync`) and `fail()` for this to work.
 
+`pipeToSync()` never falls back to the asynchronous writer methods. If
+`writer.endSync()` returns `-1` because the writer cannot close synchronously
+(for example, a `push()` writer whose consumer has not read all of the data
+yet), `pipeToSync()` throws `ERR_INVALID_STATE`. All of the data was accepted
+by then, so the writer is not failed: it can still be closed, for example with
+`await writer.end()`.
+
 ### `pull(source[, ...transforms][, options])`
 
 <!-- YAML

lib/internal/streams/iter/pull.js

@@ -1048,6 +1048,7 @@ function pipeToSync(source, ...args) {
     normalized;
 
   let totalBytes = 0;
+  let closedSync = true;
 
   try {
     for (const batch of pipeline) {
@@ -1076,10 +1077,7 @@ function pipeToSync(source, ...args) {
     }
 
     if (!options.preventClose) {
-      if (FunctionPrototypeCall(endSync, writer) < 0) {
-        throw new ERR_INVALID_STATE(
-          'Writer could not be closed synchronously');
-      }
+      closedSync = FunctionPrototypeCall(endSync, writer) >= 0;
     }
   } catch (error) {
     if (!options.preventFail) {
@@ -1088,6 +1086,15 @@ function pipeToSync(source, ...args) {
     throw error;
   }
 
+  // endSync() returning -1 only means that the writer cannot close
+  // synchronously; every chunk was accepted. pipeToSync() never falls back to
+  // the async end(), so report it, but leave the writer as it is: the caller
+  // can still close it, e.g. with `await writer.end()`.
+  if (!closedSync) {
+    throw new ERR_INVALID_STATE(
+      'Writer could not be closed synchronously');
+  }
+
   return totalBytes;
 }
 

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

@@ -5,7 +5,9 @@
 
 const common = require('../common');
 const assert = require('assert');
-const { pipeTo, pipeToSync, fromSync } = require('stream/iter');
+const {
+  pipeTo, pipeToSync, fromSync, push, text,
+} = require('stream/iter');
 
 // pipeToSync cannot complete when endSync() requires async fallback.
 async function testPipeToSyncEndSyncFailure() {
@@ -22,6 +24,29 @@ async function testPipeToSyncEndSyncFailure() {
   assert.strictEqual(endCalled, false);
 }
 
+// The data was accepted, so endSync() returning -1 does not fail the writer,
+// even without preventFail, and the caller can still close it.
+async function testPipeToSyncEndSyncFailureDoesNotFailWriter() {
+  const writer = {
+    writeSync() { return true; },
+    endSync: common.mustCall(() => -1),
+    end: common.mustNotCall(),
+    fail: common.mustNotCall(),
+  };
+  assert.throws(() => pipeToSync(fromSync('data'), writer),
+                { code: 'ERR_INVALID_STATE' });
+
+  // A push() writer whose consumer has not read yet cannot close
+  // synchronously. After the throw, the data is intact and the writer can be
+  // ended asynchronously.
+  const { writer: pushWriter, readable } = push();
+  assert.throws(() => pipeToSync(fromSync(['abc', 'def']), pushWriter),
+                { code: 'ERR_INVALID_STATE' });
+  const result = text(readable);
+  assert.strictEqual(await pushWriter.end(), 6);
+  assert.strictEqual(await result, 'abcdef');
+}
+
 // pipeToSync requires endSync() when closing is enabled.
 async function testPipeToSyncNoEndSync() {
   let writeCalled = false;
@@ -106,6 +131,7 @@ async function testFailThrowingDoesNotMaskError() {
 
 Promise.all([
   testPipeToSyncEndSyncFailure(),
+  testPipeToSyncEndSyncFailureDoesNotFailWriter(),
   testPipeToSyncNoEndSync(),
   testPipeToSyncPreventFail(),
   testPipeToSyncPreventClose(),