[BUG] Since #59, concurrent sends reach consumers in different orders and current-value/replay consumers can end on a stale value

[PaulTaykalo]

Hi @twittemb, a follow-up to #52 and #59.

Problem

Since #59, send delivers outside the lock, so two send calls made at the same time can reach consumers in different orders:

• Passthrough: consumer A gets 1, 2 while consumer B gets 2, 1.
• CurrentValue and Replay: a consumer can end on 1 while value is 2, and it stays stale until the next send.

Repro: two consumers, and two threads sending at the same time with DispatchQueue.concurrentPerform. On main, the consumers' arrays differ in almost every run. Before #59, they never did.

Suggested fix

Queue each delivery while holding the state lock, then deliver the queue in order without holding any lock:

final class OrderedDelivery: @unchecked Sendable {
  private let state = ManagedCriticalState((pending: Deque<() -> Void>(), isDelivering: false))

  /// Call while holding the subject's state lock. Returns true if the caller must `drain()` afterwards.
  func enqueue(_ delivery: @escaping () -> Void) -> Bool {
    self.state.withCriticalRegion { state in
      state.pending.append(delivery)
      defer { state.isDelivering = true }
      return !state.isDelivering
    }
  }

  /// Runs queued deliveries in order. Call without holding any lock.
  func drain() {
    while let delivery = self.state.withCriticalRegion({ state -> (() -> Void)? in
      if state.pending.isEmpty { state.isDelivering = false }
      return state.pending.popFirst()
    }) {
      delivery()
    }
  }
}

// In each subject's send(_:) and send(_ termination:):
let shouldDrain = self.state.withCriticalRegion { state -> Bool in
  // existing state updates stay here
  let channels = Array(state.channels.values)
  return self.delivery.enqueue { for channel in channels { channel.send(element) } }
}
if shouldDrain { self.delivery.drain() }

Holding a lock during delivery instead doesn't work. If a consumer's cancellation handler calls send, it deadlocks the same way as #52.

One trade-off: when several threads send at once, send can return before its value is delivered.

With this change, #59's AsyncSubjectCancellationTests and the full suite pass (152 tests). It also passes new tests for ordering and for a cancellation handler that calls send. I can open a PR with the fix and the tests.

[PaulTaykalo]

Suggested regression tests for the fix. Both files go in Tests/AsyncSubjets/.

1. Ordering. Two threads send at the same time. All consumers must get the same order, and a consumer's last value must match value or the replay buffer. These fail on main and pass with the suggested fix.

import Dispatch
@testable import AsyncExtensions
import Foundation
import XCTest

/// Two producers send to the same subject at the same time. Which producer goes first is
/// unspecified, but every subscriber should observe one agreed order, and the subject's stored
/// state (`value`, replay buffer) should match what subscribers last received.
final class AsyncSubjectConcurrentSendOrderingTests: XCTestCase {
  private let rounds = 200
  private let sendsPerProducer = 100

  func test_AsyncPassthroughSubject_subscribers_receive_concurrent_sends_in_the_same_order() async {
    var roundsWithDifferentOrder = 0
    for _ in 0..<rounds {
      let subject = AsyncPassthroughSubject<Int>()
      let received = await collect(subscribers: 2, from: subject) {
        sendConcurrently(producers: 2, sendsPerProducer: sendsPerProducer) { subject.send($0) }
        subject.send(.finished)
      }
      if received[0] != received[1] { roundsWithDifferentOrder += 1 }
    }
    XCTAssertEqual(
      roundsWithDifferentOrder, 0,
      "Subscribers saw concurrent sends in different orders in \(roundsWithDifferentOrder) of \(rounds) rounds"
    )
  }

  func test_AsyncCurrentValueSubject_subscriber_ends_on_the_subject_value_after_concurrent_sends() async {
    var roundsWithStaleSubscriber = 0
    for _ in 0..<rounds {
      let subject = AsyncCurrentValueSubject<Int>(-1)
      var value = 0
      let received = await collect(subscribers: 1, from: subject) {
        sendConcurrently(producers: 2, sendsPerProducer: sendsPerProducer) { subject.send($0) }
        value = subject.value
        subject.send(.finished)
      }
      if received[0].last != value { roundsWithStaleSubscriber += 1 }
    }
    XCTAssertEqual(
      roundsWithStaleSubscriber, 0,
      "The subscriber's last value differed from `value` in \(roundsWithStaleSubscriber) of \(rounds) rounds"
    )
  }

  func test_AsyncReplaySubject_subscriber_ends_on_the_replayed_value_after_concurrent_sends() async {
    var roundsWithStaleSubscriber = 0
    for _ in 0..<rounds {
      let subject = AsyncReplaySubject<Int>(bufferSize: 1)
      var replayedToLateSubscriber: Int?
      let received = await collect(subscribers: 1, from: subject) {
        sendConcurrently(producers: 2, sendsPerProducer: sendsPerProducer) { subject.send($0) }
        replayedToLateSubscriber = subject.state.criticalState.buffer.last
        subject.send(.finished)
      }
      if received[0].last != replayedToLateSubscriber { roundsWithStaleSubscriber += 1 }
    }
    XCTAssertEqual(
      roundsWithStaleSubscriber, 0,
      "The subscriber's last value differed from the replay buffer in \(roundsWithStaleSubscriber) of \(rounds) rounds"
    )
  }

  /// Registers `subscribers` consumers before `produce` runs, then returns what each one received.
  private func collect<Subject: AsyncSubject>(
    subscribers: Int,
    from subject: Subject,
    produce: () -> Void
  ) async -> [[Subject.Element]] where Subject.Element: Sendable {
    // Register every consumer up front so no value can be missed at setup.
    let iterators = (0..<subscribers).map { _ in subject.makeAsyncIterator() }
    let consumers = iterators.map { iterator in
      Task.detached { () -> [Subject.Element] in
        var iterator = iterator
        var values: [Subject.Element] = []
        while let value = try? await iterator.next() {
          values.append(value)
        }
        return values
      }
    }
    produce()
    var received: [[Subject.Element]] = []
    for consumer in consumers {
      received.append(await consumer.value)
    }
    return received
  }
}

/// Each producer sends distinct values from its own thread; returns when all producers are done.
private func sendConcurrently(producers: Int, sendsPerProducer: Int, send: (Int) -> Void) {
  DispatchQueue.concurrentPerform(iterations: producers) { producer in
    for index in 0..<sendsPerProducer {
      send(producer * 1_000_000 + index)
    }
  }
}

2. Cancellation handler that calls send. A consumer's cancellation handler sends to its own subject while another thread sends. This passes on main and with the suggested fix, and it deadlocks if delivery holds a lock.

import Dispatch
@testable import AsyncExtensions
import XCTest

final class AsyncSubjectCancellationHandlerSendTests: XCTestCase {
  func test_cancellation_handler_can_send_to_its_own_subject_during_a_concurrent_send() async {
    let done = expectation(description: "all races finished")
    DispatchQueue.global().async {
      for _ in 0..<50_000 {
        let subject = AsyncPassthroughSubject<Int>()
        let iterator = subject.makeAsyncIterator()
        let exited = DispatchSemaphore(value: 0)
        let consumer = Task.detached {
          defer { exited.signal() }
          await withTaskCancellationHandler {
            var iterator = iterator
            while let _ = await iterator.next() {}
          } onCancel: {
            subject.send(-1)
          }
        }
        // Wait until the consumer is suspended in next() before racing.
        while !isSuspended(subject) {
          Thread.sleep(forTimeInterval: 0.00005)
        }
        DispatchQueue.concurrentPerform(iterations: 2) { index in
          if index == 0 { consumer.cancel() } else { subject.send(1) }
        }
        _ = exited.wait(timeout: .now() + .seconds(5))
      }
      done.fulfill()
    }
    await fulfillment(of: [done], timeout: 60)
  }
}

private func isSuspended(_ subject: AsyncPassthroughSubject<Int>) -> Bool {
  guard let channel = subject.state.criticalState.channels.values.first else { return false }
  if case .awaiting = channel.state.criticalState { return true }
  return false
}

[PaulTaykalo]

Why each fix keeps breaking something

Each fix so far protected one guarantee and quietly broke another. These are the guarantees the subjects need to keep, with what broke each one and the test that now checks it:

Guarantee	Broken by	Checked by
Cancelling a consumer while send runs doesn't deadlock	original code (#52)	AsyncSubjectCancellationTests (#59)
A consumer added during send doesn't miss values	original code (#56)	test_subscription_racing_send_receives_sent_element (#57)
All consumers see concurrent sends in the same order	#59	ordering tests (comment above)
A consumer's last value matches value or the replay buffer	#59	ordering tests (comment above)
A cancellation handler can call send	holding a lock during delivery	handler test (comment above)
send has delivered before it returns	suggested fix, only while other threads are sending	not tested: an accepted trade-off

Two things would stop this from repeating:

1. Every change to the subjects should pass every test in this table, not only the test for the bug it fixes. Fix subject deadlocks during concurrent cancellation #59 only covered the first row, which is how the ordering problem slipped through.
2. One rule covers the deadlock rows: never hold a lock while resuming a consumer. The cancellation handler can take any lock that is held during a resume, and that's the cycle in [BUG] Possible deadlock in Async subjects #52. The suggested fix follows this rule, so it can't deadlock this way.