diff --git a/lib/internal/streams/iter/broadcast.js b/lib/internal/streams/iter/broadcast.js index 3592453036d303..ec4ae0b9ddcc7b 100644 --- a/lib/internal/streams/iter/broadcast.js +++ b/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](); diff --git a/lib/internal/streams/iter/classic.js b/lib/internal/streams/iter/classic.js index b4401632946f0d..22fe9318184d24 100644 --- a/lib/internal/streams/iter/classic.js +++ b/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(); diff --git a/lib/internal/streams/iter/push.js b/lib/internal/streams/iter/push.js index 23adc5edc1cd3e..98b0f6bf6806f0 100644 --- a/lib/internal/streams/iter/push.js +++ b/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 diff --git a/test/parallel/test-stream-iter-validation.js b/test/parallel/test-stream-iter-validation.js index ecefaeb171ec41..bc57197b9d6077 100644 --- a/test/parallel/test-stream-iter-validation.js +++ b/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());