Crash: `os_unfair_lock` unlocked from wrong thread in `MergeStorage`

[Arideno]

We hit an intermittent crash in production using chunks(ofCount:or:) on an AsyncStream. It comes down to the os_unfair_lock inside MergeStorage being locked on one cooperative-pool thread and then unlocked on another, which os_unfair_lock doesn't allow:

EXC_BREAKPOINT
_os_unfair_lock_unowned_abort

chunks(ofCount:or:) builds a merge of the stream and an AsyncTimerSequence under the hood, so this is really a merge issue. It looks like the same root cause as #341 — merge resumes its continuation from an unstructured task without carrying the executor, so the resume (and the unlock) lands on a different thread than the one that took the lock.

Stack trace

Thread 48, queue = 'com.apple.root.user-initiated-qos.cooperative'
stop reason = EXC_BREAKPOINT

#0  _os_unfair_lock_unowned_abort                      libsystem_platform.dylib
#1  _os_unfair_lock_unlock_slow                        libsystem_platform.dylib
#2  static Lock.unlock(_:)                             Locking.swift:103
#3  Lock.unlock()                                      Locking.swift:136
#4  closure #2 in closure #1 in MergeStorage.next(continuation:)   MergeStorage.swift:109
#7  closure #1 in MergeStorage.next()                  MergeStorage.swift:107
#9  withTaskCancellationHandler(operation:onCancel:isolation:)     libswift_Concurrency
#11 MergeStorage.next()                                MergeStorage.swift:65
#12 AsyncMerge2Sequence.Iterator.InternalClass.next()  AsyncMerge2Sequence.swift:92
#13 AsyncMerge2Sequence.Iterator.next()                AsyncMerge2Sequence.swift:103
#14 AsyncChunksOfCountOrSignalSequence.Iterator.next() AsyncChunksOfCountOrSignalSequence.swift:129
#16 AsyncIteratorProtocol.next(isolation:)             libswift_Concurrency
#17 for await batch in stream.chunks(ofCount: 64, or: .repeating(every: .seconds(30)))

The lock is taken in MergeStorage.next() (frame #11) and released in the continuation-resume closure (frame #4), which runs on a different thread.

Repro

A producer task yields into an AsyncStream while a consumer drains it through chunks(ofCount:or:). The count-vs-timer merge keeps both branches hot, so the lock acquire and release frequently land on different threads. It's a timing race, so it doesn't fail on the first iteration — running several concurrently makes it show up reliably, usually within seconds to a minute.

import AsyncAlgorithms

func reproduce() async {
    let (stream, continuation) = AsyncStream.makeStream(of: Int.self)

    let producer = Task.detached {
        for i in 0..<1_000_000 {
            continuation.yield(i)
        }
        continuation.finish()
    }

    for await batch in stream.chunks(ofCount: 64, or: .repeating(every: .milliseconds(10))) {
        _ = batch.count
    }

    await producer.value
}

await withTaskGroup(of: Void.self) { group in
    for _ in 0..<50 {
        group.addTask { await reproduce() }
    }
}

Reproduces faster on multi-core hardware. A shorter timer interval and more concurrent children widen the race window.

Environment

• swift-async-algorithms 1.1.4
• Swift 6.3.3 release toolchain
• iOS (device and simulator), arm64
• Default cooperative executor, no custom executor

[NotTheNHK]

What exact toolchain version are you using? I believe this should be resolved in Swift 6.4 with the new nonisolated(nonsending) withUnsafeThrowingContinuation version.

[Arideno]

@NotTheNHK we are currently using latest available Xcode, so it's 6.3.3

[thliu21]

I investigated this on the latest main at 3da39bbc.

The crash is consistent with MergeStorage.next() acquiring the Darwin os_unfair_lock, entering withUnsafeThrowingContinuation, and unlocking inside its closure. If the older continuation API re-enqueues the task before invoking that closure, the unlock can occur on a different thread and abort with _os_unfair_lock_unowned_abort.

There is relevant compiler-side evidence in Swift 6.4: swiftlang/swift#84944 changes the continuation APIs to nonisolated(nonsending) specifically to avoid unnecessary task re-enqueues and adds a runtime no-hop test.

I could not reproduce the crash with Xcode 26.5 / Swift 6.3.2. A stress test with 32 concurrent consumers and 250,000 yields each—8,000,000 yields total—passed in Release. A custom task executor that runs every enqueued job on a fresh thread also passed in both Debug and Release before any library change.

Since the original report uses Swift 6.3.3, this does not rule out a 6.3.3-specific regression. Could you verify whether the crash still occurs with Swift 6.4 and, if possible, share a minimal project including any custom executor or actor context? If it remains reproducible on 6.4, the likely library-side fix is to ensure MergeStorage never holds its thread-affine lock across continuation entry.

[FranzBusch]

It is correct that between 6.0 until 6.3.3 there is a bug in withContinuation APIs that re-enqueued the job breaking the guarantee that we rely on here. 6.4 should fix this. I would welcome a patch though that refactors the storage and the state machine to not hold the locker over a withContinuation and instead handles the race in the state machine.

[NotTheNHK]

I can take a look once I'm done with all the AsyncStream stuff, if no one else has picked it up by then.