Commit d4ec9ff27de for nodejs
commit d4ec9ff27de4505692fc7ef82ac4c8802e7efdfa
Author: James M Snell <jasnell@gmail.com>
Date: Sat Oct 3 14:44:05 2026 +0000
stream: reject drop-newest backpressure in shareSync()
With 'drop-newest', a shareSync() consumer that needed to pull while
the buffer was at the budget discarded one entry from the source and
then returned { done: true } even though the source was not exhausted.
for...of loops, and anything else that trusts the iterator protocol,
silently stopped consuming. There is no correct alternative in a
synchronous context: the slowest consumer cannot advance while another
consumer's next() is running, so the call can neither wait for budget
nor keep discarding until budget is released.
Reject 'drop-newest' in shareSync() with ERR_INVALID_ARG_VALUE, as is
already done for 'unbounded'. The two tests that asserted the previous
"done but not detached" behavior are replaced by one that checks the
rejection. Also document how 'unbounded' and 'drop-newest' make the
async share() wait for the slowest consumer.
Assisted-by: OpenCode
Signed-off-by: James M Snell <jasnell@gmail.com>
PR-URL: https://github.com/nodejs/node/pull/66483
Reviewed-By: Trivikram Kamat <trivikr.dev@gmail.com>
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(),