stream: fix Writer sync method return values for try-fallback pattern

All `*Sync` methods on the Writer interface are designed as
try-fallback pairs with their async counterparts: attempt the
fast synchronous path first, and fall back to the async version
only when the synchronous call indicates it could not complete.

`endSync()` was returning the byte count unconditionally, making
the fallback check in duplex `close()` incorrect (`!0 === true`).
`failSync()` always returned true even when already errored.

- `endSync()` now returns -1 when writer is not open, byte count
  (`>= 0`) on success
- `failSync()` now returns false when already errored
- `PushQueue.fail()` returns boolean to support `failSync()`
- duplex `close()` uses `endSync() < 0` for the fallback check
- Document the try-fallback pattern and sync return values

Author: jasnell

doc/api/stream_iter.md

@@ -695,7 +695,19 @@ run().catch(console.error);
 
 #### Writer
 
-The writer returned by `push()` has the following methods:
+The writer returned by `push()` has the following methods.
+
+Each async method has a synchronous `*Sync` counterpart designed for a
+try-fallback pattern: attempt the fast synchronous path first, and fall back
+to the async version only when the synchronous call indicates it could not
+complete:
+
+```mjs
+if (!writer.writeSync(chunk)) await writer.write(chunk);
+if (!writer.writevSync(chunks)) await writer.writev(chunks);
+if (writer.endSync() < 0) await writer.end();
+if (!writer.failSync(err)) await writer.fail(err);
+```
 
 ##### `writer.fail(reason)`
 
@@ -707,8 +719,10 @@ Fail the stream with an error.
 ##### `writer.failSync(reason)`
 
 * `reason` {Error}
+* Returns: {boolean} `true` if the writer was failed, `false` if already
+  errored.
 
-Synchronously fail the stream with an error. Does not return a promise.
+Synchronous variant of `writer.fail()`.
 
 ##### `writer.desiredSize`
 
@@ -726,6 +740,20 @@ Returns `null` if the writer is closed or the consumer has disconnected.
 
 Signal that no more data will be written.
 
+##### `writer.endSync()`
+
+* Returns: {number} Total bytes written, or `-1` if the writer is not open.
+
+Synchronous variant of `writer.end()`. Returns `-1` if the writer is already
+closed or errored. Can be used as a try-fallback pattern:
+
+```cjs
+const result = writer.endSync();
+if (result < 0) {
+  writer.end();
+}
+```
+
 ##### `writer.write(chunk[, options])`
 
 * `chunk` {Uint8Array|string}

lib/internal/streams/iter/duplex.js

@@ -46,7 +46,7 @@ function duplex(options) {
       if (aWriterRef === null) return;
       const writer = aWriterRef;
       aWriterRef = null;
-      if (!writer.endSync()) {
+      if (writer.endSync() < 0) {
         await writer.end();
       }
     },
@@ -62,7 +62,7 @@ function duplex(options) {
       if (bWriterRef === null) return;
       const writer = bWriterRef;
       bWriterRef = null;
-      if (!writer.endSync()) {
+      if (writer.endSync() < 0) {
         await writer.end();
       }
     },

lib/internal/streams/iter/push.js

@@ -248,7 +248,7 @@ class PushQueue {
    */
   end() {
     if (this.#writerState !== 'open') {
-      return this.#bytesWritten;
+      return -1;
     }
 
     this.#writerState = 'closed';
@@ -261,16 +261,18 @@ class PushQueue {
 
   /**
    * Put queue into terminal error state.
+   * @returns {boolean} true if the writer was failed, false if already errored.
    */
   fail(reason) {
-    if (this.#writerState === 'errored') return;
+    if (this.#writerState === 'errored') return false;
 
     this.#writerState = 'errored';
     this.#error = reason ?? new ERR_INVALID_STATE('Failed');
     this.#cleanup();
     this.#rejectPendingReads(this.#error);
     this.#rejectPendingWrites(this.#error);
     this.#rejectPendingDrains(this.#error);
+    return true;
   }
 
   get totalBytesWritten() {
@@ -512,8 +514,7 @@ class PushWriter {
   }
 
   failSync(reason) {
-    this.#queue.fail(reason);
-    return true;
+    return this.#queue.fail(reason);
   }
 }
 

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

@@ -108,6 +108,9 @@ async function testWriterEnd() {
   const totalBytes = writer.endSync();
   assert.strictEqual(totalBytes, 0);
 
+  // Calling endSync again returns -1 (already closed)
+  assert.strictEqual(writer.endSync(), -1);
+
   const batches = [];
   for await (const batch of readable) {
     batches.push(batch);