flatMapLatest: crash and intermittent hang under a single consumer

[dvi]

Environment

swift-async-algorithms 1.1.4 (commit d0b4a06)
Swift 6.3.2 (swiftlang-6.3.2.1.108, clang-2100.1.1.101), swift-driver 1.148.6
Reproduced on iOS 18.4-26.2 Simulator (arm64), Xcode 26.5 (build 17F42), Swift language mode v6

Summary

Consuming a flatMapLatest sequence with a single for await loop, where the base switches inner sequences frequently, can trap in FlatMapLatestStateMachine.next(for:):

FlatMapLatestStateMachine.swift:132: Precondition failed: Already have downstream continuation
The same scenario also intermittently hangs (the downstream continuation is never resumed) instead of trapping.

In FlatMapLatestStorage.next(), the .suspend case releases the lock then suspend() re-acquires it to call next(for:):

https://github.com/apple/swift-async-algorithms/blob/1.1.4/Sources/AsyncAlgorithms/FlatMapLatest/FlatMapLatestStorage.swift#L53-L68
next(for:) then asserts the state is unchanged:

https://github.com/apple/swift-async-algorithms/blob/1.1.4/Sources/AsyncAlgorithms/FlatMapLatest/FlatMapLatestStateMachine.swift#L132-L133

precondition(downstreamCont == nil, "Already have downstream continuation")
precondition(buffer.isEmpty, "Buffer should be empty if suspending")

The operator's internal outer/inner tasks take the same lock and run concurrently with the consumer, so they can mutate the state during the window between next() releasing the lock and next(for:) re-acquiring it.

Prod Stack trace

Crashed: com.apple.root.utility-qos.cooperative
EXC_BREAKPOINT
0  FlatMapLatestStateMachine.next(for:)
1  closure #1 in closure #1 in FlatMapLatestStorage.suspend()
2  partial apply for closure #1 in closure #1 in FlatMapLatestStorage.suspend()
3  Lock.withLock<A>(_:)
4  closure #1 in FlatMapLatestStorage.suspend()
5  partial apply for closure #1 in FlatMapLatestStorage.suspend()
6  withUnsafeThrowingContinuation<A>(isolation:_:)
7  swift::runJobInEstablishedExecutorContext(swift::Job*)

Reproduction

for _ in 0 ..< 5000 {
    let toggles = AsyncStream<Bool> { continuation in
        for index in 0 ..< 20 {
            continuation.yield(index.isMultiple(of: 2))
        }
        continuation.finish()
    }

    let sequence = toggles.flatMapLatest { enabled in
        AsyncStream<Int> { continuation in
            if enabled {
                for value in 0 ..< 200 {
                    continuation.yield(value)
                }
            }
            continuation.finish()
        }
    }

    for await _ in sequence {}
}

[shubhransh-gupta]

Hi @dvi @phausler — Looking into this:
This issue (which existed in 1.1.4 where precondition(downstreamCont == nil) / precondition(buffer.isEmpty) trapped due to the window between next() unlocking and suspend() calling next(for:)) was resolved on main in commit 9d349bc (#403).

In that cleanup pass, flatMapLatest was updated so that internal task transitions and continuation handoffs prevent this race condition. Running the 5,000-iteration reproduction script and stress tests on the latest codebase now completes cleanly with zero crashes or hangs.