stream: fix share consumer premature termination on concurrent pull

When multiple share consumers call `next()` concurrently, they all
await the same `#pullFromSource()` call. When it completes, only one
item is added to the buffer, so consumers whose cursor doesn't
find data would incorrectly return done:true even though the
source still has data.

Replace the linear fall-through in the consumer `next()` method with
a retry loop. Consumers that wake without data at their cursor now
re-pull from source. The existing `#pulling` guard ensures only one
actual source pull is in flight at a time; other consumers enqueue
on `#pullWaiters` at the cost of a single promise allocation.

Author: jasnell

lib/internal/streams/iter/share.js

@@ -115,63 +115,48 @@ class ShareImpl {
               throw self.#sourceError;
             }
 
-            if (state.detached) {
-              return { __proto__: null, done: true, value: undefined };
-            }
-
-            if (self.#cancelled) {
-              state.detached = true;
-              self.#consumers.delete(state);
-              return { __proto__: null, done: true, value: undefined };
-            }
-
-            // Check if data is available in buffer
-            const bufferIndex = state.cursor - self.#bufferStart;
-            if (bufferIndex < self.#buffer.length) {
-              const chunk = self.#buffer.get(bufferIndex);
-              state.cursor++;
-              self.#tryTrimBuffer();
-              return { __proto__: null, done: false, value: chunk };
-            }
-
-            if (self.#sourceExhausted) {
-              state.detached = true;
-              self.#consumers.delete(state);
-              return { __proto__: null, done: true, value: undefined };
-            }
+            // Loop until we get data, source is exhausted, or
+            // consumer is detached. Multiple consumers may be woken
+            // after a single pull - those that find no data at their
+            // cursor must re-pull rather than terminating prematurely.
+            for (;;) {
+              if (state.detached) {
+                return { __proto__: null, done: true, value: undefined };
+              }
 
-            // Need to pull from source - check buffer limit
-            const canPull = await self.#waitForBufferSpace(state);
-            if (!canPull) {
-              state.detached = true;
-              self.#consumers.delete(state);
-              if (self.#sourceError) throw self.#sourceError;
-              return { __proto__: null, done: true, value: undefined };
-            }
+              if (self.#cancelled) {
+                state.detached = true;
+                self.#consumers.delete(state);
+                return { __proto__: null, done: true, value: undefined };
+              }
 
-            await self.#pullFromSource();
+              // Check if data is available in buffer
+              const bufferIndex = state.cursor - self.#bufferStart;
+              if (bufferIndex < self.#buffer.length) {
+                const chunk = self.#buffer.get(bufferIndex);
+                state.cursor++;
+                self.#tryTrimBuffer();
+                return { __proto__: null, done: false, value: chunk };
+              }
 
-            if (self.#sourceError) {
-              state.detached = true;
-              self.#consumers.delete(state);
-              throw self.#sourceError;
-            }
+              if (self.#sourceExhausted) {
+                state.detached = true;
+                self.#consumers.delete(state);
+                if (self.#sourceError) throw self.#sourceError;
+                return { __proto__: null, done: true, value: undefined };
+              }
 
-            const newBufferIndex = state.cursor - self.#bufferStart;
-            if (newBufferIndex < self.#buffer.length) {
-              const chunk = self.#buffer.get(newBufferIndex);
-              state.cursor++;
-              self.#tryTrimBuffer();
-              return { __proto__: null, done: false, value: chunk };
-            }
+              // Need to pull from source - check buffer limit
+              const canPull = await self.#waitForBufferSpace(state);
+              if (!canPull) {
+                state.detached = true;
+                self.#consumers.delete(state);
+                if (self.#sourceError) throw self.#sourceError;
+                return { __proto__: null, done: true, value: undefined };
+              }
 
-            if (self.#sourceExhausted) {
-              state.detached = true;
-              self.#consumers.delete(state);
-              return { __proto__: null, done: true, value: undefined };
+              await self.#pullFromSource();
             }
-
-            return { __proto__: null, done: true, value: undefined };
           },
 
           async return() {

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

@@ -238,4 +238,31 @@ Promise.all([
   testShareSyncCancel(),
   testSyncShareFromSync(),
   testSyncShareFromRejectsNonStreamable(),
+  testShareMultipleConsumersConcurrentPull(),
 ]).then(common.mustCall());
+
+async function testShareMultipleConsumersConcurrentPull() {
+  // Regression test: multiple consumers pulling concurrently should each
+  // receive all items even when only one item is pulled from source at a time.
+  // Previously, consumers woken after a pull that found no data at their
+  // cursor would return done:true prematurely (thundering herd bug).
+  async function* slowSource() {
+    for (let i = 0; i < 5; i++) {
+      await new Promise((r) => setTimeout(r, 1));
+      yield [new TextEncoder().encode(`item-${i}`)];
+    }
+  }
+  const shared = share(slowSource());
+  const c1 = shared.pull();
+  const c2 = shared.pull();
+  const c3 = shared.pull();
+
+  const [t1, t2, t3] = await Promise.all([
+    text(c1), text(c2), text(c3),
+  ]);
+
+  const expected = 'item-0item-1item-2item-3item-4';
+  assert.strictEqual(t1, expected);
+  assert.strictEqual(t2, expected);
+  assert.strictEqual(t3, expected);
+}