stream: make additional fixups to stream/iter arg validation

Author: jasnell

lib/internal/streams/iter/broadcast.js

@@ -12,6 +12,8 @@ const {
   ArrayPrototypePush,
   ArrayPrototypeSlice,
   MathMax,
+  MathMin,
+  NumberMAX_SAFE_INTEGER,
   PromisePrototypeThen,
   PromiseResolve,
   PromiseWithResolvers,
@@ -30,6 +32,7 @@ const {
     ERR_INVALID_ARG_TYPE,
     ERR_INVALID_RETURN_VALUE,
     ERR_INVALID_STATE,
+    ERR_OPERATION_FAILED,
   },
 } = require('internal/errors');
 const {
@@ -115,7 +118,6 @@ class BroadcastImpl {
   #writer = null;
 
   constructor(options) {
-    validateBackpressure(options.backpressure);
     this.#options = options;
     this[kOnBufferDrained] = null;
   }
@@ -652,15 +654,15 @@ function broadcast(options = { __proto__: null }) {
     backpressure = 'strict',
     signal,
   } = options;
-  validateInteger(highWaterMark, 'options.highWaterMark', 1);
+  validateInteger(highWaterMark, 'options.highWaterMark');
   validateBackpressure(backpressure);
   if (signal !== undefined) {
     validateAbortSignal(signal, 'options.signal');
   }
 
   const opts = {
     __proto__: null,
-    highWaterMark: MathMax(1, highWaterMark),
+    highWaterMark: MathMax(1, MathMin(NumberMAX_SAFE_INTEGER, highWaterMark)),
     backpressure,
     signal,
   };
@@ -741,7 +743,7 @@ const Broadcast = {
         }
       } catch (error) {
         const err = isError(error) ? error :
-          new ERR_INVALID_ARG_TYPE('error', 'Error', String(error));
+          new ERR_OPERATION_FAILED(String(error));
         if (!w.failSync(err)) {
           await w.fail(err);
         }

lib/internal/streams/iter/consumers.js

@@ -251,7 +251,7 @@ function validateSyncConsumerOptions(options) {
  */
 function bytesSync(source, options = { __proto__: null }) {
   validateSyncConsumerOptions(options);
-  return concatBytes(collectSync(source, options?.limit));
+  return concatBytes(collectSync(source, options.limit));
 }
 
 /**

lib/internal/streams/iter/from.js

@@ -185,7 +185,9 @@ function tryStringCoercion(obj) {
   // Check for custom toString
   if (hasCustomToString(obj)) {
     const result = FunctionPrototypeCall(obj.toString, obj);
-    return result;
+    if (typeof result === 'string') {
+      return result;
+    }
   }
 
   return null;

lib/internal/streams/iter/pull.js

@@ -595,16 +595,15 @@ function pipeToSync(source, ...args) {
   const { transforms, writer, options } = parsePipeToArgs(args);
 
   // Handle transform-writer
-  const finalTransforms = ArrayPrototypeSlice(transforms);
   if (isTransformObject(writer)) {
-    ArrayPrototypePush(finalTransforms, writer);
+    ArrayPrototypePush(transforms, writer);
   }
 
   // Create pipeline
-  const pipeline = finalTransforms.length > 0 ?
+  const pipeline = transforms.length > 0 ?
     createSyncPipeline(
       { [SymbolIterator]: () => source[SymbolIterator]() },
-      finalTransforms) :
+      transforms) :
     source;
 
   let totalBytes = 0;
@@ -652,9 +651,8 @@ async function pipeTo(source, ...args) {
   }
 
   // Handle transform-writer
-  const finalTransforms = ArrayPrototypeSlice(transforms);
   if (isTransformObject(writer)) {
-    ArrayPrototypePush(finalTransforms, writer);
+    ArrayPrototypePush(transforms, writer);
   }
 
   const signal = options?.signal;
@@ -691,7 +689,7 @@ async function pipeTo(source, ...args) {
 
   try {
     // Fast path: no transforms - iterate directly
-    if (finalTransforms.length === 0) {
+    if (transforms.length === 0) {
       if (isAsyncIterable(source)) {
         for await (const batch of source) {
           signal?.throwIfAborted();
@@ -710,7 +708,7 @@ async function pipeTo(source, ...args) {
         { [SymbolIterator]: () => source[SymbolIterator]() };
 
       const pipeline = createAsyncPipeline(
-        streamableSource, finalTransforms, signal);
+        streamableSource, transforms, signal);
 
       for await (const batch of pipeline) {
         signal?.throwIfAborted();

lib/internal/streams/iter/push.js

@@ -10,6 +10,8 @@ const {
   ArrayPrototypePush,
   ArrayPrototypeSlice,
   MathMax,
+  MathMin,
+  NumberMAX_SAFE_INTEGER,
   PromiseResolve,
   PromiseWithResolvers,
   SymbolAsyncIterator,
@@ -79,18 +81,18 @@ class PushQueue {
   #signal;
   #abortHandler;
 
-  constructor(options = {}) {
+  constructor(options = { __proto__: null }) {
     const {
       highWaterMark = kPushDefaultHWM,
       backpressure = 'strict',
       signal,
     } = options;
-    validateInteger(highWaterMark, 'options.highWaterMark', 1);
+    validateInteger(highWaterMark, 'options.highWaterMark');
     validateBackpressure(backpressure);
     if (signal !== undefined) {
       validateAbortSignal(signal, 'options.signal');
     }
-    this.#highWaterMark = MathMax(1, highWaterMark);
+    this.#highWaterMark = MathMax(1, MathMin(NumberMAX_SAFE_INTEGER, highWaterMark));
     this.#backpressure = backpressure;
     this.#signal = signal;
     this.#abortHandler = undefined;

lib/internal/streams/iter/share.js

@@ -8,6 +8,8 @@
 const {
   ArrayPrototypePush,
   MathMax,
+  MathMin,
+  NumberMAX_SAFE_INTEGER,
   PromisePrototypeThen,
   PromiseResolve,
   PromiseWithResolvers,
@@ -158,7 +160,7 @@ class ShareImpl {
               }
 
               // Need to pull from source - check buffer limit
-              const canPull = await self.#waitForBufferSpace(state);
+              const canPull = await self.#waitForBufferSpace();
               if (!canPull) {
                 state.detached = true;
                 self.#consumers.delete(state);
@@ -230,7 +232,7 @@ class ShareImpl {
 
   // Internal methods
 
-  async #waitForBufferSpace(_state) {
+  async #waitForBufferSpace() {
     while (this.#buffer.length >= this.#options.highWaterMark) {
       if (this.#cancelled || this.#sourceError || this.#sourceExhausted) {
         return !this.#cancelled;
@@ -363,7 +365,6 @@ class SyncShareImpl {
   #cancelled = false;
 
   constructor(source, options) {
-    validateBackpressure(options.backpressure);
     this.#source = source;
     this.#options = options;
   }
@@ -387,6 +388,7 @@ class SyncShareImpl {
 
   #createRawConsumer() {
     const state = {
+      __proto__: null,
       cursor: this.#bufferStart,
       detached: false,
     };
@@ -577,15 +579,15 @@ function share(source, options = { __proto__: null }) {
     backpressure = 'strict',
     signal,
   } = options;
-  validateInteger(highWaterMark, 'options.highWaterMark', 1);
+  validateInteger(highWaterMark, 'options.highWaterMark');
   validateBackpressure(backpressure);
   if (signal !== undefined) {
     validateAbortSignal(signal, 'options.signal');
   }
 
   const opts = {
     __proto__: null,
-    highWaterMark: MathMax(1, highWaterMark),
+    highWaterMark: MathMax(1, MathMin(NumberMAX_SAFE_INTEGER, highWaterMark)),
     backpressure,
     signal,
   };
@@ -614,12 +616,12 @@ function shareSync(source, options = { __proto__: null }) {
     highWaterMark = kMultiConsumerDefaultHWM,
     backpressure = 'strict',
   } = options;
-  validateInteger(highWaterMark, 'options.highWaterMark', 1);
+  validateInteger(highWaterMark, 'options.highWaterMark');
   validateBackpressure(backpressure);
 
   const opts = {
     __proto__: null,
-    highWaterMark: MathMax(1, highWaterMark),
+    highWaterMark: MathMax(1, MathMin(NumberMAX_SAFE_INTEGER, highWaterMark)),
     backpressure,
   };
 

lib/internal/streams/iter/utils.js

@@ -110,12 +110,6 @@ function isPullOptions(value) {
   );
 }
 
-/**
- * Parse variadic arguments for pull/pullSync.
- * Returns { transforms, options }
- * @param {Array} args
- * @returns {{ transforms: Array, options: object|undefined }}
- */
 /**
  * Check if a value is a valid transform (function or transform object).
  * @param {unknown} value

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

@@ -20,9 +20,12 @@ const {
 
 // HighWaterMark must be integer >= 1
 assert.throws(() => push({ highWaterMark: 'bad' }), { code: 'ERR_INVALID_ARG_TYPE' });
-assert.throws(() => push({ highWaterMark: 0 }), { code: 'ERR_OUT_OF_RANGE' });
-assert.throws(() => push({ highWaterMark: -1 }), { code: 'ERR_OUT_OF_RANGE' });
 assert.throws(() => push({ highWaterMark: 1.5 }), { code: 'ERR_OUT_OF_RANGE' });
+// Values < 1 are clamped to 1
+push({ highWaterMark: 0 }).writer.endSync();
+push({ highWaterMark: -1 }).writer.endSync();
+// Values above MAX_SAFE_INTEGER are clamped
+push({ highWaterMark: Number.MAX_SAFE_INTEGER }).writer.endSync();
 
 // Signal must be AbortSignal
 assert.throws(() => push({ signal: 'bad' }), { code: 'ERR_INVALID_ARG_TYPE' });
@@ -80,7 +83,6 @@ assert.throws(() => pullSync(fromSync('a'), 42), { code: 'ERR_INVALID_ARG_TYPE'
 // =============================================================================
 
 assert.throws(() => broadcast({ highWaterMark: 'bad' }), { code: 'ERR_INVALID_ARG_TYPE' });
-assert.throws(() => broadcast({ highWaterMark: 0 }), { code: 'ERR_OUT_OF_RANGE' });
 assert.throws(() => broadcast({ signal: {} }), { code: 'ERR_INVALID_ARG_TYPE' });
 assert.throws(() => broadcast({ backpressure: 'bad' }), { code: 'ERR_INVALID_ARG_VALUE' });
 
@@ -94,15 +96,12 @@ assert.throws(() => Broadcast.from('bad'), { code: 'ERR_INVALID_ARG_TYPE' });
 
 assert.throws(() => share(42), { code: 'ERR_INVALID_ARG_TYPE' });
 assert.throws(() => share(from('a'), { highWaterMark: 'bad' }), { code: 'ERR_INVALID_ARG_TYPE' });
-assert.throws(() => share(from('a'), { highWaterMark: 0 }), { code: 'ERR_OUT_OF_RANGE' });
 assert.throws(() => share(from('a'), { signal: {} }), { code: 'ERR_INVALID_ARG_TYPE' });
 assert.throws(() => share(from('a'), { backpressure: 'bad' }), { code: 'ERR_INVALID_ARG_VALUE' });
 
 assert.throws(() => shareSync(42), { code: 'ERR_INVALID_ARG_TYPE' });
 assert.throws(() => shareSync(fromSync('a'), { highWaterMark: 'bad' }),
               { code: 'ERR_INVALID_ARG_TYPE' });
-assert.throws(() => shareSync(fromSync('a'), { highWaterMark: 0 }),
-              { code: 'ERR_OUT_OF_RANGE' });
 
 // Share.from / SyncShare.fromSync reject non-iterable
 assert.throws(() => Share.from(42), { code: 'ERR_INVALID_ARG_TYPE' });