stream: update broadcast to retain buffered data with zero consumers

While the Broadcast is still open, maintain buffered data when
consumer count reaches zero

Signed-off-by: James M Snell <jasnell@gmail.com>
Assisted-by: Opencode
PR-URL: #66028
Reviewed-By: Filip Skokan <panva.ip@gmail.com>
Reviewed-By: Tim Perry <pimterry@gmail.com>

Author: jasnell

lib/internal/streams/iter/broadcast.js

@@ -426,6 +426,8 @@ class BroadcastImpl {
   }
 
   #tryTrimBuffer() {
+    // Retain buffered data for consumers that attach while none are active.
+    if (this.#consumers.size === 0) return;
     if (this.#cachedMinCursorConsumers === 0) {
       this.#recomputeMinCursor();
     }

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

@@ -350,6 +350,18 @@ async function testLateJoinerSeesBufferedData() {
   assert.strictEqual(result, 'before-join');
 }
 
+async function testLateJoinerAfterDetachSeesBufferedData() {
+  const { writer, broadcast: bc } = broadcast({ budget: 16384 });
+  const first = bc.push()[Symbol.asyncIterator]();
+
+  writer.writeSync('before-detach');
+  await first.return();
+
+  const second = bc.push();
+  writer.endSync();
+  assert.strictEqual(await text(second), 'before-detach');
+}
+
 async function testOverlappingNextKeepsEarlierRead() {
   const { writer, broadcast: bc } = broadcast();
   const it = bc.push()[Symbol.asyncIterator]();
@@ -403,5 +415,6 @@ Promise.all([
   testFailDetachesConsumers(),
   testWriterFailIdempotent(),
   testLateJoinerSeesBufferedData(),
+  testLateJoinerAfterDetachSeesBufferedData(),
   testOverlappingNextKeepsEarlierRead(),
 ]).then(common.mustCall());