diff --git a/lib/internal/streams/iter/broadcast.js b/lib/internal/streams/iter/broadcast.js index 0674d4d57037..0dc6364768d5 100644 --- a/lib/internal/streams/iter/broadcast.js +++ b/lib/internal/streams/iter/broadcast.js @@ -21,7 +21,6 @@ const { SymbolAsyncDispose, SymbolAsyncIterator, SymbolDispose, - TypedArrayPrototypeGetByteLength, } = primordials; const { lazyDOMException } = require('internal/util'); @@ -52,13 +51,14 @@ const { } = require('internal/streams/iter/from'); const { - pull: pullWithTransforms, + pullWithConsumerCleanup, } = require('internal/streams/iter/pull'); const { kMultiConsumerDefaultBudget, kResolvedPromise, convertChunks, + createBatchEntry, getWriterSignal, getMinCursor, hasProtocol, @@ -67,6 +67,7 @@ const { wrapError, toUint8Array, validateBackpressure, + validateBatchEntry, } = require('internal/streams/iter/utils'); const { @@ -157,12 +158,7 @@ class BroadcastImpl { // When no transforms, return rawConsumer directly (controller elided // per PULL-02 optimization -- no transforms means no signal recipient). if (transforms.length > 0 || signal) { - const pullArgs = [...transforms]; - if (signal) { - ArrayPrototypePush(pullArgs, - { __proto__: null, signal }); - } - return pullWithTransforms(rawConsumer, ...pullArgs); + return pullWithConsumerCleanup(rawConsumer, transforms, signal); } return rawConsumer; } @@ -220,7 +216,8 @@ class BroadcastImpl { const bufferIndex = state.cursor - self.#bufferStart; if (bufferIndex < self.#buffer.length) { - const chunk = self.#buffer.get(bufferIndex); + const chunk = self.#readEntry(self.#buffer.get(bufferIndex)); + if (chunk === null) return PromiseReject(self.#error); const cursor = state.cursor; state.cursor++; if (cursor === self.#cachedMinCursor && @@ -309,10 +306,10 @@ class BroadcastImpl { // Methods accessed by BroadcastWriter via symbol keys - [kWrite](chunk) { + [kWrite](entry) { if (this.#ended || this.#cancelled) return false; - const batchSize = this.#batchByteSize(chunk); + const batchSize = entry.byteLength; // Skip empty chunks -- zero-byte writes would accumulate infinitely // without ever triggering backpressure under a byte-budget model. @@ -327,7 +324,7 @@ class BroadcastImpl { while (this.#bufferedBytes >= this.#options.budget && this.#buffer.length > 0) { const evicted = this.#buffer.shift(); - this.#bufferedBytes -= this.#batchByteSize(evicted); + this.#bufferedBytes -= evicted.byteLength; this.#bufferStart++; } for (const consumer of this.#consumers) { @@ -343,7 +340,7 @@ class BroadcastImpl { } } - this.#buffer.push(chunk); + this.#buffer.push(entry); this.#bufferedBytes += batchSize; this.#notifyConsumers(); return true; @@ -357,7 +354,8 @@ class BroadcastImpl { while (consumer.resolve) { const bufferIndex = consumer.cursor - this.#bufferStart; if (bufferIndex < this.#buffer.length) { - const chunk = this.#buffer.get(bufferIndex); + const chunk = this.#readEntry(this.#buffer.get(bufferIndex)); + if (chunk === null) return; const cursor = consumer.cursor; consumer.cursor++; if (cursor === this.#cachedMinCursor && @@ -438,7 +436,7 @@ class BroadcastImpl { if (trimCount > 0) { for (let i = 0; i < trimCount; i++) { const evicted = this.#buffer.get(i); - this.#bufferedBytes -= this.#batchByteSize(evicted); + this.#bufferedBytes -= evicted.byteLength; } this.#buffer.trimFront(trimCount); this.#bufferStart = this.#cachedMinCursor; @@ -450,12 +448,16 @@ class BroadcastImpl { } } - #batchByteSize(batch) { - let size = 0; - for (let i = 0; i < batch.length; i++) { - size += TypedArrayPrototypeGetByteLength(batch[i]); + #readEntry(entry) { + try { + return validateBatchEntry(entry); + } catch (error) { + this.#writer.fail(error); + if (this.#error === undefined) this[kAbort](error); + this.#buffer.clear(); + this.#bufferedBytes = 0; + return null; } - return size; } #notifyConsumers() { @@ -469,7 +471,8 @@ class BroadcastImpl { if (consumer.resolve) { const bufferIndex = consumer.cursor - this.#bufferStart; if (bufferIndex < this.#buffer.length) { - const chunk = this.#buffer.get(bufferIndex); + const chunk = this.#readEntry(this.#buffer.get(bufferIndex)); + if (chunk === null) return; const cursor = consumer.cursor; consumer.cursor++; if (cursor === this.#cachedMinCursor && @@ -592,8 +595,9 @@ class BroadcastWriter { // Fast path: no signal, writer open, buffer has space if (this.#canUseWriteFastPath(signal)) { const converted = toUint8Array(chunk); - this.#broadcast[kWrite]([converted]); - this.#totalBytes += TypedArrayPrototypeGetByteLength(converted); + const batch = createBatchEntry([converted]); + this.#broadcast[kWrite](batch); + this.#totalBytes += batch.byteLength; return kResolvedPromise; } return this.#writevSlow([chunk], signal); @@ -605,11 +609,12 @@ class BroadcastWriter { // Fast path: no signal, writer open, buffer has space if (this.#canUseWriteFastPath(signal)) { const converted = convertChunks(chunks); - this.#broadcast[kWrite](converted); - for (let i = 0; i < converted.length; i++) { - this.#totalBytes += TypedArrayPrototypeGetByteLength(converted[i]); + const batch = createBatchEntry(converted); + if (this.#state === 'open' && this.#broadcast[kWrite](batch)) { + this.#totalBytes += batch.byteLength; + return kResolvedPromise; } - return kResolvedPromise; + return this.#writeBatchSlow(batch, signal); } return this.#writevSlow(chunks, signal); } @@ -624,12 +629,23 @@ class BroadcastWriter { signal?.throwIfAborted(); - const converted = convertChunks(chunks); + const batch = createBatchEntry(convertChunks(chunks)); - if (this.#broadcast[kWrite](converted)) { - for (let i = 0; i < converted.length; i++) { - this.#totalBytes += TypedArrayPrototypeGetByteLength(converted[i]); - } + return this.#writeBatchSlow(batch, signal); + } + + async #writeBatchSlow(batch, signal) { + if (this.#state === 'errored') { + throw this.#error; + } + if (this.#state !== 'open') { + throw new ERR_INVALID_STATE.TypeError('Writer is closed'); + } + + signal?.throwIfAborted(); + + if (this.#broadcast[kWrite](batch)) { + this.#totalBytes += batch.byteLength; return; } @@ -641,11 +657,11 @@ class BroadcastWriter { 'Backpressure violation: too many pending writes. ' + 'Await each write() call to respect backpressure.'); } - return this.#createPendingWrite(converted, signal); + return this.#createPendingWrite(batch, signal); } // 'unbounded' policy - return this.#createPendingWrite(converted, signal); + return this.#createPendingWrite(batch, signal); } writeSync(chunk) { @@ -653,8 +669,9 @@ class BroadcastWriter { if (!this.#broadcast[kCanWrite]()) return false; const converted = toUint8Array(chunk); - if (this.#broadcast[kWrite]([converted])) { - this.#totalBytes += TypedArrayPrototypeGetByteLength(converted); + const batch = createBatchEntry([converted]); + if (this.#broadcast[kWrite](batch)) { + this.#totalBytes += batch.byteLength; return true; } return false; @@ -665,10 +682,9 @@ class BroadcastWriter { if (this.#state !== 'open') return false; if (!this.#broadcast[kCanWrite]()) return false; const converted = convertChunks(chunks); - if (this.#broadcast[kWrite](converted)) { - for (let i = 0; i < converted.length; i++) { - this.#totalBytes += TypedArrayPrototypeGetByteLength(converted[i]); - } + const batch = createBatchEntry(converted); + if (this.#broadcast[kWrite](batch)) { + this.#totalBytes += batch.byteLength; return true; } return false; @@ -757,9 +773,9 @@ class BroadcastWriter { * promise rejects. Signal listeners are cleaned up on normal resolution. * @returns {Promise} */ - #createPendingWrite(chunk, signal) { + #createPendingWrite(batch, signal) { const { promise, resolve, reject } = PromiseWithResolvers(); - const entry = { __proto__: null, chunk, resolve, reject }; + const entry = { __proto__: null, batch, resolve, reject }; this.#pendingWrites.push(entry); if (signal) { wireBroadcastWriteSignal(entry, signal, resolve, reject, this); @@ -770,14 +786,17 @@ class BroadcastWriter { #resolvePendingWrites() { while (this.#pendingWrites.length > 0 && this.#broadcast[kCanWrite]()) { const pending = this.#pendingWrites.shift(); - if (this.#broadcast[kWrite](pending.chunk)) { - for (let i = 0; i < pending.chunk.length; i++) { - this.#totalBytes += TypedArrayPrototypeGetByteLength(pending.chunk[i]); + try { + validateBatchEntry(pending.batch); + if (this.#broadcast[kWrite](pending.batch)) { + this.#totalBytes += pending.batch.byteLength; + pending.resolve(); + } else { + this.#pendingWrites.unshift(pending); + break; } - pending.resolve(); - } else { - this.#pendingWrites.unshift(pending); - break; + } catch (error) { + pending.reject(error); } } this.#finishEndIfReady(); @@ -811,18 +830,18 @@ function wireBroadcastWriteSignal(entry, signal, resolve, reject, self) { const pendingWrites = getBroadcastPendingWrites(self); const idx = pendingWrites.indexOf(entry); if (idx !== -1) pendingWrites.removeAt(idx); - entry.chunk = null; + entry.batch = null; reject(signal.reason ?? lazyDOMException('Aborted', 'AbortError')); if (idx !== -1) self[kPendingWriteRemoved](); }; entry.resolve = function() { signal.removeEventListener('abort', onAbort); - entry.chunk = null; + entry.batch = null; resolve(); }; entry.reject = function(reason) { signal.removeEventListener('abort', onAbort); - entry.chunk = null; + entry.batch = null; reject(reason); }; signal.addEventListener('abort', onAbort, { __proto__: null, once: true }); diff --git a/lib/internal/streams/iter/consumers.js b/lib/internal/streams/iter/consumers.js index e3bd7856dcd4..446fde1d88e3 100644 --- a/lib/internal/streams/iter/consumers.js +++ b/lib/internal/streams/iter/consumers.js @@ -18,6 +18,7 @@ const { Promise, PromisePrototypeThen, SafePromiseAllReturnVoid, + Symbol, SymbolAsyncIterator, TypedArrayPrototypeGetBuffer, TypedArrayPrototypeGetByteLength, @@ -54,6 +55,8 @@ const { const { concatBytes, + createBatchEntry, + validateBatchEntry, yieldAbortable, } = require('internal/streams/iter/utils'); @@ -89,6 +92,17 @@ 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 @@ -98,23 +112,23 @@ function isMergeOptions(value) { function collectSync(source, limit) { // Normalize source via fromSync() - accepts strings, ArrayBuffers, protocols, etc. const normalized = fromSync(source); - const chunks = []; + const entries = []; let totalBytes = 0; for (const batch of normalized) { - for (let i = 0; i < batch.length; i++) { - const chunk = batch[i]; - if (limit !== undefined) { - totalBytes += TypedArrayPrototypeGetByteLength(chunk); + 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); } } - ArrayPrototypePush(chunks, chunk); } + ArrayPrototypePush(entries, entry); } - return chunks; + return flattenBatchEntries(entries); } /** @@ -131,16 +145,14 @@ async function collectAsync(source, signal, limit) { const abortableSource = signal && isAsyncIterable(source) ? yieldAbortable(source, signal) : source; const normalized = from(abortableSource); - const chunks = []; + const entries = []; // Fast path: no signal and no limit if (!signal && limit === undefined) { for await (const batch of normalized) { - for (let i = 0; i < batch.length; i++) { - ArrayPrototypePush(chunks, batch[i]); - } + ArrayPrototypePush(entries, createBatchEntry(batch)); } - return chunks; + return flattenBatchEntries(entries); } // Slow path: with signal or limit checks @@ -149,19 +161,19 @@ async function collectAsync(source, signal, limit) { for await (const batch of iterable) { signal?.throwIfAborted(); - for (let i = 0; i < batch.length; i++) { - const chunk = batch[i]; - if (limit !== undefined) { - totalBytes += TypedArrayPrototypeGetByteLength(chunk); + 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); } } - ArrayPrototypePush(chunks, chunk); } + ArrayPrototypePush(entries, entry); } - return chunks; + return flattenBatchEntries(entries); } /** @@ -390,6 +402,8 @@ function ondrain(drainable) { // Merge Utility // ============================================================================= +const kNoMergeError = Symbol('kNoMergeError'); + /** * Merge multiple async iterables by yielding values in temporal order. * @param {...(AsyncIterable|object)} args @@ -440,6 +454,7 @@ function merge(...args) { let activeCount = normalized.length; let waitResolve = null; let onAbort; + let stopped = false; if (signal) { onAbort = () => { @@ -457,11 +472,13 @@ function merge(...args) { // Called when a source's .next() settles. Pushes the result into // the ready queue and wakes the consumer if it's waiting. const onSettled = (iterator, result) => { + if (stopped) return; if (result.done) { activeCount--; } else { ArrayPrototypePush(ready, { __proto__: null, + kind: 'value', iterator, value: result.value, }); @@ -472,6 +489,19 @@ function merge(...args) { } }; + const onRejected = (reason) => { + if (stopped) return; + ArrayPrototypePush(ready, { + __proto__: null, + kind: 'error', + reason, + }); + if (waitResolve) { + waitResolve(); + waitResolve = null; + } + }; + // Start one .next() per source const iterators = []; for (let i = 0; i < normalized.length; i++) { @@ -480,17 +510,12 @@ function merge(...args) { PromisePrototypeThen( iterator.next(), (r) => onSettled(iterator, r), - (err) => { - ArrayPrototypePush(ready, { __proto__: null, error: err }); - if (waitResolve) { - waitResolve(); - waitResolve = null; - } - }, + onRejected, ); } - let primaryError; + let completed = false; + let primaryError = kNoMergeError; try { while (activeCount > 0 || ready.length > 0) { signal?.throwIfAborted(); @@ -498,20 +523,14 @@ function merge(...args) { // Drain ready queue synchronously while (ready.length > 0) { const item = ArrayPrototypeShift(ready); - if (item?.error) { - throw item.error; + if (item.kind === 'error') { + throw item.reason; } yield item.value; PromisePrototypeThen( item.iterator.next(), (r) => onSettled(item.iterator, r), - (err) => { - ArrayPrototypePush(ready, { __proto__: null, error: err }); - if (waitResolve) { - waitResolve(); - waitResolve = null; - } - }, + onRejected, ); } @@ -526,9 +545,11 @@ function merge(...args) { }); } } + completed = true; } catch (err) { primaryError = err; } finally { + stopped = true; if (onAbort !== undefined) { signal.removeEventListener('abort', onAbort); } @@ -538,7 +559,7 @@ function merge(...args) { await cleanupIterators( iterators, primaryError, - signal?.aborted && primaryError === signal.reason, + !completed, ); } }, @@ -546,7 +567,7 @@ function merge(...args) { } async function cleanupIterators(iterators, primaryError, skipAwaitCleanup) { - let cleanupError; + let cleanupError = kNoMergeError; await SafePromiseAllReturnVoid(iterators, async (iterator) => { if (iterator.return) { try { @@ -558,12 +579,12 @@ async function cleanupIterators(iterators, primaryError, skipAwaitCleanup) { } } catch (err) { // Keep the first cleanup error encountered. - cleanupError ??= err; + if (cleanupError === kNoMergeError) cleanupError = err; } } }); - if (cleanupError !== undefined) { - if (primaryError !== undefined) { + if (cleanupError !== kNoMergeError) { + if (primaryError !== kNoMergeError) { // Both a primary error and a cleanup error occurred. // Wrap in SuppressedError so neither is lost: // .error = primaryError, .suppressed = cleanupError. @@ -573,7 +594,7 @@ async function cleanupIterators(iterators, primaryError, skipAwaitCleanup) { // No primary error - the cleanup error is the only error. throw cleanupError; } - if (primaryError !== undefined) { + if (primaryError !== kNoMergeError) { throw primaryError; } } diff --git a/lib/internal/streams/iter/duplex.js b/lib/internal/streams/iter/duplex.js index b37b91279232..50a04961b1f2 100644 --- a/lib/internal/streams/iter/duplex.js +++ b/lib/internal/streams/iter/duplex.js @@ -6,6 +6,7 @@ // channel's writer appears in the other channel's readable. const { + SafePromiseAllReturnVoid, SymbolAsyncDispose, SymbolAsyncIterator, } = primordials; @@ -50,68 +51,8 @@ function duplex(options = { __proto__: null }) { backpressure: b?.backpressure ?? backpressure, }); - let aClosed = false; - let bClosed = false; - // Track active iterators so close() can call .return() on them - let aReadableIterator = null; - let bReadableIterator = null; - - const channelA = { - __proto__: null, - get writer() { return aWriter; }, - // Wrap readable to track the iterator for cleanup on close() - get readable() { - return { - __proto__: null, - [SymbolAsyncIterator]() { - const iter = aReadable[SymbolAsyncIterator](); - aReadableIterator = iter; - return iter; - }, - }; - }, - async close() { - if (aClosed) return; - aClosed = true; - // End the writer (signals end-of-stream to B's readable) - aWriter.endSync(); - // Stop iteration of this channel's readable - if (aReadableIterator?.return) { - await aReadableIterator.return(); - aReadableIterator = null; - } - }, - [SymbolAsyncDispose]() { - return this.close(); - }, - }; - - const channelB = { - __proto__: null, - get writer() { return bWriter; }, - get readable() { - return { - __proto__: null, - [SymbolAsyncIterator]() { - const iter = bReadable[SymbolAsyncIterator](); - bReadableIterator = iter; - return iter; - }, - }; - }, - async close() { - if (bClosed) return; - bClosed = true; - bWriter.endSync(); - if (bReadableIterator?.return) { - await bReadableIterator.return(); - bReadableIterator = null; - } - }, - [SymbolAsyncDispose]() { - return this.close(); - }, - }; + const channelA = createDuplexChannel(aWriter, aReadable); + const channelB = createDuplexChannel(bWriter, bReadable); // Signal handler: fail both writers with the abort reason so consumers // see the error. This is an error-path shutdown, not a clean close. @@ -132,6 +73,38 @@ function duplex(options = { __proto__: null }) { return [channelA, channelB]; } +function createDuplexChannel(writer, readable) { + // A push readable has one shared consumer state. Keeping an iterator from + // creation lets close() terminate that state even if no caller has iterated. + const closeIterator = readable[SymbolAsyncIterator](); + let closePromise; + + return { + __proto__: null, + get writer() { return writer; }, + get readable() { return readable; }, + close() { + closePromise ??= closeDuplexChannel(writer, closeIterator); + return closePromise; + }, + [SymbolAsyncDispose]() { + return this.close(); + }, + }; +} + +async function closeDuplexChannel(writer, closeIterator) { + const result = writer.endSync(); + const endPromise = result < 0 ? writer.end() : undefined; + const returnPromise = closeIterator.return(); + + if (endPromise !== undefined) { + await SafePromiseAllReturnVoid([endPromise, returnPromise]); + } else { + await returnPromise; + } +} + module.exports = { duplex, }; diff --git a/lib/internal/streams/iter/pull.js b/lib/internal/streams/iter/pull.js index b6c2d9849c42..c1e42f03dc38 100644 --- a/lib/internal/streams/iter/pull.js +++ b/lib/internal/streams/iter/pull.js @@ -12,11 +12,11 @@ const { ArrayIsArray, ArrayPrototypePush, ArrayPrototypeSlice, + FunctionPrototypeCall, PromisePrototypeThen, PromiseResolve, SymbolAsyncIterator, SymbolIterator, - TypedArrayPrototypeGetByteLength, Uint8Array, } = primordials; @@ -24,6 +24,7 @@ const { codes: { ERR_INVALID_ARG_TYPE, ERR_INVALID_ARG_VALUE, + ERR_INVALID_STATE, ERR_OUT_OF_RANGE, }, } = require('internal/errors'); @@ -34,7 +35,10 @@ const { isPromise, isUint8Array, } = require('internal/util/types'); -const { AbortController } = require('internal/abort_controller'); +const { + AbortController, + AbortSignal, +} = require('internal/abort_controller'); const { arrayBufferViewToUint8Array, @@ -48,11 +52,14 @@ const { } = require('internal/streams/iter/from'); const { + createBatchEntry, isPullOptions, isTransform, isTransformObject, parsePullArgs, toUint8Array, + validateBatchEntry, + validateByteView, wrapError, yieldAbortable, } = require('internal/streams/iter/utils'); @@ -862,8 +869,103 @@ function pull(source, ...args) { return { __proto__: null, - async *[SymbolAsyncIterator]() { - yield* createAsyncPipeline(from(source), transforms, signal); + [SymbolAsyncIterator]() { + const controller = new AbortController(); + const iteratorSignal = signal === undefined ? + controller.signal : AbortSignal.any([signal, controller.signal]); + + async function* pipeline() { + yield* createAsyncPipeline(from(source), transforms, iteratorSignal); + } + const iterator = pipeline(); + + return { + __proto__: null, + next(value) { + return iterator.next(value); + }, + return(value) { + controller.abort(lazyDOMException('Aborted', 'AbortError')); + return iterator.return(value); + }, + throw(error) { + controller.abort(error); + return iterator.throw(error); + }, + [SymbolAsyncIterator]() { + return this; + }, + }; + }, + }; +} + +// Keep ownership of a bonded consumer outside the transform pipeline so it can +// be detached even when the pipeline never starts or terminates early. +function pullWithConsumerCleanup(source, transforms, signal) { + const sourceIterator = source[SymbolAsyncIterator](); + const pipelineSource = { + __proto__: null, + [SymbolAsyncIterator]() { + return sourceIterator; + }, + }; + const pipeline = signal === undefined ? + pull(pipelineSource, ...transforms) : + pull(pipelineSource, ...transforms, { __proto__: null, signal }); + let sourceClosed = false; + let abortHandler; + + function closeSource(method, value) { + if (sourceClosed) return; + sourceClosed = true; + if (abortHandler !== undefined) { + signal.removeEventListener('abort', abortHandler); + } + const close = sourceIterator[method] ?? sourceIterator.return; + if (typeof close === 'function') { + const result = FunctionPrototypeCall(close, sourceIterator, value); + PromisePrototypeThen(PromiseResolve(result), undefined, () => {}); + } + } + + if (signal !== undefined) { + abortHandler = () => closeSource('throw', signal.reason); + signal.addEventListener('abort', abortHandler, + { __proto__: null, once: true }); + if (signal.aborted) abortHandler(); + } + + return { + __proto__: null, + [SymbolAsyncIterator]() { + const iterator = pipeline[SymbolAsyncIterator](); + return { + __proto__: null, + next(value) { + return PromisePrototypeThen( + iterator.next(value), + (result) => { + if (result.done) closeSource('return'); + return result; + }, + (error) => { + closeSource('throw', error); + throw error; + }); + }, + return(value) { + closeSource('return', value); + return iterator.return(value); + }, + throw(error) { + closeSource('throw', error); + return iterator.throw(error); + }, + [SymbolAsyncIterator]() { + return this; + }, + }; }, }; } @@ -880,6 +982,13 @@ function pull(source, ...args) { */ function pipeToSync(source, ...args) { const { transforms, writer, options } = parsePipeToArgs(args, 'writeSync'); + const hasWritevSync = typeof writer.writevSync === 'function'; + const endSync = writer.endSync; + + if (!options?.preventClose && typeof endSync !== 'function') { + throw new ERR_INVALID_ARG_TYPE( + 'writer.endSync', 'Function', endSync); + } // Normalize source and create pipeline const normalized = fromSync(source); @@ -888,34 +997,37 @@ function pipeToSync(source, ...args) { normalized; let totalBytes = 0; - const hasWritevSync = typeof writer.writevSync === 'function'; - const hasEndSync = typeof writer.endSync === 'function'; try { for (const batch of pipeline) { + const entry = createBatchEntry(batch); if (hasWritevSync && batch.length > 1) { - if (writer.writevSync(batch) === false) { + const accepted = writer.writevSync(validateBatchEntry(entry)); + validateBatchEntry(entry); + if (accepted === false) { throw new ERR_OUT_OF_RANGE( 'write', 'within byte budget', 'budget exhausted'); } - for (let i = 0; i < batch.length; i++) { - totalBytes += TypedArrayPrototypeGetByteLength(batch[i]); - } + totalBytes += entry.byteLength; } else { - for (let i = 0; i < batch.length; i++) { - const chunk = batch[i]; - if (writer.writeSync(chunk) === false) { + for (let i = 0; i < entry.views.length; i++) { + const view = entry.views[i]; + const chunk = validateByteView(view); + const accepted = writer.writeSync(chunk); + validateByteView(view); + if (accepted === false) { throw new ERR_OUT_OF_RANGE( 'write', 'within byte budget', 'budget exhausted'); } - totalBytes += TypedArrayPrototypeGetByteLength(chunk); + totalBytes += view.byteLength; } } } if (!options?.preventClose) { - if (!hasEndSync || writer.endSync() < 0) { - writer.end?.(); + if (FunctionPrototypeCall(endSync, writer) < 0) { + throw new ERR_INVALID_STATE( + 'Writer could not be closed synchronously'); } } } catch (error) { @@ -957,19 +1069,26 @@ async function pipeTo(source, ...args) { // Async fallback for writeBatch when sync write fails partway through. // Continues writing from batch[startIndex] using async write(). - async function writeBatchAsyncFallback(batch, startIndex) { - for (let i = startIndex; i < batch.length; i++) { - const chunk = batch[i]; - if (hasWriteSync && writer.writeSync(chunk)) { - // Sync retry succeeded - } else { - const result = writer.write( - chunk, signal ? { __proto__: null, signal } : undefined); - if (result !== undefined) { - await result; + async function writeBatchAsyncFallback(entry, startIndex) { + for (let i = startIndex; i < entry.views.length; i++) { + const view = entry.views[i]; + if (hasWriteSync) { + const chunk = validateByteView(view); + if (writer.writeSync(chunk)) { + validateByteView(view); + totalBytes += view.byteLength; + continue; } + validateByteView(view); + } + const result = writer.write( + validateByteView(view), + signal ? { __proto__: null, signal } : undefined); + if (result !== undefined) { + await result; } - totalBytes += TypedArrayPrototypeGetByteLength(chunk); + validateByteView(view); + totalBytes += view.byteLength; } } @@ -977,34 +1096,37 @@ async function pipeTo(source, ...args) { // Returns undefined on sync success, or a Promise when async fallback // is required. Callers must check: const p = writeBatch(b); if (p) await p; function writeBatch(batch) { + const entry = createBatchEntry(batch); if (hasWritev && batch.length > 1) { - if (!hasWritevSync || !writer.writevSync(batch)) { + if (!hasWritevSync || + !writer.writevSync(validateBatchEntry(entry))) { + validateBatchEntry(entry); const opts = signal ? { __proto__: null, signal } : undefined; - const writevResult = writer.writev(batch, opts); + const writevResult = writer.writev(validateBatchEntry(entry), opts); if (writevResult === undefined) { - for (let i = 0; i < batch.length; i++) { - totalBytes += TypedArrayPrototypeGetByteLength(batch[i]); - } + validateBatchEntry(entry); + totalBytes += entry.byteLength; return; } return PromisePrototypeThen(PromiseResolve(writevResult), () => { - for (let i = 0; i < batch.length; i++) { - totalBytes += TypedArrayPrototypeGetByteLength(batch[i]); - } + validateBatchEntry(entry); + totalBytes += entry.byteLength; }); } - for (let i = 0; i < batch.length; i++) { - totalBytes += TypedArrayPrototypeGetByteLength(batch[i]); - } + validateBatchEntry(entry); + totalBytes += entry.byteLength; return; } - for (let i = 0; i < batch.length; i++) { - const chunk = batch[i]; + for (let i = 0; i < entry.views.length; i++) { + const view = entry.views[i]; + const chunk = validateByteView(view); if (!hasWriteSync || !writer.writeSync(chunk)) { + if (hasWriteSync) validateByteView(view); // Sync path failed at index i - fall back to async for the rest. - return writeBatchAsyncFallback(batch, i); + return writeBatchAsyncFallback(entry, i); } - totalBytes += TypedArrayPrototypeGetByteLength(chunk); + validateByteView(view); + totalBytes += view.byteLength; } } @@ -1084,4 +1206,5 @@ module.exports = { pipeToSync, pull, pullSync, + pullWithConsumerCleanup, }; diff --git a/lib/internal/streams/iter/push.js b/lib/internal/streams/iter/push.js index cc7bfc900cec..36a2034e5298 100644 --- a/lib/internal/streams/iter/push.js +++ b/lib/internal/streams/iter/push.js @@ -15,7 +15,6 @@ const { SymbolAsyncDispose, SymbolAsyncIterator, SymbolDispose, - TypedArrayPrototypeGetByteLength, } = primordials; const { @@ -37,16 +36,18 @@ const { const { kPushDefaultBudget, kResolvedPromise, + createBatchEntry, onSignalAbort, toUint8Array, convertChunks, getWriterSignal, parsePullArgs, validateBackpressure, + validateBatchEntry, } = require('internal/streams/iter/utils'); const { - pull: pullWithTransforms, + pullWithConsumerCleanup, } = require('internal/streams/iter/pull'); const { @@ -185,7 +186,11 @@ class PushQueue { if (this.#writerState !== 'open') return false; if (this.#consumerState !== 'active') return false; - const batchSize = this.#batchByteSize(chunks); + return this.#writeEntry(createBatchEntry(chunks)); + } + + #writeEntry(entry) { + const batchSize = entry.byteLength; // Skip empty chunks -- zero-byte writes would accumulate infinitely // without ever triggering backpressure under a byte-budget model. @@ -201,7 +206,7 @@ class PushQueue { while (this.#bufferedBytes >= this.#budget && this.#slots.length > 0) { const evicted = this.#slots.shift(); - this.#bufferedBytes -= this.#batchByteSize(evicted); + this.#bufferedBytes -= evicted.byteLength; } break; case 'drop-newest': @@ -211,7 +216,7 @@ class PushQueue { } } - this.#slots.push(chunks); + this.#slots.push(entry); this.#bufferedBytes += batchSize; this.#bytesWritten += batchSize; @@ -253,8 +258,8 @@ class PushQueue { // Check for pre-aborted signal (after state checks per spec) signal?.throwIfAborted(); - // Try sync first - if (this.writeSync(chunks)) { + const entry = createBatchEntry(chunks); + if (this.#writeEntry(entry)) { return; } @@ -266,9 +271,9 @@ class PushQueue { 'Backpressure violation: too many pending writes. ' + 'Await each write() call to respect backpressure.'); } - return this.#createPendingWrite(chunks, signal); + return this.#createPendingWrite(entry, signal); case 'unbounded': - return this.#createPendingWrite(chunks, signal); + return this.#createPendingWrite(entry, signal); default: throw new ERR_INVALID_STATE( 'Unexpected: writeSync should have handled non-strict policy'); @@ -281,9 +286,9 @@ class PushQueue { * promise rejects. Signal listeners are cleaned up on normal resolution. * @returns {Promise} */ - #createPendingWrite(chunks, signal) { + #createPendingWrite(batch, signal) { const { promise, resolve, reject } = PromiseWithResolvers(); - const entry = { __proto__: null, chunks, resolve, reject }; + const entry = { __proto__: null, batch, resolve, reject }; this.#pendingWrites.push(entry); if (signal) { @@ -425,6 +430,13 @@ class PushQueue { // =========================================================================== async read() { + if (this.#consumerState === 'returned') { + return { __proto__: null, done: true, value: undefined }; + } + if (this.#consumerState === 'thrown') { + throw this.#error; + } + // If there's data in the buffer, return it immediately if (this.#slots.length > 0) { const result = this.#drain(); @@ -458,16 +470,9 @@ class PushQueue { consumerReturn() { if (this.#consumerState !== 'active') return; this.#consumerState = 'returned'; - this.#cleanup(); + const error = new ERR_INVALID_STATE.TypeError('Stream closed by consumer'); + this.#terminateWriterFromConsumer(error); this.#resolvePendingReads(); - this.#rejectPendingWrites( - new ERR_INVALID_STATE.TypeError('Stream closed by consumer')); - // If closing, reject the pending end promise - if (this.#writerState === 'closing' && this.#pendingEnd) { - this.#pendingEnd.reject( - new ERR_INVALID_STATE.TypeError('Stream closed by consumer')); - this.#pendingEnd = null; - } // Resolve pending drains with false - no more data will be consumed this.#resolvePendingDrains(false); } @@ -476,13 +481,8 @@ class PushQueue { if (this.#consumerState !== 'active') return; this.#consumerState = 'thrown'; this.#error = error; - this.#cleanup(); + this.#terminateWriterFromConsumer(error); this.#rejectPendingReads(error); - this.#rejectPendingWrites(error); - if (this.#writerState === 'closing' && this.#pendingEnd) { - this.#pendingEnd.reject(error); - this.#pendingEnd = null; - } // Reject pending drains - the consumer errored this.#rejectPendingDrains(error); } @@ -492,37 +492,63 @@ class PushQueue { // =========================================================================== #drain() { - this.#bufferedBytes = 0; - if (this.#slots.length === 1) { - return this.#slots.shift(); - } + try { + if (this.#slots.length === 1) { + const result = validateBatchEntry(this.#slots.shift()); + this.#bufferedBytes = 0; + return result; + } - const result = []; - for (let i = 0; i < this.#slots.length; i++) { - const slot = this.#slots.get(i); - for (let j = 0; j < slot.length; j++) { - ArrayPrototypePush(result, slot[j]); + const result = []; + for (let i = 0; i < this.#slots.length; i++) { + const batch = validateBatchEntry(this.#slots.get(i)); + for (let j = 0; j < batch.length; j++) { + ArrayPrototypePush(result, batch[j]); + } } + this.#slots.clear(); + this.#bufferedBytes = 0; + return result; + } catch (error) { + this.#slots.clear(); + this.#bufferedBytes = 0; + this.fail(error); + throw error; } - this.#slots.clear(); - return result; } - #batchByteSize(batch) { - let size = 0; - for (let i = 0; i < batch.length; i++) { - size += TypedArrayPrototypeGetByteLength(batch[i]); + #terminateWriterFromConsumer(error) { + this.#slots.clear(); + this.#bufferedBytes = 0; + if (this.#writerState === 'open' || this.#writerState === 'closing') { + this.#writerState = 'errored'; + this.#error = error; + } + this.#cleanup(); + this.#rejectPendingWrites(error); + if (this.#pendingEnd) { + this.#pendingEnd.reject(error); + this.#pendingEnd = null; } - return size; } #resolvePendingReads() { while (this.#pendingReads.length > 0) { - if (this.#slots.length > 0) { + if (this.#consumerState === 'returned') { const pending = this.#pendingReads.shift(); - const result = this.#drain(); - this.#resolvePendingWrites(); - pending.resolve({ __proto__: null, done: false, value: result }); + pending.resolve({ __proto__: null, done: true, value: undefined }); + } else if (this.#consumerState === 'thrown') { + const pending = this.#pendingReads.shift(); + pending.reject(this.#error); + } else if (this.#slots.length > 0) { + const pending = this.#pendingReads.shift(); + try { + const result = this.#drain(); + this.#resolvePendingWrites(); + pending.resolve({ __proto__: null, done: false, value: result }); + } catch (error) { + pending.reject(error); + } } else if (this.#writerState === 'closing' && this.#slots.length === 0) { this.endDrained(); const pending = this.#pendingReads.shift(); @@ -533,9 +559,6 @@ class PushQueue { } else if (this.#writerState === 'errored') { const pending = this.#pendingReads.shift(); pending.reject(this.#error); - } else if (this.#consumerState === 'returned') { - const pending = this.#pendingReads.shift(); - pending.resolve({ __proto__: null, done: true, value: undefined }); } else { break; } @@ -546,11 +569,15 @@ class PushQueue { while (this.#pendingWrites.length > 0 && this.#bufferedBytes < this.#budget) { const pending = this.#pendingWrites.shift(); - const batchSize = this.#batchByteSize(pending.chunks); - this.#slots.push(pending.chunks); - this.#bufferedBytes += batchSize; - this.#bytesWritten += batchSize; - pending.resolve(); + try { + validateBatchEntry(pending.batch); + this.#slots.push(pending.batch); + this.#bufferedBytes += pending.batch.byteLength; + this.#bytesWritten += pending.batch.byteLength; + pending.resolve(); + } catch (error) { + pending.reject(error); + } } if (this.#bufferedBytes < this.#budget) { @@ -630,12 +657,10 @@ class PushWriter { writev(chunks, options) { validateArray(chunks, 'chunks'); const signal = getWriterSignal(options); - if (!signal && this.#queue.canWriteSync()) { - const bytes = convertChunks(chunks); - this.#queue.writeSync(bytes); + const bytes = convertChunks(chunks); + if (!signal && this.#queue.writeSync(bytes)) { return kResolvedPromise; } - const bytes = convertChunks(chunks); return this.#queue.writeAsync(bytes, signal); } @@ -756,12 +781,8 @@ function push(...args) { // Apply transforms lazily if provided let readable; if (transforms.length > 0) { - if (options.signal) { - readable = pullWithTransforms( - rawReadable, ...transforms, { __proto__: null, signal: options.signal }); - } else { - readable = pullWithTransforms(rawReadable, ...transforms); - } + readable = pullWithConsumerCleanup( + rawReadable, transforms, options.signal); } else { readable = rawReadable; } diff --git a/lib/internal/streams/iter/share.js b/lib/internal/streams/iter/share.js index 711abeb21b9a..00d9d387ce82 100644 --- a/lib/internal/streams/iter/share.js +++ b/lib/internal/streams/iter/share.js @@ -10,11 +10,12 @@ const { PromisePrototypeThen, PromiseResolve, PromiseWithResolvers, + SafePromiseRace, SafeSet, + Symbol, SymbolAsyncIterator, SymbolDispose, SymbolIterator, - TypedArrayPrototypeGetByteLength, } = primordials; const { @@ -30,18 +31,20 @@ const { } = require('internal/streams/iter/from'); const { - pull: pullWithTransforms, pullSync: pullSyncWithTransforms, + pullWithConsumerCleanup, } = require('internal/streams/iter/pull'); const { kMultiConsumerDefaultBudget, + createBatchEntry, getMinCursor, hasProtocol, onSignalAbort, wrapError, parsePullArgs, validateBackpressure, + validateBatchEntry, } = require('internal/streams/iter/utils'); const { @@ -65,6 +68,9 @@ const { // Async Share Implementation // ============================================================================= +const kNoShareError = Symbol('kNoShareError'); +const kShareCancelled = Symbol('kShareCancelled'); + class ShareImpl { #source; #options; @@ -77,6 +83,9 @@ class ShareImpl { #cancelled = false; #pulling = false; #pullWaiters = []; + #cancelPromise; + #resolveCancel; + #cancelError = kNoShareError; #cachedMinCursor = 0; #cachedMinCursorConsumers = 0; /** Cumulative byte size of buffered entries */ @@ -85,6 +94,9 @@ class ShareImpl { constructor(source, options) { this.#source = source; this.#options = options; + const { promise, resolve } = PromiseWithResolvers(); + this.#cancelPromise = promise; + this.#resolveCancel = resolve; } get consumerCount() { @@ -111,13 +123,7 @@ class ShareImpl { const rawConsumer = this.#createRawConsumer(); if (transforms.length > 0 || signal) { - if (signal) { - return pullWithTransforms( - rawConsumer, - ...transforms, - { __proto__: null, signal }); - } - return pullWithTransforms(rawConsumer, ...transforms); + return pullWithConsumerCleanup(rawConsumer, transforms, signal); } return rawConsumer; } @@ -129,6 +135,7 @@ class ShareImpl { resolve: null, reject: null, detached: false, + error: kNoShareError, pendingNext: PromiseResolve(), }; @@ -147,32 +154,28 @@ class ShareImpl { __proto__: null, [SymbolAsyncIterator]() { const getNext = async () => { - if (self.#sourceError !== undefined) { - state.detached = true; - self.#consumers.delete(state); - throw self.#sourceError; - } - // Loop until we get data, source is exhausted, or // consumer is detached. Multiple consumers may be woken // after a single pull - those that find no data at their // cursor must re-pull rather than terminating prematurely. for (;;) { if (state.detached) { - if (self.#sourceError !== undefined) throw self.#sourceError; + if (state.error !== kNoShareError) throw state.error; return { __proto__: null, done: true, value: undefined }; } if (self.#cancelled) { state.detached = true; + state.error = self.#cancelError; self.#deleteConsumer(state); + if (state.error !== kNoShareError) throw state.error; return { __proto__: null, done: true, value: undefined }; } // Check if data is available in buffer const bufferIndex = state.cursor - self.#bufferStart; if (bufferIndex < self.#buffer.length) { - const chunk = self.#buffer.get(bufferIndex); + const chunk = self.#readEntry(self.#buffer.get(bufferIndex)); const cursor = state.cursor; state.cursor++; if (cursor === self.#cachedMinCursor && @@ -185,7 +188,10 @@ class ShareImpl { if (self.#sourceExhausted) { state.detached = true; self.#deleteConsumer(state); - if (self.#sourceError !== undefined) throw self.#sourceError; + if (self.#sourceError !== undefined) { + state.error = self.#sourceError; + throw state.error; + } return { __proto__: null, done: true, value: undefined }; } @@ -193,8 +199,9 @@ class ShareImpl { const shouldBuffer = await self.#waitForBufferSpace(); if (shouldBuffer === null) { state.detached = true; + state.error = self.#cancelError; self.#deleteConsumer(state); - if (self.#sourceError !== undefined) throw self.#sourceError; + if (state.error !== kNoShareError) throw state.error; return { __proto__: null, done: true, value: undefined }; } @@ -246,14 +253,23 @@ class ShareImpl { this.#cancelled = true; if (reason !== undefined) { - this.#sourceError = reason; + this.#cancelError = reason; } + this.#resolveCancel(kShareCancelled); + this.#resolveCancel = null; + if (this.#sourceIterator?.return) { - PromisePrototypeThen(this.#sourceIterator.return(), undefined, () => {}); + try { + PromisePrototypeThen( + PromiseResolve(this.#sourceIterator.return()), undefined, () => {}); + } catch { + // Cancellation has precedence over source cleanup errors. + } } for (const consumer of this.#consumers) { + consumer.error = this.#cancelError; if (consumer.resolve) { if (reason !== undefined) { consumer.reject?.(reason); @@ -266,6 +282,8 @@ class ShareImpl { consumer.detached = true; } this.#consumers.clear(); + this.#buffer.clear(); + this.#bufferedBytes = 0; for (let i = 0; i < this.#pullWaiters.length; i++) { this.#pullWaiters[i](); @@ -302,7 +320,7 @@ class ShareImpl { while (this.#bufferedBytes >= this.#options.budget && this.#buffer.length > 0) { const evicted = this.#buffer.shift(); - this.#bufferedBytes -= this.#batchByteSize(evicted); + this.#bufferedBytes -= evicted.byteLength; this.#bufferStart++; } for (const consumer of this.#consumers) { @@ -369,13 +387,19 @@ class ShareImpl { } } - const result = await this.#sourceIterator.next(); + const result = await SafePromiseRace([ + this.#sourceIterator.next(), + this.#cancelPromise, + ]); + + if (this.#cancelled || result === kShareCancelled) return; if (result.done) { this.#sourceExhausted = true; } else if (!discard) { - this.#buffer.push(result.value); - this.#bufferedBytes += this.#batchByteSize(result.value); + const entry = createBatchEntry(result.value); + this.#buffer.push(entry); + this.#bufferedBytes += entry.byteLength; } } catch (error) { this.#sourceError = wrapError(error); @@ -398,7 +422,7 @@ class ShareImpl { if (trimCount > 0) { for (let i = 0; i < trimCount; i++) { const evicted = this.#buffer.get(i); - this.#bufferedBytes -= this.#batchByteSize(evicted); + this.#bufferedBytes -= evicted.byteLength; } this.#buffer.trimFront(trimCount); this.#bufferStart = this.#cachedMinCursor; @@ -409,12 +433,15 @@ class ShareImpl { } } - #batchByteSize(batch) { - let size = 0; - for (let i = 0; i < batch.length; i++) { - size += TypedArrayPrototypeGetByteLength(batch[i]); + #readEntry(entry) { + try { + return validateBatchEntry(entry); + } catch (error) { + this.cancel(error); + this.#buffer.clear(); + this.#bufferedBytes = 0; + throw error; } - return size; } #recomputeMinCursor() { @@ -517,7 +544,7 @@ class SyncShareImpl { const bufferIndex = state.cursor - self.#bufferStart; if (bufferIndex < self.#buffer.length) { - const chunk = self.#buffer.get(bufferIndex); + const chunk = self.#readEntry(self.#buffer.get(bufferIndex)); const cursor = state.cursor; state.cursor++; if (cursor === self.#cachedMinCursor && @@ -549,7 +576,7 @@ class SyncShareImpl { while (self.#bufferedBytes >= self.#options.budget && self.#buffer.length > 0) { const evicted = self.#buffer.shift(); - self.#bufferedBytes -= self.#batchByteSize(evicted); + self.#bufferedBytes -= evicted.byteLength; self.#bufferStart++; } for (const consumer of self.#consumers) { @@ -577,7 +604,7 @@ class SyncShareImpl { const newBufferIndex = state.cursor - self.#bufferStart; if (newBufferIndex < self.#buffer.length) { - const chunk = self.#buffer.get(newBufferIndex); + const chunk = self.#readEntry(self.#buffer.get(newBufferIndex)); const cursor = state.cursor; state.cursor++; if (cursor === self.#cachedMinCursor && @@ -649,8 +676,9 @@ class SyncShareImpl { if (result.done) { this.#sourceExhausted = true; } else { - this.#buffer.push(result.value); - this.#bufferedBytes += this.#batchByteSize(result.value); + const entry = createBatchEntry(result.value); + this.#buffer.push(entry); + this.#bufferedBytes += entry.byteLength; } } catch (error) { this.#sourceError = wrapError(error); @@ -666,19 +694,22 @@ class SyncShareImpl { if (trimCount > 0) { for (let i = 0; i < trimCount; i++) { const evicted = this.#buffer.get(i); - this.#bufferedBytes -= this.#batchByteSize(evicted); + this.#bufferedBytes -= evicted.byteLength; } this.#buffer.trimFront(trimCount); this.#bufferStart = this.#cachedMinCursor; } } - #batchByteSize(batch) { - let size = 0; - for (let i = 0; i < batch.length; i++) { - size += TypedArrayPrototypeGetByteLength(batch[i]); + #readEntry(entry) { + try { + return validateBatchEntry(entry); + } catch (error) { + this.cancel(error); + this.#buffer.clear(); + this.#bufferedBytes = 0; + throw error; } - return size; } #recomputeMinCursor() { diff --git a/lib/internal/streams/iter/utils.js b/lib/internal/streams/iter/utils.js index 9966f351b9d5..66e831a58523 100644 --- a/lib/internal/streams/iter/utils.js +++ b/lib/internal/streams/iter/utils.js @@ -3,6 +3,7 @@ const { Array, ArrayBufferPrototypeGetByteLength, + ArrayBufferPrototypeGetDetached, ArrayPrototypeSlice, PromiseResolve, PromiseWithResolvers, @@ -25,6 +26,7 @@ const { TextEncoder } = require('internal/encoding'); const { codes: { ERR_INVALID_ARG_TYPE, + ERR_INVALID_STATE, ERR_OPERATION_FAILED, }, } = require('internal/errors'); @@ -183,17 +185,69 @@ function toUint8Array(chunk) { return chunk; } -/** - * Check if all chunks in an array are already Uint8Array. - * Short-circuits on the first non-Uint8Array chunk found. - * @param {Array} chunks - * @returns {boolean} - */ -function allUint8Array(chunks) { +function snapshotByteView(value) { + const buffer = TypedArrayPrototypeGetBuffer(value); + const sharedBufferView = isSharedArrayBuffer(buffer) ? + new Uint8Array(buffer) : undefined; + return { + __proto__: null, + value, + buffer, + bufferByteLength: sharedBufferView === undefined ? + ArrayBufferPrototypeGetByteLength(buffer) : + TypedArrayPrototypeGetByteLength(sharedBufferView), + byteLength: TypedArrayPrototypeGetByteLength(value), + byteOffset: TypedArrayPrototypeGetByteOffset(value), + detached: sharedBufferView === undefined && + ArrayBufferPrototypeGetDetached(buffer), + sharedBufferView, + }; +} + +function validateByteView(snapshot) { + const { + value, + buffer, + bufferByteLength, + byteLength, + byteOffset, + detached, + sharedBufferView, + } = snapshot; + const currentBufferByteLength = sharedBufferView === undefined ? + ArrayBufferPrototypeGetByteLength(buffer) : + TypedArrayPrototypeGetByteLength(sharedBufferView); + const currentDetached = sharedBufferView === undefined && + ArrayBufferPrototypeGetDetached(buffer); + + if (TypedArrayPrototypeGetBuffer(value) !== buffer || + currentBufferByteLength !== bufferByteLength || + TypedArrayPrototypeGetByteLength(value) !== byteLength || + TypedArrayPrototypeGetByteOffset(value) !== byteOffset || + currentDetached !== detached) { + throw new ERR_INVALID_STATE.TypeError( + 'Byte view was resized or detached after being accepted'); + } + return value; +} + +function createBatchEntry(chunks) { + const views = new Array(chunks.length); + let byteLength = 0; for (let i = 0; i < chunks.length; i++) { - if (!isUint8Array(chunks[i])) return false; + const view = snapshotByteView(chunks[i]); + views[i] = view; + byteLength += view.byteLength; } - return true; + return { __proto__: null, views, byteLength }; +} + +function validateBatchEntry(entry) { + const chunks = new Array(entry.views.length); + for (let i = 0; i < entry.views.length; i++) { + chunks[i] = validateByteView(entry.views[i]); + } + return chunks; } function copyBytes(chunk) { @@ -250,9 +304,6 @@ function concatBytes(chunks) { * @returns {Uint8Array[]} */ function convertChunks(chunks) { - if (allUint8Array(chunks)) { - return ArrayPrototypeSlice(chunks); - } const len = chunks.length; const result = new Array(len); for (let i = 0; i < len; i++) { @@ -379,9 +430,9 @@ module.exports = { kMultiConsumerDefaultBudget, kPushDefaultBudget, kResolvedPromise, - allUint8Array, concatBytes, convertChunks, + createBatchEntry, getWriterSignal, getMinCursor, hasProtocol, @@ -392,6 +443,8 @@ module.exports = { parsePullArgs, toUint8Array, validateBackpressure, + validateBatchEntry, + validateByteView, wrapError, yieldAbortable, }; diff --git a/test/parallel/test-stream-iter-consumers-merge.js b/test/parallel/test-stream-iter-consumers-merge.js index 84aeb24b6159..e5d20d2dffa3 100644 --- a/test/parallel/test-stream-iter-consumers-merge.js +++ b/test/parallel/test-stream-iter-consumers-merge.js @@ -108,6 +108,109 @@ async function testMergeSourceError() { ); } +async function testMergeFalsySourceErrors() { + const reasons = [undefined, null, false, 0, '', NaN]; + + for (const reason of reasons) { + const noError = { __proto__: null }; + let actual = noError; + try { + await text(merge(rejectedSource(reason), from('other'))); + } catch (error) { + actual = error; + } + assert.strictEqual(Object.is(actual, reason), true); + } +} + +function rejectedSource(reason) { + return { + __proto__: null, + [Symbol.asyncIterator]() { + return this; + }, + next() { + return Promise.reject(reason); + }, + }; +} + +function pendingSource() { + return { + __proto__: null, + [Symbol.asyncIterator]() { + return this; + }, + next() { + return new Promise(() => {}); + }, + return() { + return new Promise(() => {}); + }, + }; +} + +async function testMergeSourceErrorDoesNotAwaitCleanup() { + const reason = new Error('source failed'); + + const timedOut = { __proto__: null }; + const outcome = await Promise.race([ + text(merge(rejectedSource(reason), pendingSource())).then( + () => ({ __proto__: null, status: 'fulfilled' }), + (error) => ({ __proto__: null, status: 'rejected', error }), + ), + new Promise((resolve) => setImmediate(resolve, timedOut)), + ]); + + assert.notStrictEqual(outcome, timedOut); + assert.strictEqual(outcome.status, 'rejected'); + assert.strictEqual(outcome.error, reason); +} + +async function testMergeBreakDoesNotAwaitCleanup() { + async function* readySource() { + yield [Uint8Array.of(1)]; + } + + const timedOut = { __proto__: null }; + const outcome = await Promise.race([ + (async () => { + for await (const batch of merge(readySource(), pendingSource())) { + assert.deepStrictEqual(batch, [Uint8Array.of(1)]); + break; + } + return true; + })(), + new Promise((resolve) => setImmediate(resolve, timedOut)), + ]); + + assert.strictEqual(outcome, true); +} + +async function testMergeNaNAbortDoesNotAwaitCleanup() { + const ac = new AbortController(); + const iterator = merge(pendingSource(), pendingSource(), { + __proto__: null, + signal: ac.signal, + })[Symbol.asyncIterator](); + const next = iterator.next(); + await new Promise(setImmediate); + ac.abort(NaN); + + const timedOut = { __proto__: null }; + const outcome = await Promise.race([ + next.then( + () => ({ __proto__: null, status: 'fulfilled' }), + (error) => ({ __proto__: null, status: 'rejected', error }), + ), + new Promise((resolve) => setImmediate(resolve, timedOut)), + ]); + + assert.notStrictEqual(outcome, timedOut); + assert.strictEqual(outcome.status, 'rejected'); + assert.strictEqual(Object.is(outcome.error, NaN), true); +} + async function testMergeConsumerBreak() { let source1Return = false; let source2Return = false; @@ -296,9 +399,8 @@ async function testMergeCleanupErrorOnly() { ); } -// Primary error + cleanup error: a source throws during iteration AND -// iterator.return() also throws. Should get a SuppressedError. -async function testMergePrimaryAndCleanupError() { +// A primary source error must not wait for asynchronous cleanup failures. +async function testMergePrimaryErrorPrecedesCleanupError() { async function* badSource() { yield [new TextEncoder().encode('x')]; throw new Error('primary boom'); @@ -319,15 +421,7 @@ async function testMergePrimaryAndCleanupError() { // Consume until error } }, - (err) => { - assert.ok( - err instanceof SuppressedError, - `Expected SuppressedError, got ${err.constructor.name}`, - ); - assert.strictEqual(err.error.message, 'primary boom'); - assert.strictEqual(err.suppressed.message, 'cleanup boom'); - return true; - }, + { message: 'primary boom' }, ); } @@ -360,6 +454,10 @@ Promise.all([ testMergeWithAbortSignal(), testMergeSyncSources(), testMergeSourceError(), + testMergeFalsySourceErrors(), + testMergeSourceErrorDoesNotAwaitCleanup(), + testMergeBreakDoesNotAwaitCleanup(), + testMergeNaNAbortDoesNotAwaitCleanup(), testMergeConsumerBreak(), testMergeSignalMidIteration(), testMergeSignalDuringPendingMultiSourceRead(), @@ -368,6 +466,6 @@ Promise.all([ testMergeStringSources(), testMergeObjectLikeSources(), testMergeCleanupErrorOnly(), - testMergePrimaryAndCleanupError(), + testMergePrimaryErrorPrecedesCleanupError(), testMergeBreakWithCleanupError(), ]).then(common.mustCall()); diff --git a/test/parallel/test-stream-iter-duplex.js b/test/parallel/test-stream-iter-duplex.js index 0969b91e7d21..5d617baa9882 100644 --- a/test/parallel/test-stream-iter-duplex.js +++ b/test/parallel/test-stream-iter-duplex.js @@ -14,32 +14,26 @@ async function testBasicDuplex() { // A writes, B reads await channelA.writer.write('hello from A'); - await channelA.close(); - + const closing = channelA.close(); const dataAtB = await text(channelB.readable); + await closing; assert.strictEqual(dataAtB, 'hello from A'); } async function testBidirectional() { const [channelA, channelB] = duplex(); - // A writes to B, B writes to A concurrently - const writeA = (async () => { - await channelA.writer.write('A to B'); - await channelA.close(); - })(); - - const writeB = (async () => { - await channelB.writer.write('B to A'); - await channelB.close(); - })(); - - const readAtB = text(channelB.readable); - const readAtA = text(channelA.readable); + await channelA.writer.write('A to B'); + await channelB.writer.write('B to A'); - await Promise.all([writeA, writeB]); - - const [dataAtA, dataAtB] = await Promise.all([readAtA, readAtB]); + const endA = channelA.writer.end(); + const endB = channelB.writer.end(); + const [dataAtA, dataAtB] = await Promise.all([ + text(channelA.readable), + text(channelB.readable), + ]); + await Promise.all([endA, endB]); + await Promise.all([channelA.close(), channelB.close()]); assert.strictEqual(dataAtB, 'A to B'); assert.strictEqual(dataAtA, 'B to A'); @@ -51,19 +45,29 @@ async function testMultipleWrites() { await channelA.writer.write('one'); await channelA.writer.write('two'); await channelA.writer.write('three'); - await channelA.close(); - + const closing = channelA.close(); const data = await text(channelB.readable); + await closing; assert.strictEqual(data, 'onetwothree'); } async function testChannelClose() { const [channelA, channelB] = duplex(); - - await channelA.close(); - - // Should be able to close twice without error - await channelA.close(); + const iteratorA = channelA.readable[Symbol.asyncIterator](); + const otherIteratorA = channelA.readable[Symbol.asyncIterator](); + const pendingRead = iteratorA.next(); + + const closing = channelA.close(); + assert.strictEqual(channelA.close(), closing); + await closing; + + assert.strictEqual((await pendingRead).done, true); + assert.strictEqual((await otherIteratorA.next()).done, true); + assert.strictEqual( + (await channelA.readable[Symbol.asyncIterator]().next()).done, true); + await assert.rejects(channelB.writer.write('late'), { + code: 'ERR_INVALID_STATE', + }); // B's readable should end (A -> B direction is closed) const batches = []; @@ -80,9 +84,9 @@ async function testWithOptions() { }); await channelA.writer.write('msg'); - await channelA.close(); - + const closing = channelA.close(); const data = await text(channelB.readable); + await closing; assert.strictEqual(data, 'msg'); } @@ -95,15 +99,17 @@ async function testPerChannelOptions() { // Channel A -> B direction uses A's options // Channel B -> A direction uses B's options await channelA.writer.write('from-a'); - await channelA.close(); - await channelB.writer.write('from-b'); - await channelB.close(); + + const endA = channelA.writer.end(); + const endB = channelB.writer.end(); const [dataAtA, dataAtB] = await Promise.all([ text(channelA.readable), text(channelB.readable), ]); + await Promise.all([endA, endB]); + await Promise.all([channelA.close(), channelB.close()]); assert.strictEqual(dataAtB, 'from-a'); assert.strictEqual(dataAtA, 'from-b'); @@ -146,17 +152,39 @@ async function testWriterEndWithPreAbortedSignal() { async function testEmptyDuplex() { const [channelA, channelB] = duplex(); - // Close without writing - await channelA.close(); - await channelB.close(); + await channelA.writer.end(); + await channelB.writer.end(); const dataAtA = await bytes(channelA.readable); const dataAtB = await bytes(channelB.readable); + await Promise.all([channelA.close(), channelB.close()]); assert.strictEqual(dataAtA.byteLength, 0); assert.strictEqual(dataAtB.byteLength, 0); } +async function testCloseWaitsForDrain() { + const [channelA, channelB] = duplex(); + await channelA.writer.write('buffered'); + + let closed = false; + const closing = channelA.close().then(common.mustCall(() => { + closed = true; + })); + await new Promise(setImmediate); + assert.strictEqual(closed, false); + + assert.strictEqual(await text(channelB.readable), 'buffered'); + await closing; +} + +async function testClosePropagatesWriterFailure() { + const [channelA] = duplex(); + const reason = new Error('writer failed'); + channelA.writer.fail(reason); + await assert.rejects(channelA.close(), (error) => error === reason); +} + // Channel fail propagation async function testChannelFail() { const [a, b] = duplex(); @@ -200,6 +228,8 @@ Promise.all([ testAbortSignal(), testWriterEndWithPreAbortedSignal(), testEmptyDuplex(), + testCloseWaitsForDrain(), + testClosePropagatesWriterFailure(), testChannelFail(), testAbortSignalBothChannels(), ]).then(common.mustCall()); diff --git a/test/parallel/test-stream-iter-pipeto-edge.js b/test/parallel/test-stream-iter-pipeto-edge.js index 3f09c4dfd42d..13223a226a1f 100644 --- a/test/parallel/test-stream-iter-pipeto-edge.js +++ b/test/parallel/test-stream-iter-pipeto-edge.js @@ -1,33 +1,41 @@ // Flags: --experimental-stream-iter 'use strict'; -// Edge case tests for pipeToSync: endSync fallback, preventFail. +// Edge case tests for pipeToSync close and failure behavior. const common = require('../common'); const assert = require('assert'); const { pipeToSync, fromSync } = require('stream/iter'); -// pipeToSync endSync returns negative → falls back to end() -async function testPipeToSyncEndSyncFallback() { +// pipeToSync cannot complete when endSync() requires async fallback. +async function testPipeToSyncEndSyncFailure() { let endCalled = false; const writer = { writeSync() { return true; }, - endSync() { return -1; }, // Negative → triggers end() fallback + endSync() { return -1; }, end() { endCalled = true; }, }; - pipeToSync(fromSync('data'), writer); - assert.strictEqual(endCalled, true); + assert.throws( + () => pipeToSync(fromSync('data'), writer, { preventFail: true }), + { code: 'ERR_INVALID_STATE' }, + ); + assert.strictEqual(endCalled, false); } -// pipeToSync endSync missing → falls back to end() +// pipeToSync requires endSync() when closing is enabled. async function testPipeToSyncNoEndSync() { + let writeCalled = false; let endCalled = false; const writer = { - writeSync() { return true; }, + writeSync() { writeCalled = true; return true; }, end() { endCalled = true; }, }; - pipeToSync(fromSync('data'), writer); - assert.strictEqual(endCalled, true); + assert.throws( + () => pipeToSync(fromSync('data'), writer), + { code: 'ERR_INVALID_ARG_TYPE' }, + ); + assert.strictEqual(writeCalled, false); + assert.strictEqual(endCalled, false); } // pipeToSync with preventFail: true — source error does NOT call fail() @@ -61,7 +69,7 @@ async function testPipeToSyncPreventClose() { } Promise.all([ - testPipeToSyncEndSyncFallback(), + testPipeToSyncEndSyncFailure(), testPipeToSyncNoEndSync(), testPipeToSyncPreventFail(), testPipeToSyncPreventClose(), diff --git a/test/parallel/test-stream-iter-pipeto.js b/test/parallel/test-stream-iter-pipeto.js index 5d8b5088f540..f16d7ae89972 100644 --- a/test/parallel/test-stream-iter-pipeto.js +++ b/test/parallel/test-stream-iter-pipeto.js @@ -71,6 +71,7 @@ async function testPipeToSyncSourceError() { let failCalled = false; const writer = { writeSync() { return true; }, + endSync: common.mustNotCall(), fail(reason) { failCalled = true; }, }; function* failingSource() { @@ -127,6 +128,7 @@ async function testPipeToSyncWithTransforms() { const chunks = []; const writer = { writeSync(chunk) { chunks.push(new TextDecoder().decode(chunk)); return true; }, + endSync() { return 0; }, }; const upper = (batch) => { if (batch === null) return null; @@ -160,6 +162,7 @@ async function testPipeToSyncWriterTransformMethodIgnored() { chunks.push(new TextDecoder().decode(chunk)); return true; }, + endSync() { return 0; }, }; pipeToSync(fromSync('hello'), writer); @@ -240,7 +243,7 @@ async function testPipeToSyncMinimalWriter() { }, }; - pipeToSync(fromSync('minimal-sync'), minimalWriter); + pipeToSync(fromSync('minimal-sync'), minimalWriter, { preventClose: true }); assert.strictEqual(chunks.length > 0, true); } diff --git a/test/parallel/test-stream-iter-pull-async.js b/test/parallel/test-stream-iter-pull-async.js index 3159b718f27e..86c790c85cb2 100644 --- a/test/parallel/test-stream-iter-pull-async.js +++ b/test/parallel/test-stream-iter-pull-async.js @@ -3,7 +3,15 @@ const common = require('../common'); const assert = require('assert'); -const { pull, from, text, tap } = require('stream/iter'); +const { + broadcast, + from, + pull, + push, + share, + tap, + text, +} = require('stream/iter'); async function testPullIdentity() { const data = await text(pull(from('hello-async'))); @@ -170,6 +178,61 @@ async function testPullSignalAbortWhileSourceNextPending() { await assert.rejects(next, { name: 'AbortError' }); } +async function testPullReturnWhileSourceNextPending() { + let startNext; + const nextStarted = new Promise((resolve) => { startNext = resolve; }); + const source = { + [Symbol.asyncIterator]() { + return { + next() { + startNext(); + return new Promise(() => {}); + }, + }; + }, + }; + + const iter = pull(source)[Symbol.asyncIterator](); + const next = assert.rejects(iter.next(), { name: 'AbortError' }); + await nextStarted; + + const timeout = {}; + const result = await Promise.race([ + iter.return(), + new Promise((resolve) => setImmediate(resolve, timeout)), + ]); + + assert.notStrictEqual(result, timeout); + assert.deepStrictEqual(result, { value: undefined, done: true }); + await next; +} + +async function testTransformedConsumerReturnBeforeNext() { + const identity = (chunks) => chunks; + const pushed = push(identity); + const { broadcast: bc } = broadcast(); + const broadcastConsumer = bc.push(identity); + const shared = share(from('shared')); + const sharedConsumer = shared.pull(identity); + + assert.strictEqual(bc.consumerCount, 1); + assert.strictEqual(shared.consumerCount, 1); + + const cases = [ + [pushed.readable, common.mustCall( + () => assert.strictEqual(pushed.writer.canWrite, null))], + [broadcastConsumer, common.mustCall( + () => assert.strictEqual(bc.consumerCount, 0))], + [sharedConsumer, common.mustCall( + () => assert.strictEqual(shared.consumerCount, 0))], + ]; + + for (const [readable, verify] of cases) { + await readable[Symbol.asyncIterator]().return(); + verify(); + } +} + async function testPullSignalAbortWithTransformWhileSourceNextPending() { const source = { [Symbol.asyncIterator]() { @@ -417,6 +480,8 @@ async function testTransformOptionsNotShared() { testTapCallbackError(), testPullSignalAbortMidIteration(), testPullSignalAbortWhileSourceNextPending(), + testPullReturnWhileSourceNextPending(), + testTransformedConsumerReturnBeforeNext(), testPullSignalAbortWithTransformWhileSourceNextPending(), testPullConsumerBreakCleanup(), testPullTransformReturnsPromise(), diff --git a/test/parallel/test-stream-iter-push-writer.js b/test/parallel/test-stream-iter-push-writer.js index 8ce555ca9251..dd6cf7d494b2 100644 --- a/test/parallel/test-stream-iter-push-writer.js +++ b/test/parallel/test-stream-iter-push-writer.js @@ -424,6 +424,20 @@ async function testConsumerReturnResolvesPendingRead() { assert.strictEqual(readResult.done, true); } +async function testEndRejectsAfterConsumerReturn() { + const { writer, readable } = push(); + writer.writeSync('data'); + const iter = readable[Symbol.asyncIterator](); + + await iter.return(); + + await assert.rejects( + writer.end({ signal: AbortSignal.timeout(common.platformTimeout(100)) }), + { code: 'ERR_INVALID_STATE' }, + ); + assert.strictEqual((await iter.next()).done, true); +} + // iterator.throw() rejects a pending read with the thrown error async function testConsumerThrowRejectsPendingRead() { const { readable } = push(); @@ -599,6 +613,7 @@ Promise.all([ testFailRejectsFutureReadWithFalsyReason(), testFailRejectsPendingReadWithFalsyReason(), testConsumerReturnResolvesPendingRead(), + testEndRejectsAfterConsumerReturn(), testConsumerThrowRejectsPendingRead(), testEndRejectsPendingWrites(), testEndIdempotentWhenClosed(), diff --git a/test/parallel/test-stream-iter-resizable-buffers.js b/test/parallel/test-stream-iter-resizable-buffers.js new file mode 100644 index 000000000000..b66973393cac --- /dev/null +++ b/test/parallel/test-stream-iter-resizable-buffers.js @@ -0,0 +1,178 @@ +// Flags: --experimental-stream-iter +'use strict'; + +const common = require('../common'); +const assert = require('assert'); +const { + array, + arraySync, + broadcast, + pipeTo, + pipeToSync, + push, + share, + shareSync, +} = require('stream/iter'); + +const kResizeError = { + code: 'ERR_INVALID_STATE', + message: /resized or detached/, +}; + +async function testBufferedViewMutationRejected() { + const resizable = new ArrayBuffer(1, { maxByteLength: 2 }); + const growable = new SharedArrayBuffer(1, { maxByteLength: 2 }); + const detachable = new ArrayBuffer(1); + const cases = [ + [new Uint8Array(resizable), () => resizable.resize(2)], + [new Uint8Array(growable), () => growable.grow(2)], + [new Uint8Array(detachable), () => { + structuredClone(detachable, { transfer: [detachable] }); + }], + ]; + + for (const [view, mutate] of cases) { + const { writer, readable } = push(); + assert.strictEqual(writer.writeSync(view), true); + mutate(); + await assert.rejects( + readable[Symbol.asyncIterator]().next(), + kResizeError, + ); + } +} + +async function testDropOldestUsesAcceptedByteLength() { + const buffer = new ArrayBuffer(16384, { maxByteLength: 16384 }); + const { writer, broadcast: bc } = broadcast({ + budget: 16384, + backpressure: 'drop-oldest', + }); + const iterator = bc.push()[Symbol.asyncIterator](); + + assert.strictEqual(writer.writeSync(new Uint8Array(buffer)), true); + buffer.resize(0); + assert.strictEqual(writer.writeSync(Uint8Array.of(2)), true); + assert.strictEqual(writer.writeSync(Uint8Array.of(3)), true); + writer.endSync(); + + assert.strictEqual((await iterator.next()).value[0][0], 2); + assert.strictEqual((await iterator.next()).value[0][0], 3); + assert.strictEqual((await iterator.next()).done, true); +} + +async function testPendingWritesRejectResizedViews() { + const pushResult = push({ budget: 16384, backpressure: 'unbounded' }); + assert.strictEqual( + pushResult.writer.writeSync(new Uint8Array(16384)), true); + const pushBuffer = new ArrayBuffer(1, { maxByteLength: 2 }); + const pushPending = pushResult.writer.write(new Uint8Array(pushBuffer)); + const pushRejected = assert.rejects(pushPending, kResizeError); + pushBuffer.resize(2); + const pushIterator = pushResult.readable[Symbol.asyncIterator](); + assert.strictEqual((await pushIterator.next()).done, false); + await pushRejected; + pushResult.writer.endSync(); + assert.strictEqual((await pushIterator.next()).done, true); + + const { writer, broadcast: bc } = broadcast({ + budget: 16384, + backpressure: 'unbounded', + }); + const broadcastIterator = bc.push()[Symbol.asyncIterator](); + assert.strictEqual(writer.writeSync(new Uint8Array(16384)), true); + const broadcastBuffer = new ArrayBuffer(1, { maxByteLength: 2 }); + const broadcastPending = writer.write(new Uint8Array(broadcastBuffer)); + const broadcastRejected = assert.rejects(broadcastPending, kResizeError); + broadcastBuffer.resize(2); + assert.strictEqual((await broadcastIterator.next()).done, false); + await broadcastRejected; + writer.endSync(); + assert.strictEqual((await broadcastIterator.next()).done, true); +} + +async function testBroadcastRejectsResizedBufferedView() { + const buffer = new ArrayBuffer(1, { maxByteLength: 2 }); + const { writer, broadcast: bc } = broadcast(); + const iterator = bc.push()[Symbol.asyncIterator](); + + assert.strictEqual(writer.writeSync(new Uint8Array(buffer)), true); + buffer.resize(2); + + await assert.rejects(iterator.next(), kResizeError); + await assert.rejects(writer.end(), kResizeError); +} + +async function testShareRejectsResizedBufferedView() { + const asyncBuffer = new ArrayBuffer(1, { maxByteLength: 2 }); + const shared = share([[new Uint8Array(asyncBuffer)]]); + const first = shared.pull()[Symbol.asyncIterator](); + const second = shared.pull()[Symbol.asyncIterator](); + + assert.strictEqual((await first.next()).done, false); + asyncBuffer.resize(2); + await assert.rejects(second.next(), kResizeError); + + const syncBuffer = new ArrayBuffer(1, { maxByteLength: 2 }); + const sharedSync = shareSync([[new Uint8Array(syncBuffer)]]); + const firstSync = sharedSync.pull()[Symbol.iterator](); + const secondSync = sharedSync.pull()[Symbol.iterator](); + + assert.strictEqual(firstSync.next().done, false); + syncBuffer.resize(2); + assert.throws(() => secondSync.next(), kResizeError); +} + +async function testConsumersRejectResizedViews() { + const asyncBuffer = new ArrayBuffer(1, { maxByteLength: 2 }); + async function* asyncSource() { + yield [new Uint8Array(asyncBuffer)]; + asyncBuffer.resize(2); + } + await assert.rejects(array(asyncSource(), { limit: 1 }), kResizeError); + + const syncBuffer = new ArrayBuffer(1, { maxByteLength: 2 }); + function* syncSource() { + yield [new Uint8Array(syncBuffer)]; + syncBuffer.resize(2); + } + assert.throws(() => arraySync(syncSource(), { limit: 1 }), kResizeError); +} + +async function testPipeRejectsWriterResize() { + const asyncBuffer = new ArrayBuffer(1, { maxByteLength: 2 }); + const asyncWriter = { + write() { + asyncBuffer.resize(2); + }, + fail: common.mustCall(), + }; + await assert.rejects( + pipeTo([new Uint8Array(asyncBuffer)], asyncWriter), + kResizeError, + ); + + const syncBuffer = new ArrayBuffer(1, { maxByteLength: 2 }); + const syncWriter = { + writeSync() { + syncBuffer.resize(2); + return true; + }, + fail: common.mustCall(), + }; + assert.throws( + () => pipeToSync( + [new Uint8Array(syncBuffer)], syncWriter, { preventClose: true }), + kResizeError, + ); +} + +Promise.all([ + testBufferedViewMutationRejected(), + testDropOldestUsesAcceptedByteLength(), + testPendingWritesRejectResizedViews(), + testBroadcastRejectsResizedBufferedView(), + testShareRejectsResizedBufferedView(), + testConsumersRejectResizedViews(), + testPipeRejectsWriterResize(), +]).then(common.mustCall()); diff --git a/test/parallel/test-stream-iter-share-async.js b/test/parallel/test-stream-iter-share-async.js index c96a0cb0f3c3..dafb157a7512 100644 --- a/test/parallel/test-stream-iter-share-async.js +++ b/test/parallel/test-stream-iter-share-async.js @@ -143,6 +143,70 @@ async function testShareCancelWithFalsyReason() { } } +async function testShareCancelWhileSourcePullPending() { + const noReason = { __proto__: null }; + + for (const reason of [noReason, 0]) { + const sourceStarted = Promise.withResolvers(); + const sourceNext = Promise.withResolvers(); + let nextCalls = 0; + let returnCalls = 0; + const source = { + __proto__: null, + [Symbol.asyncIterator]() { + return { + __proto__: null, + next() { + nextCalls++; + sourceStarted.resolve(); + return sourceNext.promise; + }, + async return() { + returnCalls++; + return { __proto__: null, done: true, value: undefined }; + }, + }; + }, + }; + const shared = share(source); + const iterator = shared.pull()[Symbol.asyncIterator](); + const read = iterator.next().then( + (value) => ({ __proto__: null, rejected: false, value }), + (error) => ({ __proto__: null, rejected: true, error }), + ); + + await sourceStarted.promise; + if (reason === noReason) { + shared.cancel(); + } else { + shared.cancel(reason); + } + + const timedOut = { __proto__: null }; + const outcome = await Promise.race([ + read, + new Promise((resolve) => setImmediate(resolve, timedOut)), + ]); + assert.notStrictEqual(outcome, timedOut); + if (reason === noReason) { + assert.strictEqual(outcome.rejected, false); + assert.strictEqual(outcome.value.done, true); + } else { + assert.strictEqual(outcome.rejected, true); + assert.strictEqual(outcome.error, reason); + } + assert.strictEqual(nextCalls, 1); + + sourceNext.resolve({ + __proto__: null, + done: false, + value: [Uint8Array.of(1)], + }); + await new Promise(setImmediate); + assert.strictEqual(returnCalls, 1); + } +} + async function testShareAbortSignal() { const ac = new AbortController(); const reason = new Error('share aborted'); @@ -272,6 +336,27 @@ async function testShareSourceError() { }, { message: 'share source boom' }); } +async function testShareSourceErrorFollowsBufferedData() { + const reason = new Error('share source boom'); + async function* failingSource() { + yield [Uint8Array.of(1)]; + throw reason; + } + + const shared = share(failingSource()); + const fast = shared.pull()[Symbol.asyncIterator](); + const slow = shared.pull()[Symbol.asyncIterator](); + const returned = shared.pull()[Symbol.asyncIterator](); + await returned.return(); + + assert.deepStrictEqual((await fast.next()).value, [Uint8Array.of(1)]); + await assert.rejects(fast.next(), (error) => error === reason); + + assert.deepStrictEqual((await slow.next()).value, [Uint8Array.of(1)]); + await assert.rejects(slow.next(), (error) => error === reason); + assert.strictEqual((await returned.next()).done, true); +} + async function testShareLateJoiningConsumer() { // A consumer that joins after some data has been consumed should only // see data remaining in the buffer (not items already trimmed). @@ -380,12 +465,14 @@ Promise.all([ testShareCancelMidIteration(), testShareCancelWithReason(), testShareCancelWithFalsyReason(), + testShareCancelWhileSourcePullPending(), testShareAbortSignal(), testShareAbortSignalWhileSourcePullPending(), testSharePullAbortSignalRejectsPendingNext(), testSharePullPreAbortedSignalDoesNotAddConsumer(), testShareAlreadyAborted(), testShareSourceError(), + testShareSourceErrorFollowsBufferedData(), testShareLateJoiningConsumer(), testShareConsumerBreak(), testShareMultipleConsumersConcurrentPull(), diff --git a/test/parallel/test-stream-iter-validation.js b/test/parallel/test-stream-iter-validation.js index 8dcfb46f173f..b93e6575490f 100644 --- a/test/parallel/test-stream-iter-validation.js +++ b/test/parallel/test-stream-iter-validation.js @@ -99,16 +99,16 @@ assert.throws(() => duplex({ budget: 0 }), { code: 'ERR_OUT_OF_RANGE' }); const [a, b] = duplex({ budget: Number.MAX_SAFE_INTEGER }); assert.strictEqual(a.writer.canWrite, true); assert.strictEqual(b.writer.canWrite, true); - a.close(); - b.close(); + a.writer.endSync(); + b.writer.endSync(); } // Per-direction overrides { const [a, b] = duplex({ a: { budget: 16384 }, b: { budget: 32768 } }); assert.strictEqual(a.writer.canWrite, true); assert.strictEqual(b.writer.canWrite, true); - a.close(); - b.close(); + a.writer.endSync(); + b.writer.endSync(); } assert.throws(() => duplex({ signal: {} }), { code: 'ERR_INVALID_ARG_TYPE' }); @@ -397,8 +397,8 @@ async function testAsyncValidation() { // Duplex with valid options { const [a, b] = duplex({ budget: 16384 }); - a.close(); - b.close(); + a.writer.endSync(); + b.writer.endSync(); } // Broadcast with valid options diff --git a/test/parallel/test-stream-iter-writer-reentrancy.js b/test/parallel/test-stream-iter-writer-reentrancy.js new file mode 100644 index 000000000000..119bda269239 --- /dev/null +++ b/test/parallel/test-stream-iter-writer-reentrancy.js @@ -0,0 +1,65 @@ +// Flags: --experimental-stream-iter +'use strict'; + +const common = require('../common'); +const assert = require('assert'); +const { broadcast, push } = require('stream/iter'); + +const factories = [ + () => push({ budget: 16384 }), + () => { + const { writer, broadcast: bc } = broadcast({ budget: 16384 }); + return { __proto__: null, writer, readable: bc.push() }; + }, +]; + +async function testWritevReentrancy() { + for (const factory of factories) { + const { writer, readable } = factory(); + const chunks = []; + Object.defineProperty(chunks, 0, { + __proto__: null, + enumerable: true, + get: common.mustCall(() => { + assert.strictEqual( + writer.writeSync(new Uint8Array(16384)), true); + return Uint8Array.of(42); + }), + }); + + let resolved = false; + const write = writer.writev(chunks).then(common.mustCall(() => { + resolved = true; + })); + await new Promise(setImmediate); + assert.strictEqual(resolved, false); + + const iterator = readable[Symbol.asyncIterator](); + assert.strictEqual((await iterator.next()).value[0].byteLength, 16384); + await write; + assert.strictEqual((await iterator.next()).value[0][0], 42); + writer.endSync(); + assert.strictEqual((await iterator.next()).done, true); + } + + for (const factory of factories) { + const { writer, readable } = factory(); + const chunks = []; + Object.defineProperty(chunks, 0, { + __proto__: null, + enumerable: true, + get: common.mustCall(() => { + writer.endSync(); + return Uint8Array.of(42); + }), + }); + + await assert.rejects(writer.writev(chunks), { + code: 'ERR_INVALID_STATE', + }); + assert.strictEqual( + (await readable[Symbol.asyncIterator]().next()).done, true); + } +} + +testWritevReentrancy().then(common.mustCall());