diff --git a/doc/api/stream_iter.md b/doc/api/stream_iter.md index f2fbba2286e0..80579fba9820 100644 --- a/doc/api/stream_iter.md +++ b/doc/api/stream_iter.md @@ -1497,11 +1497,19 @@ added: * `options` {Object} * `budget` {number} Must be >= 16384. **Default:** `65536`. - * `backpressure` {string} **Default:** `'strict'`. + * `backpressure` {string} `'strict'`, `'drop-oldest'`, or `'drop-newest'`. + **Default:** `'strict'`. * Returns: {SyncShare} Synchronous version of [`share()`][]. +Because there is no way to wait in a synchronous context, `'unbounded'` is not +supported and throws `ERR_INVALID_ARG_VALUE`. With `'drop-newest'`, a consumer +that reaches the end of the buffer while the budget is exhausted discards a +single entry from the source and then returns `{ done: true }` without a +value; the consumer is not detached, so it can resume once the slowest +consumer advances and releases budget. + ### Class: `SyncShare` #### Static method: `SyncShare.fromSync(input[, options])` diff --git a/lib/internal/quic/quic.js b/lib/internal/quic/quic.js index 2ee2529b1c81..632369fc9086 100644 --- a/lib/internal/quic/quic.js +++ b/lib/internal/quic/quic.js @@ -2640,15 +2640,10 @@ class QuicStream { error = inner.destroyError ?? error; if (error !== undefined) { inner.pendingClose.reject(error); + inner.pendingStream.reject(error); } else { inner.pendingClose.resolve(); - } - if (inner.state.pending) { - if (error !== undefined) { - inner.pendingStream.reject(error); - } else { - inner.pendingStream.resolve(error); - } + inner.pendingStream.resolve(); } debug('stream closed'); if (onStreamClosedChannel.hasSubscribers) { diff --git a/lib/internal/streams/iter/broadcast.js b/lib/internal/streams/iter/broadcast.js index ddce0d12d42c..673f21f11866 100644 --- a/lib/internal/streams/iter/broadcast.js +++ b/lib/internal/streams/iter/broadcast.js @@ -24,6 +24,10 @@ const { } = primordials; const { lazyDOMException } = require('internal/util'); +const { + AbortController, + abortSignal, +} = require('internal/abort_controller'); const { codes: { @@ -63,6 +67,7 @@ const { parsePullArgs, toWriterUint8Array, validateBatchEntry, + yieldAbortable, } = require('internal/streams/iter/utils'); const { converters, @@ -79,6 +84,7 @@ const kAbort = Symbol('kAbort'); const kCanWrite = Symbol('kCanWrite'); const kOnBufferDrained = Symbol('kOnBufferDrained'); const kOnEndDrained = Symbol('kOnEndDrained'); +const kOnCancel = Symbol('kOnCancel'); const kPendingWriteRemoved = Symbol('kPendingWriteRemoved'); const kNoBroadcastError = Symbol('kNoBroadcastError'); @@ -120,6 +126,7 @@ class BroadcastImpl { this.#options = options; this[kOnBufferDrained] = null; this[kOnEndDrained] = null; + this[kOnCancel] = null; } setWriter(writer) { @@ -307,6 +314,9 @@ class BroadcastImpl { } this.#consumers.clear(); this.#cachedMinCursorConsumers = 0; + const onCancel = this[kOnCancel]; + this[kOnCancel] = null; + onCancel?.(reason); } [SymbolDispose]() { @@ -444,6 +454,8 @@ class BroadcastImpl { } #tryTrimBuffer() { + // Retain buffered data for consumers that attach while none are active. + if (this.#consumers.size === 0) return; if (this.#cachedMinCursorConsumers === 0) { this.#recomputeMinCursor(); } @@ -913,13 +925,23 @@ const Broadcast = { }); const result = broadcast(options); const { signal } = options; + const controller = new AbortController(); + if (signal?.aborted) { + abortSignal(controller.signal, signal.reason); + } + const onCancel = (reason) => { + if (!controller.signal.aborted) { + abortSignal(controller.signal, reason); + } + }; + result.broadcast[kOnCancel] = onCancel; const pump = async () => { const w = result.writer; try { if (isAsyncIterable(source)) { - for await (const chunks of source) { - signal?.throwIfAborted(); + for await (const chunks of yieldAbortable(source, controller.signal)) { + controller.signal.throwIfAborted(); if (ArrayIsArray(chunks)) { if (!w.writevSync(chunks)) { await w.writev(chunks, signal ? { signal } : undefined); @@ -930,7 +952,7 @@ const Broadcast = { } } else if (isSyncIterable(source)) { for (const chunks of source) { - signal?.throwIfAborted(); + controller.signal.throwIfAborted(); if (ArrayIsArray(chunks)) { if (!w.writevSync(chunks)) { await w.writev(chunks, signal ? { signal } : undefined); @@ -944,7 +966,13 @@ const Broadcast = { await w.end(signal ? { signal } : undefined); } } catch (error) { - w.fail(error); + if (!controller.signal.aborted) { + w.fail(error); + } + } finally { + if (result.broadcast[kOnCancel] === onCancel) { + result.broadcast[kOnCancel] = null; + } } }; PromisePrototypeThen(pump(), undefined, () => {}); diff --git a/lib/internal/streams/iter/share.js b/lib/internal/streams/iter/share.js index 6154509a5b64..5e813a8de51d 100644 --- a/lib/internal/streams/iter/share.js +++ b/lib/internal/streams/iter/share.js @@ -56,6 +56,7 @@ const { const { codes: { ERR_INVALID_ARG_TYPE, + ERR_INVALID_ARG_VALUE, ERR_INVALID_RETURN_VALUE, ERR_OUT_OF_RANGE, }, @@ -200,7 +201,16 @@ class ShareImpl { } // Need to pull from source - check buffer limit - const shouldBuffer = await self.#waitForBufferSpace(); + let shouldBuffer; + try { + shouldBuffer = await self.#waitForBufferSpace(); + } catch (error) { + state.detached = true; + if (self.#deleteConsumer(state)) { + self.#tryTrimBuffer(); + } + throw error; + } if (shouldBuffer === null) { state.detached = true; state.error = self.#cancelError; @@ -573,17 +583,13 @@ class SyncShareImpl { } // Check buffer limit + let dropped = false; if (self.#bufferedBytes >= self.#options.budget) { switch (self.#options.backpressure) { case 'strict': throw new ERR_OUT_OF_RANGE( 'buffered bytes', `< ${self.#options.budget}`, self.#bufferedBytes); - case 'unbounded': - throw new ERR_OUT_OF_RANGE( - 'buffered bytes', `< ${self.#options.budget} ` + - '(unbounded not available in sync context)', - self.#bufferedBytes); case 'drop-oldest': while (self.#bufferedBytes >= self.#options.budget && self.#buffer.length > 0) { @@ -600,13 +606,20 @@ class SyncShareImpl { self.#recomputeMinCursor(); break; case 'drop-newest': - state.detached = true; - self.#deleteConsumer(state); - return { __proto__: null, done: true, value: undefined }; + // Discarding does not reclaim budget, and the slowest + // consumer cannot advance while this synchronous next() is + // running, so at most one entry may be dropped per call. + // Looping here would spin forever on an unbounded source + // and drain a finite one in a single call. + self.#pullFromSource(true); + dropped = true; + break; } } - self.#pullFromSource(); + if (!dropped) { + self.#pullFromSource(); + } if (self.#sourceError !== kNoShareError) { state.detached = true; @@ -685,7 +698,7 @@ class SyncShareImpl { this.cancel(); } - #pullFromSource() { + #pullFromSource(discard = false) { if (this.#sourceExhausted || this.#cancelled) return; try { @@ -695,7 +708,7 @@ class SyncShareImpl { if (result.done) { this.#sourceExhausted = true; - } else { + } else if (!discard) { const entry = createBatchEntry(result.value); this.#buffer.push(entry); this.#bufferedBytes += entry.byteLength; @@ -805,6 +818,11 @@ function shareSync(source, options = { __proto__: null }) { backpressure = 'strict', } = options; validateInteger(budget, 'options.budget', 16384); + if (backpressure === 'unbounded') { + throw new ERR_INVALID_ARG_VALUE( + 'options.backpressure', backpressure, + 'unbounded is not supported by shareSync()'); + } const opts = { __proto__: null, diff --git a/test/parallel/test-stream-iter-broadcast-basic.js b/test/parallel/test-stream-iter-broadcast-basic.js index 3dbd3ce97512..4135aea6d97d 100644 --- a/test/parallel/test-stream-iter-broadcast-basic.js +++ b/test/parallel/test-stream-iter-broadcast-basic.js @@ -350,6 +350,18 @@ async function testLateJoinerSeesBufferedData() { assert.strictEqual(result, 'before-join'); } +async function testLateJoinerAfterDetachSeesBufferedData() { + const { writer, broadcast: bc } = broadcast({ budget: 16384 }); + const first = bc.push()[Symbol.asyncIterator](); + + writer.writeSync('before-detach'); + await first.return(); + + const second = bc.push(); + writer.endSync(); + assert.strictEqual(await text(second), 'before-detach'); +} + async function testOverlappingNextKeepsEarlierRead() { const { writer, broadcast: bc } = broadcast(); const it = bc.push()[Symbol.asyncIterator](); @@ -403,5 +415,6 @@ Promise.all([ testFailDetachesConsumers(), testWriterFailIdempotent(), testLateJoinerSeesBufferedData(), + testLateJoinerAfterDetachSeesBufferedData(), testOverlappingNextKeepsEarlierRead(), ]).then(common.mustCall()); diff --git a/test/parallel/test-stream-iter-broadcast-from.js b/test/parallel/test-stream-iter-broadcast-from.js index 928d4d472f06..6062f5244ec1 100644 --- a/test/parallel/test-stream-iter-broadcast-from.js +++ b/test/parallel/test-stream-iter-broadcast-from.js @@ -4,6 +4,7 @@ const common = require('../common'); const assert = require('assert'); const { broadcast, Broadcast, from, text } = require('stream/iter'); +const { setImmediate } = require('timers/promises'); // ============================================================================= // Broadcast.from @@ -117,34 +118,46 @@ async function testAlreadyAbortedSignal() { // ============================================================================= async function testBroadcastFromCancelWhileBlocked() { - // Create a slow async source that blocks between yields - let sourceFinished = false; - async function* slowSource() { - const enc = new TextEncoder(); - yield [enc.encode('chunk1')]; - // Simulate a long delay without keeping the cancelled source alive. - await new Promise((resolve) => setTimeout(resolve, 10000).unref()); - yield [enc.encode('chunk2')]; - sourceFinished = true; - } - - const { broadcast: bc } = Broadcast.from(slowSource()); - const consumer = bc.push(); + let resolveNext; + let sourceReturned = false; + const source = { + [Symbol.asyncIterator]() { + return { + next() { + const { promise, resolve } = Promise.withResolvers(); + resolveNext = resolve; + return promise; + }, + return() { + sourceReturned = true; + return Promise.resolve({ __proto__: null, done: true }); + }, + }; + }, + }; - // Read the first chunk - const iter = consumer[Symbol.asyncIterator](); - const first = await iter.next(); - assert.strictEqual(first.done, false); + const { writer, broadcast: bc } = Broadcast.from(source); + const iter = bc.push()[Symbol.asyncIterator](); + const pendingRead = iter.next(); + await setImmediate(); - // Cancel while the source is blocked waiting to yield the next chunk + let writesAfterCancel = 0; + writer.writevSync = () => { writesAfterCancel++; return true; }; bc.cancel(); - - // The iteration should complete (not hang) - const next = await iter.next(); - assert.strictEqual(next.done, true); - - // Source should NOT have finished (we cancelled before chunk2) - assert.strictEqual(sourceFinished, false); + assert.deepStrictEqual(await pendingRead, { + __proto__: null, + done: true, + value: undefined, + }); + + resolveNext({ + __proto__: null, + done: false, + value: [new TextEncoder().encode('late')], + }); + await setImmediate(); + assert.strictEqual(writesAfterCancel, 0); + assert.strictEqual(sourceReturned, true); } // ============================================================================= diff --git a/test/parallel/test-stream-iter-share-from.js b/test/parallel/test-stream-iter-share-from.js index 806e30876302..69b782ce5608 100644 --- a/test/parallel/test-stream-iter-share-from.js +++ b/test/parallel/test-stream-iter-share-from.js @@ -221,23 +221,29 @@ async function testShareDropNewest() { // ============================================================================= async function testShareStrictBackpressure() { - async function* source() { - for (let i = 0; i < 10; i++) { - yield [new Uint8Array(16384)]; + for (const transformed of [false, true]) { + async function* source() { + for (let i = 0; i < 10; i++) { + yield [new Uint8Array(16384)]; + } } + const shared = share(source(), { + budget: 32768, + backpressure: 'strict', + }); + const consumer = transformed ? + shared.pull((chunks) => chunks) : shared.pull(); + const fast = consumer[Symbol.asyncIterator](); + // This consumer prevents the buffer from being trimmed. + shared.pull(); + + await fast.next(); + await fast.next(); + await assert.rejects(fast.next(), { code: 'ERR_OUT_OF_RANGE' }); + assert.strictEqual(shared.consumerCount, 1); + assert.strictEqual((await fast.next()).done, true); + shared.cancel(); } - const shared = share(source(), { budget: 32768, backpressure: 'strict' }); - const fast = shared.pull(); - // Create a second consumer that never reads — this prevents buffer trimming - shared.pull(); - - // The fast consumer's pulls will eventually cause the buffer to exceed - // the budget (since the slow consumer prevents trimming), - // triggering an ERR_OUT_OF_RANGE error. - await assert.rejects(async () => { - // eslint-disable-next-line no-unused-vars - for await (const _ of fast) { /* consume */ } - }, { code: 'ERR_OUT_OF_RANGE' }); } Promise.all([ diff --git a/test/parallel/test-stream-iter-share-sync.js b/test/parallel/test-stream-iter-share-sync.js index 20c98d134206..fd0184df9bce 100644 --- a/test/parallel/test-stream-iter-share-sync.js +++ b/test/parallel/test-stream-iter-share-sync.js @@ -139,6 +139,78 @@ function testShareSyncSourceError() { }, { message: 'sync share boom' }); } +function testShareSyncRejectsUnbounded() { + assert.throws( + () => shareSync(fromSync('data'), { backpressure: 'unbounded' }), + { code: 'ERR_INVALID_ARG_VALUE' }, + ); +} + +function testShareSyncDropNewest() { + let pulls = 0; + function* source() { + for (let i = 0; i < 4; i++) { + pulls++; + const chunk = new Uint8Array(16384); + chunk[0] = i; + yield [chunk]; + } + } + + const shared = shareSync(source(), { + budget: 16384, + backpressure: 'drop-newest', + }); + const fast = shared.pull()[Symbol.iterator](); + const slow = shared.pull()[Symbol.iterator](); + + assert.strictEqual(fast.next().value[0][0], 0); + + // The budget is exhausted and the slow consumer cannot advance while this + // call is running, so exactly one entry is dropped and no value is + // available. The consumer is not detached. + assert.strictEqual(fast.next().done, true); + assert.strictEqual(pulls, 2); + + // The slow consumer still sees the buffered entry, which releases budget. + assert.strictEqual(slow.next().value[0][0], 0); + + // Entry 1 was dropped for every consumer, so both resume at entry 2. + assert.strictEqual(slow.next().value[0][0], 2); + assert.strictEqual(pulls, 3); + assert.strictEqual(fast.next().value[0][0], 2); +} + +// Regression test: a full buffer must not spin pulling-and-discarding from an +// unbounded source, since discarding never reclaims budget. +function testShareSyncDropNewestUnboundedSource() { + let pulls = 0; + function* source() { + for (;;) { + pulls++; + yield [new Uint8Array(16384)]; + } + } + + const shared = shareSync(source(), { + budget: 16384, + backpressure: 'drop-newest', + }); + const fast = shared.pull()[Symbol.iterator](); + shared.pull(); + + assert.strictEqual(fast.next().done, false); + assert.strictEqual(pulls, 1); + + // Each blocked call drops at most one entry and returns without a value. + for (let i = 0; i < 3; i++) { + assert.strictEqual(fast.next().done, true); + assert.strictEqual(pulls, 2 + i); + } + + shared.cancel(); +} + // shareSync() accepts string source directly (normalized via fromSync()) function testShareSyncStringSource() { const shared = shareSync('hello-sync-share'); @@ -154,5 +226,8 @@ Promise.all([ testShareSyncCancelWithReason(), testShareSyncCancelWithFalsyReason(), testShareSyncSourceError(), + testShareSyncRejectsUnbounded(), + testShareSyncDropNewest(), + testShareSyncDropNewestUnboundedSource(), testShareSyncStringSource(), ]).then(common.mustCall());