oximedia_workflow/
parallel_steps.rs

1//! Parallel and sequential step execution within workflow stages.
2//!
3//! A [`WorkflowStage`] is a logical grouping of [`WorkflowStep`]s that either
4//! run **in parallel** (all steps spawned concurrently via `std::thread::scope`)
5//! or **sequentially** (each step blocks the next).
6//!
7//! # Example
8//!
9//! ```rust
10//! use oximedia_workflow::parallel_steps::{
11//!     ParallelSteps, StepType, WorkflowStage, WorkflowStep,
12//! };
13//!
14//! let mut parallel = ParallelSteps::new("encode-all");
15//! parallel.add_step(WorkflowStep::new("pass-1", StepType::Compute { value: 1 }));
16//! parallel.add_step(WorkflowStep::new("pass-2", StepType::Compute { value: 2 }));
17//! let results = parallel.execute_parallel().expect("all steps should succeed");
18//! assert_eq!(results.len(), 2);
19//! ```
20
21use serde::{Deserialize, Serialize};
22use std::time::{Duration, Instant};
23
24// ---------------------------------------------------------------------------
25// Step types
26// ---------------------------------------------------------------------------
27
28/// The unit of work a [`WorkflowStep`] performs.
29#[derive(Debug, Clone, Serialize, Deserialize)]
30pub enum StepType {
31    /// Produce an integer value (useful for testing).
32    Compute {
33        /// The integer to return as output.
34        value: i64,
35    },
36    /// Fail intentionally with a reason string.
37    Fail {
38        /// Human-readable failure reason.
39        reason: String,
40    },
41    /// Sleep for a duration (simulates I/O-bound work).
42    Wait {
43        /// Duration to sleep, in milliseconds.
44        duration_ms: u64,
45    },
46    /// Multiply `value` by `factor` and return the product.
47    Transform {
48        /// Input value.
49        value: i64,
50        /// Multiplication factor.
51        factor: i64,
52    },
53}
54
55// ---------------------------------------------------------------------------
56// WorkflowStep
57// ---------------------------------------------------------------------------
58
59/// A single unit of work inside a [`WorkflowStage`].
60#[derive(Debug, Clone, Serialize, Deserialize)]
61pub struct WorkflowStep {
62    /// Human-readable step name.
63    pub name: String,
64    /// What this step does.
65    pub step_type: StepType,
66    /// Optional per-step timeout.  When `Some(ms)`, a `Wait` step that
67    /// exceeds the timeout is marked as timed-out.
68    pub timeout_ms: Option<u64>,
69}
70
71impl WorkflowStep {
72    /// Create a step with no timeout.
73    #[must_use]
74    pub fn new(name: impl Into<String>, step_type: StepType) -> Self {
75        Self {
76            name: name.into(),
77            step_type,
78            timeout_ms: None,
79        }
80    }
81
82    /// Set an optional timeout (milliseconds).
83    #[must_use]
84    pub fn with_timeout(mut self, timeout_ms: u64) -> Self {
85        self.timeout_ms = Some(timeout_ms);
86        self
87    }
88}
89
90// ---------------------------------------------------------------------------
91// StepResult
92// ---------------------------------------------------------------------------
93
94/// The outcome of executing a single [`WorkflowStep`].
95#[derive(Debug, Clone)]
96pub struct StepResult {
97    /// Name of the step that produced this result.
98    pub name: String,
99    /// Whether the step completed without error.
100    pub success: bool,
101    /// Integer output, if the step produced one.
102    pub output: Option<i64>,
103    /// Error message, if the step failed.
104    pub error: Option<String>,
105    /// Approximate wall-clock time the step took, in milliseconds.
106    pub duration_ms: u64,
107}
108
109// ---------------------------------------------------------------------------
110// ParallelStepError
111// ---------------------------------------------------------------------------
112
113/// Errors from parallel (or sequential) step execution.
114#[derive(Debug, thiserror::Error)]
115pub enum ParallelStepError {
116    /// A single named step failed.
117    #[error("Step '{name}' failed: {reason}")]
118    StepFailed {
119        /// Name of the failing step.
120        name: String,
121        /// Failure reason.
122        reason: String,
123    },
124    /// Multiple steps failed (reported when `fail_fast = false`).
125    #[error("{count} of {total} steps failed")]
126    MultipleStepsFailed {
127        /// Number of failed steps.
128        count: usize,
129        /// Total number of steps in the group.
130        total: usize,
131        /// Individual step results (both successes and failures).
132        results: Vec<StepResult>,
133    },
134    /// A step exceeded its per-step timeout.
135    #[error("Step timed out: {name}")]
136    Timeout {
137        /// Name of the step that timed out.
138        name: String,
139    },
140}
141
142// ---------------------------------------------------------------------------
143// Step execution
144// ---------------------------------------------------------------------------
145
146/// Execute a single step and return a [`StepResult`].
147///
148/// This function is `Send + Sync`-safe and is designed to be called from
149/// `std::thread::scope` worker threads.
150pub fn execute_step(step: &WorkflowStep) -> StepResult {
151    let start = Instant::now();
152
153    let (success, output, error) = match &step.step_type {
154        StepType::Compute { value } => (true, Some(*value), None),
155
156        StepType::Fail { reason } => (false, None, Some(reason.clone())),
157
158        StepType::Wait { duration_ms } => {
159            let wait = Duration::from_millis(*duration_ms);
160            // Honour per-step timeout if set.
161            let timed_out = if let Some(limit_ms) = step.timeout_ms {
162                *duration_ms > limit_ms
163            } else {
164                false
165            };
166
167            if timed_out {
168                let elapsed = start.elapsed().as_millis() as u64;
169                return StepResult {
170                    name: step.name.clone(),
171                    success: false,
172                    output: None,
173                    error: Some(format!("step '{}' timed out", step.name)),
174                    duration_ms: elapsed,
175                };
176            }
177
178            std::thread::sleep(wait);
179            (true, None, None)
180        }
181
182        StepType::Transform { value, factor } => {
183            let result = value.saturating_mul(*factor);
184            (true, Some(result), None)
185        }
186    };
187
188    let duration_ms = start.elapsed().as_millis() as u64;
189    StepResult {
190        name: step.name.clone(),
191        success,
192        output,
193        error,
194        duration_ms,
195    }
196}
197
198// ---------------------------------------------------------------------------
199// ParallelSteps
200// ---------------------------------------------------------------------------
201
202/// A named group of steps that execute concurrently.
203pub struct ParallelSteps {
204    /// Group name (for logging and error messages).
205    pub name: String,
206    /// The steps in this group.
207    pub steps: Vec<WorkflowStep>,
208    /// When `true` (the default): return an error as soon as the first
209    /// failure is detected rather than waiting for all steps to finish.
210    ///
211    /// Note: because this implementation uses `std::thread::scope`, all
212    /// spawned threads must complete before the scope returns regardless
213    /// of `fail_fast`.  The flag controls only whether the *error type*
214    /// returned is [`ParallelStepError::StepFailed`] (single, fail-fast)
215    /// or [`ParallelStepError::MultipleStepsFailed`] (aggregated).
216    pub fail_fast: bool,
217}
218
219impl ParallelSteps {
220    /// Create a new group with `fail_fast = true` and no steps.
221    #[must_use]
222    pub fn new(name: impl Into<String>) -> Self {
223        Self {
224            name: name.into(),
225            steps: Vec::new(),
226            fail_fast: true,
227        }
228    }
229
230    /// Set the `fail_fast` flag.
231    #[must_use]
232    pub fn with_fail_fast(mut self, fail_fast: bool) -> Self {
233        self.fail_fast = fail_fast;
234        self
235    }
236
237    /// Append a step to the group.
238    pub fn add_step(&mut self, step: WorkflowStep) -> &mut Self {
239        self.steps.push(step);
240        self
241    }
242
243    /// Execute all steps in parallel using `std::thread::scope`.
244    ///
245    /// All worker threads are joined before this function returns.
246    /// Results are collected in the same order as [`Self::steps`].
247    ///
248    /// # Errors
249    ///
250    /// Returns [`ParallelStepError::StepFailed`] if `fail_fast = true` and
251    /// one or more steps fail, or [`ParallelStepError::MultipleStepsFailed`]
252    /// if `fail_fast = false` and multiple steps fail.
253    pub fn execute_parallel(&self) -> Result<Vec<StepResult>, ParallelStepError> {
254        if self.steps.is_empty() {
255            return Ok(Vec::new());
256        }
257
258        // Collect results using thread::scope so all threads are joined before
259        // we return.  We use a Mutex-protected Vec to gather results.
260        let results: std::sync::Mutex<Vec<(usize, StepResult)>> =
261            std::sync::Mutex::new(Vec::with_capacity(self.steps.len()));
262
263        std::thread::scope(|scope| {
264            for (idx, step) in self.steps.iter().enumerate() {
265                let results_ref = &results;
266                scope.spawn(move || {
267                    let result = execute_step(step);
268                    if let Ok(mut guard) = results_ref.lock() {
269                        guard.push((idx, result));
270                    }
271                });
272            }
273        });
274
275        // Reconstruct in original order.
276        let mut collected = results.into_inner().unwrap_or_default();
277        collected.sort_by_key(|(idx, _)| *idx);
278        let ordered: Vec<StepResult> = collected.into_iter().map(|(_, r)| r).collect();
279
280        // Check for failures.
281        let failed: Vec<&StepResult> = ordered.iter().filter(|r| !r.success).collect();
282
283        if failed.is_empty() {
284            return Ok(ordered);
285        }
286
287        if self.fail_fast {
288            // Return the first failure.
289            let first = &failed[0];
290            return Err(ParallelStepError::StepFailed {
291                name: first.name.clone(),
292                reason: first
293                    .error
294                    .clone()
295                    .unwrap_or_else(|| "unknown error".to_string()),
296            });
297        }
298
299        // Not fail-fast: return aggregated error.
300        Err(ParallelStepError::MultipleStepsFailed {
301            count: failed.len(),
302            total: ordered.len(),
303            results: ordered,
304        })
305    }
306}
307
308// ---------------------------------------------------------------------------
309// WorkflowStage
310// ---------------------------------------------------------------------------
311
312/// A logical stage within a workflow, containing steps that run either
313/// sequentially or in parallel.
314pub enum WorkflowStage {
315    /// Steps run one after another; fails on the first error.
316    Sequential(Vec<WorkflowStep>),
317    /// All steps start at the same time; results are joined.
318    Parallel(Vec<WorkflowStep>),
319}
320
321impl WorkflowStage {
322    /// Execute this stage and return all step results.
323    ///
324    /// # Errors
325    ///
326    /// Returns [`ParallelStepError`] if any step fails.
327    pub fn execute(&self) -> Result<Vec<StepResult>, ParallelStepError> {
328        match self {
329            Self::Sequential(steps) => {
330                let mut results = Vec::with_capacity(steps.len());
331                for step in steps {
332                    let r = execute_step(step);
333                    if !r.success {
334                        let reason = r
335                            .error
336                            .clone()
337                            .unwrap_or_else(|| "unknown error".to_string());
338                        let name = r.name.clone();
339                        results.push(r);
340                        return Err(ParallelStepError::StepFailed { name, reason });
341                    }
342                    results.push(r);
343                }
344                Ok(results)
345            }
346            Self::Parallel(steps) => {
347                let mut group = ParallelSteps::new("stage");
348                group.fail_fast = true;
349                for step in steps {
350                    group.steps.push(step.clone());
351                }
352                group.execute_parallel()
353            }
354        }
355    }
356
357    /// Return the number of steps in this stage.
358    #[must_use]
359    pub fn step_count(&self) -> usize {
360        match self {
361            Self::Sequential(steps) | Self::Parallel(steps) => steps.len(),
362        }
363    }
364}
365
366// ---------------------------------------------------------------------------
367// Tests
368// ---------------------------------------------------------------------------
369
370#[cfg(test)]
371mod tests {
372    use super::*;
373
374    // -----------------------------------------------------------------------
375    // ParallelSteps — success paths
376    // -----------------------------------------------------------------------
377
378    #[test]
379    fn parallel_all_success_returns_results() {
380        let mut group = ParallelSteps::new("grp");
381        group.add_step(WorkflowStep::new("a", StepType::Compute { value: 10 }));
382        group.add_step(WorkflowStep::new("b", StepType::Compute { value: 20 }));
383        group.add_step(WorkflowStep::new(
384            "c",
385            StepType::Transform {
386                value: 3,
387                factor: 7,
388            },
389        ));
390        let results = group.execute_parallel().expect("all should succeed");
391        assert_eq!(results.len(), 3);
392        assert_eq!(results[0].output, Some(10));
393        assert_eq!(results[1].output, Some(20));
394        assert_eq!(results[2].output, Some(21));
395    }
396
397    #[test]
398    fn parallel_empty_steps_returns_empty() {
399        let group = ParallelSteps::new("empty");
400        let results = group
401            .execute_parallel()
402            .expect("empty group should succeed");
403        assert!(results.is_empty());
404    }
405
406    #[test]
407    fn parallel_single_step_success() {
408        let mut group = ParallelSteps::new("single");
409        group.add_step(WorkflowStep::new("only", StepType::Compute { value: 42 }));
410        let results = group.execute_parallel().expect("should succeed");
411        assert_eq!(results[0].output, Some(42));
412        assert!(results[0].success);
413    }
414
415    // -----------------------------------------------------------------------
416    // ParallelSteps — failure paths
417    // -----------------------------------------------------------------------
418
419    #[test]
420    fn parallel_one_fail_fast_gives_step_failed_error() {
421        let mut group = ParallelSteps::new("fail-fast-grp");
422        group.fail_fast = true;
423        group.add_step(WorkflowStep::new("ok", StepType::Compute { value: 1 }));
424        group.add_step(WorkflowStep::new(
425            "bad",
426            StepType::Fail {
427                reason: "oops".to_string(),
428            },
429        ));
430        let err = group.execute_parallel().expect_err("should fail");
431        assert!(matches!(err, ParallelStepError::StepFailed { .. }));
432    }
433
434    #[test]
435    fn parallel_one_fail_not_fast_gives_multiple_failed() {
436        let mut group = ParallelSteps::new("no-ff");
437        group.fail_fast = false;
438        group.add_step(WorkflowStep::new("ok", StepType::Compute { value: 1 }));
439        group.add_step(WorkflowStep::new(
440            "bad",
441            StepType::Fail {
442                reason: "nope".to_string(),
443            },
444        ));
445        let err = group.execute_parallel().expect_err("should fail");
446        if let ParallelStepError::MultipleStepsFailed { count, total, .. } = err {
447            assert_eq!(count, 1);
448            assert_eq!(total, 2);
449        } else {
450            panic!("expected MultipleStepsFailed");
451        }
452    }
453
454    #[test]
455    fn parallel_all_fail_no_fast_aggregates() {
456        let mut group = ParallelSteps::new("all-fail");
457        group.fail_fast = false;
458        group.add_step(WorkflowStep::new(
459            "a",
460            StepType::Fail {
461                reason: "err-a".to_string(),
462            },
463        ));
464        group.add_step(WorkflowStep::new(
465            "b",
466            StepType::Fail {
467                reason: "err-b".to_string(),
468            },
469        ));
470        let err = group.execute_parallel().expect_err("should fail");
471        if let ParallelStepError::MultipleStepsFailed { count, total, .. } = err {
472            assert_eq!(count, 2);
473            assert_eq!(total, 2);
474        } else {
475            panic!("expected MultipleStepsFailed");
476        }
477    }
478
479    // -----------------------------------------------------------------------
480    // execute_step — direct tests
481    // -----------------------------------------------------------------------
482
483    #[test]
484    fn execute_compute_step() {
485        let step = WorkflowStep::new("s", StepType::Compute { value: -7 });
486        let r = execute_step(&step);
487        assert!(r.success);
488        assert_eq!(r.output, Some(-7));
489        assert!(r.error.is_none());
490    }
491
492    #[test]
493    fn execute_fail_step() {
494        let step = WorkflowStep::new(
495            "s",
496            StepType::Fail {
497                reason: "bad".to_string(),
498            },
499        );
500        let r = execute_step(&step);
501        assert!(!r.success);
502        assert_eq!(r.error.as_deref(), Some("bad"));
503    }
504
505    #[test]
506    fn execute_transform_step() {
507        let step = WorkflowStep::new(
508            "s",
509            StepType::Transform {
510                value: 6,
511                factor: 9,
512            },
513        );
514        let r = execute_step(&step);
515        assert!(r.success);
516        assert_eq!(r.output, Some(54));
517    }
518
519    #[test]
520    fn execute_wait_step_timeout() {
521        // A Wait step whose duration exceeds the per-step timeout should fail.
522        let step = WorkflowStep::new("s", StepType::Wait { duration_ms: 500 }).with_timeout(10);
523        let r = execute_step(&step);
524        assert!(!r.success, "should time out");
525        assert!(r.error.is_some());
526    }
527
528    // -----------------------------------------------------------------------
529    // WorkflowStage
530    // -----------------------------------------------------------------------
531
532    #[test]
533    fn sequential_stage_executes_in_order() {
534        let steps = vec![
535            WorkflowStep::new("first", StepType::Compute { value: 1 }),
536            WorkflowStep::new("second", StepType::Compute { value: 2 }),
537            WorkflowStep::new("third", StepType::Compute { value: 3 }),
538        ];
539        let stage = WorkflowStage::Sequential(steps);
540        let results = stage.execute().expect("should succeed");
541        assert_eq!(results.len(), 3);
542        assert_eq!(results[0].name, "first");
543        assert_eq!(results[1].name, "second");
544        assert_eq!(results[2].name, "third");
545    }
546
547    #[test]
548    fn sequential_stage_stops_on_first_failure() {
549        let steps = vec![
550            WorkflowStep::new("ok", StepType::Compute { value: 1 }),
551            WorkflowStep::new(
552                "bad",
553                StepType::Fail {
554                    reason: "stop".to_string(),
555                },
556            ),
557            WorkflowStep::new("never", StepType::Compute { value: 99 }),
558        ];
559        let stage = WorkflowStage::Sequential(steps);
560        let err = stage.execute().expect_err("should fail");
561        assert!(matches!(err, ParallelStepError::StepFailed { .. }));
562    }
563
564    #[test]
565    fn parallel_stage_all_success() {
566        let steps = vec![
567            WorkflowStep::new("p1", StepType::Compute { value: 10 }),
568            WorkflowStep::new(
569                "p2",
570                StepType::Transform {
571                    value: 3,
572                    factor: 4,
573                },
574            ),
575        ];
576        let stage = WorkflowStage::Parallel(steps);
577        let results = stage.execute().expect("should succeed");
578        assert_eq!(results.len(), 2);
579    }
580
581    #[test]
582    fn stage_step_count() {
583        let seq = WorkflowStage::Sequential(vec![
584            WorkflowStep::new("a", StepType::Compute { value: 0 }),
585            WorkflowStep::new("b", StepType::Compute { value: 0 }),
586        ]);
587        assert_eq!(seq.step_count(), 2);
588
589        let par =
590            WorkflowStage::Parallel(vec![WorkflowStep::new("x", StepType::Compute { value: 0 })]);
591        assert_eq!(par.step_count(), 1);
592    }
593
594    #[test]
595    fn parallel_and_sequential_same_results_all_success() {
596        let steps = vec![
597            WorkflowStep::new("a", StepType::Compute { value: 5 }),
598            WorkflowStep::new("b", StepType::Compute { value: 10 }),
599        ];
600        let seq_results = WorkflowStage::Sequential(steps.clone())
601            .execute()
602            .expect("seq ok");
603        let par_results = WorkflowStage::Parallel(steps).execute().expect("par ok");
604
605        let seq_outputs: Vec<Option<i64>> = seq_results.iter().map(|r| r.output).collect();
606        let par_outputs: Vec<Option<i64>> = par_results.iter().map(|r| r.output).collect();
607        assert_eq!(seq_outputs, par_outputs);
608    }
609}
oximedia_workflow/parallel_steps.rs

oximedia_workflow/
parallel_steps.rs
