stream: apply spec conformance fixes (part 1)

Author: jasnell

doc/api/fs.md

@@ -946,11 +946,8 @@ added: REPLACEME
     * `options` {Object}
       * `signal` {AbortSignal} If the signal is already aborted, `end()`
         rejects with `AbortError` and the writer remains open.
-  * `fail(reason)` {Function} Returns {Promise\<void>}. Puts the writer
-    into a terminal error state.
-  * `failSync(reason)` {Function} Returns {boolean}. Synchronous best-effort
-    cleanup. Marks the writer as closed so subsequent writes fail immediately.
-    Cannot honor `autoClose` (requires async I/O).
+  * `fail(reason)` {Function} Puts the writer into a terminal error state.
+    Synchronous. If the writer is already closed or errored, this is a no-op.
 
 Return a [`node:stream/iter`][] writer backed by this file handle.
 

doc/api/stream_iter.md

@@ -409,7 +409,7 @@ complete:
 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(err);  // Always synchronous, no fallback needed
 ```
 
 ### `writer.desiredSize`
@@ -446,18 +446,12 @@ if (result < 0) {
 
 ### `writer.fail(reason)`
 
-* `reason` {Error}
-* Returns: {Promise\<void>}
-
-Fail the stream with an error.
-
-### `writer.failSync(reason)`
-
-* `reason` {Error}
-* Returns: {boolean} `true` if the writer was failed, `false` if already
-  errored.
+* `reason` {any}
 
-Synchronous variant of `writer.fail()`.
+Put the writer into a terminal error state. If the writer is already closed
+or errored, this is a no-op. Unlike `write()` and `end()`, `fail()` is
+unconditionally synchronous because failing a writer is a pure state
+transition with no async work to perform.
 
 ### `writer.write(chunk[, options])`
 
@@ -604,10 +598,10 @@ Pipe a source through transforms into a writer. If the writer has a
 scatter/gather I/O).
 
 If the writer implements the optional `*Sync` methods (`writeSync`, `writevSync`,
-`endSync`, `failSync`), `pipeTo()` will attempt to use the synchronous methods
+`endSync`), `pipeTo()` will attempt to use the synchronous methods
 first as a fast path, and fall back to the async versions only when the sync
 methods indicate they cannot complete (e.g., backpressure or waiting for the
-next tick).
+next tick). `fail()` is always called synchronously.
 
 ```mjs
 import { from, pipeTo, compressGzip } from 'node:stream/iter';
@@ -655,7 +649,7 @@ Synchronous version of [`pipeTo()`][]. The `source`, all transforms, and the
 `writer` must be synchronous. Cannot accept async iterables or promises.
 
 The `writer` must have the `*Sync` methods (`writeSync`, `writevSync`,
-`endSync`, `failSync`) for this to work.
+`endSync`) and `fail()` for this to work.
 
 ### `pull(source[, ...transforms][, options])`
 

lib/internal/fs/promises.js

@@ -541,7 +541,7 @@ if (getOptionValue('--experimental-stream-iter')) {
    *   autoClose?: boolean;
    *   start?: number;
    * }} [options]
-   * @returns {{ write, writev, end, fail, failSync }}
+   * @returns {{ write, writev, end, fail }}
    */
   FileHandle.prototype.writer = function writer(options) {
     if (this[kFd] === -1)
@@ -558,6 +558,7 @@ if (getOptionValue('--experimental-stream-iter')) {
     let pos = options?.start ?? -1;
     let totalBytesWritten = 0;
     let closed = false;
+    let error = null;
 
     if (pos !== -1) {
       validateInteger(pos, 'options.start', 0);
@@ -643,9 +644,12 @@ if (getOptionValue('--experimental-stream-iter')) {
     return {
       __proto__: null,
       write(chunk, options) {
+        if (error) {
+          return PromiseReject(error);
+        }
         if (closed) {
           return PromiseReject(
-            new ERR_INVALID_STATE('The writer is closed'));
+            new ERR_INVALID_STATE.TypeError('The writer is closed'));
         }
         if (options?.signal?.aborted) {
           return PromiseReject(
@@ -657,9 +661,12 @@ if (getOptionValue('--experimental-stream-iter')) {
       },
 
       writev(chunks, options) {
+        if (error) {
+          return PromiseReject(error);
+        }
         if (closed) {
           return PromiseReject(
-            new ERR_INVALID_STATE('The writer is closed'));
+            new ERR_INVALID_STATE.TypeError('The writer is closed'));
         }
         if (options?.signal?.aborted) {
           return PromiseReject(
@@ -682,19 +689,12 @@ if (getOptionValue('--experimental-stream-iter')) {
         return totalBytesWritten;
       },
 
-      async fail(reason) {
-        await cleanup();
-      },
-
-      failSync(reason) {
-        // Synchronous cleanup is best-effort for file handles.
-        // Mark as closed so subsequent writes fail immediately.
-        if (closed) return true;
+      fail(reason) {
+        if (closed || error) return;
+        error = reason ?? new ERR_INVALID_STATE('Failed');
         closed = true;
         handle[kLocked] = false;
         handle[kUnref]();
-        // autoClose cannot be handled synchronously - skip it.
-        return true;
       },
 
       async [SymbolAsyncDispose]() {

lib/internal/streams/iter/broadcast.js

@@ -21,8 +21,9 @@ const {
   SafeSet,
   String,
   Symbol,
-  SymbolAsyncIterator,
-  SymbolDispose,
+   SymbolAsyncDispose,
+   SymbolAsyncIterator,
+   SymbolDispose,
   TypedArrayPrototypeGetByteLength,
 } = primordials;
 
@@ -565,26 +566,23 @@ class BroadcastWriter {
     return this.#totalBytes;
   }
 
-  fail(reason) {
-    if (this.#aborted) return kResolvedPromise;
-    this.#aborted = true;
-    this.#closed = true;
-    const error = reason ?? new ERR_INVALID_STATE('Failed');
-    this.#rejectPendingWrites(error);
-    this.#rejectPendingDrains(error);
-    this.#broadcast[kAbort](error);
-    return kResolvedPromise;
+   fail(reason) {
+     if (this.#aborted || this.#closed) return;
+     this.#aborted = true;
+     this.#closed = true;
+     const error = reason ?? new ERR_INVALID_STATE('Failed');
+     this.#rejectPendingWrites(error);
+     this.#rejectPendingDrains(error);
+     this.#broadcast[kAbort](error);
+   }
+
+  [SymbolAsyncDispose]() {
+    this.fail();
+    return PromiseResolve();
   }
 
-  failSync(reason) {
-    if (this.#aborted) return true;
-    this.#aborted = true;
-    this.#closed = true;
-    const error = reason ?? new ERR_INVALID_STATE('Failed');
-    this.#rejectPendingWrites(error);
-    this.#rejectPendingDrains(error);
-    this.#broadcast[kAbort](error);
-    return true;
+  [SymbolDispose]() {
+    this.fail();
   }
 
   [kCancelWriter]() {
@@ -778,9 +776,7 @@ const Broadcast = {
       } catch (error) {
         const err = isError(error) ? error :
           new ERR_OPERATION_FAILED(String(error));
-        if (!w.failSync(err)) {
-          await w.fail(err);
-        }
+         w.fail(err);
       }
     };
     PromisePrototypeThen(pump(), undefined, () => {});

lib/internal/streams/iter/pull.js

@@ -588,9 +588,8 @@ function pipeToSync(source, ...args) {
   const hasWriteSync = typeof writer.writeSync === 'function';
   const hasWritevSync = typeof writer.writevSync === 'function';
   const hasEndSync = typeof writer.endSync === 'function';
-  const hasFailSync = typeof writer.failSync === 'function';
 
-  try {
+   try {
     for (const batch of pipeline) {
       if (hasWritevSync && batch.length > 1) {
         writer.writevSync(batch);
@@ -616,17 +615,15 @@ function pipeToSync(source, ...args) {
       }
     }
   } catch (error) {
-    if (!options?.preventFail) {
-      const err = isError(error) ? error :
-        new ERR_OPERATION_FAILED(String(error));
-      if (!hasFailSync || !writer.failSync(err)) {
-        writer.fail?.(err);
-      }
-    }
-    throw error;
-  }
-
-  return totalBytes;
+   if (!options?.preventFail) {
+       const err = isError(error) ? error :
+         new ERR_OPERATION_FAILED(String(error));
+       writer.fail(err);
+     }
+     throw error;
+   }
+
+   return totalBytes;
 }
 
 /**
@@ -656,8 +653,7 @@ async function pipeTo(source, ...args) {
   const hasWriteSync = typeof writer.writeSync === 'function';
   const hasWritevSync = typeof writer.writevSync === 'function';
   const hasEndSync = typeof writer.endSync === 'function';
-  const hasFailSync = typeof writer.failSync === 'function';
-  // Write a batch using try-fallback: sync first, async if needed.
+   // Write a batch using try-fallback: sync first, async if needed.
   async function writeBatch(batch) {
     if (hasWritev && batch.length > 1) {
       if (!hasWritevSync || !writer.writevSync(batch)) {
@@ -727,17 +723,15 @@ async function pipeTo(source, ...args) {
       }
     }
   } catch (error) {
-    if (!options?.preventFail) {
-      const err = isError(error) ? error :
-        new ERR_OPERATION_FAILED(String(error));
-      if (!hasFailSync || !writer.failSync(err)) {
-        await writer.fail?.(err);
-      }
-    }
-    throw error;
-  }
-
-  return totalBytes;
+     if (!options?.preventFail) {
+       const err = isError(error) ? error :
+         new ERR_OPERATION_FAILED(String(error));
+       writer.fail(err);
+     }
+     throw error;
+   }
+
+   return totalBytes;
 }
 
 module.exports = {