aster_forge_tasks/
steps.rs

1//! Background task step state helpers.
2
3use chrono::{DateTime, Utc};
4use serde::{Deserialize, Serialize};
5#[cfg(all(debug_assertions, feature = "openapi"))]
6use utoipa::ToSchema;
7
8use crate::{Result, TaskCoreError};
9
10/// Runtime status for a task step.
11#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
12#[cfg_attr(all(debug_assertions, feature = "openapi"), derive(ToSchema))]
13#[serde(rename_all = "snake_case")]
14pub enum TaskStepStatus {
15    /// The step has not started.
16    Pending,
17    /// The step is currently running.
18    Active,
19    /// The step completed successfully.
20    Succeeded,
21    /// The step failed.
22    Failed,
23    /// The step was intentionally skipped.
24    Skipped,
25    /// The step was canceled.
26    Canceled,
27}
28
29/// Serialized task step shown in task APIs.
30#[derive(Debug, Clone, Serialize, Deserialize)]
31#[cfg_attr(all(debug_assertions, feature = "openapi"), derive(ToSchema))]
32pub struct TaskStepInfo {
33    /// Stable step key.
34    pub key: String,
35    /// Human-readable step title.
36    pub title: String,
37    /// Current step status.
38    pub status: TaskStepStatus,
39    /// Current progress amount.
40    pub progress_current: i64,
41    /// Total progress amount.
42    pub progress_total: i64,
43    /// Optional detail text.
44    pub detail: Option<String>,
45    /// Step start time.
46    #[cfg_attr(all(debug_assertions, feature = "openapi"), schema(value_type = Option<String>))]
47    pub started_at: Option<DateTime<Utc>>,
48    /// Step finish time.
49    #[cfg_attr(all(debug_assertions, feature = "openapi"), schema(value_type = Option<String>))]
50    pub finished_at: Option<DateTime<Utc>>,
51}
52
53/// Static step definition used to create initial task steps.
54#[derive(Debug, Clone, Copy)]
55pub struct TaskStepSpec {
56    /// Stable step key.
57    pub key: &'static str,
58    /// Human-readable step title.
59    pub title: &'static str,
60}
61
62fn new_task_step(spec: TaskStepSpec, status: TaskStepStatus, detail: Option<&str>) -> TaskStepInfo {
63    let now = (status == TaskStepStatus::Active).then(Utc::now);
64    TaskStepInfo {
65        key: spec.key.to_string(),
66        title: spec.title.to_string(),
67        status,
68        progress_current: 0,
69        progress_total: 0,
70        detail: detail.map(str::to_string),
71        started_at: now,
72        finished_at: None,
73    }
74}
75
76/// Creates initial task step state from static specs.
77#[must_use]
78pub fn initial_task_steps_from_specs(specs: &[TaskStepSpec]) -> Vec<TaskStepInfo> {
79    specs
80        .iter()
81        .enumerate()
82        .map(|(index, spec)| {
83            new_task_step(
84                *spec,
85                if index == 0 {
86                    TaskStepStatus::Active
87                } else {
88                    TaskStepStatus::Pending
89                },
90                if index == 0 {
91                    Some("Waiting for worker")
92                } else {
93                    None
94                },
95            )
96        })
97        .collect()
98}
99
100fn find_task_step_mut<'a>(
101    steps: &'a mut [TaskStepInfo],
102    key: &str,
103) -> Result<&'a mut TaskStepInfo> {
104    steps
105        .iter_mut()
106        .find(|step| step.key == key)
107        .ok_or_else(|| TaskCoreError::invalid_value(format!("task step '{key}' not found")))
108}
109
110/// Marks a task step active.
111///
112/// # Errors
113///
114/// Returns [`TaskCoreError`] when the step key or progress transition is invalid.
115pub fn set_task_step_active(
116    steps: &mut [TaskStepInfo],
117    key: &str,
118    detail: Option<&str>,
119    progress: Option<(i64, i64)>,
120) -> Result<()> {
121    let now = Utc::now();
122    let step = find_task_step_mut(steps, key)?;
123    step.status = TaskStepStatus::Active;
124    if step.started_at.is_none() {
125        step.started_at = Some(now);
126    }
127    step.finished_at = None;
128    step.detail = detail.map(str::to_string);
129    if let Some((current, total)) = progress {
130        step.progress_current = current;
131        step.progress_total = total;
132    }
133    Ok(())
134}
135
136/// Marks a task step succeeded.
137///
138/// # Errors
139///
140/// Returns [`TaskCoreError`] when the step key or progress transition is invalid.
141pub fn set_task_step_succeeded(
142    steps: &mut [TaskStepInfo],
143    key: &str,
144    detail: Option<&str>,
145    progress: Option<(i64, i64)>,
146) -> Result<()> {
147    let now = Utc::now();
148    let step = find_task_step_mut(steps, key)?;
149    step.status = TaskStepStatus::Succeeded;
150    if step.started_at.is_none() {
151        step.started_at = Some(now);
152    }
153    step.finished_at = Some(now);
154    step.detail = detail.map(str::to_string);
155    if let Some((current, total)) = progress {
156        step.progress_current = current;
157        step.progress_total = total;
158    } else if step.progress_total > 0 {
159        step.progress_current = step.progress_total;
160    }
161    Ok(())
162}
163
164/// Marks a task step skipped.
165///
166/// # Errors
167///
168/// Returns [`TaskCoreError`] when the step key is unknown or the transition is invalid.
169pub fn set_task_step_skipped(
170    steps: &mut [TaskStepInfo],
171    key: &str,
172    detail: Option<&str>,
173) -> Result<()> {
174    let now = Utc::now();
175    let step = find_task_step_mut(steps, key)?;
176    step.status = TaskStepStatus::Skipped;
177    if step.started_at.is_none() {
178        step.started_at = Some(now);
179    }
180    step.finished_at = Some(now);
181    step.detail = detail.map(str::to_string);
182    Ok(())
183}
184
185/// Marks the active step failed, or the last pending step when no step is active.
186pub fn mark_active_step_failed(steps: &mut [TaskStepInfo], detail: Option<&str>) {
187    let now = Utc::now();
188    if let Some(step) = steps
189        .iter_mut()
190        .find(|step| step.status == TaskStepStatus::Active)
191    {
192        step.status = TaskStepStatus::Failed;
193        if step.started_at.is_none() {
194            step.started_at = Some(now);
195        }
196        step.finished_at = Some(now);
197        step.detail = detail.map(str::to_string);
198        return;
199    }
200    if let Some(step) = steps
201        .iter_mut()
202        .rev()
203        .find(|step| step.status == TaskStepStatus::Pending)
204    {
205        step.status = TaskStepStatus::Failed;
206        step.started_at = Some(now);
207        step.finished_at = Some(now);
208        step.detail = detail.map(str::to_string);
209    }
210}
211
212#[cfg(test)]
213mod tests {
214    use super::{
215        TaskStepInfo, TaskStepSpec, TaskStepStatus, initial_task_steps_from_specs,
216        mark_active_step_failed, set_task_step_active, set_task_step_skipped,
217        set_task_step_succeeded,
218    };
219
220    fn step(key: &str, status: TaskStepStatus) -> TaskStepInfo {
221        TaskStepInfo {
222            key: key.to_string(),
223            title: key.to_string(),
224            status,
225            progress_current: 0,
226            progress_total: 1,
227            detail: None,
228            started_at: None,
229            finished_at: None,
230        }
231    }
232
233    #[test]
234    fn initial_steps_activate_first_spec_and_leave_rest_pending() {
235        let steps = initial_task_steps_from_specs(&[
236            TaskStepSpec {
237                key: "prepare",
238                title: "Prepare",
239            },
240            TaskStepSpec {
241                key: "finish",
242                title: "Finish",
243            },
244        ]);
245
246        assert_eq!(steps.len(), 2);
247        assert_eq!(steps[0].key, "prepare");
248        assert_eq!(steps[0].title, "Prepare");
249        assert_eq!(steps[0].status, TaskStepStatus::Active);
250        assert_eq!(steps[0].detail.as_deref(), Some("Waiting for worker"));
251        assert!(steps[0].started_at.is_some());
252        assert_eq!(steps[1].status, TaskStepStatus::Pending);
253        assert_eq!(steps[1].detail, None);
254        assert!(steps[1].started_at.is_none());
255    }
256
257    #[test]
258    fn step_state_helpers_update_timestamps_progress_and_detail() {
259        let mut steps = vec![step("prepare", TaskStepStatus::Pending)];
260
261        set_task_step_active(&mut steps, "prepare", Some("running"), Some((2, 5))).unwrap();
262        assert_eq!(steps[0].status, TaskStepStatus::Active);
263        assert_eq!(steps[0].detail.as_deref(), Some("running"));
264        assert_eq!(steps[0].progress_current, 2);
265        assert_eq!(steps[0].progress_total, 5);
266        assert!(steps[0].started_at.is_some());
267        assert!(steps[0].finished_at.is_none());
268
269        set_task_step_succeeded(&mut steps, "prepare", Some("done"), None).unwrap();
270        assert_eq!(steps[0].status, TaskStepStatus::Succeeded);
271        assert_eq!(steps[0].detail.as_deref(), Some("done"));
272        assert_eq!(steps[0].progress_current, 5);
273        assert!(steps[0].finished_at.is_some());
274
275        set_task_step_skipped(&mut steps, "prepare", Some("skip")).unwrap();
276        assert_eq!(steps[0].status, TaskStepStatus::Skipped);
277        assert_eq!(steps[0].detail.as_deref(), Some("skip"));
278    }
279
280    #[test]
281    fn mark_active_step_failed_updates_active_step_first() {
282        let mut steps = vec![
283            step("prepare", TaskStepStatus::Succeeded),
284            step("process", TaskStepStatus::Active),
285            step("finish", TaskStepStatus::Pending),
286        ];
287
288        mark_active_step_failed(&mut steps, Some("failed"));
289
290        assert_eq!(steps[1].status, TaskStepStatus::Failed);
291        assert_eq!(steps[1].detail.as_deref(), Some("failed"));
292        assert!(steps[1].started_at.is_some());
293        assert!(steps[1].finished_at.is_some());
294        assert_eq!(steps[2].status, TaskStepStatus::Pending);
295    }
296
297    #[test]
298    fn mark_active_step_failed_falls_back_to_last_pending_step() {
299        let mut steps = vec![
300            step("prepare", TaskStepStatus::Succeeded),
301            step("process", TaskStepStatus::Pending),
302            step("finish", TaskStepStatus::Pending),
303        ];
304
305        mark_active_step_failed(&mut steps, Some("pending failed"));
306
307        assert_eq!(steps[1].status, TaskStepStatus::Pending);
308        assert_eq!(steps[2].status, TaskStepStatus::Failed);
309        assert_eq!(steps[2].detail.as_deref(), Some("pending failed"));
310    }
311}
aster_forge_tasks/steps.rs

aster_forge_tasks/
steps.rs
