stream: validate options.signal in iter Writer methods

Signed-off-by: Sai Asish Y <say.apm35@gmail.com>

Author: SAY-5

lib/internal/streams/iter/broadcast.js

@@ -532,6 +532,7 @@ class BroadcastWriter {
   }
 
   write(chunk, options) {
+    validateAbortSignal(options?.signal, 'options.signal');
     // Fast path: no signal, writer open, buffer has space
     if (this.#canUseWriteFastPath(options)) {
       const converted = toUint8Array(chunk);
@@ -546,6 +547,7 @@ class BroadcastWriter {
     if (!ArrayIsArray(chunks)) {
       throw new ERR_INVALID_ARG_TYPE('chunks', 'Array', chunks);
     }
+    validateAbortSignal(options?.signal, 'options.signal');
     // Fast path: no signal, writer open, buffer has space
     if (this.#canUseWriteFastPath(options)) {
       const converted = convertChunks(chunks);
@@ -623,6 +625,7 @@ class BroadcastWriter {
 
   // end() is synchronous internally - signal accepted for interface compliance.
   end(options) {
+    validateAbortSignal(options?.signal, 'options.signal');
     if (this.#isClosed()) return this.#closed;
     this.#closed = PromiseResolve(this.#totalBytes);
     this.#broadcast[kEnd]();

lib/internal/streams/iter/classic.js

@@ -42,6 +42,7 @@ const {
 } = require('internal/errors');
 
 const {
+  validateAbortSignal,
   validateInteger,
   validateObject,
 } = require('internal/validators');
@@ -572,10 +573,11 @@ function fromWritable(writable, options = kNullPrototype) {
     // as 'error' events caught by our generic error handler, rejecting
     // the next pending operation rather than the already-resolved one.
     //
-    // The options.signal parameter from the Writer interface is ignored.
-    // Classic stream.Writable has no per-write abort signal support;
-    // cancellation should be handled at the pipeline level instead.
-    write(chunk) {
+    // The options.signal parameter from the Writer interface is validated
+    // but otherwise ignored. Classic stream.Writable has no per-write abort
+    // signal support; cancellation should be handled at the pipeline level.
+    write(chunk, options) {
+      validateAbortSignal(options?.signal, 'options.signal');
       if (!isWritable()) {
         return PromiseReject(new ERR_STREAM_WRITE_AFTER_END());
       }
@@ -617,10 +619,11 @@ function fromWritable(writable, options = kNullPrototype) {
       return PromiseResolve();
     },
 
-    writev(chunks) {
+    writev(chunks, options) {
       if (!ArrayIsArray(chunks)) {
         throw new ERR_INVALID_ARG_TYPE('chunks', 'Array', chunks);
       }
+      validateAbortSignal(options?.signal, 'options.signal');
       if (!isWritable()) {
         return PromiseReject(new ERR_STREAM_WRITE_AFTER_END());
       }
@@ -666,8 +669,9 @@ function fromWritable(writable, options = kNullPrototype) {
       return -1;
     },
 
-    // options.signal is ignored for the same reason as write().
-    end() {
+    // options.signal is validated but otherwise ignored, as in write().
+    end(options) {
+      validateAbortSignal(options?.signal, 'options.signal');
       if ((writable.writableFinished ?? false) ||
           (writable.destroyed ?? false)) {
         cleanup();

lib/internal/streams/iter/push.js

@@ -565,6 +565,7 @@ class PushWriter {
   }
 
   write(chunk, options) {
+    validateAbortSignal(options?.signal, 'options.signal');
     if (!options?.signal && this.#queue.canWriteSync()) {
       const bytes = toUint8Array(chunk);
       this.#queue.writeSync([bytes]);
@@ -578,6 +579,7 @@ class PushWriter {
     if (!ArrayIsArray(chunks)) {
       throw new ERR_INVALID_ARG_TYPE('chunks', 'Array', chunks);
     }
+    validateAbortSignal(options?.signal, 'options.signal');
     if (!options?.signal && this.#queue.canWriteSync()) {
       const bytes = convertChunks(chunks);
       this.#queue.writeSync(bytes);
@@ -601,6 +603,7 @@ class PushWriter {
   }
 
   end(options) {
+    validateAbortSignal(options?.signal, 'options.signal');
     const result = this.#queue.end();
     if (result === -2) {
       // Errored: reject with stored error

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

@@ -353,4 +353,42 @@ async function testAsyncValidation() {
   assert.strictEqual(typeof transform.transform, 'function');
 }
 
+// Writer methods validate options.signal on every surface
+{
+  const { writer } = push();
+  assert.throws(() => writer.write('a', { signal: 'bad' }),
+                { code: 'ERR_INVALID_ARG_TYPE' });
+  assert.throws(() => writer.writev(['a'], { signal: {} }),
+                { code: 'ERR_INVALID_ARG_TYPE' });
+  assert.throws(() => writer.end({ signal: 'bad' }),
+                { code: 'ERR_INVALID_ARG_TYPE' });
+  // A valid signal and an absent signal are both accepted.
+  writer.write('a', { signal: new AbortController().signal });
+  writer.write('b');
+  writer.endSync();
+}
+
+{
+  const { writer } = broadcast();
+  assert.throws(() => writer.write('a', { signal: 'bad' }),
+                { code: 'ERR_INVALID_ARG_TYPE' });
+  assert.throws(() => writer.writev(['a'], { signal: {} }),
+                { code: 'ERR_INVALID_ARG_TYPE' });
+  assert.throws(() => writer.end({ signal: 'bad' }),
+                { code: 'ERR_INVALID_ARG_TYPE' });
+  writer.endSync();
+}
+
+{
+  const { Writable } = require('stream');
+  const { fromWritable } = require('stream/iter');
+  const writer = fromWritable(new Writable({ write(chunk, enc, cb) { cb(); } }));
+  assert.throws(() => writer.write('a', { signal: 'bad' }),
+                { code: 'ERR_INVALID_ARG_TYPE' });
+  assert.throws(() => writer.writev(['a'], { signal: {} }),
+                { code: 'ERR_INVALID_ARG_TYPE' });
+  assert.throws(() => writer.end({ signal: 'bad' }),
+                { code: 'ERR_INVALID_ARG_TYPE' });
+}
+
 testAsyncValidation().then(common.mustCall());