stream: prevent share from eagerly draining source

lib/internal/streams/iter/share.js

@@ -181,6 +181,9 @@ class ShareImpl {
             }
 
             await self.#pullFromSource(!shouldBuffer);
+            if (!shouldBuffer) {
+              await self.#waitForBufferSpaceAfterDrop();
+            }
           }
         };
 
@@ -299,6 +302,17 @@ class ShareImpl {
     return true;
   }
 
+  async #waitForBufferSpaceAfterDrop() {
+    while (this.#bufferedBytes >= this.#options.budget &&
+           !this.#cancelled &&
+           this.#sourceError === undefined &&
+           !this.#sourceExhausted) {
+      const { promise, resolve } = PromiseWithResolvers();
+      ArrayPrototypePush(this.#pullWaiters, resolve);
+      await promise;
+    }
+  }
+
   #pullFromSource(discard = false) {
     if (this.#sourceExhausted || this.#cancelled) {
       return PromiseResolve();

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

@@ -170,40 +170,50 @@ async function testShareDropOldest() {
 }
 
 async function testShareDropNewest() {
-  // With drop-newest and a stalled consumer, the async path allows the
-  // buffer to grow beyond budget (the "drop" applies to the
-  // backpressure signal, not the buffer contents). Both consumers
-  // ultimately see all items.
+  let pulls = 0;
+  let secondPull;
+  const secondPullStarted = new Promise((resolve) => {
+    secondPull = resolve;
+  });
+
   async function* source() {
-    for (let i = 0; i < 4; i++) {
+    for (let i = 0; i < 7; i++) {
+      pulls++;
+      if (pulls === 2) secondPull();
       const chunk = new Uint8Array(16384);
       chunk[0] = i;
       yield [chunk];
     }
   }
-  const shared = share(source(), { budget: 32768, backpressure: 'drop-newest' });
-  const fast = shared.pull();
-  const slow = shared.pull();
+  const shared = share(source(), {
+    budget: 16384,
+    backpressure: 'drop-newest',
+  });
+  const fast = shared.pull()[Symbol.asyncIterator]();
+  const slow = shared.pull()[Symbol.asyncIterator]();
 
-  // Fast consumer reads all items
-  const fastIndices = [];
-  for await (const batch of fast) {
-    for (const chunk of batch) {
-      fastIndices.push(chunk[0]);
-    }
-  }
-  assert.strictEqual(fastIndices.length, 2);
+  const first = await fast.next();
+  assert.strictEqual(first.value[0][0], 0);
 
-  // Slow consumer also sees all items (buffer grew past budget)
-  const slowIndices = [];
-  for await (const batch of slow) {
-    for (const chunk of batch) {
-      slowIndices.push(chunk[0]);
-    }
-  }
-  assert.strictEqual(slowIndices.length, 2);
-  assert.strictEqual(slowIndices[0], 0);
-  assert.strictEqual(slowIndices[1], 1);
+  let nextSettled = false;
+  const next = fast.next().then((result) => {
+    nextSettled = true;
+    return result;
+  });
+
+  await secondPullStarted;
+  await new Promise(setImmediate);
+  assert.strictEqual(pulls, 2);
+  assert.strictEqual(nextSettled, false);
+
+  const slowResult = await slow.next();
+  assert.strictEqual(slowResult.value[0][0], 0);
+
+  const nextResult = await next;
+  assert.strictEqual(nextResult.value[0][0], 2);
+  assert.strictEqual(pulls, 3);
+
+  shared.cancel();
 }
 
 // =============================================================================