stream: use webidl validation semantics for args

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

Author: jasnell

doc/api/stream_iter.md

@@ -401,6 +401,12 @@ const { writer, readable } = push({
 A writer is any object conforming to the Writer interface. Only `write()` is
 required; all other methods are optional.
 
+Writer arguments use Web IDL conversion semantics. A non-`Uint8Array` chunk is
+converted to a `USVString` and then UTF-8 encoded. `writev()` and
+`writevSync()` accept any iterable object whose values can be converted to
+chunks. Writer option dictionaries treat `null` as an empty dictionary and
+ignore unknown members.
+
 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
@@ -492,7 +498,7 @@ Synchronous write. Does not block; returns `false` if backpressure is active.
 
 #### `writer.writev(chunks[, options])`
 
-* `chunks` {Uint8Array\[]|string\[]}
+* `chunks` {Iterable} of {Uint8Array|string} values
 * `options` {Object}
   * `signal` {AbortSignal} Cancel just this write operation. The signal cancels
     only the pending `writev()` call; it does not fail the writer itself.
@@ -502,7 +508,7 @@ Write multiple chunks as a single batch.
 
 #### `writer.writevSync(chunks)`
 
-* `chunks` {Uint8Array\[]|string\[]}
+* `chunks` {Iterable} of {Uint8Array|string} values
 * Returns: {boolean} `true` if the write was accepted, `false` if the
   buffer is full.
 
@@ -521,6 +527,12 @@ import { from, pull, bytes, Stream } from 'node:stream/iter';
 Stream.from('hello');
 ```
 
+Options dictionaries defined by the Iterable Streams API use Web IDL
+conversion semantics. `null` is treated as an empty dictionary, unknown
+members are ignored, and known members are converted to their declared types
+before the operation runs. Conversion failures use Node.js error codes such as
+`ERR_INVALID_ARG_TYPE`, `ERR_INVALID_ARG_VALUE`, and `ERR_OUT_OF_RANGE`.
+
 ```cjs
 // Named exports
 const { from, pull, bytes, Stream } = require('node:stream/iter');
@@ -860,7 +872,7 @@ added:
 
 * `options` {Object}
   * `budget` {number} Buffer size in bytes for both directions.
-    **Default:** `16384`.
+    Must be >= 16384. **Default:** `16384`.
   * `backpressure` {string} Policy for both directions.
     **Default:** `'strict'`.
   * `signal` {AbortSignal} Cancellation signal for both channels.
@@ -1379,6 +1391,7 @@ added:
     **Default:** `65536`.
   * `backpressure` {string} `'strict'`, `'unbounded'`, `'drop-oldest'`, or
     `'drop-newest'`. **Default:** `'strict'`.
+  * `signal` {AbortSignal}
 * Returns: {Share}
 
 Create a pull-model multi-consumer shared stream. Unlike `broadcast()`, the

lib/internal/fs/promises.js

@@ -164,17 +164,19 @@ const lazyReadableStream = getLazy(() =>
 let newStreamsPull;
 let newStreamsPullSync;
 let newStreamsParsePullArgs;
-let newStreamsToUint8Array;
+let newStreamsToWriterUint8Array;
 let newStreamsConvertChunks;
+let newStreamsGetWriterSignal;
 function lazyNewStreams() {
   if (newStreamsPull === undefined) {
     const pullModule = require('internal/streams/iter/pull');
     newStreamsPull = pullModule.pull;
     newStreamsPullSync = pullModule.pullSync;
     const utils = require('internal/streams/iter/utils');
     newStreamsParsePullArgs = utils.parsePullArgs;
-    newStreamsToUint8Array = utils.toUint8Array;
+    newStreamsToWriterUint8Array = utils.toWriterUint8Array;
     newStreamsConvertChunks = utils.convertChunks;
+    newStreamsGetWriterSignal = utils.getWriterSignal;
   }
 }
 
@@ -885,24 +887,18 @@ if (getOptionValue('--experimental-stream-iter')) {
     return {
       __proto__: null,
       write(chunk, options = kNullPrototo) {
+        chunk = newStreamsToWriterUint8Array(chunk);
+        const signal = newStreamsGetWriterSignal(options);
         if (error) {
           return PromiseReject(error);
         }
         if (closed) {
           return PromiseReject(
             new ERR_INVALID_STATE.TypeError('The writer is closed'));
         }
-        validateObject(options, 'options');
-        const {
-          signal,
-        } = options;
-        if (signal !== undefined) {
-          validateAbortSignal(signal, 'options.signal');
-          if (signal.aborted) {
-            return PromiseReject(signal.reason);
-          }
+        if (signal?.aborted) {
+          return PromiseReject(signal.reason);
         }
-        chunk = newStreamsToUint8Array(chunk);
         if (bytesRemaining >= 0 && chunk.byteLength > bytesRemaining) {
           return PromiseReject(
             new ERR_OUT_OF_RANGE('write', `<= ${bytesRemaining} bytes`,
@@ -915,24 +911,18 @@ if (getOptionValue('--experimental-stream-iter')) {
       },
 
       writev(chunks, options = kNullPrototo) {
+        chunks = newStreamsConvertChunks(chunks);
+        const signal = newStreamsGetWriterSignal(options);
         if (error) {
           return PromiseReject(error);
         }
         if (closed) {
           return PromiseReject(
             new ERR_INVALID_STATE.TypeError('The writer is closed'));
         }
-        validateObject(options, 'options');
-        const {
-          signal,
-        } = options;
-        if (signal !== undefined) {
-          validateAbortSignal(signal, 'options.signal');
-          if (signal?.aborted) {
-            return PromiseReject(signal.reason);
-          }
+        if (signal?.aborted) {
+          return PromiseReject(signal.reason);
         }
-        chunks = newStreamsConvertChunks(chunks);
         let totalSize = 0;
         for (let i = 0; i < chunks.length; i++) {
           totalSize += chunks[i].byteLength;
@@ -949,8 +939,8 @@ if (getOptionValue('--experimental-stream-iter')) {
       },
 
       writeSync(chunk) {
+        chunk = newStreamsToWriterUint8Array(chunk);
         if (error || closed || asyncPending) return false;
-        chunk = newStreamsToUint8Array(chunk);
         const length = chunk.byteLength;
         if (length > syncWriteThreshold) return false;
         if (length === 0) return true;
@@ -980,8 +970,8 @@ if (getOptionValue('--experimental-stream-iter')) {
       },
 
       writevSync(chunks) {
-        if (error || closed || asyncPending) return false;
         chunks = newStreamsConvertChunks(chunks);
+        if (error || closed || asyncPending) return false;
         let totalSize = 0;
         for (let i = 0; i < chunks.length; i++) {
           totalSize += chunks[i].byteLength;
@@ -1016,6 +1006,7 @@ if (getOptionValue('--experimental-stream-iter')) {
       },
 
       end(options = kNullPrototo) {
+        const signal = newStreamsGetWriterSignal(options);
         if (error) {
           return PromiseReject(error);
         }
@@ -1025,15 +1016,8 @@ if (getOptionValue('--experimental-stream-iter')) {
         if (closing) {
           return pendingEndPromise;
         }
-        validateObject(options, 'options');
-        const {
-          signal,
-        } = options;
-        if (signal !== undefined) {
-          validateAbortSignal(signal, 'options.signal');
-          if (signal.aborted) {
-            return PromiseReject(signal.reason);
-          }
+        if (signal?.aborted) {
+          return PromiseReject(signal.reason);
         }
         closing = true;
         pendingEndPromise = PromisePrototypeThen(

lib/internal/quic/quic.js

@@ -138,8 +138,9 @@ const {
 } = require('internal/streams/iter/types');
 
 const {
-  toUint8Array,
   convertChunks,
+  getWriterSignal,
+  toWriterUint8Array,
 } = require('internal/streams/iter/utils');
 
 const {
@@ -164,7 +165,6 @@ const {
 } = require('internal/fs/promises');
 
 const {
-  validateAbortSignal,
   validateBoolean,
   validateFunction,
   validateInteger,
@@ -2197,15 +2197,14 @@ class QuicStream {
     // signals backpressure additional writes are rejected until the buffer has
     // capacity again.
 
-    function writeSync(chunk) {
+    function writeConvertedSync(chunk) {
       // If the stream is closed, errored, or write-ended, we cannot accept
       // more data. Refuse the sync write.
       // If a drain is already pending, another operation is waiting
       // for capacity. Refuse the sync write.
       if (closed || errored || stream.#inner.state.writeEnded || drainWakeup != null) {
         return false;
       }
-      chunk = toUint8Array(chunk);
       const len = TypedArrayPrototypeGetByteLength(chunk);
       if (len === 0) return true;
       // Refuse the write only when there is no available capacity at
@@ -2223,13 +2222,18 @@ class QuicStream {
       return true;
     }
 
-    async function write(chunk, options = kEmptyObject) {
-      validateObject(options, 'options');
-      const { signal } = options;
-      if (signal !== undefined) {
-        validateAbortSignal(signal, 'options.signal');
-        signal.throwIfAborted();
-      }
+    function writeSync(chunk) {
+      return writeConvertedSync(toWriterUint8Array(chunk));
+    }
+
+    function write(chunk, options = kEmptyObject) {
+      chunk = toWriterUint8Array(chunk);
+      const signal = getWriterSignal(options);
+      return writeAsync(chunk, signal);
+    }
+
+    async function writeAsync(chunk, signal) {
+      signal?.throwIfAborted();
       if (errored) throw error;
       if (closed || stream.#inner.state.writeEnded) {
         throw new ERR_INVALID_STATE('Writer is closed');
@@ -2242,16 +2246,15 @@ class QuicStream {
         throw new ERR_INVALID_STATE('Stream write buffer is full');
       }
 
-      if (!writeSync(chunk)) {
+      if (!writeConvertedSync(chunk)) {
         throw new ERR_INVALID_STATE('Stream write buffer is full');
       }
     }
 
-    function writevSync(chunks) {
+    function writevConvertedSync(chunks) {
       if (closed || errored || stream.#inner.state.writeEnded || drainWakeup != null) {
         return false;
       }
-      chunks = convertChunks(chunks);
       let len = 0;
       for (const c of chunks) len += TypedArrayPrototypeGetByteLength(c);
       if (len === 0) return true;
@@ -2262,13 +2265,18 @@ class QuicStream {
       return true;
     }
 
-    async function writev(chunks, options = kEmptyObject) {
-      validateObject(options, 'options');
-      const { signal } = options;
-      if (signal !== undefined) {
-        validateAbortSignal(signal, 'options.signal');
-        signal.throwIfAborted();
-      }
+    function writevSync(chunks) {
+      return writevConvertedSync(convertChunks(chunks));
+    }
+
+    function writev(chunks, options = kEmptyObject) {
+      chunks = convertChunks(chunks);
+      const signal = getWriterSignal(options);
+      return writevAsync(chunks, signal);
+    }
+
+    async function writevAsync(chunks, signal) {
+      signal?.throwIfAborted();
 
       if (errored) throw error;
       if (closed || stream.#inner.state.writeEnded) {
@@ -2283,7 +2291,7 @@ class QuicStream {
         throw new ERR_INVALID_STATE('Stream write buffer is full');
       }
 
-      if (!writevSync(chunks)) {
+      if (!writevConvertedSync(chunks)) {
         throw new ERR_INVALID_STATE('Stream write buffer is full');
       }
     }
@@ -2308,11 +2316,13 @@ class QuicStream {
       return totalBytesWritten;
     }
 
-    async function end(options = kEmptyObject) {
-      validateObject(options, 'options');
-      const { signal } = options;
+    function end(options = kEmptyObject) {
+      const signal = getWriterSignal(options);
+      return endAsync(signal);
+    }
+
+    async function endAsync(signal) {
       if (signal !== undefined) {
-        validateAbortSignal(signal, 'options.signal');
         signal.throwIfAborted();
         // TODO(@jasnell): The stream/iter spec allows individual sync end
         // calls to be canceled via an AbortSignal. We currently do not support