Fix race conditions in PrimitiveSequence+Concurrency (#2641)

This caused continuations to be resumed more than once, crashing clients due to a violation of the Swift Concurrency contract

Author: 0xpablo

CHANGELOG.md

@@ -12,6 +12,10 @@ All notable changes to this project will be documented in this file.
 * Provides `Infallible` versions of `CombineLatest+Collection` helpers.
 * Explicitly declare `APPLICATION_EXTENSION_API_ONLY` for CocoaPods 
 
+### Anomalies
+
+* Fixes a crash that could occur when awaiting a `Single`, `Maybe`, or `Completable` that was disposed.
+
 ## 6.5.0
 
 You can now use `await` on `Observable`-conforming objects (as well as `Driver`, `Signal`, `Infallible`, `Single`, `Completable`) using the following syntax:

RxSwift/Traits/PrimitiveSequence/PrimitiveSequence+Concurrency.swift

@@ -62,19 +62,29 @@ public extension PrimitiveSequenceType where Trait == SingleTrait {
  operation: {
  try await withCheckedThrowingContinuation { continuation in
  var didResume = false
+ let lock = RecursiveLock()
  disposable.setDisposable(
  self.subscribe(
- onSuccess: {
- didResume = true
- continuation.resume(returning: $0)
+ onSuccess: { value in
+ lock.withLock {
+ guard !didResume else { return }
+ didResume = true
+ continuation.resume(returning: value)
+ }
  },
- onFailure: {
- didResume = true
- continuation.resume(throwing: $0)
+ onFailure: { error in
+ lock.withLock {
+ guard !didResume else { return }
+ didResume = true
+ continuation.resume(throwing: error)
+ }
  },
  onDisposed: {
- guard !didResume else { return }
- continuation.resume(throwing: CancellationError())
+ lock.withLock {
+ guard !didResume else { return }
+ didResume = true
+ continuation.resume(throwing: CancellationError())
+ }
  }
  )
  )
@@ -111,27 +121,37 @@ public extension PrimitiveSequenceType where Trait == MaybeTrait {
  return try await withTaskCancellationHandler(
  operation: {
  try await withCheckedThrowingContinuation { continuation in
- var didEmit = false
  var didResume = false
+ let lock = RecursiveLock()
  disposable.setDisposable(
  self.subscribe(
  onSuccess: { value in
- didEmit = true
- didResume = true
- continuation.resume(returning: value)
+ lock.withLock {
+ guard !didResume else { return }
+ didResume = true
+ continuation.resume(returning: value)
+ }
  },
  onError: { error in
- didResume = true
- continuation.resume(throwing: error)
+ lock.withLock {
+ guard !didResume else { return }
+ didResume = true
+ continuation.resume(throwing: error)
+ }
  },
  onCompleted: {
- guard !didEmit else { return }
- didResume = true
- continuation.resume(returning: nil)
+ lock.withLock {
+ guard !didResume else { return }
+ didResume = true
+ continuation.resume(returning: nil)
+ }
  },
  onDisposed: {
- guard !didResume else { return }
- continuation.resume(throwing: CancellationError())
+ lock.withLock {
+ guard !didResume else { return }
+ didResume = true
+ continuation.resume(throwing: CancellationError())
+ }
  }
  )
  )
@@ -168,19 +188,29 @@ public extension PrimitiveSequenceType where Trait == CompletableTrait, Element
  operation: {
  try await withCheckedThrowingContinuation { continuation in
  var didResume = false
+ let lock = RecursiveLock()
  disposable.setDisposable(
  self.subscribe(
  onCompleted: {
- didResume = true
- continuation.resume()
+ lock.withLock {
+ guard !didResume else { return }
+ didResume = true
+ continuation.resume()
+ }
  },
  onError: { error in
- didResume = true
- continuation.resume(throwing: error)
+ lock.withLock {
+ guard !didResume else { return }
+ didResume = true
+ continuation.resume(throwing: error)
+ }
  },
  onDisposed: {
- guard !didResume else { return }
- continuation.resume(throwing: CancellationError())
+ lock.withLock {
+ guard !didResume else { return }
+ didResume = true
+ continuation.resume(throwing: CancellationError())
+ }
  }
  )
  )

Tests/RxSwiftTests/PrimitiveSequence+ConcurrencyTests.swift

@@ -86,6 +86,40 @@ extension PrimitiveSequenceConcurrencyTests {
  [randomResult]
  )
  }
+
+ /// A previous implementation of the `Single` to swift concurrency bridge had a bug where it would sometimes call the continuation twice.
+ /// The current number of iterations is a sweet spot to not make the tests too slow while still catching the bug in most runs.
+ /// If you are debugging this issue you might want to increase the iterations and/or run this test repeatedly.
+ func testSingleContinuationIsNotResumedTwice() {
+ let expectation = XCTestExpectation()
+ let iterations = 10000
+ for i in 0 ..< iterations {
+ DispatchQueue.global(qos: .userInitiated).async {
+ let single = Single<Int>.create { observer in
+ DispatchQueue.global(qos: .userInitiated).asyncAfter(deadline: .now() + 0.005) {
+ observer(.success(42))
+ }
+ return Disposables.create()
+ }
+
+ let task = Task {
+ _ = try await single.value
+ }
+
+ DispatchQueue.global(qos: .userInitiated).asyncAfter(deadline: .now() + 0.005) {
+ task.cancel()
+ }
+
+ self.sleep(Double.random(in: 0.004...0.006))
+
+ if i == iterations - 1 {
+ expectation.fulfill()
+ }
+ }
+ }
+
+ wait(for: [expectation], timeout: 10)
+ }
 }
 
 // MARK: - Maybe
@@ -167,6 +201,40 @@ extension PrimitiveSequenceConcurrencyTests {
  try await Task.sleep(nanoseconds: 1_000_000)
  task.cancel()
  }
+
+ /// A previous implementation of the `Single` to swift concurrency bridge had a bug where it would sometimes call the continuation twice.
+ /// The current number of iterations is a sweet spot to not make the tests too slow while still catching the bug in most runs.
+ /// If you are debugging this issue you might want to increase the iterations and/or run this test repeatedly.
+ func testMaybeContinuationIsNotResumedTwice() {
+ let expectation = XCTestExpectation()
+ let iterations = 10000
+ for i in 0 ..< iterations {
+ DispatchQueue.global(qos: .userInitiated).async {
+ let maybe = Maybe<Bool>.create { observer in
+ DispatchQueue.global(qos: .userInitiated).asyncAfter(deadline: .now() + 0.005) {
+ observer(.success(true))
+ }
+ return Disposables.create()
+ }
+
+ let task = Task {
+ _ = try await maybe.value
+ }
+
+ DispatchQueue.global(qos: .userInitiated).asyncAfter(deadline: .now() + 0.005) {
+ task.cancel()
+ }
+
+ self.sleep(Double.random(in: 0.004...0.006))
+
+ if i == iterations - 1 {
+ expectation.fulfill()
+ }
+ }
+ }
+
+ wait(for: [expectation], timeout: 10)
+ }
 }
 
 // MARK: - Completable
@@ -220,6 +288,40 @@ extension PrimitiveSequenceConcurrencyTests {
  }
  }.cancel()
  }
+
+ /// A previous implementation of the `Single` to swift concurrency bridge had a bug where it would sometimes call the continuation twice.
+ /// The current number of iterations is a sweet spot to not make the tests too slow while still catching the bug in most runs.
+ /// If you are debugging this issue you might want to increase the iterations and/or run this test repeatedly.
+ func testCompletableContinuationIsNotResumedTwice() {
+ let expectation = XCTestExpectation()
+ let iterations = 10000
+ for i in 0 ..< iterations {
+ DispatchQueue.global(qos: .userInitiated).async {
+ let completable = Completable.create { observer in
+ DispatchQueue.global(qos: .userInitiated).asyncAfter(deadline: .now() + 0.005) {
+ observer(.completed)
+ }
+ return Disposables.create()
+ }
+
+ let task = Task {
+ _ = try await completable.value
+ }
+
+ DispatchQueue.global(qos: .userInitiated).asyncAfter(deadline: .now() + 0.005) {
+ task.cancel()
+ }
+
+ self.sleep(Double.random(in: 0.004...0.006))
+
+ if i == iterations - 1 {
+ expectation.fulfill()
+ }
+ }
+ }
+
+ wait(for: [expectation], timeout: 10)
+ }
 }
 #endif