Synchronize ScanSink's accumulator to fix a data race (fixes #2679)

[jadhavgaurav]

Fixes #2679

Root cause

ScanSink<Element, Observer>.on(_:) in RxSwift/Observables/Scan.swift mutates self.accumulate, a plain var, directly on every .next event with no lock:

func on(_ event: Event<Element>) {
    switch event {
    case let .next(element):
        do {
            try parent.accumulator(&accumulate, element)
            forwardOn(.next(accumulate))
        ...

Every other sink with comparable cross-event mutable state (Merge, Zip, Multicast, ReplaySubject, Buffer, and others) guards that state with a lock via the LockOwnerType / SynchronizedOnType pattern. ScanSink is the odd one out.

This is reachable in practice, not just in theory: PublishSubject.on(_:) only takes its lock around synchronized_on, which builds a snapshot of the current observers. The actual delivery to those observers happens in dispatch(_:_:) (RxSwift/Extensions/Bag+Rx.swift), outside that lock:

public func on(_ event: Event<Element>) {
    dispatch(synchronized_on(event), event) // dispatch() runs unlocked
}

So two threads calling subject.onNext(_:) concurrently on a shared PublishSubject really can invoke ScanSink.on(_:) at the same time on two different threads, racing on the read-modify-write of accumulate.

Reproduction

I built a stress test that subscribes .scan(0) { acc, _ in acc + 1 } to a shared PublishSubject<Int> and fires 20,000-50,000 onNext calls concurrently via DispatchQueue.concurrentPerform, then checks whether every accumulated value from 1...N was observed exactly once.

Without the fix, this is genuinely flaky (as data races are), but reliably catchable:

• Plain debug runs at 50,000 iterations show duplicated and skipped accumulator values in roughly 1 out of every 2-3 runs (e.g. one run: 49,999 unique values out of 50,000 observed, with one value duplicated and another silently skipped — a classic lost update).
• RxSwift's own debug-only SynchronizationTracker (built into PublishSubject, unrelated to my test) independently flags the exact same run with 50,000+ "Synchronization anomaly" warnings, confirming overlapping event delivery from multiple unsynchronized threads.
• Under Thread Sanitizer (swift build -Xswiftc -sanitize=thread), the race is caught on every run, with a precise report pointing at the exact line:

WARNING: ThreadSanitizer: Swift access race (pid=85966)
  Modifying access of Swift variable at 0x00010ac01078 by thread T10:
    #0 ScanSink.on(_:) Scan.swift:65
    ...
    #8 dispatch<A>(_:_:) Bag+Rx.swift:13
    #9 PublishSubject.on(_:) PublishSubject.swift:62
    ...
  Previous modifying access of Swift variable at 0x00010ac01078 by thread T1:
    #0 ScanSink.on(_:) Scan.swift:65
    ...
  Thread T10 (tid=1090523, running) is a GCD worker thread
  Thread T1 (tid=1090513, running) is a GCD worker thread
SUMMARY: ThreadSanitizer: Swift access race Scan.swift:65 in ScanSink.on(_:)

Two different GCD worker threads modifying the same accumulate storage, both stacks going through PublishSubject.on → dispatch → ScanSink.on(_:) → Scan.swift:65 (the try parent.accumulator(&accumulate, element) line).

With the fix applied, the same harness is clean: 0 TSan warnings across multiple runs, 0 duplicate/lost values, deterministic pass every time.

To confirm the fix is actually what resolves it (not just a coincidence of timing), I reverted only the source change in Scan.swift (via git apply -R on a saved patch, keeping the stress test) and reran under the identical conditions: the TSan race report and functional corruption both came back immediately. Reapplying the fix made them disappear again.

The fix

Mirrors the existing LockOwnerType / SynchronizedOnType pattern already used by BufferTimeCountSink (and the same pattern used across Merge, Zip, etc.): on(_:) now delegates to synchronizedOn(_:), which takes the sink's RecursiveLock before calling into synchronized_on(_:), which holds the original, unchanged logic.

Test

Added testScan_ConcurrentNextEvents_DoesNotCorruptAccumulator to Observable+ScanTests.swift, following the concurrency-stress-test style already established in Observable+ConcurrencyTests.swift (DispatchQueue-based, real threads, not the virtual TestScheduler). It fires 20,000 concurrent onNext calls through scan() and asserts every value from 1 to 20,000 is observed exactly once, with no duplicates.

I wasn't able to run this through swift test or the Xcode AllTests-macOS scheme in my environment (no Xcode installation available, only Command Line Tools, so the RxTest/XCTest-linked targets can't build locally here) — this matches why the project's own CI never runs swift test either, only xcodebuild against the Xcode schemes. I validated the equivalent logic instead via a standalone SwiftPM executable that links only RxSwift (no XCTest dependency), which is what produced the plain-run and TSan results above. swift build --target RxSwift (and RxRelay/RxBlocking/RxCocoa) succeed cleanly with the fix in place. I'd appreciate a maintainer running the added test through the Xcode-based CI to confirm.

I wasn't able to run SwiftLint locally either (it depends on SourceKit from a full Xcode install, also unavailable here), but the change is styled to match BufferTimeCountSink exactly, and Tests/ isn't included in .swiftlint.yml's lint scope regardless.

Disclosure

I used AI assistance (Claude) to help investigate this issue and prepare this PR. The root cause analysis, the reproduction, and the fix were all verified directly against RxSwift's source and by running the stress test and TSan reports shown above; I'm not asking anyone to take these claims on faith without the evidence included here.

[jadhavgaurav]

Friendly ping. This fixes a data race in ScanSink and includes a regression test; CI is green. Happy to adjust anything.