oxide_batch_core/domain/
lifecycle.rs

1use std::error::Error;
2use std::fmt;
3use std::time::SystemTime;
4
5use super::{BatchStatus, DomainError, FailureSummary};
6
7/// A database-agnostic optimistic-lock version for an execution record.
8#[derive(Clone, Copy, Debug, Default, Eq, Hash, Ord, PartialEq, PartialOrd)]
9pub struct ExecutionVersion(u64);
10
11impl ExecutionVersion {
12    /// The version assigned to a newly created execution attempt.
13    pub const INITIAL: Self = Self(0);
14
15    /// Reconstructs a version stored by a repository.
16    #[must_use]
17    pub const fn new(value: u64) -> Self {
18        Self(value)
19    }
20
21    /// Returns the repository-independent numeric value.
22    #[must_use]
23    pub const fn get(self) -> u64 {
24        self.0
25    }
26
27    /// Returns the next optimistic version.
28    ///
29    /// # Errors
30    ///
31    /// Returns [`LifecycleError::VersionExhausted`] at the representable limit.
32    pub fn next(self) -> Result<Self, LifecycleError> {
33        self.0
34            .checked_add(1)
35            .map(Self)
36            .ok_or(LifecycleError::VersionExhausted { version: self })
37    }
38}
39
40impl fmt::Display for ExecutionVersion {
41    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
42        self.0.fmt(formatter)
43    }
44}
45
46/// A requested framework lifecycle transition and its deterministic timestamp.
47#[derive(Clone, Copy, Debug, Eq, PartialEq)]
48pub struct LifecycleTransition {
49    target: BatchStatus,
50    transitioned_at: SystemTime,
51    failure: Option<FailureSummary>,
52    terminal_rollback: bool,
53}
54
55impl LifecycleTransition {
56    /// Requests a transition that does not introduce a failure.
57    #[must_use]
58    pub const fn new(target: BatchStatus, transitioned_at: SystemTime) -> Self {
59        Self {
60            target,
61            transitioned_at,
62            failure: None,
63            terminal_rollback: false,
64        }
65    }
66
67    /// Requests a transition to `FAILED` with a redacted failure summary.
68    #[must_use]
69    pub const fn failed(transitioned_at: SystemTime, failure: FailureSummary) -> Self {
70        Self {
71            target: BatchStatus::Failed,
72            transitioned_at,
73            failure: Some(failure),
74            terminal_rollback: false,
75        }
76    }
77
78    /// Returns the requested framework status.
79    #[must_use]
80    pub const fn target(self) -> BatchStatus {
81        self.target
82    }
83
84    /// Returns the deterministic transition instant supplied by the caller.
85    #[must_use]
86    pub const fn transitioned_at(self) -> SystemTime {
87        self.transitioned_at
88    }
89
90    /// Marks the transition as one whose terminal work was rolled back.
91    #[must_use]
92    pub const fn with_terminal_rollback(mut self) -> Self {
93        self.terminal_rollback = true;
94        self
95    }
96
97    pub(crate) const fn terminal_rollback(self) -> bool {
98        self.terminal_rollback
99    }
100
101    pub(crate) const fn failure(self) -> Option<FailureSummary> {
102        self.failure
103    }
104}
105
106/// A typed lifecycle-policy or optimistic-concurrency failure.
107#[derive(Clone, Debug, Eq, PartialEq)]
108#[non_exhaustive]
109pub enum LifecycleError {
110    /// The caller observed an older or otherwise different execution version.
111    StaleVersion {
112        /// The version supplied by the caller.
113        expected: ExecutionVersion,
114        /// The current version of the execution record.
115        actual: ExecutionVersion,
116    },
117    /// The requested in-place status transition is not legal.
118    IllegalTransition {
119        /// The current framework status.
120        from: BatchStatus,
121        /// The requested framework status.
122        to: BatchStatus,
123    },
124    /// Restart is valid only by creating another execution attempt.
125    RestartRequiresNewAttempt {
126        /// The finished status from which a restart was requested.
127        from: BatchStatus,
128    },
129    /// The current execution outcome cannot be restarted.
130    NotRestartable {
131        /// The status that prevents restart.
132        status: BatchStatus,
133    },
134    /// A restart reused a prior job- or step-attempt identifier.
135    AttemptIdentifierReused,
136    /// A transition to `FAILED` did not include a redacted failure summary.
137    FailedTransitionMissingFailure,
138    /// A supplied transition instant violated timestamp ordering.
139    InvalidTransitionTime {
140        /// The facade-owned validation failure.
141        source: DomainError,
142    },
143    /// The optimistic version cannot be incremented.
144    VersionExhausted {
145        /// The maximum current version.
146        version: ExecutionVersion,
147    },
148    /// A durable execution counter cannot be incremented.
149    CountExhausted,
150}
151
152impl fmt::Display for LifecycleError {
153    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
154        match self {
155            Self::StaleVersion { expected, actual } => {
156                write!(
157                    formatter,
158                    "stale execution version: expected {expected}, actual {actual}"
159                )
160            }
161            Self::IllegalTransition { from, to } => {
162                write!(
163                    formatter,
164                    "illegal lifecycle transition from {from} to {to}"
165                )
166            }
167            Self::RestartRequiresNewAttempt { from } => write!(
168                formatter,
169                "restart from {from} requires a new execution attempt"
170            ),
171            Self::NotRestartable { status } => {
172                write!(formatter, "an execution in {status} is not restartable")
173            }
174            Self::AttemptIdentifierReused => {
175                formatter.write_str("a restart requires a distinct execution identifier")
176            }
177            Self::FailedTransitionMissingFailure => {
178                formatter.write_str("a transition to FAILED requires a failure summary")
179            }
180            Self::InvalidTransitionTime { .. } => {
181                formatter.write_str("the lifecycle transition timestamp is out of order")
182            }
183            Self::VersionExhausted { version } => {
184                write!(
185                    formatter,
186                    "execution version {version} cannot be incremented"
187                )
188            }
189            Self::CountExhausted => formatter.write_str("an execution counter is exhausted"),
190        }
191    }
192}
193
194impl Error for LifecycleError {
195    fn source(&self) -> Option<&(dyn Error + 'static)> {
196        match self {
197            Self::InvalidTransitionTime { source } => Some(source),
198            _ => None,
199        }
200    }
201}
202
203pub(crate) fn validate_expected_version(
204    expected: ExecutionVersion,
205    actual: ExecutionVersion,
206) -> Result<(), LifecycleError> {
207    if expected != actual {
208        return Err(LifecycleError::StaleVersion { expected, actual });
209    }
210    Ok(())
211}
212
213pub(crate) const fn is_legal_in_place_transition(from: BatchStatus, to: BatchStatus) -> bool {
214    matches!(
215        (from, to),
216        (
217            BatchStatus::Starting,
218            BatchStatus::Started
219                | BatchStatus::Stopping
220                | BatchStatus::Failed
221                | BatchStatus::Unknown
222        ) | (
223            BatchStatus::Started,
224            BatchStatus::Stopping
225                | BatchStatus::Stopped
226                | BatchStatus::Failed
227                | BatchStatus::Completed
228                | BatchStatus::Unknown
229        ) | (
230            BatchStatus::Stopping,
231            BatchStatus::Stopped | BatchStatus::Failed | BatchStatus::Unknown
232        ) | (
233            BatchStatus::Stopped | BatchStatus::Failed | BatchStatus::Unknown,
234            BatchStatus::Abandoned
235        ) | (BatchStatus::Unknown, BatchStatus::Failed)
236    )
237}
238
239pub(crate) fn validate_transition(
240    from: BatchStatus,
241    transition: LifecycleTransition,
242) -> Result<(), LifecycleError> {
243    let to = transition.target();
244    if matches!(from, BatchStatus::Stopped | BatchStatus::Failed)
245        && matches!(to, BatchStatus::Starting)
246    {
247        return Err(LifecycleError::RestartRequiresNewAttempt { from });
248    }
249    if !is_legal_in_place_transition(from, to) {
250        return Err(LifecycleError::IllegalTransition { from, to });
251    }
252    if matches!(to, BatchStatus::Failed) && transition.failure().is_none() {
253        return Err(LifecycleError::FailedTransitionMissingFailure);
254    }
255    Ok(())
256}
257
258pub(crate) fn validate_restart(status: BatchStatus) -> Result<(), LifecycleError> {
259    if !matches!(status, BatchStatus::Stopped | BatchStatus::Failed) {
260        return Err(LifecycleError::NotRestartable { status });
261    }
262    Ok(())
263}
oxide_batch_core/domain/lifecycle.rs

oxide_batch_core/domain/
lifecycle.rs
