ironflow_engine/fsm/
run_fsm.rs

1//! [`RunFsm`] — Finite state machine for the run lifecycle.
2//!
3//! Events drive transitions; the FSM rejects invalid ones and keeps
4//! a full history of state changes.
5
6use chrono::Utc;
7use ironflow_store::entities::RunStatus;
8use serde::{Deserialize, Serialize};
9use strum::Display;
10
11use super::{Transition, TransitionError};
12
13/// Events that drive [`RunFsm`] transitions.
14///
15/// Each event represents something that happened during execution.
16///
17/// # Examples
18///
19/// ```
20/// use ironflow_engine::fsm::RunEvent;
21///
22/// let event = RunEvent::PickedUp;
23/// assert_eq!(event.to_string(), "picked_up");
24/// ```
25#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize, Display)]
26#[serde(rename_all = "snake_case")]
27#[strum(serialize_all = "snake_case")]
28pub enum RunEvent {
29    /// Worker or inline executor picked up the run.
30    PickedUp,
31    /// All steps completed successfully.
32    AllStepsCompleted,
33    /// A step failed and the error is not retryable, or max retries exhausted.
34    StepFailed,
35    /// A step failed but a retry is possible.
36    StepFailedRetryable,
37    /// Retry attempt started.
38    RetryStarted,
39    /// Maximum retries exhausted after a retryable failure.
40    MaxRetriesExceeded,
41    /// User or API requested cancellation.
42    CancelRequested,
43    /// A step requires human approval before continuing.
44    ApprovalRequested,
45    /// Human approved the run to continue.
46    Approved,
47    /// Human rejected the run.
48    Rejected,
49    /// A delay step suspended the run until a scheduled time.
50    DelaySleeping,
51    /// The delay elapsed and the run is re-queued.
52    DelayElapsed,
53}
54
55/// Finite state machine for a workflow run.
56///
57/// Wraps a [`RunStatus`] and enforces valid transitions via typed
58/// [`RunEvent`]s. Records every transition in a history log.
59///
60/// # Transition table
61///
62/// | From | Event | To |
63/// |------|-------|----|
64/// | Pending | PickedUp | Running |
65/// | Pending | CancelRequested | Cancelled |
66/// | Running | AllStepsCompleted | Completed |
67/// | Running | StepFailed | Failed |
68/// | Running | StepFailedRetryable | Retrying |
69/// | Running | CancelRequested | Cancelled |
70/// | Retrying | RetryStarted | Running |
71/// | Retrying | MaxRetriesExceeded | Failed |
72/// | Retrying | CancelRequested | Cancelled |
73/// | Running | ApprovalRequested | AwaitingApproval |
74/// | AwaitingApproval | Approved | Running |
75/// | AwaitingApproval | Rejected | Failed |
76/// | AwaitingApproval | CancelRequested | Cancelled |
77/// | Running | DelaySleeping | Sleeping |
78/// | Sleeping | DelayElapsed | Pending |
79/// | Sleeping | CancelRequested | Cancelled |
80///
81/// # Examples
82///
83/// ```
84/// use ironflow_engine::fsm::{RunFsm, RunEvent};
85/// use ironflow_store::entities::RunStatus;
86///
87/// let mut fsm = RunFsm::new();
88/// assert_eq!(fsm.state(), RunStatus::Pending);
89///
90/// fsm.apply(RunEvent::PickedUp).unwrap();
91/// assert_eq!(fsm.state(), RunStatus::Running);
92///
93/// fsm.apply(RunEvent::AllStepsCompleted).unwrap();
94/// assert_eq!(fsm.state(), RunStatus::Completed);
95/// assert_eq!(fsm.history().len(), 2);
96/// ```
97#[derive(Debug, Clone)]
98pub struct RunFsm {
99    state: RunStatus,
100    history: Vec<Transition<RunStatus, RunEvent>>,
101}
102
103impl RunFsm {
104    /// Create a new FSM in `Pending` state.
105    ///
106    /// # Examples
107    ///
108    /// ```
109    /// use ironflow_engine::fsm::RunFsm;
110    /// use ironflow_store::entities::RunStatus;
111    ///
112    /// let fsm = RunFsm::new();
113    /// assert_eq!(fsm.state(), RunStatus::Pending);
114    /// ```
115    pub fn new() -> Self {
116        Self {
117            state: RunStatus::Pending,
118            history: Vec::new(),
119        }
120    }
121
122    /// Create a FSM from an existing state (e.g. loaded from DB).
123    ///
124    /// # Examples
125    ///
126    /// ```
127    /// use ironflow_engine::fsm::RunFsm;
128    /// use ironflow_store::entities::RunStatus;
129    ///
130    /// let fsm = RunFsm::from_state(RunStatus::Running);
131    /// assert_eq!(fsm.state(), RunStatus::Running);
132    /// ```
133    pub fn from_state(state: RunStatus) -> Self {
134        Self {
135            state,
136            history: Vec::new(),
137        }
138    }
139
140    /// Returns the current state.
141    pub fn state(&self) -> RunStatus {
142        self.state
143    }
144
145    /// Returns the full transition history.
146    pub fn history(&self) -> &[Transition<RunStatus, RunEvent>] {
147        &self.history
148    }
149
150    /// Returns `true` if the FSM is in a terminal state.
151    pub fn is_terminal(&self) -> bool {
152        self.state.is_terminal()
153    }
154
155    /// Apply an event, transitioning to a new state if valid.
156    ///
157    /// # Errors
158    ///
159    /// Returns [`TransitionError`] if the event is not allowed in the current state.
160    ///
161    /// # Examples
162    ///
163    /// ```
164    /// use ironflow_engine::fsm::{RunFsm, RunEvent};
165    /// use ironflow_store::entities::RunStatus;
166    ///
167    /// let mut fsm = RunFsm::new();
168    ///
169    /// // Valid transition
170    /// assert!(fsm.apply(RunEvent::PickedUp).is_ok());
171    ///
172    /// // Invalid: can't pick up a running run
173    /// assert!(fsm.apply(RunEvent::PickedUp).is_err());
174    /// ```
175    pub fn apply(
176        &mut self,
177        event: RunEvent,
178    ) -> Result<RunStatus, TransitionError<RunStatus, RunEvent>> {
179        let next = next_state(self.state, event).ok_or(TransitionError {
180            from: self.state,
181            event,
182        })?;
183
184        let transition = Transition {
185            from: self.state,
186            to: next,
187            event,
188            at: Utc::now(),
189        };
190
191        self.history.push(transition);
192        self.state = next;
193        Ok(next)
194    }
195
196    /// Check if an event would be accepted without applying it.
197    ///
198    /// # Examples
199    ///
200    /// ```
201    /// use ironflow_engine::fsm::{RunFsm, RunEvent};
202    ///
203    /// let fsm = RunFsm::new();
204    /// assert!(fsm.can_apply(RunEvent::PickedUp));
205    /// assert!(!fsm.can_apply(RunEvent::AllStepsCompleted));
206    /// ```
207    pub fn can_apply(&self, event: RunEvent) -> bool {
208        next_state(self.state, event).is_some()
209    }
210}
211
212impl Default for RunFsm {
213    fn default() -> Self {
214        Self::new()
215    }
216}
217
218/// Pure transition function — returns the next state for a given (state, event)
219/// pair, or `None` if the transition is invalid.
220fn next_state(from: RunStatus, event: RunEvent) -> Option<RunStatus> {
221    match (from, event) {
222        // Pending
223        (RunStatus::Pending, RunEvent::PickedUp) => Some(RunStatus::Running),
224        (RunStatus::Pending, RunEvent::CancelRequested) => Some(RunStatus::Cancelled),
225
226        // Running
227        (RunStatus::Running, RunEvent::AllStepsCompleted) => Some(RunStatus::Completed),
228        (RunStatus::Running, RunEvent::StepFailed) => Some(RunStatus::Failed),
229        (RunStatus::Running, RunEvent::StepFailedRetryable) => Some(RunStatus::Retrying),
230        (RunStatus::Running, RunEvent::CancelRequested) => Some(RunStatus::Cancelled),
231
232        // Retrying
233        (RunStatus::Retrying, RunEvent::RetryStarted) => Some(RunStatus::Running),
234        (RunStatus::Retrying, RunEvent::MaxRetriesExceeded) => Some(RunStatus::Failed),
235        (RunStatus::Retrying, RunEvent::CancelRequested) => Some(RunStatus::Cancelled),
236
237        // Approval
238        (RunStatus::Running, RunEvent::ApprovalRequested) => Some(RunStatus::AwaitingApproval),
239        (RunStatus::AwaitingApproval, RunEvent::Approved) => Some(RunStatus::Running),
240        (RunStatus::AwaitingApproval, RunEvent::Rejected) => Some(RunStatus::Failed),
241        (RunStatus::AwaitingApproval, RunEvent::CancelRequested) => Some(RunStatus::Cancelled),
242
243        // Delay
244        (RunStatus::Running, RunEvent::DelaySleeping) => Some(RunStatus::Sleeping),
245        (RunStatus::Sleeping, RunEvent::DelayElapsed) => Some(RunStatus::Pending),
246        (RunStatus::Sleeping, RunEvent::CancelRequested) => Some(RunStatus::Cancelled),
247
248        // Terminal states and all other combos → invalid
249        _ => None,
250    }
251}
252
253#[cfg(test)]
254mod tests {
255    use super::*;
256
257    // ---- Happy paths ----
258
259    #[test]
260    fn pending_to_running() {
261        let mut fsm = RunFsm::new();
262        let result = fsm.apply(RunEvent::PickedUp);
263        assert!(result.is_ok());
264        assert_eq!(fsm.state(), RunStatus::Running);
265    }
266
267    #[test]
268    fn full_success_path() {
269        let mut fsm = RunFsm::new();
270        fsm.apply(RunEvent::PickedUp).unwrap();
271        fsm.apply(RunEvent::AllStepsCompleted).unwrap();
272        assert_eq!(fsm.state(), RunStatus::Completed);
273        assert!(fsm.is_terminal());
274        assert_eq!(fsm.history().len(), 2);
275    }
276
277    #[test]
278    fn full_failure_path() {
279        let mut fsm = RunFsm::new();
280        fsm.apply(RunEvent::PickedUp).unwrap();
281        fsm.apply(RunEvent::StepFailed).unwrap();
282        assert_eq!(fsm.state(), RunStatus::Failed);
283        assert!(fsm.is_terminal());
284    }
285
286    #[test]
287    fn retry_then_success() {
288        let mut fsm = RunFsm::new();
289        fsm.apply(RunEvent::PickedUp).unwrap();
290        fsm.apply(RunEvent::StepFailedRetryable).unwrap();
291        assert_eq!(fsm.state(), RunStatus::Retrying);
292
293        fsm.apply(RunEvent::RetryStarted).unwrap();
294        assert_eq!(fsm.state(), RunStatus::Running);
295
296        fsm.apply(RunEvent::AllStepsCompleted).unwrap();
297        assert_eq!(fsm.state(), RunStatus::Completed);
298        assert_eq!(fsm.history().len(), 4);
299    }
300
301    #[test]
302    fn retry_then_max_retries_exceeded() {
303        let mut fsm = RunFsm::new();
304        fsm.apply(RunEvent::PickedUp).unwrap();
305        fsm.apply(RunEvent::StepFailedRetryable).unwrap();
306        fsm.apply(RunEvent::MaxRetriesExceeded).unwrap();
307        assert_eq!(fsm.state(), RunStatus::Failed);
308    }
309
310    #[test]
311    fn cancel_from_pending() {
312        let mut fsm = RunFsm::new();
313        fsm.apply(RunEvent::CancelRequested).unwrap();
314        assert_eq!(fsm.state(), RunStatus::Cancelled);
315        assert!(fsm.is_terminal());
316    }
317
318    #[test]
319    fn cancel_from_running() {
320        let mut fsm = RunFsm::new();
321        fsm.apply(RunEvent::PickedUp).unwrap();
322        fsm.apply(RunEvent::CancelRequested).unwrap();
323        assert_eq!(fsm.state(), RunStatus::Cancelled);
324    }
325
326    #[test]
327    fn cancel_from_retrying() {
328        let mut fsm = RunFsm::new();
329        fsm.apply(RunEvent::PickedUp).unwrap();
330        fsm.apply(RunEvent::StepFailedRetryable).unwrap();
331        fsm.apply(RunEvent::CancelRequested).unwrap();
332        assert_eq!(fsm.state(), RunStatus::Cancelled);
333    }
334
335    // ---- Invalid transitions ----
336
337    #[test]
338    fn cannot_complete_from_pending() {
339        let mut fsm = RunFsm::new();
340        let result = fsm.apply(RunEvent::AllStepsCompleted);
341        assert!(result.is_err());
342        assert_eq!(fsm.state(), RunStatus::Pending);
343    }
344
345    #[test]
346    fn cannot_pick_up_running() {
347        let mut fsm = RunFsm::new();
348        fsm.apply(RunEvent::PickedUp).unwrap();
349        let result = fsm.apply(RunEvent::PickedUp);
350        assert!(result.is_err());
351    }
352
353    #[test]
354    fn cannot_transition_from_terminal() {
355        let mut fsm = RunFsm::new();
356        fsm.apply(RunEvent::PickedUp).unwrap();
357        fsm.apply(RunEvent::AllStepsCompleted).unwrap();
358
359        assert!(fsm.apply(RunEvent::PickedUp).is_err());
360        assert!(fsm.apply(RunEvent::CancelRequested).is_err());
361        assert!(fsm.apply(RunEvent::StepFailed).is_err());
362    }
363
364    // ---- can_apply ----
365
366    #[test]
367    fn can_apply_checks_without_mutation() {
368        let fsm = RunFsm::new();
369        assert!(fsm.can_apply(RunEvent::PickedUp));
370        assert!(fsm.can_apply(RunEvent::CancelRequested));
371        assert!(!fsm.can_apply(RunEvent::AllStepsCompleted));
372        assert!(!fsm.can_apply(RunEvent::StepFailed));
373        assert_eq!(fsm.state(), RunStatus::Pending);
374    }
375
376    // ---- from_state ----
377
378    #[test]
379    fn from_state_resumes_at_given_state() {
380        let mut fsm = RunFsm::from_state(RunStatus::Running);
381        assert_eq!(fsm.state(), RunStatus::Running);
382        assert!(fsm.history().is_empty());
383
384        fsm.apply(RunEvent::AllStepsCompleted).unwrap();
385        assert_eq!(fsm.state(), RunStatus::Completed);
386    }
387
388    // ---- History ----
389
390    #[test]
391    fn history_records_transitions() {
392        let mut fsm = RunFsm::new();
393        fsm.apply(RunEvent::PickedUp).unwrap();
394        fsm.apply(RunEvent::StepFailedRetryable).unwrap();
395        fsm.apply(RunEvent::RetryStarted).unwrap();
396
397        let history = fsm.history();
398        assert_eq!(history.len(), 3);
399
400        assert_eq!(history[0].from, RunStatus::Pending);
401        assert_eq!(history[0].to, RunStatus::Running);
402        assert_eq!(history[0].event, RunEvent::PickedUp);
403
404        assert_eq!(history[1].from, RunStatus::Running);
405        assert_eq!(history[1].to, RunStatus::Retrying);
406        assert_eq!(history[1].event, RunEvent::StepFailedRetryable);
407
408        assert_eq!(history[2].from, RunStatus::Retrying);
409        assert_eq!(history[2].to, RunStatus::Running);
410        assert_eq!(history[2].event, RunEvent::RetryStarted);
411    }
412
413    // ---- Approval transitions ----
414
415    #[test]
416    fn running_to_awaiting_approval() {
417        let mut fsm = RunFsm::new();
418        fsm.apply(RunEvent::PickedUp).unwrap();
419        fsm.apply(RunEvent::ApprovalRequested).unwrap();
420        assert_eq!(fsm.state(), RunStatus::AwaitingApproval);
421        assert!(!fsm.is_terminal());
422    }
423
424    #[test]
425    fn awaiting_approval_approved_resumes_running() {
426        let mut fsm = RunFsm::new();
427        fsm.apply(RunEvent::PickedUp).unwrap();
428        fsm.apply(RunEvent::ApprovalRequested).unwrap();
429        fsm.apply(RunEvent::Approved).unwrap();
430        assert_eq!(fsm.state(), RunStatus::Running);
431    }
432
433    #[test]
434    fn awaiting_approval_rejected_fails() {
435        let mut fsm = RunFsm::new();
436        fsm.apply(RunEvent::PickedUp).unwrap();
437        fsm.apply(RunEvent::ApprovalRequested).unwrap();
438        fsm.apply(RunEvent::Rejected).unwrap();
439        assert_eq!(fsm.state(), RunStatus::Failed);
440        assert!(fsm.is_terminal());
441    }
442
443    #[test]
444    fn awaiting_approval_cancel() {
445        let mut fsm = RunFsm::new();
446        fsm.apply(RunEvent::PickedUp).unwrap();
447        fsm.apply(RunEvent::ApprovalRequested).unwrap();
448        fsm.apply(RunEvent::CancelRequested).unwrap();
449        assert_eq!(fsm.state(), RunStatus::Cancelled);
450        assert!(fsm.is_terminal());
451    }
452
453    #[test]
454    fn cannot_approve_from_pending() {
455        let mut fsm = RunFsm::new();
456        assert!(fsm.apply(RunEvent::Approved).is_err());
457    }
458
459    #[test]
460    fn approval_then_complete() {
461        let mut fsm = RunFsm::new();
462        fsm.apply(RunEvent::PickedUp).unwrap();
463        fsm.apply(RunEvent::ApprovalRequested).unwrap();
464        fsm.apply(RunEvent::Approved).unwrap();
465        fsm.apply(RunEvent::AllStepsCompleted).unwrap();
466        assert_eq!(fsm.state(), RunStatus::Completed);
467        assert_eq!(fsm.history().len(), 4);
468    }
469
470    // ---- TransitionError Display ----
471
472    #[test]
473    fn transition_error_display() {
474        let mut fsm = RunFsm::new();
475        let err = fsm.apply(RunEvent::AllStepsCompleted).unwrap_err();
476        let msg = err.to_string();
477        assert!(msg.contains("all_steps_completed"));
478        assert!(msg.contains("Pending"));
479    }
480}
ironflow_engine/fsm/run_fsm.rs

ironflow_engine/fsm/
run_fsm.rs
