diff --git a/lib/internal/streams/iter/consumers.js b/lib/internal/streams/iter/consumers.js index d2ecf5f3e21..a18794aa732 100644 --- a/lib/internal/streams/iter/consumers.js +++ b/lib/internal/streams/iter/consumers.js @@ -54,9 +54,9 @@ const { const { concatBytes, - createBatchEntry, getProtocolMethod, - validateBatchEntry, + recordChunk, + validateRecordedChunks, yieldAbortable, } = require('internal/streams/iter/utils'); @@ -95,17 +95,6 @@ function isMergeOptions(value) { // Shared chunk collection helpers // ============================================================================= -function flattenBatchEntries(entries) { - const chunks = []; - for (let i = 0; i < entries.length; i++) { - const batch = validateBatchEntry(entries[i]); - for (let j = 0; j < batch.length; j++) { - ArrayPrototypePush(chunks, batch[j]); - } - } - return chunks; -} - /** * Collect chunks from a sync source into an array. * @param {Iterable} source @@ -115,23 +104,20 @@ function flattenBatchEntries(entries) { function collectSync(source, limit) { // Normalize source via fromSync() - accepts strings, ArrayBuffers, protocols, etc. const normalized = fromSync(source); - const entries = []; + const chunks = []; + const checks = []; let totalBytes = 0; for (const batch of normalized) { - const entry = createBatchEntry(batch); - if (limit !== undefined) { - for (let i = 0; i < entry.views.length; i++) { - totalBytes += entry.views[i].byteLength; - if (totalBytes > limit) { - throw new ERR_OUT_OF_RANGE('totalBytes', `<= ${limit}`, totalBytes); - } + for (let i = 0; i < batch.length; i++) { + totalBytes += recordChunk(chunks, checks, batch[i]); + if (limit !== undefined && totalBytes > limit) { + throw new ERR_OUT_OF_RANGE('totalBytes', `<= ${limit}`, totalBytes); } } - ArrayPrototypePush(entries, entry); } - return flattenBatchEntries(entries); + return validateRecordedChunks(chunks, checks); } /** @@ -146,14 +132,17 @@ async function collectAsync(source, signal, limit) { // Normalize source via from() - accepts strings, ArrayBuffers, protocols, etc. const normalized = from(source); - const entries = []; + const chunks = []; + const checks = []; // Fast path: no signal and no limit if (!signal && limit === undefined) { for await (const batch of normalized) { - ArrayPrototypePush(entries, createBatchEntry(batch)); + for (let i = 0; i < batch.length; i++) { + recordChunk(chunks, checks, batch[i]); + } } - return flattenBatchEntries(entries); + return validateRecordedChunks(chunks, checks); } // Slow path: with signal or limit checks @@ -162,19 +151,15 @@ async function collectAsync(source, signal, limit) { for await (const batch of iterable) { signal?.throwIfAborted(); - const entry = createBatchEntry(batch); - if (limit !== undefined) { - for (let i = 0; i < entry.views.length; i++) { - totalBytes += entry.views[i].byteLength; - if (totalBytes > limit) { - throw new ERR_OUT_OF_RANGE('totalBytes', `<= ${limit}`, totalBytes); - } + for (let i = 0; i < batch.length; i++) { + totalBytes += recordChunk(chunks, checks, batch[i]); + if (limit !== undefined && totalBytes > limit) { + throw new ERR_OUT_OF_RANGE('totalBytes', `<= ${limit}`, totalBytes); } } - ArrayPrototypePush(entries, entry); } - return flattenBatchEntries(entries); + return validateRecordedChunks(chunks, checks); } /** diff --git a/lib/internal/streams/iter/utils.js b/lib/internal/streams/iter/utils.js index f84d663e917..99e5c5cd093 100644 --- a/lib/internal/streams/iter/utils.js +++ b/lib/internal/streams/iter/utils.js @@ -4,6 +4,7 @@ const { Array, ArrayBufferPrototypeGetByteLength, ArrayBufferPrototypeGetDetached, + ArrayBufferPrototypeGetResizable, ArrayPrototypePush, ArrayPrototypeSlice, PromiseResolve, @@ -235,6 +236,51 @@ function validateByteView(snapshot) { return value; } +/** + * Append a chunk to `chunks` for later concatenation, and the information + * needed to detect that it was resized or detached in the meantime to + * `checks`. This provides the same guarantee as snapshotByteView() and + * validateByteView() without allocating a snapshot for every chunk in the + * common case: a non-empty view of a fixed-length, non-shared ArrayBuffer can + * only change by the buffer being detached, which makes its byteLength 0, so + * its byteLength is all that needs to be recorded. + * @param {Uint8Array[]} chunks + * @param {Array} checks + * @param {Uint8Array} value + * @returns {number} The byteLength of `value`. + */ +function recordChunk(chunks, checks, value) { + const buffer = TypedArrayPrototypeGetBuffer(value); + const byteLength = TypedArrayPrototypeGetByteLength(value); + ArrayPrototypePush(chunks, value); + if (byteLength === 0 || isSharedArrayBuffer(buffer) || + ArrayBufferPrototypeGetResizable(buffer)) { + ArrayPrototypePush(checks, snapshotByteView(value)); + } else { + ArrayPrototypePush(checks, byteLength); + } + return byteLength; +} + +/** + * Validate chunks recorded with recordChunk(). + * @param {Uint8Array[]} chunks + * @param {Array} checks + * @returns {Uint8Array[]} `chunks` + */ +function validateRecordedChunks(chunks, checks) { + for (let i = 0; i < chunks.length; i++) { + const check = checks[i]; + if (typeof check !== 'number') { + validateByteView(check); + } else if (TypedArrayPrototypeGetByteLength(chunks[i]) !== check) { + throw new ERR_INVALID_STATE.TypeError( + 'Byte view was resized or detached after being accepted'); + } + } + return chunks; +} + function createBatchEntry(chunks) { const views = new Array(chunks.length); let byteLength = 0; @@ -489,6 +535,7 @@ module.exports = { concatBytes, convertChunks, createBatchEntry, + recordChunk, splitBatchEntry, getProtocolMethod, getWriterSignal, @@ -503,6 +550,7 @@ module.exports = { toWriterUint8Array, validateBackpressure, validateBatchEntry, + validateRecordedChunks, validateByteView, yieldAbortable, }; diff --git a/test/parallel/test-stream-iter-resizable-buffers.js b/test/parallel/test-stream-iter-resizable-buffers.js index b66973393ca..91f7429ec12 100644 --- a/test/parallel/test-stream-iter-resizable-buffers.js +++ b/test/parallel/test-stream-iter-resizable-buffers.js @@ -167,6 +167,35 @@ async function testPipeRejectsWriterResize() { ); } +async function testConsumersRejectDetachedViews() { + // Views of fixed-length buffers are tracked without a full snapshot; they + // must still be rejected when detached after being accepted. + const asyncBuffer = new ArrayBuffer(2); + async function* asyncSource() { + yield [new Uint8Array(asyncBuffer), Uint8Array.of(1)]; + asyncBuffer.transfer(); + } + await assert.rejects(array(asyncSource()), kResizeError); + const limitedBuffer = new ArrayBuffer(2); + async function* limitedSource() { + yield [new Uint8Array(limitedBuffer)]; + limitedBuffer.transfer(); + } + await assert.rejects(array(limitedSource(), { limit: 10 }), kResizeError); + + const syncBuffer = new ArrayBuffer(2); + function* syncSource() { + yield [new Uint8Array(syncBuffer, 1)]; + syncBuffer.transfer(); + } + assert.throws(() => arraySync(syncSource()), kResizeError); + + // Unchanged views are returned as-is. + const chunk = new Uint8Array(4); + const [result] = arraySync([[chunk]]); + assert.strictEqual(result, chunk); +} + Promise.all([ testBufferedViewMutationRejected(), testDropOldestUsesAcceptedByteLength(), @@ -174,5 +203,6 @@ Promise.all([ testBroadcastRejectsResizedBufferedView(), testShareRejectsResizedBufferedView(), testConsumersRejectResizedViews(), + testConsumersRejectDetachedViews(), testPipeRejectsWriterResize(), ]).then(common.mustCall());