operon 0.7.0

A workflow engine for parallel, incremental scheduling of DAG-defined multiplex tasks.
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145

use std::str::FromStr;

use chrono::{DateTime, Utc};
use uuid::Uuid;

/// The trace a run leaves behind, used to check that a restart picks up consistent data.
///
/// A finishing run writes the same footprint to the entity storage through
/// [`OperonStorage::put_footprint`](crate::OperonStorage::put_footprint) and to the metadata
/// storage.
/// The next run compares the two, and resumes only when they agree — timestamp included.
///
/// [`Uuid`], [`DateTime`], and [`Utc`] are re-exported from the `uuid` and `chrono` crates for
/// construction of the footprint.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct RunFootprint {
    /// The run this footprint belongs to.
    pub metadata: RunMetadata,
    /// When the footprint was taken.
    pub at: DateTime<Utc>,
}

/// The identity of a run and how far it got.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct RunMetadata {
    /// Distinguishes this run from other runs over the same data.
    pub run_id: Uuid,
    /// The state the run had reached when it was recorded.
    pub state: RunState,
}

/// How far a run got. Decides whether a restart resumes it or starts over.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum RunState {
    /// No run has been recorded yet.
    Fresh,
    /// The run is under way.
    Running,
    /// The run stopped gracefully and is resumable immediately.
    Stopped,
    /// Every task of the run finished.
    Completed,
    /// The run stopped without recording a graceful stop, so its progress is not trusted.
    Aborted,
}

impl RunFootprint {
    pub fn new(run_id: Uuid, state: RunState) -> Self {
        let metadata = RunMetadata::new(run_id, state);
        Self::from(metadata)
    }

    pub fn at(run_id: Uuid, state: RunState, at: DateTime<Utc>) -> Self {
        let metadata = RunMetadata::new(run_id, state);
        Self { metadata, at }
    }
}

impl From<RunMetadata> for RunFootprint {
    fn from(metadata: RunMetadata) -> Self {
        Self {
            metadata,
            at: Utc::now(),
        }
    }
}

impl RunMetadata {
    pub fn new(run_id: Uuid, state: RunState) -> Self {
        Self { run_id, state }
    }
}

impl RunState {
    /// The states a recorded run can be in.
    pub(crate) const RECORDED: [Self; 4] =
        [Self::Running, Self::Stopped, Self::Completed, Self::Aborted];

    /// The states an ended execution can be in.
    pub(crate) const ENDED: [Self; 3] = [Self::Stopped, Self::Completed, Self::Aborted];
}

impl Default for RunMetadata {
    fn default() -> Self {
        Self {
            run_id: Uuid::new_v4(),
            state: RunState::Fresh,
        }
    }
}

impl std::fmt::Display for RunState {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        let state_str = match self {
            RunState::Fresh => "fresh",
            RunState::Running => "running",
            RunState::Stopped => "stopped",
            RunState::Completed => "completed",
            RunState::Aborted => "aborted",
        };
        write!(f, "{}", state_str)
    }
}

impl FromStr for RunState {
    type Err = String;

    fn from_str(s: &str) -> Result<Self, Self::Err> {
        if s.eq_ignore_ascii_case("fresh") {
            Ok(RunState::Fresh)
        } else if s.eq_ignore_ascii_case("running") {
            Ok(RunState::Running)
        } else if s.eq_ignore_ascii_case("stopped") {
            Ok(RunState::Stopped)
        } else if s.eq_ignore_ascii_case("completed") {
            Ok(RunState::Completed)
        } else if s.eq_ignore_ascii_case("aborted") {
            Ok(RunState::Aborted)
        } else {
            Err(s.to_owned())
        }
    }
}

#[cfg(test)]
mod tests {
    use rstest::rstest;

    use super::*;

    #[rstest]
    #[case::fresh(RunState::Fresh)]
    #[case::running(RunState::Running)]
    #[case::stopped(RunState::Stopped)]
    #[case::completed(RunState::Completed)]
    #[case::aborted(RunState::Aborted)]
    fn test_run_state_round_trip(#[case] state: RunState) {
        assert_eq!(RunState::from_str(&state.to_string()), Ok(state));
    }

    #[test]
    fn test_parse_paused_is_rejected() {
        assert_eq!(RunState::from_str("paused"), Err("paused".to_owned()));
    }
}