aion/engine/startup_deferred.rs
1//! Deferred startup recovery (#266): the seam that lets an embedding host
2//! install the engine-backed collaborators its activity dispatcher consults
3//! BEFORE recovery replay re-dispatches in-flight work.
4//!
5//! By default [`crate::EngineBuilder::build`] performs the four startup
6//! recovery steps itself, exactly as it always has. A host that decorates the
7//! activity dispatcher with seams it can only fill once the engine exists
8//! (the server's declared-body source is the motivating case) calls
9//! [`crate::EngineBuilder::defer_startup_recovery`]; `build()` then skips the
10//! four steps and stows what they need in a one-shot slot, and the host runs
11//! [`Engine::run_startup_recovery`] after its seams are installed. The four
12//! steps run in the same order either way.
13
14use std::sync::{Arc, Mutex};
15
16use chrono::Utc;
17
18use crate::EngineError;
19use crate::durability::ActiveWorkflowRecoverySeam;
20
21use super::api::Engine;
22use super::startup::{
23 StartupRecoveryContext, recover_active_workflows_on_startup, recover_timers_on_startup,
24};
25
26/// What `build()` stows when startup recovery is deferred: the two recovery
27/// inputs the constructed [`Engine`] does not itself hold.
28pub(crate) struct DeferredStartupRecovery {
29 /// The active-workflow recovery seam override configured on the builder.
30 pub(crate) recovery: Option<Arc<dyn ActiveWorkflowRecoverySeam>>,
31 /// Whether this node seeds the schedule coordinator's history.
32 pub(crate) bootstrap_schedule_coordinator: bool,
33}
34
35/// The one-shot slot the staged startup-recovery entry points consume.
36pub(super) enum DeferredRecoverySlot {
37 /// `build()` ran recovery itself (the default); nothing is owed.
38 NotDeferred,
39 /// Recovery was deferred and has not run yet.
40 Pending(DeferredStartupRecovery),
41 /// [`Engine::recover_workflows_on_startup`] ran (or is running); the
42 /// catch-up legs — timer recovery, schedule-coordinator catch-up,
43 /// schedule recovery — are still owed to
44 /// [`Engine::run_startup_catchup`].
45 WorkflowsRecovered,
46 /// Every recovery leg already ran (or is running).
47 Completed,
48}
49
50impl DeferredRecoverySlot {
51 /// The slot state `build()` hands [`Engine::new`].
52 pub(super) fn from_build(deferred: Option<DeferredStartupRecovery>) -> Mutex<Self> {
53 Mutex::new(deferred.map_or(Self::NotDeferred, Self::Pending))
54 }
55}
56
57impl Engine {
58 /// Run the startup recovery steps a deferred build skipped, in the same
59 /// order `build()` runs them: active-workflow recovery replay, timer
60 /// recovery, schedule-coordinator catch-up, schedule recovery.
61 ///
62 /// Call this exactly once, after every seam the activity dispatcher
63 /// consults is installed — recovery replay re-dispatches in-flight
64 /// activities through the dispatcher immediately.
65 ///
66 /// # Errors
67 ///
68 /// Returns [`EngineError::StartupRecoveryNotDeferred`] when the engine was
69 /// built without [`crate::EngineBuilder::defer_startup_recovery`] (its
70 /// build already ran recovery), [`EngineError::StartupRecoveryAlreadyRan`]
71 /// on a second call, [`EngineError::StartupRecoverySlotPoisoned`] when the
72 /// slot lock was poisoned, and any recovery-step error the deferred steps
73 /// themselves surface.
74 pub async fn run_startup_recovery(&self) -> Result<(), EngineError> {
75 self.recover_workflows_on_startup().await?;
76 self.run_startup_catchup().await
77 }
78
79 /// Run ONLY the active-workflow recovery leg of a deferred build: every
80 /// durably-active workflow is replayed to residency and registered, so
81 /// signals, queries, and cancels answer correctly the moment transports
82 /// serve. The catch-up legs — owed timer fires, schedule-coordinator
83 /// catch-up, schedule recovery — remain owed to
84 /// [`Engine::run_startup_catchup`], which a host may run behind already-
85 /// open doors: an owed-fire backlog has no upper bound, and every fire it
86 /// delivers goes through the same idempotent record-once path the live
87 /// timer wheel uses, so serving during catch-up is the steady-state
88 /// contract, not a special mode.
89 ///
90 /// # Errors
91 ///
92 /// Returns [`EngineError::StartupRecoveryNotDeferred`] when the engine was
93 /// built without [`crate::EngineBuilder::defer_startup_recovery`],
94 /// [`EngineError::StartupRecoveryAlreadyRan`] on a second call,
95 /// [`EngineError::StartupRecoverySlotPoisoned`] when the slot lock was
96 /// poisoned, and any error the workflow-recovery leg itself surfaces.
97 pub async fn recover_workflows_on_startup(&self) -> Result<(), EngineError> {
98 let deferred = {
99 let mut slot = self
100 .deferred_startup_recovery
101 .lock()
102 .map_err(|_| EngineError::StartupRecoverySlotPoisoned)?;
103 match std::mem::replace(&mut *slot, DeferredRecoverySlot::WorkflowsRecovered) {
104 DeferredRecoverySlot::Pending(deferred) => deferred,
105 DeferredRecoverySlot::NotDeferred => {
106 *slot = DeferredRecoverySlot::NotDeferred;
107 return Err(EngineError::StartupRecoveryNotDeferred);
108 }
109 DeferredRecoverySlot::WorkflowsRecovered => {
110 *slot = DeferredRecoverySlot::WorkflowsRecovered;
111 return Err(EngineError::StartupRecoveryAlreadyRan);
112 }
113 DeferredRecoverySlot::Completed => {
114 *slot = DeferredRecoverySlot::Completed;
115 return Err(EngineError::StartupRecoveryAlreadyRan);
116 }
117 }
118 };
119 recover_active_workflows_on_startup(StartupRecoveryContext {
120 store: Arc::clone(&self.store),
121 visibility_store: Arc::clone(&self.visibility_store),
122 runtime: Arc::clone(&self.runtime),
123 catalog: Arc::clone(&self.catalog),
124 registry: Arc::clone(&self.registry),
125 supervision: Arc::clone(&self.supervision),
126 recovery: deferred.recovery,
127 search_attribute_schema: Arc::clone(&self.search_attribute_schema),
128 bootstrap_schedule_coordinator: deferred.bootstrap_schedule_coordinator,
129 })
130 .await
131 }
132
133 /// Run the catch-up legs a [`Engine::recover_workflows_on_startup`] call
134 /// left owed, in the order `build()` runs them: timer recovery (owed
135 /// fires and future re-arms), schedule-coordinator catch-up, schedule
136 /// recovery. Safe to run while the host is serving — every leg is the
137 /// same idempotent machinery the live paths run against a serving engine.
138 ///
139 /// # Errors
140 ///
141 /// Returns [`EngineError::StartupCatchupBeforeWorkflowRecovery`] when the
142 /// workflow-recovery leg has not run,
143 /// [`EngineError::StartupRecoveryNotDeferred`] when the build was not
144 /// deferred, [`EngineError::StartupRecoveryAlreadyRan`] on a second call,
145 /// [`EngineError::StartupRecoverySlotPoisoned`] when the slot lock was
146 /// poisoned, and any error the catch-up legs themselves surface.
147 pub async fn run_startup_catchup(&self) -> Result<(), EngineError> {
148 {
149 let mut slot = self
150 .deferred_startup_recovery
151 .lock()
152 .map_err(|_| EngineError::StartupRecoverySlotPoisoned)?;
153 match std::mem::replace(&mut *slot, DeferredRecoverySlot::Completed) {
154 DeferredRecoverySlot::WorkflowsRecovered => {}
155 DeferredRecoverySlot::Pending(deferred) => {
156 *slot = DeferredRecoverySlot::Pending(deferred);
157 return Err(EngineError::StartupCatchupBeforeWorkflowRecovery);
158 }
159 DeferredRecoverySlot::NotDeferred => {
160 *slot = DeferredRecoverySlot::NotDeferred;
161 return Err(EngineError::StartupRecoveryNotDeferred);
162 }
163 DeferredRecoverySlot::Completed => {
164 return Err(EngineError::StartupRecoveryAlreadyRan);
165 }
166 }
167 }
168 recover_timers_on_startup(self.runtime.nif_state(), Arc::clone(&self.store)).await?;
169 self.catchup_schedule_coordinator().await?;
170 self.recover_schedules_on_startup(Utc::now()).await?;
171 Ok(())
172 }
173}
174
175#[cfg(test)]
176#[path = "startup_deferred_tests.rs"]
177mod tests;