diff --git a/doc/api/stream_iter.md b/doc/api/stream_iter.md index be9f220590b..7124d5e8fcf 100644 --- a/doc/api/stream_iter.md +++ b/doc/api/stream_iter.md @@ -1499,6 +1499,13 @@ Detaching keeps a consumer that is not retried (for example, one read with `for await...of`, which does not call `return()` when a read rejects) from holding buffered data and blocking the other consumers. +With `'unbounded'`, such a consumer waits until the slowest consumer releases +budget. With `'drop-newest'`, the entry pulled from the source is discarded +and the consumer then waits in the same way, so in both cases a stalled +consumer also stalls the consumers that are ahead of it. Only `'drop-oldest'` +lets consumers that are ahead continue, by discarding the oldest buffered +entries that the slowest consumer has not read yet. + ```mjs import { from, share, text } from 'node:stream/iter'; @@ -1596,18 +1603,16 @@ added: * `options` {Object} * `budget` {number} Must be >= 16384. **Default:** `65536`. - * `backpressure` {string} `'strict'`, `'drop-oldest'`, or `'drop-newest'`. + * `backpressure` {string} `'strict'` or `'drop-oldest'`. **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. +A synchronous consumer cannot wait for the slowest consumer to release budget, +and the slowest consumer cannot advance while another consumer's read is +running. `'unbounded'` and `'drop-newest'` are therefore not supported and +throw `ERR_INVALID_ARG_VALUE`. ### Class: `SyncShare` diff --git a/lib/internal/streams/iter/share.js b/lib/internal/streams/iter/share.js index 7c524655152..5439dfeb3d6 100644 --- a/lib/internal/streams/iter/share.js +++ b/lib/internal/streams/iter/share.js @@ -611,8 +611,8 @@ class SyncShareImpl { return { __proto__: null, done: true, value: undefined }; } - // Check buffer limit - let dropped = false; + // Check buffer limit. 'unbounded' and 'drop-newest' are rejected + // by shareSync(). if (self.#bufferedBytes >= self.#options.budget) { switch (self.#options.backpressure) { case 'strict': { @@ -644,21 +644,10 @@ class SyncShareImpl { } self.#recomputeMinCursor(); break; - case 'drop-newest': - // 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; } } - if (!dropped) { - self.#pullFromSource(); - } + self.#pullFromSource(); if (self.#sourceError !== kNoShareError) { state.detached = true; @@ -737,7 +726,7 @@ class SyncShareImpl { this.cancel(); } - #pullFromSource(discard = false) { + #pullFromSource() { if (this.#sourceExhausted || this.#cancelled) return; try { @@ -747,7 +736,7 @@ class SyncShareImpl { if (result.done) { this.#sourceExhausted = true; - } else if (!discard) { + } else { const entry = createBatchEntry(result.value); this.#buffer.push(entry); this.#bufferedBytes += entry.byteLength; @@ -858,10 +847,14 @@ function shareSync(source, options = { __proto__: null }) { backpressure = 'strict', } = options; validateInteger(budget, 'options.budget', 16384); - if (backpressure === 'unbounded') { + // A synchronous consumer can neither wait for the slowest consumer to + // release budget ('unbounded') nor keep pulling and discarding until it + // does ('drop-newest'): the slowest consumer cannot advance while the + // call is running. + if (backpressure === 'unbounded' || backpressure === 'drop-newest') { throw new ERR_INVALID_ARG_VALUE( 'options.backpressure', backpressure, - 'unbounded is not supported by shareSync()'); + `${backpressure} is not supported by shareSync()`); } const opts = { diff --git a/test/parallel/test-stream-iter-share-sync.js b/test/parallel/test-stream-iter-share-sync.js index 2d70ea45456..2721905bc93 100644 --- a/test/parallel/test-stream-iter-share-sync.js +++ b/test/parallel/test-stream-iter-share-sync.js @@ -146,69 +146,14 @@ function testShareSyncRejectsUnbounded() { ); } -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(); +function testShareSyncRejectsDropNewest() { + // A synchronous consumer can neither wait for the slowest consumer nor keep + // discarding until it advances, so 'drop-newest' is rejected like + // 'unbounded'. + assert.throws( + () => shareSync(fromSync('data'), { backpressure: 'drop-newest' }), + { code: 'ERR_INVALID_ARG_VALUE' }, + ); } // shareSync() accepts string source directly (normalized via fromSync()) @@ -297,8 +242,7 @@ Promise.all([ testShareSyncCancelWithFalsyReason(), testShareSyncSourceError(), testShareSyncRejectsUnbounded(), - testShareSyncDropNewest(), - testShareSyncDropNewestUnboundedSource(), + testShareSyncRejectsDropNewest(), testShareSyncStringSource(), testShareSyncRetainsBufferWhenAllConsumersDetach(), testShareSyncStrictBackpressureDetaches(),