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..39552d7324856e 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 @@ -668,7 +672,7 @@ function createReadable(queue) { }, async throw(error) { queue.consumerThrow(error); - return { __proto__: null, done: true, value: undefined }; + throw error; }, }; }, diff --git a/lib/internal/streams/iter/utils.js b/lib/internal/streams/iter/utils.js index ec02eca43f3385..dfcc103cda95b3 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(); @@ -267,6 +270,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 @@ -378,6 +392,7 @@ module.exports = { clampHWM, concatBytes, convertChunks, + getWriterSignal, getMinCursor, hasProtocol, isPullOptions, diff --git a/test/parallel/test-stream-iter-push-writer.js b/test/parallel/test-stream-iter-push-writer.js index f085f465951449..32961612804211 100644 --- a/test/parallel/test-stream-iter-push-writer.js +++ b/test/parallel/test-stream-iter-push-writer.js @@ -147,10 +147,16 @@ async function testOndrainRejectsOnConsumerThrow() { // Consumer throws via iterator.throw() before draining enough // to clear backpressure. The drain should reject. const iter = readable[Symbol.asyncIterator](); - await iter.throw(new Error('consumer error')); + const err = new Error('consumer error'); + const drainRejects = assert.rejects(drainPromise, (e) => e === err); + const pendingWriteRejects = pendingWrite.catch(() => {}); + await assert.rejects( + () => iter.throw(err), + (e) => e === err, + ); - await assert.rejects(drainPromise, /consumer error/); - await pendingWrite.catch(() => {}); // Ignore write rejection + await drainRejects; + await pendingWriteRejects; // Ignore write rejection } async function testWritev() { @@ -303,7 +309,11 @@ async function testConsumerThrowRejectsWrites() { writer.writeSync('a'); const iter = readable[Symbol.asyncIterator](); - await iter.throw(new Error('consumer boom')); + const err = new Error('consumer boom'); + await assert.rejects( + () => iter.throw(err), + (e) => e === err, + ); // Subsequent async writes should reject with the consumer's error await assert.rejects( @@ -312,6 +322,18 @@ async function testConsumerThrowRejectsWrites() { ); } +async function testConsumerThrowRejectsWithThrownError() { + const { readable } = push(); + + const iter = readable[Symbol.asyncIterator](); + const err = new Error('boom'); + + await assert.rejects( + () => iter.throw(err), + (e) => e === err, + ); +} + // end() resolves a pending read as done:true async function testEndResolvesPendingRead() { const { writer, readable } = push(); @@ -373,14 +395,16 @@ async function testConsumerThrowRejectsPendingRead() { await new Promise(setImmediate); const err = new Error('consumer read boom'); - const throwResult = await iter.throw(err); - assert.strictEqual(throwResult.value, undefined); - assert.strictEqual(throwResult.done, true); - - await assert.rejects( + const readRejects = assert.rejects( () => readPromise, (e) => e === err, ); + await assert.rejects( + () => iter.throw(err), + (e) => e === err, + ); + + await readRejects; } // end() while writes are pending rejects those writes @@ -527,6 +551,7 @@ Promise.all([ testWriteUint8Array(), testOndrainWaitsForDrain(), testConsumerThrowRejectsWrites(), + testConsumerThrowRejectsWithThrownError(), testEndResolvesPendingRead(), testFailRejectsPendingRead(), testFailRejectsFutureReadWithFalsyReason(), 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 // =============================================================================