Preserve ordered delivery for concurrent subject sends

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).
+- Subjects: preserve delivery order for concurrent sends while keeping consumer resumption outside the state lock (https://github.com/sideeffect-io/AsyncExtensions/issues/61).
 - Multicast: preserve upstream element order and prevent termination from overtaking values when consumers advance concurrently.
 
 **v0.5.2 - Oxygen:**

Sources/AsyncSubjects/AsyncCurrentValueSubject.swift

@@ -45,6 +45,7 @@ public final class AsyncCurrentValueSubject<Element>: AsyncSubject where Element
   }
 
   let state: ManagedCriticalState<State>
+  let delivery = OrderedDelivery()
 
   public var value: Element {
     get {
@@ -67,28 +68,32 @@ public final class AsyncCurrentValueSubject<Element>: AsyncSubject where Element
   /// Sends a value to all consumers
   /// - Parameter element: the value to send
   public func send(_ element: Element) {
-    let channels = self.state.withCriticalRegion { state in
+    let shouldDrain = self.state.withCriticalRegion { state in
       state.current = element
-      return Array(state.channels.values)
-    }
-    // Resuming a consumer must not hold the lock used by its cancellation handler.
-    for channel in channels {
-      channel.send(element)
+      let channels = Array(state.channels.values)
+      return self.delivery.enqueue {
+        for channel in channels {
+          channel.send(element)
+        }
+      }
     }
+    if shouldDrain { self.delivery.drain() }
   }
 
   /// Finishes the async sequences with a normal ending.
   /// - Parameter termination: The termination to finish the subject.
   public func send(_ termination: Termination<Failure>) {
-    let channels = self.state.withCriticalRegion { state in
+    let shouldDrain = self.state.withCriticalRegion { state in
       state.terminalState = termination
       let channels = Array(state.channels.values)
       state.channels.removeAll()
-      return channels
-    }
-    for channel in channels {
-      channel.finish()
+      return self.delivery.enqueue {
+        for channel in channels {
+          channel.finish()
+        }
+      }
     }
+    if shouldDrain { self.delivery.drain() }
   }
 
   func handleNewConsumer() -> (iterator: AsyncBufferedChannel<Element>.Iterator, unregister: @Sendable () -> Void) {

Sources/AsyncSubjects/AsyncPassthroughSubject.swift

@@ -42,6 +42,7 @@ public final class AsyncPassthroughSubject<Element: Sendable>: AsyncSubject {
   }
 
   let state: ManagedCriticalState<State>
+  let delivery = OrderedDelivery()
 
   public init() {
     self.state = ManagedCriticalState(
@@ -52,27 +53,31 @@ public final class AsyncPassthroughSubject<Element: Sendable>: AsyncSubject {
   /// Sends a value to all consumers
   /// - Parameter element: the value to send
   public func send(_ element: Element) {
-    let channels = self.state.withCriticalRegion { state in
-      return Array(state.channels.values)
-    }
-    // Resuming a consumer must not hold the lock used by its cancellation handler.
-    for channel in channels {
-      channel.send(element)
+    let shouldDrain = self.state.withCriticalRegion { state in
+      let channels = Array(state.channels.values)
+      return self.delivery.enqueue {
+        for channel in channels {
+          channel.send(element)
+        }
+      }
     }
+    if shouldDrain { self.delivery.drain() }
   }
 
   /// Finishes the subject with a normal ending.
   /// - Parameter termination: The termination to finish the subject
   public func send(_ termination: Termination<Failure>) {
-    let channels = self.state.withCriticalRegion { state in
+    let shouldDrain = self.state.withCriticalRegion { state in
       state.terminalState = termination
       let channels = Array(state.channels.values)
       state.channels.removeAll()
-      return channels
-    }
-    for channel in channels {
-      channel.finish()
+      return self.delivery.enqueue {
+        for channel in channels {
+          channel.finish()
+        }
+      }
     }
+    if shouldDrain { self.delivery.drain() }
   }
 
   func handleNewConsumer() -> (iterator: AsyncBufferedChannel<Element>.Iterator, unregister: @Sendable () -> Void) {

Sources/AsyncSubjects/AsyncReplaySubject.swift

@@ -34,6 +34,7 @@ public final class AsyncReplaySubject<Element>: AsyncSubject where Element: Send
   }
 
   let state: ManagedCriticalState<State>
+  let delivery = OrderedDelivery()
 
   /// Creates a replay subject with a buffer of replayable values
   /// - Parameter bufferSize: the number of values that will be replayed to new consumers
@@ -46,33 +47,37 @@ public final class AsyncReplaySubject<Element>: AsyncSubject where Element: Send
   /// Sends a value to all consumers
   /// - Parameter element: the value to send
   public func send(_ element: Element) {
-    let channels = self.state.withCriticalRegion { state in
+    let shouldDrain = self.state.withCriticalRegion { state in
       if state.buffer.count >= state.bufferSize && !state.buffer.isEmpty {
         state.buffer.removeFirst()
       }
       state.buffer.append(element)
-      return Array(state.channels.values)
-    }
-    // Resuming a consumer must not hold the lock used by its cancellation handler.
-    for channel in channels {
-      channel.send(element)
+      let channels = Array(state.channels.values)
+      return self.delivery.enqueue {
+        for channel in channels {
+          channel.send(element)
+        }
+      }
     }
+    if shouldDrain { self.delivery.drain() }
   }
 
   /// Finishes the subject with a normal ending.
   /// - Parameter termination: The termination to finish the subject.
   public func send(_ termination: Termination<Failure>) {
-    let channels = self.state.withCriticalRegion { state in
+    let shouldDrain = self.state.withCriticalRegion { state in
       state.terminalState = termination
       let channels = Array(state.channels.values)
       state.channels.removeAll()
       state.buffer.removeAll()
       state.bufferSize = 0
-      return channels
-    }
-    for channel in channels {
-      channel.finish()
+      return self.delivery.enqueue {
+        for channel in channels {
+          channel.finish()
+        }
+      }
     }
+    if shouldDrain { self.delivery.drain() }
   }
 
   func handleNewConsumer() -> (iterator: AsyncBufferedChannel<Element>.Iterator, unregister: @Sendable () -> Void) {

Sources/AsyncSubjects/AsyncThrowingCurrentValueSubject.swift

@@ -47,6 +47,7 @@ public final class AsyncThrowingCurrentValueSubject<Element, Failure: Error>: As
   }
 
   let state: ManagedCriticalState<State>
+  let delivery = OrderedDelivery()
 
   public var value: Element {
     get {
@@ -67,33 +68,35 @@ public final class AsyncThrowingCurrentValueSubject<Element, Failure: Error>: As
   /// Sends a value to all consumers
   /// - Parameter element: the value to send
   public func send(_ element: Element) {
-    let channels = self.state.withCriticalRegion { state in
+    let shouldDrain = self.state.withCriticalRegion { state in
       state.current = element
-      return Array(state.channels.values)
-    }
-    // Resuming a consumer must not hold the lock used by its cancellation handler.
-    for channel in channels {
-      channel.send(element)
+      let channels = Array(state.channels.values)
+      return self.delivery.enqueue {
+        for channel in channels {
+          channel.send(element)
+        }
+      }
     }
+    if shouldDrain { self.delivery.drain() }
   }
 
   /// Finishes the subject with either a normal ending or an error.
   /// - Parameter termination: The termination to finish the subject.
   public func send(_ termination: Termination<Failure>) {
-    let channels = self.state.withCriticalRegion { state in
+    let shouldDrain = self.state.withCriticalRegion { state in
       state.terminalState = termination
       let channels = Array(state.channels.values)
       state.channels.removeAll()
-      return channels
-    }
-    for channel in channels {
-      switch termination {
-        case .finished:
-          channel.finish()
-        case .failure(let error):
-          channel.fail(error)
+      return self.delivery.enqueue {
+        switch termination {
+          case .finished:
+            for channel in channels { channel.finish() }
+          case .failure(let error):
+            for channel in channels { channel.fail(error) }
+        }
       }
     }
+    if shouldDrain { self.delivery.drain() }
   }
 
   func handleNewConsumer(

Sources/AsyncSubjects/AsyncThrowingPassthroughSubject.swift

@@ -43,6 +43,7 @@ public final class AsyncThrowingPassthroughSubject<Element, Failure: Error>: Asy
   }
 
   let state: ManagedCriticalState<State>
+  let delivery = OrderedDelivery()
 
   public init() {
     self.state = ManagedCriticalState(
@@ -53,32 +54,36 @@ public final class AsyncThrowingPassthroughSubject<Element, Failure: Error>: Asy
   /// Sends a value to all consumers
   /// - Parameter element: the value to send
   public func send(_ element: Element) {
-    let channels = self.state.withCriticalRegion { state in
-      return Array(state.channels.values)
-    }
-    // Resuming a consumer must not hold the lock used by its cancellation handler.
-    for channel in channels {
-      channel.send(element)
+    let shouldDrain = self.state.withCriticalRegion { state in
+      let channels = Array(state.channels.values)
+      return self.delivery.enqueue {
+        for channel in channels {
+          channel.send(element)
+        }
+      }
     }
+    if shouldDrain { self.delivery.drain() }
   }
 
   /// Finishes the subject with either a normal ending or an error.
   /// - Parameter termination: The termination to finish the subject
   public func send(_ termination: Termination<Failure>) {
-    let channels = self.state.withCriticalRegion { state in
+    let shouldDrain = self.state.withCriticalRegion { state in
       state.terminalState = termination
       let channels = Array(state.channels.values)
       state.channels.removeAll()
-      return channels
-    }
-    for channel in channels {
-      switch termination {
-        case .finished:
-          channel.finish()
-        case .failure(let error):
-          channel.fail(error)
+      return self.delivery.enqueue {
+        for channel in channels {
+          switch termination {
+            case .finished:
+              channel.finish()
+            case .failure(let error):
+              channel.fail(error)
+          }
+        }
       }
     }
+    if shouldDrain { self.delivery.drain() }
   }
 
   func handleNewConsumer(

Sources/AsyncSubjects/AsyncThrowingReplaySubject.swift

@@ -35,6 +35,7 @@ public final class AsyncThrowingReplaySubject<Element, Failure: Error>: AsyncSub
   }
 
   let state: ManagedCriticalState<State>
+  let delivery = OrderedDelivery()
 
   public init(bufferSize: UInt) {
     self.state = ManagedCriticalState(
@@ -45,38 +46,40 @@ public final class AsyncThrowingReplaySubject<Element, Failure: Error>: AsyncSub
   /// Sends a value to all consumers
   /// - Parameter element: the value to send
   public func send(_ element: Element) {
-    let channels = self.state.withCriticalRegion { state in
+    let shouldDrain = self.state.withCriticalRegion { state in
       if state.buffer.count >= state.bufferSize && !state.buffer.isEmpty {
         state.buffer.removeFirst()
       }
       state.buffer.append(element)
-      return Array(state.channels.values)
-    }
-    // Resuming a consumer must not hold the lock used by its cancellation handler.
-    for channel in channels {
-      channel.send(element)
+      let channels = Array(state.channels.values)
+      return self.delivery.enqueue {
+        for channel in channels {
+          channel.send(element)
+        }
+      }
     }
+    if shouldDrain { self.delivery.drain() }
   }
 
   /// Finishes the subject with either a normal ending or an error.
   /// - Parameter termination: The termination to finish the subject
   public func send(_ termination: Termination<Failure>) {
-    let channels = self.state.withCriticalRegion { state in
+    let shouldDrain = self.state.withCriticalRegion { state in
       state.terminalState = termination
       let channels = Array(state.channels.values)
       state.channels.removeAll()
       state.buffer.removeAll()
       state.bufferSize = 0
-      return channels
-    }
-    for channel in channels {
-      switch termination {
-        case .finished:
-          channel.finish()
-        case .failure(let error):
-          channel.fail(error)
+      return self.delivery.enqueue {
+        switch termination {
+          case .finished:
+            for channel in channels { channel.finish() }
+          case .failure(let error):
+            for channel in channels { channel.fail(error) }
+        }
       }
     }
+    if shouldDrain { self.delivery.drain() }
   }
 
   func handleNewConsumer(

Sources/Supporting/OrderedDelivery.swift

@@ -0,0 +1,35 @@
+import DequeModule
+
+/// Delivers subject events in the same order that their state updates are recorded.
+/// The queue lock is released before a delivery resumes any consumer.
+final class OrderedDelivery: @unchecked Sendable {
+  private struct State {
+    var pending: Deque<@Sendable () -> Void> = []
+    var isDelivering = false
+  }
+
+  private let state = ManagedCriticalState(State())
+
+  /// Queues a delivery and returns true when the caller should start draining.
+  func enqueue(_ delivery: @escaping @Sendable () -> Void) -> Bool {
+    self.state.withCriticalRegion { state in
+      state.pending.append(delivery)
+      guard !state.isDelivering else { return false }
+      state.isDelivering = true
+      return true
+    }
+  }
+
+  /// Drains queued deliveries without holding the queue lock while they run.
+  func drain() {
+    while let delivery = self.state.withCriticalRegion({ state -> (@Sendable () -> Void)? in
+      guard !state.pending.isEmpty else {
+        state.isDelivering = false
+        return nil
+      }
+      return state.pending.popFirst()
+    }) {
+      delivery()
+    }
+  }
+}