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>

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`
 

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 = {

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(),