Add reproduction for TxPubSub issue

.changeset/fix-txpubsub-publish-all-iterables.md

@@ -0,0 +1,5 @@
+---
+"effect": patch
+---
+
+Fix `TxPubSub.publishAll` dropping values from one-shot iterables when a transaction retries.

packages/effect/src/TxPubSub.ts

@@ -9,6 +9,7 @@
  *
  * @since 4.0.0
  */
+import * as Arr from "./Array.ts"
 import * as Effect from "./Effect.ts"
 import { dual } from "./Function.ts"
 import type { Inspectable } from "./Inspectable.ts"
@@ -471,17 +472,19 @@ export const publishAll: {
   <A>(self: TxPubSub<A>, values: Iterable<A>): Effect.Effect<boolean>
 } = dual(
   2,
-  <A>(self: TxPubSub<A>, values: Iterable<A>): Effect.Effect<boolean> =>
-    Effect.gen(function*() {
+  <A>(self: TxPubSub<A>, values: Iterable<A>): Effect.Effect<boolean> => {
+    const valuesArray = Arr.fromIterable(values)
+    return Effect.gen(function*() {
       if (yield* TxRef.get(self.shutdownRef)) return false
 
       let allAccepted = true
-      for (const value of values) {
+      for (const value of valuesArray) {
         const accepted = yield* publish(self, value)
         if (!accepted) allAccepted = false
       }
       return allAccepted
     }).pipe(Effect.tx)
+  }
 )
 
 /**

packages/effect/test/TxPubSub.test.ts

@@ -1,5 +1,5 @@
 import { assert, describe, it } from "@effect/vitest"
-import { Effect, Fiber, TxPubSub, TxQueue } from "effect"
+import { Effect, Fiber, Option, TxPubSub, TxQueue } from "effect"
 
 describe("TxPubSub", () => {
   describe("constructors", () => {
@@ -95,6 +95,33 @@ describe("TxPubSub", () => {
         )
       }))
 
+    it.effect("publishAll preserves a one-shot iterable across retries", () =>
+      Effect.gen(function*() {
+        const hub = yield* TxPubSub.bounded<number>(1)
+
+        yield* Effect.scoped(
+          Effect.gen(function*() {
+            const sub = yield* TxPubSub.subscribe(hub)
+            yield* TxPubSub.publish(hub, 1)
+
+            let iterations = 0
+            const hasIterated = () => iterations > 0
+            const values = (function*() {
+              iterations++
+              yield 2
+            })()
+            const fiber = yield* Effect.forkChild(TxPubSub.publishAll(hub, values))
+            while (!hasIterated()) {
+              yield* Effect.yieldNow
+            }
+
+            assert.strictEqual(yield* TxQueue.take(sub), 1)
+            assert.strictEqual(yield* Fiber.join(fiber), true)
+            assert.deepStrictEqual(yield* TxQueue.poll(sub), Option.some(2))
+          })
+        )
+      }))
+
     it.effect("subscriber only receives messages published after subscription", () =>
       Effect.gen(function*() {
         const hub = yield* Effect.tx(TxPubSub.unbounded<number>())