Fix a race condition when a system is started on a different queue from its event serialising queue.

ReactiveFeedback/SignalProducer+System.swift

@@ -29,8 +29,23 @@ extension SignalProducer where Error == NoError {
 
             return SignalProducer<Event, NoError>(Signal.merge(events))
                 .scan(initial, reduce)
-                .prefix(value: initial)
                 .on(value: stateObserver.send(value:))
+                .prefix(value: initial)
+                .on(started: {
+                    // NOTE: Due to the nature of `prefix` lazily starting the producer being prefixed, we cannot rely
+                    //       on `on(value:)` to ignite the feedbacks with the initial state.
+                    //
+                    //       At the time `prefix(value:)` calls `on(value:)` for the initial value, the events-reducer
+                    //       producer has not yet been started yet. Consequentially, it would lead to dropped events
+                    //       when the system is instantiated on a queue different from the queue used for
+                    //       serializing events.
+                    //
+                    //       Having said that, `prefix(value:)` is guaranteed to have started the prefixed producer as
+                    //       part of the synchronous producer starting process. So we can address the issue by applying
+                    //       `on(started:)` after `prefix(value:)` to ignite the system, while having `on(value:)`
+                    //       instead applied before `prefix(value:)` to keep the reducer-to-feedbacks path open.
+                    stateObserver.send(value: initial)
+                })
         }
     }
 

ReactiveFeedbackTests/SystemTests.swift

@@ -181,4 +181,39 @@ class SystemTests: XCTestCase {
         expect(value) == "initial_a"
         expect(startCount) == 2
     }
+
+    func test_should_not_miss_delivery_to_reducer_when_started_asynchronously() {
+        let creationScheduler = QueueScheduler()
+        let systemScheduler = QueueScheduler()
+
+        let observedState: Atomic<[String]> = Atomic([])
+
+        let semaphore = DispatchSemaphore(value: 0)
+
+        creationScheduler.schedule {
+             SignalProducer<String, NoError>
+                .system(
+                    initial: "initial",
+                    scheduler: systemScheduler,
+                    reduce: { (state: String, event: String) -> String in
+                        return state + event
+                    },
+                    feedbacks: [
+                        Feedback { scheduler, state in
+                            return state
+                                .take(first: 1)
+                                .map(value: "_event")
+                                .observe(on: scheduler)
+                                .on(terminated: { semaphore.signal() })
+                        }
+                    ]
+                )
+                .startWithValues { state in
+                    observedState.modify { $0.append(state) }
+                }
+        }
+
+        semaphore.wait()
+        expect(observedState.value).toEventually(equal(["initial", "initial_event"]))
+    }
 }