From 076a59ee09029baf17401f829e883aa98b084b47 Mon Sep 17 00:00:00 2001
From: Thibault Wittemberg <thibault.wittemberg@gmail.com>
Date: Sat, 3 Oct 2026 10:53:01 +0200
Subject: [PATCH] Fix multicast delivery ordering under concurrent demand

Broadcast each upstream result before another consumer can advance the iterator, preventing values from being reordered or overtaken by termination. Add concurrent subscriber regressions for normal completion and failure.
---
 CHANGELOG.md                                  |  1 +
 .../Operators/AsyncMulticastSequence.swift    | 13 +++---
 .../AsyncMulticastSequenceTests.swift         | 46 +++++++++++++++++++
 3 files changed, 54 insertions(+), 6 deletions(-)

diff --git a/CHANGELOG.md b/CHANGELOG.md
index ba628b4..5a02bc5 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -1,6 +1,7 @@
 **Unreleased:**
 
 - Subjects: fix a deadlock when sending values or termination concurrently with consumer cancellation (https://github.com/sideeffect-io/AsyncExtensions/issues/52).
+- Multicast: preserve upstream element order and prevent termination from overtaking values when consumers advance concurrently.
 
 **v0.5.2 - Oxygen:**
 
diff --git a/Sources/Operators/AsyncMulticastSequence.swift b/Sources/Operators/AsyncMulticastSequence.swift
index 1cc4971..697912a 100644
--- a/Sources/Operators/AsyncMulticastSequence.swift
+++ b/Sources/Operators/AsyncMulticastSequence.swift
@@ -128,14 +128,15 @@ where Base.Element == Subject.Element, Subject.Failure == Error, Base.AsyncItera
       }
 
       self.state.withCriticalRegion { state in
+        // Broadcast before another consumer can advance upstream, including its termination.
+        // This lock belongs to multicast and is not acquired by subject cancellation handlers.
+        switch toSend {
+          case .success(.some(let element)): self.subject.send(element)
+          case .success(.none): self.subject.send(.finished)
+          case .failure(let error): self.subject.send(.failure(error))
+        }
         state = .available(iterator)
       }
-
-      switch toSend {
-        case .success(.some(let element)): self.subject.send(element)
-        case .success(.none): self.subject.send(.finished)
-        case .failure(let error): self.subject.send(.failure(error))
-      }
     }.value
   }
 
diff --git a/Tests/Operators/AsyncMulticastSequenceTests.swift b/Tests/Operators/AsyncMulticastSequenceTests.swift
index ffe9828..8693e11 100644
--- a/Tests/Operators/AsyncMulticastSequenceTests.swift
+++ b/Tests/Operators/AsyncMulticastSequenceTests.swift
@@ -40,6 +40,52 @@ private class SpyAsyncSequenceForNumberOfIterators<Element>: AsyncSequence {
 }
 
 final class AsyncMulticastSequenceTests: XCTestCase {
+  func test_concurrent_consumers_receive_all_elements_in_order_before_finish() async {
+    await assertConcurrentDelivery()
+  }
+
+  func test_concurrent_consumers_receive_all_elements_in_order_before_failure() async {
+    await assertConcurrentDelivery(failure: MockError(code: 1701))
+  }
+
+  private func assertConcurrentDelivery(failure: MockError? = nil) async {
+    let elements = Array(0..<100)
+
+    for _ in 0..<20 {
+      let upstream = AsyncThrowingStream<Int, Error> { continuation in
+        elements.forEach { continuation.yield($0) }
+        continuation.finish(throwing: failure)
+      }
+      let sut = upstream.multicast(AsyncThrowingPassthroughSubject<Int, Error>())
+      // Register every subscriber before allowing any of them to advance upstream.
+      let iterators = (0..<8).map { _ in sut.makeAsyncIterator() }
+      let finished = expectation(description: "All concurrent subscribers finish")
+      finished.expectedFulfillmentCount = iterators.count
+      sut.connect()
+
+      let consumers = iterators.map { iterator in
+        Task {
+          defer { finished.fulfill() }
+          var iterator = iterator
+          var received = [Int]()
+          do {
+            while let element = try await iterator.next() {
+              received.append(element)
+            }
+            XCTAssertNil(failure)
+          } catch {
+            XCTAssertNotNil(failure)
+            XCTAssertEqual(error as? MockError, failure)
+          }
+          XCTAssertEqual(received, elements)
+        }
+      }
+
+      await fulfillment(of: [finished], timeout: 5)
+      consumers.forEach { $0.cancel() }
+    }
+  }
+
   func test_multiple_loops_receive_elements_from_single_baseIterator() {
     let taskHaveIterators = expectation(description: "All tasks have their iterator")
     taskHaveIterators.expectedFulfillmentCount = 2