wfe-core 1.10.0

Core traits, models, builder, and executor for the WFE workflow engine
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

use async_trait::async_trait;
use serde_json::json;

use crate::models::ExecutionResult;
use crate::traits::step::{StepBody, StepExecutionContext};

/// A container step that executes its children sequentially.
/// Completes when all children have finished.
#[derive(Default)]
pub struct SequenceStep;

#[async_trait]
impl StepBody for SequenceStep {
    async fn run(&mut self, context: &StepExecutionContext<'_>) -> crate::Result<ExecutionResult> {
        let mut scope = context.execution_pointer.scope.clone();
        scope.push(context.execution_pointer.id.clone());

        // If this container has declared children and they haven't been
        // spawned yet, branch to create execution pointers for all of them.
        let has_children = !context.step.children.is_empty();
        let children_spawned = context
            .workflow
            .execution_pointers
            .iter()
            .any(|p| p.scope == scope);

        if has_children && !children_spawned {
            Ok(ExecutionResult::branch(vec![json!(null)], None))
        } else if context.workflow.is_branch_complete(&scope) {
            Ok(ExecutionResult::next())
        } else {
            Ok(ExecutionResult::persist(json!({"children_active": true})))
        }
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::models::{ExecutionPointer, PointerStatus};
    use crate::primitives::test_helpers::*;

    #[tokio::test]
    async fn children_complete_proceeds() {
        let mut step = SequenceStep;
        let pointer = ExecutionPointer::new(0);
        let wf_step = default_step();

        let mut workflow = default_workflow();
        let mut child = ExecutionPointer::new(1);
        child.scope = vec![pointer.id.clone()];
        child.status = PointerStatus::Complete;
        workflow.execution_pointers.push(child);

        let ctx = make_context(&pointer, &wf_step, &workflow);
        let result = step.run(&ctx).await.unwrap();
        assert!(result.proceed);
    }

    #[tokio::test]
    async fn children_incomplete_persists() {
        let mut step = SequenceStep;
        let pointer = ExecutionPointer::new(0);
        let wf_step = default_step();

        let mut workflow = default_workflow();
        let mut child = ExecutionPointer::new(1);
        child.scope = vec![pointer.id.clone()];
        child.status = PointerStatus::Running;
        workflow.execution_pointers.push(child);

        let ctx = make_context(&pointer, &wf_step, &workflow);
        let result = step.run(&ctx).await.unwrap();
        assert!(!result.proceed);
        assert_eq!(
            result.persistence_data,
            Some(json!({"children_active": true}))
        );
    }

    #[tokio::test]
    async fn no_children_in_scope_proceeds() {
        let mut step = SequenceStep;
        let pointer = ExecutionPointer::new(0);
        let wf_step = default_step();
        let workflow = default_workflow();

        let ctx = make_context(&pointer, &wf_step, &workflow);
        let result = step.run(&ctx).await.unwrap();
        // No children in scope means is_branch_complete returns true (vacuously).
        assert!(result.proceed);
    }

    #[tokio::test]
    async fn spawns_children_when_defined() {
        let mut step = SequenceStep;
        let pointer = ExecutionPointer::new(0);
        let mut wf_step = default_step();
        wf_step.children = vec![1, 2];

        let workflow = default_workflow();

        let ctx = make_context(&pointer, &wf_step, &workflow);
        let result = step.run(&ctx).await.unwrap();
        assert!(!result.proceed);
        assert!(result.branch_values.is_some());
    }
}