stream: validate backpressure option at construction time

Invalid backpressure values like `'banana'` would fall through
switch statements at write time with a confusing `ERR_INVALID_STATE`
error about "`writeSync` should have handled non-strict policy".

Add `validateBackpressure()` in utils.js and call it from the
`PushQueue`, `BroadcastImpl`, `ShareImpl`, and `SyncShareImpl`
constructors. Invalid values now throw `ERR_INVALID_ARG_VALUE`
immediately at construction.

Author: jasnell

lib/internal/streams/iter/broadcast.js

@@ -49,6 +49,7 @@ const {
 const {
   allUint8Array,
   toUint8Array,
+  validateBackpressure,
 } = require('internal/streams/iter/utils');
 
 const {
@@ -111,6 +112,7 @@ class BroadcastImpl {
   #writer = null;
 
   constructor(options) {
+    validateBackpressure(options.backpressure);
     this.#options = options;
     this[kOnBufferDrained] = null;
   }

lib/internal/streams/iter/push.js

@@ -29,6 +29,7 @@ const {
 const {
   toUint8Array,
   allUint8Array,
+  validateBackpressure,
 } = require('internal/streams/iter/utils');
 
 const {
@@ -73,6 +74,7 @@ class PushQueue {
   constructor(options = {}) {
     this.#highWaterMark = MathMax(1, options.highWaterMark ?? 4);
     this.#backpressure = options.backpressure ?? 'strict';
+    validateBackpressure(this.#backpressure);
     this.#signal = options.signal;
     this.#abortHandler = undefined;
 

lib/internal/streams/iter/share.js

@@ -37,6 +37,7 @@ const {
 
 const {
   parsePullArgs,
+  validateBackpressure,
 } = require('internal/streams/iter/utils');
 
 const {
@@ -350,6 +351,7 @@ class SyncShareImpl {
   #cancelled = false;
 
   constructor(source, options) {
+    validateBackpressure(options.backpressure);
     this.#source = source;
     this.#options = options;
   }

lib/internal/streams/iter/utils.js

@@ -10,6 +10,11 @@ const {
 } = primordials;
 
 const { TextEncoder } = require('internal/encoding');
+const {
+  codes: {
+    ERR_INVALID_ARG_VALUE,
+  },
+} = require('internal/errors');
 
 // Shared TextEncoder instance for string conversion.
 const encoder = new TextEncoder();
@@ -112,10 +117,26 @@ function parsePullArgs(args) {
   return { transforms: args, options: undefined };
 }
 
+/**
+ * Validate backpressure option value.
+ * @param {string} value
+ */
+function validateBackpressure(value) {
+  if (value !== 'strict' &&
+      value !== 'block' &&
+      value !== 'drop-oldest' &&
+      value !== 'drop-newest') {
+    throw new ERR_INVALID_ARG_VALUE(
+      'options.backpressure', value,
+      'must be "strict", "block", "drop-oldest", or "drop-newest"');
+  }
+}
+
 module.exports = {
   toUint8Array,
   allUint8Array,
   concatBytes,
   isPullOptions,
   parsePullArgs,
+  validateBackpressure,
 };

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

@@ -352,4 +352,19 @@ Promise.all([
   testCancelledWriteRemovedFromQueue(),
   testOndrainResolvesFalseOnConsumerBreak(),
   testOndrainRejectsOnConsumerThrow(),
+  testInvalidBackpressure(),
 ]).then(common.mustCall());
+
+async function testInvalidBackpressure() {
+  assert.throws(() => push({ backpressure: 'banana' }), {
+    code: 'ERR_INVALID_ARG_VALUE',
+  });
+  assert.throws(() => push({ backpressure: '' }), {
+    code: 'ERR_INVALID_ARG_VALUE',
+  });
+
+  // Valid values should not throw
+  for (const bp of ['strict', 'block', 'drop-oldest', 'drop-newest']) {
+    push({ backpressure: bp });
+  }
+}