From a56cebfdd6e358d30c03a59243f90a82d74cffad Mon Sep 17 00:00:00 2001 From: "Kamat, Trivikram" <16024985+trivikr@users.noreply.github.com> Date: Wed, 8 Jul 2026 22:09:15 -0700 Subject: [PATCH] stream: validate writer options signal Validate options.signal for stream/iter writer write(), writev(), and end() methods across push, broadcast, and fromWritable. Signed-off-by: Kamat, Trivikram <16024985+trivikr@users.noreply.github.com> Assisted-by: openai:gpt-5.5 --- lib/internal/streams/iter/broadcast.js | 20 +++++---- lib/internal/streams/iter/classic.js | 19 +++++--- lib/internal/streams/iter/push.js | 12 +++-- lib/internal/streams/iter/utils.js | 17 ++++++- test/parallel/test-stream-iter-validation.js | 47 +++++++++++++++++++- 5 files changed, 93 insertions(+), 22 deletions(-) diff --git a/lib/internal/streams/iter/broadcast.js b/lib/internal/streams/iter/broadcast.js index 3592453036d303..16635d965dc257 100644 --- a/lib/internal/streams/iter/broadcast.js +++ b/lib/internal/streams/iter/broadcast.js @@ -58,6 +58,7 @@ const { kResolvedPromise, clampHWM, convertChunks, + getWriterSignal, getMinCursor, hasProtocol, onSignalAbort, @@ -526,28 +527,30 @@ class BroadcastWriter { return this.#isClosedOrAborted() ? null : this.#broadcast[kGetDesiredSize](); } - #canUseWriteFastPath(options) { - return !options?.signal && !this.#isClosed() && !this.#aborted && + #canUseWriteFastPath(signal) { + return !signal && !this.#isClosed() && !this.#aborted && this.#broadcast[kCanWrite](); } write(chunk, options) { + const signal = getWriterSignal(options); // Fast path: no signal, writer open, buffer has space - if (this.#canUseWriteFastPath(options)) { + if (this.#canUseWriteFastPath(signal)) { const converted = toUint8Array(chunk); this.#broadcast[kWrite]([converted]); this.#totalBytes += TypedArrayPrototypeGetByteLength(converted); return kResolvedPromise; } - return this.#writevSlow([chunk], options); + return this.#writevSlow([chunk], signal); } writev(chunks, options) { if (!ArrayIsArray(chunks)) { throw new ERR_INVALID_ARG_TYPE('chunks', 'Array', chunks); } + const signal = getWriterSignal(options); // Fast path: no signal, writer open, buffer has space - if (this.#canUseWriteFastPath(options)) { + if (this.#canUseWriteFastPath(signal)) { const converted = convertChunks(chunks); this.#broadcast[kWrite](converted); for (let i = 0; i < converted.length; i++) { @@ -555,12 +558,10 @@ class BroadcastWriter { } return kResolvedPromise; } - return this.#writevSlow(chunks, options); + return this.#writevSlow(chunks, signal); } - async #writevSlow(chunks, options) { - const signal = options?.signal; - + async #writevSlow(chunks, signal) { // Check for pre-aborted signal?.throwIfAborted(); @@ -623,6 +624,7 @@ class BroadcastWriter { // end() is synchronous internally - signal accepted for interface compliance. end(options) { + getWriterSignal(options); 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..162f3c6b273e8f 100644 --- a/lib/internal/streams/iter/classic.js +++ b/lib/internal/streams/iter/classic.js @@ -61,6 +61,7 @@ const { } = require('internal/streams/iter/types'); const { + getWriterSignal, validateBackpressure, toUint8Array, } = require('internal/streams/iter/utils'); @@ -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 instead. + write(chunk, options) { + getWriterSignal(options); 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); } + getWriterSignal(options); if (!isWritable()) { return PromiseReject(new ERR_STREAM_WRITE_AFTER_END()); } @@ -666,8 +669,10 @@ 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 for the same reason as + // write(). + end(options) { + getWriterSignal(options); 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..944265c14c3f44 100644 --- a/lib/internal/streams/iter/push.js +++ b/lib/internal/streams/iter/push.js @@ -42,6 +42,7 @@ const { onSignalAbort, toUint8Array, convertChunks, + getWriterSignal, parsePullArgs, validateBackpressure, } = require('internal/streams/iter/utils'); @@ -565,26 +566,28 @@ class PushWriter { } write(chunk, options) { - if (!options?.signal && this.#queue.canWriteSync()) { + const signal = getWriterSignal(options); + if (!signal && this.#queue.canWriteSync()) { const bytes = toUint8Array(chunk); this.#queue.writeSync([bytes]); return kResolvedPromise; } const bytes = toUint8Array(chunk); - return this.#queue.writeAsync([bytes], options?.signal); + return this.#queue.writeAsync([bytes], signal); } writev(chunks, options) { if (!ArrayIsArray(chunks)) { throw new ERR_INVALID_ARG_TYPE('chunks', 'Array', chunks); } - if (!options?.signal && this.#queue.canWriteSync()) { + const signal = getWriterSignal(options); + if (!signal && this.#queue.canWriteSync()) { const bytes = convertChunks(chunks); this.#queue.writeSync(bytes); return kResolvedPromise; } const bytes = convertChunks(chunks); - return this.#queue.writeAsync(bytes, options?.signal); + return this.#queue.writeAsync(bytes, signal); } writeSync(chunk) { @@ -601,6 +604,7 @@ class PushWriter { } end(options) { + getWriterSignal(options); const result = this.#queue.end(); if (result === -2) { // Errored: reject with stored error diff --git a/lib/internal/streams/iter/utils.js b/lib/internal/streams/iter/utils.js index ca3d2531d6f9e0..fcfe488c0c62dc 100644 --- a/lib/internal/streams/iter/utils.js +++ b/lib/internal/streams/iter/utils.js @@ -35,7 +35,10 @@ const { isError } = require('internal/util'); const { isSharedArrayBuffer, isUint8Array } = require('internal/util/types'); -const { validateOneOf } = require('internal/validators'); +const { + validateAbortSignal, + validateOneOf, +} = require('internal/validators'); // Cached resolved promise to avoid allocating a new one on every sync fast-path. const kResolvedPromise = PromiseResolve(); @@ -263,6 +266,17 @@ function convertChunks(chunks) { return result; } +/** + * Validate Writer options and return options.signal. + * @param {object|undefined} options + * @returns {AbortSignal|undefined} + */ +function getWriterSignal(options) { + const signal = options?.signal; + validateAbortSignal(signal, 'options.signal'); + return signal; +} + /** * Wrap a caught value as an Error, converting non-Error values. * @param {unknown} error @@ -374,6 +388,7 @@ module.exports = { clampHWM, concatBytes, convertChunks, + getWriterSignal, getMinCursor, hasProtocol, isPullOptions, diff --git a/test/parallel/test-stream-iter-validation.js b/test/parallel/test-stream-iter-validation.js index ecefaeb171ec41..08eeaeee685656 100644 --- a/test/parallel/test-stream-iter-validation.js +++ b/test/parallel/test-stream-iter-validation.js @@ -3,8 +3,9 @@ const common = require('../common'); const assert = require('assert'); +const { Writable } = require('stream'); const { - from, fromSync, pull, pullSync, pipeTo, + from, fromSync, pull, pullSync, pipeTo, fromWritable, push, duplex, broadcast, Broadcast, share, shareSync, Share, SyncShare, bytes, bytesSync, text, textSync, @@ -42,6 +43,19 @@ assert.throws(() => push({ signal: {} }), { code: 'ERR_INVALID_ARG_TYPE' }); assert.throws(() => push(42, {}), { code: 'ERR_INVALID_ARG_TYPE' }); assert.throws(() => push('bad', {}), { code: 'ERR_INVALID_ARG_TYPE' }); +// Writer options.signal must be AbortSignal +{ + const { writer } = push(); + const badOptions = { signal: 'bad' }; + assert.throws(() => writer.write('a', badOptions), + { code: 'ERR_INVALID_ARG_TYPE' }); + assert.throws(() => writer.writev(['b'], badOptions), + { code: 'ERR_INVALID_ARG_TYPE' }); + assert.throws(() => writer.end(badOptions), + { code: 'ERR_INVALID_ARG_TYPE' }); + writer.endSync(); +} + // Writer.writev requires array { const { writer } = push(); @@ -147,6 +161,19 @@ assert.throws(() => broadcast({ highWaterMark: Number.MAX_SAFE_INTEGER + 1 }), assert.throws(() => broadcast({ signal: {} }), { code: 'ERR_INVALID_ARG_TYPE' }); assert.throws(() => broadcast({ backpressure: 'bad' }), { code: 'ERR_INVALID_ARG_VALUE' }); +// BroadcastWriter options.signal must be AbortSignal +{ + const { writer } = broadcast(); + const badOptions = { signal: 'bad' }; + assert.throws(() => writer.write('a', badOptions), + { code: 'ERR_INVALID_ARG_TYPE' }); + assert.throws(() => writer.writev(['b'], badOptions), + { code: 'ERR_INVALID_ARG_TYPE' }); + assert.throws(() => writer.end(badOptions), + { code: 'ERR_INVALID_ARG_TYPE' }); + writer.endSync(); +} + // BroadcastWriter.writev requires array { const { writer } = broadcast(); @@ -160,6 +187,24 @@ assert.throws(() => broadcast({ backpressure: 'bad' }), { code: 'ERR_INVALID_ARG // Broadcast.from rejects non-streamable input assert.throws(() => Broadcast.from(42), { code: 'ERR_INVALID_ARG_TYPE' }); +// fromWritable Writer options.signal must be AbortSignal +{ + const writable = new Writable({ + write(chunk, encoding, callback) { + callback(); + }, + }); + const writer = fromWritable(writable); + const badOptions = { signal: 'bad' }; + assert.throws(() => writer.write('a', badOptions), + { code: 'ERR_INVALID_ARG_TYPE' }); + assert.throws(() => writer.writev(['b'], badOptions), + { code: 'ERR_INVALID_ARG_TYPE' }); + assert.throws(() => writer.end(badOptions), + { code: 'ERR_INVALID_ARG_TYPE' }); + writable.destroy(); +} + // ============================================================================= // share() / shareSync() validation // =============================================================================