Fix Effect.throttle data races and scheduler value return (#669)

* Fix `Effect.throttle` data races and scheduler value return

The `Effect.throttle` operator had multiple data races when updating
`throttleTimes` and `throttleValues` shared state, because:

1. `Effect.throttle` can be called from any scheduler.
2. The internal `flatMap` runs on the current chain scheduler, but the
throttled (delayed) value runs on the passed in `scheduler` parameter
(which can be different).

By protecting shared state with a lock (similar to the cancellables'),
these data races should be addressed.

Additionally, the `Effect.throttle` should return all values in the
`scheduler` passed in, so that the API contract is honored and values
come from where callers expect them to.

## Changes

- Add new `throttleLock` `recursive lock to protect `throttleTimes`
and `throttleValues` shared state in `Effect.throttle` operator.

- Ensure all values in `Effect.throttle` come from the passed in
`scheduler`.

* Fix `Effect.throttle` tests, Use `sync` lock helper

Author: p4checo

Sources/ComposableArchitecture/Effects/Throttling.swift

@@ -19,34 +19,41 @@ extension Effect {
     scheduler: S,
     latest: Bool
   ) -> Effect where S: Scheduler {
-    self.flatMap { value -> AnyPublisher<Output, Failure> in
-      guard let throttleTime = throttleTimes[id] as! S.SchedulerTimeType? else {
-        throttleTimes[id] = scheduler.now
-        throttleValues[id] = nil
-        return Just(value).setFailureType(to: Failure.self).eraseToAnyPublisher()
-      }
+    self.receive(on: scheduler)
+      .flatMap { value -> AnyPublisher<Output, Failure> in
+        throttleLock.lock()
+        defer { throttleLock.unlock() }
 
-      let value = latest ? value : (throttleValues[id] as! Output? ?? value)
-      throttleValues[id] = value
+        guard let throttleTime = throttleTimes[id] as! S.SchedulerTimeType? else {
+          throttleTimes[id] = scheduler.now
+          throttleValues[id] = nil
+          return Just(value).setFailureType(to: Failure.self).eraseToAnyPublisher()
+        }
 
-      guard throttleTime.distance(to: scheduler.now) < interval else {
-        throttleTimes[id] = scheduler.now
-        throttleValues[id] = nil
-        return Just(value).setFailureType(to: Failure.self).eraseToAnyPublisher()
-      }
+        let value = latest ? value : (throttleValues[id] as! Output? ?? value)
+        throttleValues[id] = value
 
-      return Just(value)
-        .delay(
-          for: scheduler.now.distance(to: throttleTime.advanced(by: interval)), scheduler: scheduler
-        )
-        .handleEvents(receiveOutput: { _ in throttleTimes[id] = scheduler.now })
-        .setFailureType(to: Failure.self)
-        .eraseToAnyPublisher()
-    }
-    .eraseToEffect()
-    .cancellable(id: id, cancelInFlight: true)
+        guard throttleTime.distance(to: scheduler.now) < interval else {
+          throttleTimes[id] = scheduler.now
+          throttleValues[id] = nil
+          return Just(value).setFailureType(to: Failure.self).eraseToAnyPublisher()
+        }
+
+        return Just(value)
+          .delay(
+            for: scheduler.now.distance(to: throttleTime.advanced(by: interval)), scheduler: scheduler
+          )
+          .handleEvents(
+            receiveOutput: { _ in throttleLock.sync { throttleTimes[id] = scheduler.now } }
+          )
+          .setFailureType(to: Failure.self)
+          .eraseToAnyPublisher()
+      }
+      .eraseToEffect()
+      .cancellable(id: id, cancelInFlight: true)
   }
 }
 
 var throttleTimes: [AnyHashable: Any] = [:]
 var throttleValues: [AnyHashable: Any] = [:]
+let throttleLock = NSRecursiveLock()

Tests/ComposableArchitectureTests/EffectThrottleTests.swift

@@ -26,11 +26,15 @@ final class EffectThrottleTests: XCTestCase {
 
     runThrottledEffect(value: 1)
 
+    scheduler.advance()
+
     // A value emits right away.
     XCTAssertEqual(values, [1])
 
     runThrottledEffect(value: 2)
 
+    scheduler.advance()
+
     // A second value is throttled.
     XCTAssertEqual(values, [1])
 
@@ -76,11 +80,15 @@ final class EffectThrottleTests: XCTestCase {
 
     runThrottledEffect(value: 1)
 
+    scheduler.advance()
+
     // A value emits right away.
     XCTAssertEqual(values, [1])
 
     runThrottledEffect(value: 2)
 
+    scheduler.advance()
+
     // A second value is throttled.
     XCTAssertEqual(values, [1])
 
@@ -124,13 +132,17 @@ final class EffectThrottleTests: XCTestCase {
 
     runThrottledEffect(value: 1)
 
+    scheduler.advance()
+
     // A value emits right away.
     XCTAssertEqual(values, [1])
 
     scheduler.advance(by: 2)
 
     runThrottledEffect(value: 2)
 
+    scheduler.advance()
+
     // A second value is emitted right away.
     XCTAssertEqual(values, [1, 2])
   }
@@ -156,6 +168,8 @@ final class EffectThrottleTests: XCTestCase {
 
     runThrottledEffect(value: 1)
 
+    scheduler.advance()
+
     // A value emits right away.
     XCTAssertEqual(values, [1])