stream: keep overlapping broadcast reads pending

lib/internal/streams/iter/broadcast.js

@@ -9,6 +9,7 @@
 const {
   ArrayIsArray,
   ArrayPrototypePush,
+  ArrayPrototypeShift,
   MathMax,
   PromisePrototypeThen,
   PromiseReject,
@@ -146,6 +147,7 @@ class BroadcastImpl {
       cursor: this.#bufferStart,
       resolve: null,
       reject: null,
+      pending: [],
       detached: false,
     };
 
@@ -165,8 +167,10 @@ class BroadcastImpl {
 
     function detach() {
       state.detached = true;
-      state.resolve = null;
-      state.reject = null;
+      if (state.resolve) {
+        state.resolve({ __proto__: null, done: true, value: undefined });
+      }
+      self.#resolvePendingDone(state);
       if (self.#deleteConsumer(state)) {
         self.#tryTrimBuffer();
       }
@@ -207,6 +211,13 @@ class BroadcastImpl {
               return kDone;
             }
 
+            if (state.resolve) {
+              const { promise, resolve, reject } = PromiseWithResolvers();
+              ArrayPrototypePush(state.pending,
+                                 { __proto__: null, resolve, reject });
+              return promise;
+            }
+
             const { promise, resolve, reject } = PromiseWithResolvers();
             state.resolve = resolve;
             state.reject = reject;
@@ -250,6 +261,11 @@ class BroadcastImpl {
         consumer.resolve = null;
         consumer.reject = null;
       }
+      if (reason !== undefined) {
+        this.#rejectPending(consumer, reason);
+      } else {
+        this.#resolvePendingDone(consumer);
+      }
       consumer.detached = true;
     }
     this.#consumers.clear();
@@ -296,7 +312,7 @@ class BroadcastImpl {
     this.#ended = true;
 
     for (const consumer of this.#consumers) {
-      if (consumer.resolve) {
+      while (consumer.resolve) {
         const bufferIndex = consumer.cursor - this.#bufferStart;
         if (bufferIndex < this.#buffer.length) {
           const chunk = this.#buffer.get(bufferIndex);
@@ -309,9 +325,15 @@ class BroadcastImpl {
           consumer.resolve({ __proto__: null, done: false, value: chunk });
         } else {
           consumer.resolve({ __proto__: null, done: true, value: undefined });
+          this.#resolvePendingDone(consumer);
+          consumer.detached = true;
         }
         consumer.resolve = null;
         consumer.reject = null;
+        if (consumer.detached && this.#deleteConsumer(consumer)) {
+          this.#tryTrimBuffer();
+          break;
+        }
       }
     }
   }
@@ -328,6 +350,7 @@ class BroadcastImpl {
         consumer.resolve = null;
         consumer.reject = null;
       }
+      this.#rejectPending(consumer, reason);
       consumer.detached = true;
     }
     this.#consumers.clear();
@@ -396,6 +419,11 @@ class BroadcastImpl {
           consumer.resolve = null;
           consumer.reject = null;
           resolve({ __proto__: null, done: false, value: chunk });
+          if (consumer.detached && this.#deleteConsumer(consumer)) {
+            this.#tryTrimBuffer();
+          } else if (this.#promotePending(consumer)) {
+            ArrayPrototypePush(this.#waiters, consumer);
+          }
         } else {
           // Still waiting -- put back
           ArrayPrototypePush(this.#waiters, consumer);
@@ -418,6 +446,31 @@ class BroadcastImpl {
     }
     return false;
   }
+
+  #promotePending(consumer) {
+    const next = ArrayPrototypeShift(consumer.pending);
+    if (next === undefined) return false;
+    consumer.resolve = next.resolve;
+    consumer.reject = next.reject;
+    return true;
+  }
+
+  #resolvePendingDone(consumer) {
+    if (consumer.resolve) {
+      consumer.resolve = null;
+      consumer.reject = null;
+    }
+    while (consumer.pending.length > 0) {
+      ArrayPrototypeShift(consumer.pending).resolve(
+        { __proto__: null, done: true, value: undefined });
+    }
+  }
+
+  #rejectPending(consumer, reason) {
+    while (consumer.pending.length > 0) {
+      ArrayPrototypeShift(consumer.pending).reject(reason);
+    }
+  }
 }
 
 // =============================================================================

test/parallel/test-stream-iter-broadcast-basic.js

@@ -3,6 +3,7 @@
 
 const common = require('../common');
 const assert = require('assert');
+const { setTimeout } = require('timers/promises');
 const { broadcast, text } = require('stream/iter');
 
 // =============================================================================
@@ -243,6 +244,38 @@ async function testLateJoinerSeesBufferedData() {
   assert.strictEqual(result, 'before-join');
 }
 
+async function testOverlappingNextKeepsEarlierRead() {
+  const { writer, broadcast: bc } = broadcast();
+  const it = bc.push()[Symbol.asyncIterator]();
+
+  const first = it.next();
+  const second = it.next();
+
+  await writer.write('x');
+
+  const secondResult = await Promise.race([
+    second.then((value) => ({ __proto__: null, settled: true, value })),
+    setTimeout(common.platformTimeout(50),
+               { __proto__: null, settled: false }),
+  ]);
+  assert.deepStrictEqual(secondResult, {
+    __proto__: null,
+    settled: false,
+  });
+
+  const result = await first;
+  assert.strictEqual(result.done, false);
+  assert.strictEqual(Buffer.concat(result.value).toString(), 'x');
+
+  writer.endSync();
+  assert.deepStrictEqual(await second, {
+    __proto__: null,
+    done: true,
+    value: undefined,
+  });
+  assert.strictEqual(bc.consumerCount, 0);
+}
+
 Promise.all([
   testBasicBroadcast(),
   testMultipleWrites(),
@@ -257,4 +290,5 @@ Promise.all([
   testFailDetachesConsumers(),
   testWriterFailIdempotent(),
   testLateJoinerSeesBufferedData(),
+  testOverlappingNextKeepsEarlierRead(),
 ]).then(common.mustCall());