stream: reject unbounded at SyncShare construction

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

Author: jasnell

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,
   },
@@ -588,11 +589,6 @@ class SyncShareImpl {
                   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) {
@@ -609,9 +605,13 @@ class SyncShareImpl {
                   self.#recomputeMinCursor();
                   break;
                 case 'drop-newest':
-                  state.detached = true;
-                  self.#deleteConsumer(state);
-                  return { __proto__: null, done: true, value: undefined };
+                  while (self.#bufferedBytes >= self.#options.budget &&
+                         !self.#sourceExhausted &&
+                         !self.#cancelled &&
+                         self.#sourceError === kNoShareError) {
+                    self.#pullFromSource(true);
+                  }
+                  break;
               }
             }
 
@@ -694,7 +694,7 @@ class SyncShareImpl {
     this.cancel();
   }
 
-  #pullFromSource() {
+  #pullFromSource(discard = false) {
     if (this.#sourceExhausted || this.#cancelled) return;
 
     try {
@@ -704,7 +704,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;
@@ -814,6 +814,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,

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

@@ -139,6 +139,39 @@ 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 < 3; 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);
+  assert.strictEqual(fast.next().done, true);
+  assert.strictEqual(pulls, 3);
+
+  assert.strictEqual(slow.next().value[0][0], 0);
+  assert.strictEqual(slow.next().done, true);
+}
+
 // shareSync() accepts string source directly (normalized via fromSync())
 function testShareSyncStringSource() {
   const shared = shareSync('hello-sync-share');
@@ -154,5 +187,7 @@ Promise.all([
   testShareSyncCancelWithReason(),
   testShareSyncCancelWithFalsyReason(),
   testShareSyncSourceError(),
+  testShareSyncRejectsUnbounded(),
+  testShareSyncDropNewest(),
   testShareSyncStringSource(),
 ]).then(common.mustCall());