stream: add pipeToSync() failOnIncompleteClose option

pipeToSync() does not fail the writer when endSync() returns -1, so
that a caller can still close it asynchronously. A caller that cannot,
e.g. because the writer is sync-only and has no end(), would be left
with a writer that is neither closed nor failed.

Add a failOnIncompleteClose option (a Node.js extension) that fails
the writer with the thrown ERR_INVALID_STATE error in that case.
preventFail takes precedence over it.

Assisted-by: OpenCode
Signed-off-by: James M Snell <jasnell@gmail.com>

Author: jasnell

doc/api/stream_iter.md

@@ -716,6 +716,10 @@ added:
 * `...transforms` {Function|Object} Zero or more sync transforms.
 * `writer` {Object} Destination with `write(chunk)` method.
 * `options` {Object}
+  * `failOnIncompleteClose` {boolean} If `true`, call `writer.fail()` when
+    `writer.endSync()` cannot close the writer synchronously. Ignored when
+    `preventFail` is `true`. This option is a Node.js extension.
+    **Default:** `false`.
   * `preventClose` {boolean} **Default:** `false`.
   * `preventFail` {boolean} **Default:** `false`.
 * Returns: {number} Total bytes written.
@@ -730,8 +734,10 @@ The `writer` must have the `*Sync` methods (`writeSync`, `writevSync`,
 `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()`.
+by then, so by default the writer is not failed: it can still be closed, for
+example with `await writer.end()`. If the writer cannot be closed any other
+way (for example, it has no `end()` method), or the caller will not close it,
+set `failOnIncompleteClose` to fail it with the thrown error instead.
 
 ### `pull(source[, ...transforms][, options])`
 

lib/internal/streams/iter/pull.js

@@ -1088,11 +1088,16 @@ function pipeToSync(source, ...args) {
 
   // 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()`.
+  // the async end(), so report it. By default the writer is left as it is so
+  // that the caller can still close it (e.g. with `await writer.end()`);
+  // `failOnIncompleteClose` fails it instead, for callers that cannot.
   if (!closedSync) {
-    throw new ERR_INVALID_STATE(
+    const error = new ERR_INVALID_STATE(
       'Writer could not be closed synchronously');
+    if (options.failOnIncompleteClose && !options.preventFail) {
+      failWriterQuietly(writer, error);
+    }
+    throw error;
   }
 
   return totalBytes;

lib/internal/streams/iter/webidl.js

@@ -104,6 +104,13 @@ converters.PipeToOptions = createDictionaryConverter('PipeToOptions', [
 ]);
 converters.PipeToSyncOptions = createDictionaryConverter(
   'PipeToSyncOptions', [
+    // Node.js extension.
+    {
+      __proto__: null,
+      key: 'failOnIncompleteClose',
+      converter: baseConverters.boolean,
+      defaultValue: () => false,
+    },
     {
       __proto__: null,
       key: 'preventClose',

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

@@ -129,9 +129,57 @@ async function testFailThrowingDoesNotMaskError() {
     (error) => error === signal.reason);
 }
 
+// failOnIncompleteClose fails a writer that cannot be closed synchronously,
+// e.g. a sync-only writer that has no end().
+async function testPipeToSyncFailOnIncompleteClose() {
+  let failReason;
+  const writer = {
+    writeSync() { return true; },
+    endSync: common.mustCall(() => -1),
+    fail: common.mustCall((reason) => { failReason = reason; }),
+  };
+  assert.throws(
+    () => pipeToSync(fromSync('data'), writer, { failOnIncompleteClose: true }),
+    (error) => {
+      assert.strictEqual(error.code, 'ERR_INVALID_STATE');
+      assert.strictEqual(error, failReason);
+      return true;
+    });
+
+  // preventFail takes precedence.
+  assert.throws(
+    () => pipeToSync(fromSync('data'), {
+      writeSync() { return true; },
+      endSync: common.mustCall(() => -1),
+      fail: common.mustNotCall(),
+    }, { failOnIncompleteClose: true, preventFail: true }),
+    { code: 'ERR_INVALID_STATE' });
+
+  // It has no effect when the writer closes synchronously.
+  assert.strictEqual(pipeToSync(fromSync('data'), {
+    writeSync() { return true; },
+    endSync: common.mustCall(() => 4),
+    fail: common.mustNotCall(),
+  }, { failOnIncompleteClose: true }), 4);
+
+  // A push() writer is failed with the error, so its consumer sees it.
+  const { writer: pushWriter, readable } = push();
+  let thrown;
+  assert.throws(() => {
+    try {
+      pipeToSync(fromSync('abc'), pushWriter, { failOnIncompleteClose: true });
+    } catch (error) {
+      thrown = error;
+      throw error;
+    }
+  }, { code: 'ERR_INVALID_STATE' });
+  await assert.rejects(text(readable), (error) => error === thrown);
+}
+
 Promise.all([
   testPipeToSyncEndSyncFailure(),
   testPipeToSyncEndSyncFailureDoesNotFailWriter(),
+  testPipeToSyncFailOnIncompleteClose(),
   testPipeToSyncNoEndSync(),
   testPipeToSyncPreventFail(),
   testPipeToSyncPreventClose(),