[fix]: add host-level logic for reentrant event handling (#381)

#### Terminology

1. `Sink`: public action handler type vended out to client code. these
are the primary channel through which events from the 'outside world'
are sent into a Workflow runtime. is a value type that wraps a (weak)
reference to internal event handling infrastructure.
1. `ReusableSink`: internal action handler type that receives actions
forwarded through from `Sink`s. it is a reference type that is weakly
referenced by `Sink`s.
1. `EventPipe`: internal action handler type that implements an event
handling state machine. it is a reference type that is owned by either a
`ReusableSink` to propagate 'outside' events, or by a `SubtreeManager`
to propagate outputs from child workflows.
1. `SubtreeManager`: type that coordinates interactions between a
workflow node, its children, and the 'outside world'.

#### Background

the current event handling system in Workflow tracks event processing
state locally – each node has a 'subtree manager' which is responsible
for orchestrating interactions with both child workflow outputs and
events from the 'outside world' sent to the corresponding node. each
`SubtreeManager` coordinates a collection of 'event pipes' which handle
event propagation; the 'outside world' has (indirect) access to them via
'sinks', and child workflow's communicate 'up' to their parents through
`EventPipe` instances. these event pipes also encode the state machine
transitions for event handling – if they're sent events in an invalid
state they will generally halt the program.

the `EventPipe` state machine consists of the following states:

1. `preparing`: the pipe has been created, but this state indicates a
call to the corresponding node's `render()` method is still ongoing.
1. `pending`: the corresponding node's `render()` method is complete,
but the full 'render pass' over the entire tree may not yet be.
1. `enabled`: the tree's render pass is finished, and the event pipe is
valid to use. this state has a corresponding event handler closure that
can be invoked when handling an event.
1. `invalid`: the event pipe is no longer valid to use. a node's event
pipes are invalidated as one of the first parts of a render pass.

the currently expected state transitions are:

1. <none> -> `preparing`
    - `preparing` is the initial state
1. `preparing` -> `pending`
    - after a node finishes `render()`
1. `pending` -> `enabled`
    - after the tree finishes a render pass
1. `enabled` -> `invalid`
    - when a node is about to be re-rendered[^1]

[^1]: as an aside, it's reasonable to wonder if we may be allowing nodes
to 'linger on' without invalidating their event handling infrastructure
appropriately. what happens if a node is rendered in one pass and not
rendered in the next? i think in most cases it just deallocates and so
implicitly ends up ignoring any subsequent events, but... would probably
be good to verify that and formalize what should be happening...

the way current event pipe state machine works is:

1. during a render pass, every node gets new event pipes created,
initially in the `preparing` state. any existing event pipes are set to
the `invalid` state.
1. after a node is rendered, the event pipes that were created for that
pass are moved to the `pending` state.
1. after a render pass is 'complete' and any Output and a new Rendering
have been emitted, the nodes in the tree are walked and all event pipes
are moved to the `enabled` state and hooked up with the right callbacks
to invoke when they're sent events.

some additional notes about the current implementation:

1. a number of invalid state transitions are banned outright and will
cause a runtime trap if attempted.
1. the `ReusableSink` type contains logic to attempt to detect certain
forms of reentrancy. specifically it will check if the event pipe state
is `pending` and if it is, will enqueue the forwarding of the event into
the future.
1. there is some limited reentrancy detection implemented when event
pipes in the `invalid` state are reentrantly messaged.

#### Issue

the existing implementation seems to have generally worked reasonably
well in practice, but there are cases when it falls down. in particular,
two problematic cases that have been seen 'in the wild' are:

1. reentrant action emissions from synchronous side effects. perhaps the
'canonical' example of this is when an action is being processed that,
during processing, leads to a change in `UIResponder` state (.e.g
`resignFirstResponder()` is called), and some _other_ part of the
Workflow tree responds to that change by emitting another action.

1. APIs that do manual `RunLoop` spinning, leading to reentrant action
handling. one instance of this we've seen is when the UI layer responds
to a new rendering by deriving an attributed string from HTML via some
the `UIKit` extension methods it can end up spinning the main thread's
run loop waiting for `WebKit` to do HTML processing, and if there are
other Workflow events that have been enqueued they will cause the
runtime to attempt to reentrantly process an event (generally leading to
a `fatalError()`).

as the existing implementation models the event handling state machine
at the node level, it seems ill-equipped to deal with this problem
holistically since the 'is the tree currently processing an event' bit
of information is inherently a tree-level fact. we could try to augment
the existing code with a state that represents 'the runtime is currently
handling an event', but that seems somewhat awkward to fit into the
existing model since a node would have to walk to the root and then
update the whole tree with this new state[^2].

[^2]: except maybe _not_ the path from node to root? see, it seems kind
of awkward...

#### Proposed Solution

in this PR, the approach taken to solve this is:

- introduce a new `SinkEventHandler` type to track tree-level event
processing state
- plumb a new callback closure through the tree down to the
`ReusableSink` instances (which are responsible to forwarding 'external'
events into the `EventPipe` machinery)
- the new callback takes two parameters: a closure to be immediately
invoked if there is no event currently being handled, and a closure to
be enqueued and called in the future if there is.
- the callback implementation checks the current state and either
invokes the 'run now' closure (adjusting processing state as
appropriate) or enqueues the 'run later' closure.

#### Alternatives

i also drafted a more minimal change to address the new test case that i
added which simulates the 'spinning the run loop during a render pass'
problem in #379. instead of
changing the plumbing and adding tree-level state tracking, it just
changes how the enqueing logic in the `ReusableSink` implementation
works. previously the enqueuing logic would defer handling and
unconditionally forward the event through after the async block was
processed. after the change, the method becomes fully recursive, so will
check whether the original action should be enqueued again. while this
approach requires changing less code, it is also less of a 'real' fix,
as it won't solve cases in which someone, say, emits a second sink event
targeting an ancestor node in the tree while a first sink event is being
handled.

---

#### Updates

after initial feedback, made the following changes:

- made the new event handling behavior conditional and added a
`RuntimeConfig` value to opt into it
- added a queue precondition check into the new `ReusableSink` event
handling logic
- removed the `initializing` state from `SinkEventHandler` in favor of
just 'busy' and 'ready'
- added a mechanism to explicitly enter the `busy` state while invoking
a block so that the `WorkflowHost` can update the root node and ensure
no event handling can synchronously occur during that process
- added several new test cases (and minor refactoring of existing test
utilities)

Author: jamieQ

Workflow/Sources/RuntimeConfiguration.swift

@@ -88,6 +88,10 @@ extension Runtime {
 
         /// Note: this doesn't control anything yet, but is here as a placeholder
         public var renderOnlyIfStateChanged: Bool = false
+
+        /// Whether action handling should be delegated to the `SinkEventHandler` type.
+        /// This is expected to eventually be removed and become the default behavior.
+        public var useSinkEventHandler: Bool = false
     }
 
     struct BootstrappableConfiguration {

Workflow/Sources/SubtreeManager.swift

@@ -295,7 +295,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)
@@ -334,7 +337,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: OnSinkEvent?
+        ) -> ReusableSink<Action> {
             let key = ObjectIdentifier(actionType)
 
             let reusableSink: ReusableSink<Action>
@@ -348,7 +354,7 @@ extension WorkflowNode.SubtreeManager {
                 reusableSink = usedSink
             } else {
                 // Create a new reusable sink.
-                reusableSink = ReusableSink<Action>()
+                reusableSink = ReusableSink<Action>(onSinkEvent: onSinkEvent)
             }
 
             usedSinks[key] = reusableSink
@@ -359,15 +365,24 @@ 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: OnSinkEvent?) {
+            self.onSinkEvent = onSinkEvent
             self.eventPipe = EventPipe()
         }
     }
 
     fileprivate final class ReusableSink<Action: WorkflowAction>: AnyReusableSink where Action.WorkflowType == WorkflowType {
         func handle(action: Action) {
+            if let onSinkEvent {
+                handleWithSinkEventHandler(action: action, onSinkEvent: onSinkEvent)
+                return
+            }
+
+            // Prior logic
             let output = Output.update(
                 action,
                 source: .external,
@@ -384,6 +399,32 @@ extension WorkflowNode.SubtreeManager {
             }
             eventPipe.handle(event: output)
         }
+
+        private func handleWithSinkEventHandler(
+            action: Action,
+            onSinkEvent: OnSinkEvent
+        ) {
+            // new `SinkEventHandler` logic
+            dispatchPrecondition(condition: .onQueue(DispatchQueue.workflowExecution))
+
+            // If we can process now, forward through the `EventPipe`
+            let immediatePerform: () -> Void = {
+                let output = Output.update(
+                    action,
+                    source: .external,
+                    subtreeInvalidated: false // initial state
+                )
+
+                self.eventPipe.handle(event: output)
+            }
+
+            // Otherwise, try to recurse again in the future
+            let deferredPerform: () -> Void = { [weak self] in
+                self?.handle(action: action)
+            }
+
+            onSinkEvent(immediatePerform, deferredPerform)
+        }
     }
 }
 

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 sinkEventHandler: SinkEventHandler
+
     /// Initializes a new host with the given workflow at the root.
     ///
     /// - Parameter workflow: The root workflow in the hierarchy
@@ -61,15 +64,22 @@ public final class WorkflowHost<WorkflowType: Workflow> {
         observers: [WorkflowObserver] = [],
         debugger: WorkflowDebugger? = nil
     ) {
+        self.sinkEventHandler = SinkEventHandler()
+        defer { sinkEventHandler.state = .ready }
+
         let observer = WorkflowObservation
             .sharedObserversInterceptor
             .workflowObservers(for: observers)
             .chained()
 
+        let config = Runtime.configuration
+        let sinkEventCallback = config.useSinkEventHandler ? sinkEventHandler.makeOnSinkEventCallback() : nil
+
         self.context = HostContext(
             observer: observer,
             debugger: debugger,
-            runtimeConfig: Runtime.configuration
+            runtimeConfig: config,
+            onSinkEvent: sinkEventCallback
         )
 
         self.rootNode = WorkflowNode(
@@ -91,6 +101,16 @@ public final class WorkflowHost<WorkflowType: Workflow> {
 
     /// Update the input for the workflow. Will cause a render pass.
     public func update(workflow: WorkflowType) {
+        if context.runtimeConfig.useSinkEventHandler {
+            sinkEventHandler.withEventHandlingSuspended {
+                updateRootNode(workflow: workflow)
+            }
+        } else {
+            updateRootNode(workflow: workflow)
+        }
+    }
+
+    private func updateRootNode(workflow: WorkflowType) {
         rootNode.update(workflow: workflow)
 
         // Treat the update as an "output" from the workflow originating from an external event to force a render pass.
@@ -158,14 +178,19 @@ struct HostContext {
     let debugger: WorkflowDebugger?
     let runtimeConfig: Runtime.Configuration
 
+    /// Event handler to be plumbed through the runtime down to the (reusable) Sinks.
+    let onSinkEvent: OnSinkEvent?
+
     init(
         observer: WorkflowObserver?,
         debugger: WorkflowDebugger?,
-        runtimeConfig: Runtime.Configuration
+        runtimeConfig: Runtime.Configuration,
+        onSinkEvent: OnSinkEvent?
     ) {
         self.observer = observer
         self.debugger = debugger
         self.runtimeConfig = runtimeConfig
+        self.onSinkEvent = onSinkEvent
     }
 }
 
@@ -176,3 +201,88 @@ extension HostContext {
         debugger != nil ? perform() : nil
     }
 }
+
+// MARK: - SinkEventHandler
+
+/// Callback signature for the internal `ReusableSink` types to invoke when
+/// they receive an event from the 'outside world'.
+/// - Parameter immediatePerform: The event handler to invoke if the event can be processed immediately.
+/// - Parameter deferredPerform: The event handler to invoke in the future if the event cannot currently be processed.
+typealias OnSinkEvent = (
+    _ immediatePerform: () -> Void,
+    _ deferredPerform: @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 {
+        /// Ready to handle an event.
+        case ready
+
+        /// The event handler is busy. Usually this indicates another event is being
+        /// processed, but it may also be set when some other condition prevents
+        /// event handling (e.g. a `WorkflowHost` was told to update its root node).
+        case busy
+    }
+
+    fileprivate(set) var state: State
+
+    init(state: State = .busy) {
+        self.state = state
+    }
+
+    /// Synchronously performs or enqueues the specified event handlers based on the current
+    /// event handler state.
+    /// - Parameters:
+    ///   - immediate: The event handling action to perform immediately if possible.
+    ///   - deferred: The event handling action to enqueue if the event handler is already processing an event.
+    func performOrEnqueueEvent(
+        immediate: () -> Void,
+        deferred: @escaping () -> Void
+    ) {
+        switch state {
+        case .ready:
+            withEventHandlingSuspended(immediate)
+
+        case .busy:
+            DispatchQueue.workflowExecution.async(execute: deferred)
+        }
+    }
+
+    /// Invokes the given closure with event handling explicitly set to the `busy` state, so
+    /// any incoming events produced while executing the closure's body will be enqueued.
+    /// - Parameter body: The closure to invoke.
+    func withEventHandlingSuspended(_ body: () -> Void) {
+        switch state {
+        case .ready:
+            state = .busy
+            defer { state = .ready }
+            body()
+
+        case .busy:
+            body()
+        }
+    }
+
+    /// Creates the callback that should be invoked by Sinks to handle their event appropriately
+    /// given the `SinkEventHandler`'s current state.
+    /// - Returns: The callback that should be invoked.
+    func makeOnSinkEventCallback() -> OnSinkEvent {
+        // We may not actually need the weak ref, but it's more defensive to keep it.
+        let onSinkEvent: OnSinkEvent = { [weak self] immediate, deferred in
+            guard let self else {
+                // We just drop the events here. Should we signal this somehow?
+                // Maybe as a debug-only thing? Or is it just noise?
+                return
+            }
+
+            performOrEnqueueEvent(
+                immediate: immediate,
+                deferred: deferred
+            )
+        }
+
+        return onSinkEvent
+    }
+}