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 [`Engine::run_startup_recovery`] consumes.
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::run_startup_recovery`] already ran (or is running).
42    Completed,
43}
44
45impl DeferredRecoverySlot {
46    /// The slot state `build()` hands [`Engine::new`].
47    pub(super) fn from_build(deferred: Option<DeferredStartupRecovery>) -> Mutex<Self> {
48        Mutex::new(deferred.map_or(Self::NotDeferred, Self::Pending))
49    }
50}
51
52impl Engine {
53    /// Run the startup recovery steps a deferred build skipped, in the same
54    /// order `build()` runs them: active-workflow recovery replay, timer
55    /// recovery, schedule-coordinator catch-up, schedule recovery.
56    ///
57    /// Call this exactly once, after every seam the activity dispatcher
58    /// consults is installed — recovery replay re-dispatches in-flight
59    /// activities through the dispatcher immediately.
60    ///
61    /// # Errors
62    ///
63    /// Returns [`EngineError::StartupRecoveryNotDeferred`] when the engine was
64    /// built without [`crate::EngineBuilder::defer_startup_recovery`] (its
65    /// build already ran recovery), [`EngineError::StartupRecoveryAlreadyRan`]
66    /// on a second call, [`EngineError::StartupRecoverySlotPoisoned`] when the
67    /// slot lock was poisoned, and any recovery-step error the deferred steps
68    /// themselves surface.
69    pub async fn run_startup_recovery(&self) -> Result<(), EngineError> {
70        let deferred = {
71            let mut slot = self
72                .deferred_startup_recovery
73                .lock()
74                .map_err(|_| EngineError::StartupRecoverySlotPoisoned)?;
75            match std::mem::replace(&mut *slot, DeferredRecoverySlot::Completed) {
76                DeferredRecoverySlot::Pending(deferred) => deferred,
77                DeferredRecoverySlot::NotDeferred => {
78                    *slot = DeferredRecoverySlot::NotDeferred;
79                    return Err(EngineError::StartupRecoveryNotDeferred);
80                }
81                DeferredRecoverySlot::Completed => {
82                    return Err(EngineError::StartupRecoveryAlreadyRan);
83                }
84            }
85        };
86        recover_active_workflows_on_startup(StartupRecoveryContext {
87            store: Arc::clone(&self.store),
88            visibility_store: Arc::clone(&self.visibility_store),
89            runtime: Arc::clone(&self.runtime),
90            catalog: Arc::clone(&self.catalog),
91            registry: Arc::clone(&self.registry),
92            supervision: Arc::clone(&self.supervision),
93            recovery: deferred.recovery,
94            search_attribute_schema: Arc::clone(&self.search_attribute_schema),
95            bootstrap_schedule_coordinator: deferred.bootstrap_schedule_coordinator,
96        })
97        .await?;
98        recover_timers_on_startup(self.runtime.nif_state(), Arc::clone(&self.store)).await?;
99        self.catchup_schedule_coordinator().await?;
100        self.recover_schedules_on_startup(Utc::now()).await?;
101        Ok(())
102    }
103}
aion/engine/startup_deferred.rs

aion/engine/
startup_deferred.rs
