solti_model/resource/
task.rs

1//! # Task resource.
2//!
3//! [`Task`] is the K8s-style aggregate: metadata + spec + status.
4
5use serde::{Deserialize, Serialize};
6
7use crate::{
8    Labels, ObjectMeta, Slot, TaskId, TaskPhase, TaskSpec, TaskStatus,
9    error::{ModelError, ModelResult},
10};
11
12/// Unified task resource.
13///
14/// Every task is represented as a single resource with three sections:
15/// - `status`   - observed state: current phase, attempts, errors ([`TaskStatus`])
16/// - `spec`     - desired state: what to run and how ([`TaskSpec`])
17/// - `metadata` - identity, versioning, timestamps ([`ObjectMeta`])
18///
19/// ## Also
20///
21/// - [`TaskSpec`] desired state (what to run, how to restart).
22/// - [`TaskStatus`] observed state (phase, attempt, exit code).
23/// - [`TaskRun`](crate::TaskRun) per-attempt execution record.
24/// - [`ObjectMeta`] identity and versioning.
25/// - [`TaskPhase`] lifecycle state machine.
26#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
27#[serde(rename_all = "camelCase")]
28pub struct Task {
29    metadata: ObjectMeta,
30    status: TaskStatus,
31    spec: TaskSpec,
32}
33
34impl Task {
35    /// Create a new task in [`TaskPhase::Pending`] phase.
36    pub fn new(id: TaskId, spec: TaskSpec) -> Self {
37        Self {
38            metadata: ObjectMeta::new(id),
39            status: TaskStatus::pending(),
40            spec,
41        }
42    }
43
44    /// Resource metadata (identity, `resource_version`, timestamps).
45    #[inline]
46    pub fn metadata(&self) -> &ObjectMeta {
47        &self.metadata
48    }
49
50    /// Observed state (phase, attempt, exit code, error).
51    #[inline]
52    pub fn status(&self) -> &TaskStatus {
53        &self.status
54    }
55
56    /// Desired state (what to run, how to restart).
57    #[inline]
58    pub fn spec(&self) -> &TaskSpec {
59        &self.spec
60    }
61
62    /// Destructure into `(metadata, spec, status)`. Used by transport
63    /// layers that need owned fields for serialization into wire types.
64    #[inline]
65    pub fn into_parts(self) -> (ObjectMeta, TaskSpec, TaskStatus) {
66        (self.metadata, self.spec, self.status)
67    }
68
69    /// Transition the task into a new attempt: bumps attempt counter, sets phase to `Running`, clears error/exit_code.
70    pub fn transition_starting(&mut self) {
71        self.increment_attempt();
72        self.update_phase(TaskPhase::Running, None, None);
73    }
74
75    /// Transition the current attempt into a terminal phase with optional error and exit code.
76    ///
77    /// Rejects illegal transitions:
78    /// - target phase must be terminal (see [`TaskPhase::is_terminal`]);
79    ///   finishing into `Pending` or `Running` is a logic bug upstream.
80    ///
81    /// Sticky terminals: once an attempt has reached a specific final state
82    /// (`Succeeded`, `Canceled`, or `Timeout`), a later, conflicting terminal
83    /// event must not overwrite it. taskvisor emits two terminal events per run (the
84    /// per-attempt `TaskStopped`/`TaskCanceled`/`TaskFailed` followed by the
85    /// actor-level `ActorExhausted`/`ActorDead`); without this guard a trailing
86    /// `ActorExhausted` would flip a `Canceled` run into `Exhausted`. A
87    /// failure-class phase (`Failed`) is still allowed to be refined into a more
88    /// specific disposition (`Exhausted`/`Timeout`), and re-applying the same
89    /// phase is a harmless no-op.
90    ///
91    /// Bumps `resource_version` only when the phase actually changes.
92    pub fn transition_finished(
93        &mut self,
94        phase: TaskPhase,
95        error: Option<String>,
96        exit_code: Option<i32>,
97    ) -> ModelResult<()> {
98        if !phase.is_terminal() {
99            return Err(ModelError::Invalid(
100                format!("transition_finished requires a terminal phase, got {phase}").into(),
101            ));
102        }
103        if matches!(
104            self.status.phase,
105            TaskPhase::Succeeded | TaskPhase::Canceled | TaskPhase::Timeout
106        ) && self.status.phase != phase
107        {
108            // Keep the intentional/specific terminal; ignore the conflicting
109            // overwrite (e.g. a trailing ActorExhausted after a Timeout/Cancel).
110            return Ok(());
111        }
112        self.update_phase(phase, error, exit_code);
113        Ok(())
114    }
115
116    /// Raw phase setter.
117    pub(crate) fn update_phase(
118        &mut self,
119        phase: TaskPhase,
120        error: Option<String>,
121        exit_code: Option<i32>,
122    ) {
123        self.metadata.bump_resource_version();
124        self.status.phase = phase;
125        self.status.error = error;
126        self.status.exit_code = exit_code;
127    }
128
129    /// Raw attempt bump. Crate-private (see [`update_phase`]).
130    pub(crate) fn increment_attempt(&mut self) {
131        self.metadata.bump_resource_version();
132        self.status.attempt += 1;
133    }
134
135    /// Task identifier (shortcut for `metadata.id`).
136    #[inline]
137    pub fn id(&self) -> &TaskId {
138        &self.metadata.id
139    }
140
141    /// Slot (shortcut for `spec.slot()`).
142    #[inline]
143    pub fn slot(&self) -> &Slot {
144        self.spec.slot()
145    }
146
147    /// Labels (shortcut for `spec.labels()`).
148    #[inline]
149    pub fn labels(&self) -> &Labels {
150        self.spec.labels()
151    }
152
153    /// Current phase (shortcut for `status.phase`).
154    #[inline]
155    pub fn phase(&self) -> &TaskPhase {
156        &self.status.phase
157    }
158}
159
160#[cfg(test)]
161mod tests {
162    use super::*;
163    use crate::TaskKind;
164
165    fn test_spec() -> TaskSpec {
166        TaskSpec::builder("slot-a", TaskKind::Embedded, 5_000u64)
167            .build()
168            .expect("test spec must be valid")
169    }
170
171    #[test]
172    fn new_creates_pending_task() {
173        let task = Task::new("task-1".into(), test_spec());
174
175        assert_eq!(task.status().phase, TaskPhase::Pending);
176        assert_eq!(task.metadata().resource_version, 1);
177        assert_eq!(task.metadata().id, "task-1");
178        assert!(task.status().error.is_none());
179        assert_eq!(task.status().attempt, 0);
180        assert_eq!(task.slot(), "slot-a");
181    }
182
183    #[test]
184    fn transition_starting_sets_running_and_bumps() {
185        let mut task = Task::new("task-1".into(), test_spec());
186        task.transition_starting();
187
188        assert_eq!(task.status().phase, TaskPhase::Running);
189        assert_eq!(task.status().attempt, 1);
190        assert_eq!(task.metadata().resource_version, 3);
191    }
192
193    #[test]
194    fn transition_finished_accepts_terminal_and_carries_error() {
195        let mut task = Task::new("task-1".into(), test_spec());
196        task.transition_starting();
197        task.transition_finished(TaskPhase::Failed, Some("boom".into()), Some(1))
198            .unwrap();
199
200        assert_eq!(task.status().phase, TaskPhase::Failed);
201        assert_eq!(task.status().error.as_deref(), Some("boom"));
202        assert_eq!(task.status().exit_code, Some(1));
203    }
204
205    #[test]
206    fn transition_finished_rejects_non_terminal_phase() {
207        let mut task = Task::new("task-1".into(), test_spec());
208        let err = task
209            .transition_finished(TaskPhase::Running, None, None)
210            .unwrap_err();
211        assert!(err.to_string().contains("terminal phase"));
212    }
213
214    #[test]
215    fn canceled_is_not_overwritten_by_a_later_terminal_event() {
216        // A self-canceling task emits TaskCanceled (-> Canceled) and then an
217        // actor-level ActorExhausted: that trailing terminal must not flip the
218        // intentional Canceled into Exhausted/Failed.
219        let mut task = Task::new("task-1".into(), test_spec());
220        task.transition_starting();
221        task.transition_finished(TaskPhase::Canceled, None, None)
222            .unwrap();
223
224        task.transition_finished(TaskPhase::Exhausted, Some("budget".into()), Some(1))
225            .unwrap();
226
227        assert_eq!(task.status().phase, TaskPhase::Canceled);
228        assert!(task.status().error.is_none());
229        assert_eq!(task.status().exit_code, None);
230    }
231
232    #[test]
233    fn succeeded_is_not_overwritten_by_a_later_terminal_event() {
234        let mut task = Task::new("task-1".into(), test_spec());
235        task.transition_starting();
236        task.transition_finished(TaskPhase::Succeeded, None, None)
237            .unwrap();
238
239        task.transition_finished(TaskPhase::Failed, Some("late".into()), Some(2))
240            .unwrap();
241
242        assert_eq!(task.status().phase, TaskPhase::Succeeded);
243        assert!(task.status().error.is_none());
244    }
245
246    #[test]
247    fn timeout_is_not_overwritten_by_a_later_terminal_event() {
248        // A timed-out attempt is finalized as Timeout by the TimeoutHit/TaskFailed
249        // pairing; the trailing actor-level ActorExhausted must not erase it.
250        let mut task = Task::new("task-1".into(), test_spec());
251        task.transition_starting();
252        task.transition_finished(TaskPhase::Timeout, Some("deadline".into()), None)
253            .unwrap();
254
255        task.transition_finished(TaskPhase::Exhausted, Some("budget".into()), None)
256            .unwrap();
257
258        assert_eq!(task.status().phase, TaskPhase::Timeout);
259    }
260
261    #[test]
262    fn failed_can_be_refined_to_exhausted() {
263        // The per-attempt TaskFailed lands as Failed; the actor-level
264        // ActorExhausted refines it to Exhausted. This refinement must survive.
265        let mut task = Task::new("task-1".into(), test_spec());
266        task.transition_starting();
267        task.transition_finished(TaskPhase::Failed, Some("attempt".into()), None)
268            .unwrap();
269
270        task.transition_finished(TaskPhase::Exhausted, Some("max_retries".into()), None)
271            .unwrap();
272
273        assert_eq!(task.status().phase, TaskPhase::Exhausted);
274        assert_eq!(task.status().error.as_deref(), Some("max_retries"));
275    }
276
277    #[test]
278    fn convenience_accessors() {
279        let spec = TaskSpec::builder("slot-1", TaskKind::Embedded, 5_000u64)
280            .build()
281            .unwrap();
282        let task = Task::new("id-1".into(), spec);
283
284        assert_eq!(task.slot(), &Slot::from("slot-1"));
285        assert_eq!(task.id(), &TaskId::from("id-1"));
286        assert_eq!(*task.phase(), TaskPhase::Pending);
287    }
288
289    #[test]
290    fn serde_roundtrip() {
291        let spec = TaskSpec::builder("slot-1", TaskKind::Embedded, 5_000u64)
292            .build()
293            .unwrap();
294        let task = Task::new("id-1".into(), spec);
295        let json = serde_json::to_string(&task).unwrap();
296        let back: Task = serde_json::from_str(&json).unwrap();
297
298        assert_eq!(back.status().phase, TaskPhase::Pending);
299        assert_eq!(back.metadata().resource_version, 1);
300        assert_eq!(back.metadata().id, "id-1");
301    }
302}
solti_model/resource/task.rs

solti_model/resource/
task.rs
