stream: fixup stream/iter to drop at most one entry per share call

Signed-off-by: James M Snell <jasnell@gmail.com>
Assisted-by: Opencode

Author: jasnell

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])`

lib/internal/streams/iter/share.js

@@ -583,6 +583,7 @@ class SyncShareImpl {
             }
 
             // Check buffer limit
+            let dropped = false;
             if (self.#bufferedBytes >= self.#options.budget) {
               switch (self.#options.backpressure) {
                 case 'strict':
@@ -605,17 +606,20 @@ class SyncShareImpl {
                   self.#recomputeMinCursor();
                   break;
                 case 'drop-newest':
-                  while (self.#bufferedBytes >= self.#options.budget &&
-                         !self.#sourceExhausted &&
-                         !self.#cancelled &&
-                         self.#sourceError === kNoShareError) {
-                    self.#pullFromSource(true);
-                  }
+                  // 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;

test/parallel/test-stream-iter-share-sync.js

@@ -149,7 +149,7 @@ function testShareSyncRejectsUnbounded() {
 function testShareSyncDropNewest() {
   let pulls = 0;
   function* source() {
-    for (let i = 0; i < 3; i++) {
+    for (let i = 0; i < 4; i++) {
       pulls++;
       const chunk = new Uint8Array(16384);
       chunk[0] = i;
@@ -165,11 +165,50 @@ function testShareSyncDropNewest() {
   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, 3);
+  assert.strictEqual(pulls, 2);
 
+  // The slow consumer still sees the buffered entry, which releases budget.
   assert.strictEqual(slow.next().value[0][0], 0);
-  assert.strictEqual(slow.next().done, true);
+
+  // 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())
@@ -189,5 +228,6 @@ Promise.all([
   testShareSyncSourceError(),
   testShareSyncRejectsUnbounded(),
   testShareSyncDropNewest(),
+  testShareSyncDropNewestUnboundedSource(),
   testShareSyncStringSource(),
 ]).then(common.mustCall());