aion/engine/
api_schedule.rs

1//! `Engine` durable-schedule surface: create/update/pause/resume/delete,
2//! listing and description, timer-fired handling, startup recovery, and the
3//! schedule coordinator's evaluator/recorder assembly.
4
5use std::collections::HashMap;
6use std::sync::Arc;
7
8use aion_core::{
9    Event, EventEnvelope, Payload, RunId, ScheduleConfig, ScheduleId, SearchAttributeSchema,
10    SearchAttributeValue, WorkflowId,
11};
12use async_trait::async_trait;
13use chrono::{DateTime, Utc};
14use tokio::sync::Mutex as AsyncMutex;
15
16use aion_store::EventStore;
17use aion_store::visibility::VisibilityStore;
18
19use crate::durability::Recorder;
20use crate::lifecycle::start::{self, StartWorkflowContext};
21use crate::schedule::{
22    NoopScheduleCanceller, ScheduleEvaluator, ScheduleEvaluatorError, ScheduleEventSink,
23    ScheduleEventSource, ScheduleExecution, ScheduleState, ScheduleTimer, ScheduleWorkflowStarter,
24    StoreScheduleTimer, TimerEvaluationOutcome,
25};
26use crate::{EngineError, Registry, RuntimeHandle, SupervisionTree, WorkflowCatalog};
27
28use super::api::Engine;
29
30impl Engine {
31    /// Workflow history used for durable schedule events and timer ownership.
32    #[must_use]
33    pub const fn schedule_coordinator_workflow_id(&self) -> &WorkflowId {
34        &self.schedule_coordinator_workflow_id
35    }
36
37    /// Create a durable schedule and arm its first timer.
38    ///
39    /// # Errors
40    ///
41    /// Returns shutdown, durability, schedule projection, or timer arming errors.
42    pub async fn create_schedule(&self, config: ScheduleConfig) -> Result<ScheduleId, EngineError> {
43        let operation = self.shutdown_gate.begin_start()?;
44        let recorded_at = Utc::now();
45        let schedule_id = ScheduleId::new_v4();
46        let result = self
47            .create_schedule_inner(schedule_id, config, recorded_at)
48            .await;
49        drop(operation);
50        result
51    }
52
53    /// Update an existing schedule's configuration and re-arm it.
54    ///
55    /// # Errors
56    ///
57    /// Returns [`EngineError::ShuttingDown`] after shutdown begins,
58    /// [`EngineError::ScheduleNotFound`] for absent/deleted schedules, or typed durability,
59    /// projection, and timer errors.
60    pub async fn update_schedule(
61        &self,
62        schedule_id: &ScheduleId,
63        config: ScheduleConfig,
64    ) -> Result<(), EngineError> {
65        let operation = self.shutdown_gate.begin_operation()?;
66        let result = self.update_schedule_inner(schedule_id, config).await;
67        drop(operation);
68        result
69    }
70
71    /// Pause an existing schedule.
72    ///
73    /// # Errors
74    ///
75    /// Returns [`EngineError::ShuttingDown`] after shutdown begins,
76    /// [`EngineError::ScheduleNotFound`] for absent/deleted schedules, or typed durability
77    /// and projection errors.
78    pub async fn pause_schedule(&self, schedule_id: &ScheduleId) -> Result<(), EngineError> {
79        let operation = self.shutdown_gate.begin_operation()?;
80        let result = self.pause_schedule_inner(schedule_id).await;
81        drop(operation);
82        result
83    }
84
85    /// Resume an existing schedule and re-arm it.
86    ///
87    /// # Errors
88    ///
89    /// Returns [`EngineError::ShuttingDown`] after shutdown begins,
90    /// [`EngineError::ScheduleNotFound`] for absent/deleted schedules, or typed durability,
91    /// projection, and timer errors.
92    pub async fn resume_schedule(&self, schedule_id: &ScheduleId) -> Result<(), EngineError> {
93        let operation = self.shutdown_gate.begin_operation()?;
94        let result = self.resume_schedule_inner(schedule_id).await;
95        drop(operation);
96        result
97    }
98
99    /// Delete an existing schedule so it is no longer listed or armed.
100    ///
101    /// # Errors
102    ///
103    /// Returns [`EngineError::ShuttingDown`] after shutdown begins,
104    /// [`EngineError::ScheduleNotFound`] for absent/deleted schedules, or typed durability
105    /// and projection errors.
106    pub async fn delete_schedule(&self, schedule_id: &ScheduleId) -> Result<(), EngineError> {
107        let operation = self.shutdown_gate.begin_operation()?;
108        let result = self.delete_schedule_inner(schedule_id).await;
109        drop(operation);
110        result
111    }
112
113    /// List all non-deleted schedules from projected state.
114    ///
115    /// # Errors
116    ///
117    /// Currently returns only infallible projected state, wrapped for API consistency.
118    pub async fn list_schedules(&self) -> Result<Vec<ScheduleState>, EngineError> {
119        let evaluator = self.schedule_evaluator.lock().await;
120        let mut schedules = evaluator
121            .states()
122            .filter(|state| !state.is_deleted)
123            .cloned()
124            .collect::<Vec<_>>();
125        schedules.sort_by(|left, right| {
126            left.next_trigger_at
127                .cmp(&right.next_trigger_at)
128                .then_with(|| {
129                    left.schedule_id
130                        .to_string()
131                        .cmp(&right.schedule_id.to_string())
132                })
133        });
134        Ok(schedules)
135    }
136
137    /// Describe one non-deleted schedule.
138    ///
139    /// # Errors
140    ///
141    /// Returns [`EngineError::ScheduleNotFound`] for absent or deleted schedules.
142    pub async fn describe_schedule(
143        &self,
144        schedule_id: &ScheduleId,
145    ) -> Result<ScheduleState, EngineError> {
146        self.schedule_evaluator
147            .lock()
148            .await
149            .state(schedule_id)
150            .filter(|state| !state.is_deleted)
151            .cloned()
152            .ok_or_else(|| EngineError::ScheduleNotFound {
153                schedule_id: schedule_id.clone(),
154            })
155    }
156
157    async fn create_schedule_inner(
158        &self,
159        schedule_id: ScheduleId,
160        config: ScheduleConfig,
161        recorded_at: DateTime<Utc>,
162    ) -> Result<ScheduleId, EngineError> {
163        self.schedule_recorder
164            .lock()
165            .await
166            .record_schedule_created(recorded_at, schedule_id.clone(), config.clone())
167            .await?;
168
169        let state = ScheduleState::created(schedule_id.clone(), config, recorded_at)?;
170        let mut evaluator = self.schedule_evaluator.lock().await;
171        evaluator.upsert_state(state);
172        evaluator.arm_active_schedule(&schedule_id).await?;
173        Ok(schedule_id)
174    }
175
176    async fn update_schedule_inner(
177        &self,
178        schedule_id: &ScheduleId,
179        config: ScheduleConfig,
180    ) -> Result<(), EngineError> {
181        self.ensure_schedule_exists(schedule_id).await?;
182        let recorded_at = Utc::now();
183        self.schedule_recorder
184            .lock()
185            .await
186            .record_schedule_updated(recorded_at, schedule_id.clone(), config.clone())
187            .await?;
188        let event = Event::ScheduleUpdated {
189            envelope: schedule_event_envelope(recorded_at),
190            schedule_id: schedule_id.clone(),
191            config,
192        };
193        self.apply_schedule_event(schedule_id, &event, true).await
194    }
195
196    async fn pause_schedule_inner(&self, schedule_id: &ScheduleId) -> Result<(), EngineError> {
197        self.ensure_schedule_exists(schedule_id).await?;
198        let recorded_at = Utc::now();
199        self.schedule_recorder
200            .lock()
201            .await
202            .record_schedule_paused(recorded_at, schedule_id.clone())
203            .await?;
204        let event = Event::SchedulePaused {
205            envelope: schedule_event_envelope(recorded_at),
206            schedule_id: schedule_id.clone(),
207        };
208        self.apply_schedule_event(schedule_id, &event, false).await
209    }
210
211    async fn resume_schedule_inner(&self, schedule_id: &ScheduleId) -> Result<(), EngineError> {
212        self.ensure_schedule_exists(schedule_id).await?;
213        let recorded_at = Utc::now();
214        self.schedule_recorder
215            .lock()
216            .await
217            .record_schedule_resumed(recorded_at, schedule_id.clone())
218            .await?;
219        let event = Event::ScheduleResumed {
220            envelope: schedule_event_envelope(recorded_at),
221            schedule_id: schedule_id.clone(),
222        };
223        self.apply_schedule_event(schedule_id, &event, true).await
224    }
225
226    async fn delete_schedule_inner(&self, schedule_id: &ScheduleId) -> Result<(), EngineError> {
227        self.ensure_schedule_exists(schedule_id).await?;
228        let recorded_at = Utc::now();
229        self.schedule_recorder
230            .lock()
231            .await
232            .record_schedule_deleted(recorded_at, schedule_id.clone())
233            .await?;
234        let event = Event::ScheduleDeleted {
235            envelope: schedule_event_envelope(recorded_at),
236            schedule_id: schedule_id.clone(),
237        };
238        self.apply_schedule_event(schedule_id, &event, false).await
239    }
240
241    async fn ensure_schedule_exists(&self, schedule_id: &ScheduleId) -> Result<(), EngineError> {
242        self.schedule_evaluator
243            .lock()
244            .await
245            .state(schedule_id)
246            .filter(|state| !state.is_deleted)
247            .map(|_| ())
248            .ok_or_else(|| EngineError::ScheduleNotFound {
249                schedule_id: schedule_id.clone(),
250            })
251    }
252
253    async fn apply_schedule_event(
254        &self,
255        schedule_id: &ScheduleId,
256        event: &Event,
257        should_arm: bool,
258    ) -> Result<(), EngineError> {
259        let mut evaluator = self.schedule_evaluator.lock().await;
260        let mut state = evaluator
261            .state(schedule_id)
262            .filter(|state| !state.is_deleted)
263            .cloned()
264            .ok_or_else(|| EngineError::ScheduleNotFound {
265                schedule_id: schedule_id.clone(),
266            })?;
267        state.apply(event)?;
268        evaluator.upsert_state(state);
269        if should_arm {
270            evaluator.arm_active_schedule(schedule_id).await?;
271        }
272        Ok(())
273    }
274
275    /// Handles a fired durable schedule timer through the schedule evaluator.
276    ///
277    /// # Errors
278    ///
279    /// Returns schedule evaluator or shutdown errors.
280    pub async fn handle_schedule_timer_fired(
281        &self,
282        schedule_id: &ScheduleId,
283        fire_at: DateTime<Utc>,
284    ) -> Result<TimerEvaluationOutcome, EngineError> {
285        let operation = self.shutdown_gate.begin_operation()?;
286        let result = self
287            .schedule_evaluator
288            .lock()
289            .await
290            .handle_timer_fired(schedule_id, fire_at)
291            .await
292            .map_err(EngineError::from);
293        drop(operation);
294        result
295    }
296
297    /// Rebuilds schedule state from durable coordinator history and re-arms active schedules.
298    ///
299    /// # Errors
300    ///
301    /// Returns schedule projection, catch-up, timer, or workflow-start errors.
302    pub async fn recover_schedules_on_startup(
303        &self,
304        now: DateTime<Utc>,
305    ) -> Result<(), EngineError> {
306        let source = StoreScheduleEventSource {
307            store: self.store(),
308            coordinator_workflow_id: self.schedule_coordinator_workflow_id.clone(),
309        };
310        self.schedule_evaluator
311            .lock()
312            .await
313            .recover_on_startup(&source, now)
314            .await
315            .map_err(EngineError::from)
316    }
317}
318
319/// The fixed `WorkflowId` of the durable schedule coordinator.
320///
321/// Exposed so a multi-shard deployment can compute which shard owns the
322/// coordinator stream and gate [`EngineBuilder::bootstrap_schedule_coordinator`]
323/// on real ownership (SS-2 / AA-4-4): only the node owning this workflow's shard
324/// may seed and serve the coordinator.
325///
326/// [`EngineBuilder::bootstrap_schedule_coordinator`]: crate::EngineBuilder::bootstrap_schedule_coordinator
327#[must_use]
328pub fn schedule_coordinator_workflow_id() -> WorkflowId {
329    WorkflowId::new(uuid::Uuid::from_u128(
330        0x0000_0000_a10a_0000_0000_0000_0000_0004,
331    ))
332}
333
334pub(crate) fn schedule_coordinator_run_id() -> RunId {
335    RunId::new(uuid::Uuid::from_u128(
336        0x0000_0000_a10a_0000_0000_0000_0000_0005,
337    ))
338}
339
340pub(crate) const fn schedule_coordinator_workflow_type() -> &'static str {
341    "aion.schedule_coordinator"
342}
343
344/// Reserved package version recorded for the virtual schedule coordinator.
345///
346/// The coordinator is an engine-internal history with no loaded package or
347/// runtime process; recovery special-cases it before any version resolution,
348/// so this sentinel is never parsed as a content hash.
349pub(crate) fn schedule_coordinator_package_version() -> aion_core::PackageVersion {
350    aion_core::PackageVersion::new("aion:schedule-coordinator")
351}
352
353fn schedule_event_envelope(recorded_at: DateTime<Utc>) -> EventEnvelope {
354    EventEnvelope {
355        seq: 0,
356        recorded_at,
357        workflow_id: schedule_coordinator_workflow_id(),
358    }
359}
360
361pub(super) fn default_schedule_evaluator(
362    coordinator_workflow_id: WorkflowId,
363    recorder: Arc<AsyncMutex<Recorder>>,
364    deps: ScheduleRuntimeDeps,
365) -> ScheduleEvaluator {
366    let timer: Arc<dyn ScheduleTimer> = Arc::new(StoreScheduleTimer::new(
367        Arc::clone(&deps.store),
368        coordinator_workflow_id,
369    ));
370    let starter: Arc<dyn ScheduleWorkflowStarter> = Arc::new(EngineScheduleStarter { deps });
371    let canceller: Arc<dyn crate::schedule::ScheduleWorkflowCanceller> =
372        Arc::new(NoopScheduleCanceller);
373    let events: Arc<dyn ScheduleEventSink> = recorder;
374    ScheduleEvaluator::new(timer, starter, canceller, events)
375}
376
377pub(super) struct ScheduleRuntimeDeps {
378    pub(super) store: Arc<dyn EventStore>,
379    pub(super) visibility_store: Arc<dyn VisibilityStore>,
380    pub(super) runtime: Arc<RuntimeHandle>,
381    pub(super) catalog: Arc<WorkflowCatalog>,
382    pub(super) registry: Arc<Registry>,
383    pub(super) supervision: Arc<SupervisionTree>,
384    pub(super) search_attribute_schema: Arc<SearchAttributeSchema>,
385}
386
387struct EngineScheduleStarter {
388    deps: ScheduleRuntimeDeps,
389}
390
391#[async_trait]
392impl ScheduleWorkflowStarter for EngineScheduleStarter {
393    async fn start_scheduled_workflow(
394        &self,
395        workflow_type: &str,
396        input: Payload,
397        search_attributes: HashMap<String, SearchAttributeValue>,
398    ) -> Result<ScheduleExecution, ScheduleEvaluatorError> {
399        let handle = start::start_workflow_with_options(
400            StartWorkflowContext {
401                store: Arc::clone(&self.deps.store),
402                visibility_store: Arc::clone(&self.deps.visibility_store),
403                catalog: Arc::clone(&self.deps.catalog),
404                runtime: Arc::clone(&self.deps.runtime),
405                supervision: Arc::clone(&self.deps.supervision),
406                registry: Arc::clone(&self.deps.registry),
407                signal_handoff: None,
408                search_attribute_schema: Arc::clone(&self.deps.search_attribute_schema),
409                monitor_tokio_handle: tokio::runtime::Handle::current(),
410            },
411            workflow_type,
412            input,
413            start::StartWorkflowOptions {
414                search_attributes,
415                ..start::StartWorkflowOptions::default()
416            },
417        )
418        .await
419        .map_err(|error| ScheduleEvaluatorError::side_effect(error.to_string()))?;
420        Ok(ScheduleExecution::new(
421            handle.workflow_id().clone(),
422            handle.run_id().clone(),
423        ))
424    }
425}
426
427struct StoreScheduleEventSource {
428    store: Arc<dyn EventStore>,
429    coordinator_workflow_id: WorkflowId,
430}
431
432#[async_trait]
433impl ScheduleEventSource for StoreScheduleEventSource {
434    async fn schedule_events(&self) -> Result<Vec<Event>, ScheduleEvaluatorError> {
435        self.store
436            .read_history(&self.coordinator_workflow_id)
437            .await
438            .map_err(|error| ScheduleEvaluatorError::side_effect(error.to_string()))
439    }
440}
441
442#[async_trait]
443impl ScheduleEventSink for AsyncMutex<Recorder> {
444    async fn record_schedule_triggered(
445        &self,
446        schedule_id: &ScheduleId,
447        execution: &ScheduleExecution,
448        recorded_at: DateTime<Utc>,
449    ) -> Result<(), ScheduleEvaluatorError> {
450        self.lock()
451            .await
452            .record_schedule_triggered(
453                recorded_at,
454                schedule_id.clone(),
455                execution.workflow_id.clone(),
456                execution.run_id.clone(),
457            )
458            .await
459            .map_err(|error| ScheduleEvaluatorError::side_effect(error.to_string()))
460    }
461}
aion/engine/api_schedule.rs

aion/engine/
api_schedule.rs
