Fix multicast delivery ordering under concurrent demand

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:**
 

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
   }
 

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