refactor(shared): improve replicateAsyncIterator (#923)

<!-- This is an auto-generated comment: release notes by coderabbit.ai
-->
## Summary by CodeRabbit

- Bug Fixes
- Replicated async streams now propagate errors to all consumers
reliably and coordinate termination to avoid premature stops or
redundant cancellations.
- Performance
- Per-consumer buffering improves handling of multiple concurrent
consumers, reducing contention and preventing dropped items.
- Chores
- Internal lifecycle and coordination logic streamlined for more robust
behavior without changing public APIs.
- Notes
  - No user-facing API changes.
<!-- end of auto-generated comment: release notes by coderabbit.ai -->

Author: dinwwwh

packages/shared/src/iterator.ts

@@ -101,69 +101,66 @@ export function replicateAsyncIterator<T, TReturn, TNext>(
   source: AsyncIterator<T, TReturn, TNext>,
   count: number,
 ): (AsyncIteratorClass<T, TReturn, TNext>)[] {
-  const queue = new AsyncIdQueue<IteratorResult<T, TReturn>>()
+  const queue = new AsyncIdQueue<
+    { next: IteratorResult<T, TReturn> } | { next?: never, error: unknown }
+  >()
 
-  const replicated: AsyncIteratorClass<T, TReturn, TNext>[] = []
-
-  let error: undefined | { value: unknown }
+  const ids = Array.from({ length: count }, (_, i) => i.toString())
+  let isSourceFinished = false
 
   const start = once(async () => {
     try {
       while (true) {
         const item = await source.next()
 
-        for (let i = 0; i < count; i++) {
-          const id = i.toString()
-
+        ids.forEach((id) => {
           if (queue.isOpen(id)) {
-            queue.push(id, item)
+            queue.push(id, { next: item })
           }
-        }
+        })
 
         if (item.done) {
           break
         }
       }
     }
-    catch (reason) {
-      error = { value: reason }
-
-      queue.waiterIds.forEach((id) => {
-        queue.close({ id, reason })
+    catch (error) {
+      ids.forEach((id) => {
+        if (queue.isOpen(id)) {
+          queue.push(id, { error })
+        }
       })
     }
+    finally {
+      isSourceFinished = true
+    }
   })
 
-  for (let i = 0; i < count; i++) {
-    const id = i.toString()
-
+  const replicated: AsyncIteratorClass<T, TReturn, TNext>[] = ids.map((id) => {
     queue.open(id)
-    replicated.push(new AsyncIteratorClass(
-      () => {
+
+    return new AsyncIteratorClass(
+      async () => {
         start()
 
-        return new Promise((resolve, reject) => {
-          if (!error || queue.hasBufferedItems(id)) {
-            queue.pull(id)
-              .then(resolve)
-              .catch(reject)
-          }
-          else {
-            reject(error.value)
-          }
-        })
+        const item = await queue.pull(id)
+
+        if (item.next) {
+          return item.next
+        }
+
+        throw item.error
       },
       async (reason) => {
         queue.close({ id })
 
-        if (reason !== 'next') {
-          if (!queue.length) {
-            await source?.return?.()
-          }
+        if (reason !== 'next' && !queue.length && !isSourceFinished) {
+          isSourceFinished = true
+          await source?.return?.()
         }
       },
-    ))
-  }
+    )
+  })
 
   return replicated
 }