[fix]: add host-level logic for reentrant event handling

Workflow/Sources/SubtreeManager.swift

@@ -281,7 +281,10 @@ extension WorkflowNode.SubtreeManager {
         func makeSink<Action: WorkflowAction>(
             of actionType: Action.Type
         ) -> Sink<Action> where WorkflowType == Action.WorkflowType {
-            let reusableSink = sinkStore.findOrCreate(actionType: Action.self)
+            let reusableSink = sinkStore.findOrCreate(
+                actionType: Action.self,
+                onSinkEvent: hostContext.onSinkEvent
+            )
 
             let sink = Sink<Action> { [weak reusableSink] action in
                 WorkflowLogger.logSinkEvent(ref: SignpostRef(), action: action)
@@ -320,7 +323,10 @@ extension WorkflowNode.SubtreeManager {
             self.usedSinks = [:]
         }
 
-        mutating func findOrCreate<Action: WorkflowAction>(actionType: Action.Type) -> ReusableSink<Action> {
+        mutating func findOrCreate<Action: WorkflowAction>(
+            actionType: Action.Type,
+            onSinkEvent: @escaping OnSinkEvent
+        ) -> ReusableSink<Action> {
             let key = ObjectIdentifier(actionType)
 
             let reusableSink: ReusableSink<Action>
@@ -334,7 +340,7 @@ extension WorkflowNode.SubtreeManager {
                 reusableSink = usedSink
             } else {
                 // Create a new reusable sink.
-                reusableSink = ReusableSink<Action>()
+                reusableSink = ReusableSink<Action>(onSinkEvent: onSinkEvent)
             }
 
             usedSinks[key] = reusableSink
@@ -345,30 +351,33 @@ extension WorkflowNode.SubtreeManager {
 
     /// Type-erased base class for reusable sinks.
     fileprivate class AnyReusableSink {
+        /// The callback to invoke when an event is to be handled.
+        let onSinkEvent: OnSinkEvent
         var eventPipe: EventPipe
 
-        init() {
+        init(onSinkEvent: @escaping OnSinkEvent) {
+            self.onSinkEvent = onSinkEvent
             self.eventPipe = EventPipe()
         }
     }
 
     fileprivate final class ReusableSink<Action: WorkflowAction>: AnyReusableSink where Action.WorkflowType == WorkflowType {
         func handle(action: Action) {
-            let output = Output.update(
-                action,
-                source: .external,
-                subtreeInvalidated: false // initial state
-            )
+            let perform: () -> Void = {
+                let output = Output.update(
+                    action,
+                    source: .external,
+                    subtreeInvalidated: false // initial state
+                )
 
-            if case .pending = eventPipe.validationState {
-                // Workflow is currently processing an `event`.
-                // Scheduling it to be processed after.
-                DispatchQueue.workflowExecution.async { [weak self] in
-                    self?.eventPipe.handle(event: output)
-                }
-                return
+                self.eventPipe.handle(event: output)
             }
-            eventPipe.handle(event: output)
+
+            let enqueue: () -> Void = { [weak self] in
+                self?.handle(action: action)
+            }
+
+            onSinkEvent(perform, enqueue)
         }
     }
 }

Workflow/Sources/WorkflowHost.swift

@@ -14,6 +14,7 @@
  * limitations under the License.
  */
 
+import Dispatch
 import ReactiveSwift
 
 /// Defines a type that receives debug information about a running workflow hierarchy.
@@ -50,6 +51,8 @@ public final class WorkflowHost<WorkflowType: Workflow> {
         context.debugger
     }
 
+    let eventHandler: SinkEventHandler
+
     /// Initializes a new host with the given workflow at the root.
     ///
     /// - Parameter workflow: The root workflow in the hierarchy
@@ -61,6 +64,13 @@ public final class WorkflowHost<WorkflowType: Workflow> {
         observers: [WorkflowObserver] = [],
         debugger: WorkflowDebugger? = nil
     ) {
+        self.eventHandler = SinkEventHandler()
+        assert(
+            eventHandler.state == .initializing,
+            "EventHandler must begin in the `.initializing` state"
+        )
+        defer { eventHandler.state = .ready }
+
         let observer = WorkflowObservation
             .sharedObserversInterceptor
             .workflowObservers(for: observers)
@@ -69,7 +79,8 @@ public final class WorkflowHost<WorkflowType: Workflow> {
         self.context = HostContext(
             observer: observer,
             debugger: debugger,
-            runtimeConfig: Runtime.configuration
+            runtimeConfig: Runtime.configuration,
+            onSinkEvent: eventHandler.makeOnSinkEventCallback()
         )
 
         self.rootNode = WorkflowNode(
@@ -158,14 +169,19 @@ struct HostContext {
     let debugger: WorkflowDebugger?
     let runtimeConfig: Runtime.Configuration
 
+    /// Event handler to be plumbed through the runtime down to the Sinks
+    let onSinkEvent: OnSinkEvent
+
     init(
         observer: WorkflowObserver?,
         debugger: WorkflowDebugger?,
-        runtimeConfig: Runtime.Configuration
+        runtimeConfig: Runtime.Configuration,
+        onSinkEvent: @escaping OnSinkEvent
     ) {
         self.observer = observer
         self.debugger = debugger
         self.runtimeConfig = runtimeConfig
+        self.onSinkEvent = onSinkEvent
     }
 }
 
@@ -176,3 +192,72 @@ extension HostContext {
         debugger != nil ? perform() : nil
     }
 }
+
+// MARK: - EventHandler
+
+/// Callback signature for the internal `ReusableSink` types to invoke when
+/// they receive an event from the 'outside world'.
+/// - Parameter perform: The event handler to invoke if the event can be processed immediately.
+/// - Parameter enqueue: The event handler to invoke in the future if the event cannot currently be processed.
+typealias OnSinkEvent = (
+    _ perform: () -> Void,
+    _ enqueue: @escaping () -> Void
+) -> Void
+
+/// Handles events from 'Sinks' such that runtime-level event handling state is appropriately
+/// managed, and attempts to perform reentrant action handling can be detected and dealt with.
+final class SinkEventHandler {
+    enum State {
+        /// The handler (and related components) are being
+        /// initialized, and are not yet ready to process events.
+        /// Attempts to do so in this state will fail with a fatal error.
+        case initializing
+
+        /// An event is currently being processed.
+        case processingEvent
+
+        /// Ready to handle an event.
+        case ready
+    }
+
+    fileprivate(set) var state: State = .initializing
+
+    /// Synchronously performs or enqueues the specified event handlers based on the current
+    /// event handler state.
+    /// - Parameters:
+    ///   - perform: The event handling action to perform immediately if possible.
+    ///   - enqueue: The event handling action to enqueue if the event handler is already processing an event.
+    func performOrEnqueueEvent(
+        perform: () -> Void,
+        enqueue: @escaping () -> Void
+    ) {
+        switch state {
+        case .initializing:
+            fatalError("Tried to handle event before finishing initialization.")
+
+        case .processingEvent:
+            DispatchQueue.workflowExecution.async(execute: enqueue)
+
+        case .ready:
+            state = .processingEvent
+            defer { state = .ready }
+            perform()
+        }
+    }
+
+    /// Creates the callback that should be invoked by Sinks to handle their event appropriately
+    /// given the `EventHandler`'s current state.
+    /// - Returns: The callback that should be invoked.
+    func makeOnSinkEventCallback() -> OnSinkEvent {
+        // TODO: do we need the weak ref?
+        let onSinkEvent: OnSinkEvent = { [weak self] perform, enqueue in
+            guard let self else {
+                return // TODO: what's the appropriate handling?
+            }
+
+            performOrEnqueueEvent(perform: perform, enqueue: enqueue)
+        }
+
+        return onSinkEvent
+    }
+}

Workflow/Tests/TestUtilities.swift

@@ -68,7 +68,8 @@ extension HostContext {
         HostContext(
             observer: observer,
             debugger: debugger,
-            runtimeConfig: runtimeConfig
+            runtimeConfig: runtimeConfig,
+            onSinkEvent: { perform, _ in perform() }
         )
     }
 }

Workflow/Tests/WorkflowHostTests.swift

@@ -14,7 +14,9 @@
  * limitations under the License.
  */
 
+import ReactiveSwift
 import XCTest
+
 @_spi(WorkflowRuntimeConfig) @testable import Workflow
 
 final class WorkflowHostTests: XCTestCase {
@@ -58,32 +60,84 @@ final class WorkflowHostTests: XCTestCase {
 final class WorkflowHost_EventEmissionTests: XCTestCase {
     // Previous versions of Workflow would fatalError under this scenario
     func test_event_sent_to_invalidated_sink_during_action_handling() {
-        let root = Parent()
-        let host = WorkflowHost(workflow: root)
+        let host = WorkflowHost(workflow: Parent())
+        let (lifetime, token) = ReactiveSwift.Lifetime.make()
+        defer { _ = token }
         let initialRendering = host.rendering.value
         var observedRenderCount = 0
 
         XCTAssertEqual(initialRendering.eventCount, 0)
 
-        let disposable = host.rendering.signal.observeValues { rendering in
-            XCTAssertEqual(rendering.eventCount, 1)
+        host
+            .rendering
+            .signal
+            .take(during: lifetime)
+            .observeValues { rendering in
+                XCTAssertEqual(rendering.eventCount, 1)
+
+                // emit another event using an old rendering
+                // while the first is still being processed, but
+                // the workflow that handles the event has been
+                // removed from the tree
+                if observedRenderCount == 0 {
+                    initialRendering.eventHandler()
+                }
 
-            // emit another event using an old rendering
-            // while the first is still being processed, but
-            // the workflow that handles the event has been
-            // removed from the tree
-            if observedRenderCount == 0 {
-                initialRendering.eventHandler()
+                observedRenderCount += 1
             }
 
-            observedRenderCount += 1
-        }
-        defer { disposable?.dispose() }
-
         // send an event and cause a re-render
         initialRendering.eventHandler()
 
         XCTAssertEqual(observedRenderCount, 1)
+
+        drainMainQueueBySpinningRunLoop()
+
+        // Ensure the invalidated sink doesn't process the event
+        let nextRendering = host.rendering.value
+        XCTAssertEqual(nextRendering.eventCount, 1)
+        XCTAssertEqual(observedRenderCount, 1)
+    }
+
+    func test_reentrant_event_during_render() {
+        let host = WorkflowHost(workflow: ReentrancyWorkflow())
+        let (lifetime, token) = ReactiveSwift.Lifetime.make()
+        defer { _ = token }
+        let initialRendering = host.rendering.value
+
+        var emitReentrantEvent = false
+
+        let renderExpectation = expectation(description: "render")
+        renderExpectation.expectedFulfillmentCount = 2
+
+        host
+            .rendering
+            .signal
+            .take(during: lifetime)
+            .observeValues { val in
+                defer { renderExpectation.fulfill() }
+                defer { emitReentrantEvent = true }
+                guard !emitReentrantEvent else { return }
+
+                // In a prior implementation, this would check state local
+                // to the underlying EventPipe and defer event handling
+                // into the future. If the RunLoop was spun after that
+                // point, the action could attempt to be handled and an
+                // we'd hit a trap when sending a sink an action in an
+                // invalid state.
+                //
+                // 'Real world' code could hit this case as there are some
+                // UI bindings that fire when a rendering/output is updated
+                // that call into system API that do sometimes spin the
+                // RunLoop manually (e.g. stuff calling into WebKit).
+                initialRendering.sink.send(.event)
+                drainMainQueueBySpinningRunLoop()
+            }
+
+        // Send an event and cause a re-render
+        initialRendering.sink.send(.event)
+
+        waitForExpectations(timeout: 1)
     }
 }
 
@@ -115,6 +169,35 @@ extension WorkflowHostTests {
 
 // MARK: Utility Types
 
+extension WorkflowHost_EventEmissionTests {
+    struct ReentrancyWorkflow: Workflow {
+        typealias State = Void
+        typealias Output = Never
+
+        struct Rendering {
+            var sink: Sink<Action>!
+        }
+
+        func render(state: Void, context: RenderContext<Self>) -> Rendering {
+            let sink = context.makeSink(of: Action.self)
+            return Rendering(sink: sink)
+        }
+
+        enum Action: WorkflowAction {
+            typealias WorkflowType = ReentrancyWorkflow
+
+            case event
+
+            func apply(
+                toState state: inout WorkflowType.State,
+                context: ApplyContext<WorkflowType>
+            ) -> WorkflowType.Output? {
+                nil
+            }
+        }
+    }
+}
+
 extension WorkflowHost_EventEmissionTests {
     struct Parent: Workflow {
         struct Rendering {
@@ -182,3 +265,13 @@ extension WorkflowHost_EventEmissionTests {
         }
     }
 }
+
+private func drainMainQueueBySpinningRunLoop(timeoutSeconds: UInt = 1) {
+    var done = false
+    DispatchQueue.main.async { done = true }
+
+    let deadline = ContinuousClock.now + .seconds(timeoutSeconds)
+    while !done, ContinuousClock.now < deadline {
+        RunLoop.current.run(until: .now.addingTimeInterval(0.01))
+    }
+}