ironflow_engine/context/steps/
skip.rs

1//! Explicitly skipped step for [`WorkflowContext`].
2
3use chrono::Utc;
4use serde_json::json;
5use tracing::info;
6
7use ironflow_store::models::{
8    NewStep, NewStepDependency, StepKind, StepStatus, StepUpdate, step_trace_id,
9};
10
11use crate::context::WorkflowContext;
12use crate::error::EngineError;
13use crate::plan::{ConditionResult, lock_plan};
14
15impl WorkflowContext {
16    /// Record a step as explicitly skipped.
17    ///
18    /// Use this inside an `if`/`else` branch when a step should not execute
19    /// but must still appear in the DAG and timeline with its reason.
20    ///
21    /// The step is created directly in [`StepStatus::Skipped`] state and the
22    /// reason is stored in the output as `{"reason": "..."}`.
23    ///
24    /// On resume, a skip already recorded in a prior execution of the current
25    /// attempt is replayed instead of creating a second `Skipped` step.
26    ///
27    /// # Errors
28    ///
29    /// Returns [`EngineError`] if the store fails.
30    ///
31    /// # Examples
32    ///
33    /// ```no_run
34    /// use ironflow_engine::context::WorkflowContext;
35    /// use ironflow_engine::error::EngineError;
36    ///
37    /// # async fn example(ctx: &mut WorkflowContext) -> Result<(), EngineError> {
38    /// let tests_passed = false;
39    /// if tests_passed {
40    ///     // ctx.shell("deploy", ...).await?;
41    /// } else {
42    ///     ctx.skip("deploy", "tests failed").await?;
43    /// }
44    /// # Ok(())
45    /// # }
46    /// ```
47    pub async fn skip(&mut self, name: &str, reason: &str) -> Result<(), EngineError> {
48        // Plan mode: the skip and its reason become the step's condition.
49        if let Some(plan) = self.plan().cloned() {
50            self.position += 1;
51            let mut recorder = lock_plan(&plan);
52            recorder.set_condition(ConditionResult::Skipped {
53                reason: reason.to_string(),
54            });
55            recorder.record_or_stop(
56                name,
57                StepKind::Custom("skip".to_string()),
58                &self.workflow_name,
59                None,
60            )?;
61            recorder.set_last(vec![name.to_string()]);
62            return Ok(());
63        }
64
65        let position = self.position;
66        self.position += 1;
67
68        if let Some(existing) = self.replay_steps.get(&position)
69            && existing.name == name
70            && existing.kind == StepKind::Custom("skip".to_string())
71            && existing.status.state == StepStatus::Skipped
72        {
73            self.last_step_ids = vec![existing.id];
74            info!(
75                run_id = %self.run_id,
76                step = %name,
77                position,
78                "step replayed from previous execution"
79            );
80            return Ok(());
81        }
82
83        let trace_id = step_trace_id(self.run_id, name, position);
84        let step = self
85            .store
86            .create_step(NewStep {
87                run_id: self.run_id,
88                trace_id,
89                name: name.to_string(),
90                kind: StepKind::Custom("skip".to_string()),
91                position,
92                input: None,
93                is_error_handler: false,
94            })
95            .await?;
96
97        if !self.last_step_ids.is_empty() {
98            let deps: Vec<NewStepDependency> = self
99                .last_step_ids
100                .iter()
101                .map(|&depends_on| NewStepDependency {
102                    step_id: step.id,
103                    depends_on,
104                })
105                .collect();
106            self.store.create_step_dependencies(deps).await?;
107        }
108
109        let now = Utc::now();
110        self.store
111            .update_step(
112                step.id,
113                StepUpdate {
114                    status: Some(StepStatus::Skipped),
115                    output: Some(json!({"reason": reason})),
116                    completed_at: Some(now),
117                    ..StepUpdate::default()
118                },
119            )
120            .await?;
121
122        self.last_step_ids = vec![step.id];
123
124        info!(
125            run_id = %self.run_id,
126            step = %name,
127            reason,
128            "step skipped"
129        );
130
131        Ok(())
132    }
133}
ironflow_engine/context/steps/skip.rs

ironflow_engine/context/steps/
skip.rs
