ironflow_engine/
control_flow.rs

1//! Control-flow step implementations for [`WorkflowContext`].
2//!
3//! Adds the [`delay`](WorkflowContext::delay) method for persistent
4//! timed pauses that survive server restarts.
5
6use chrono::{Duration, Utc};
7use serde_json::json;
8use tracing::info;
9
10use ironflow_store::models::{NewStep, StepKind, StepStatus, StepUpdate, step_trace_id};
11
12use crate::config::delay::DelayConfig;
13use crate::context::WorkflowContext;
14use crate::error::EngineError;
15
16impl WorkflowContext {
17    /// Execute a delay (timed pause) step.
18    ///
19    /// A zero-duration delay completes immediately. Otherwise, the
20    /// delay step is marked completed and the method returns
21    /// [`EngineError::DelaySleeping`] so the engine transitions the
22    /// run to [`Sleeping`](ironflow_store::entities::RunStatus::Sleeping).
23    ///
24    /// On resume (after the worker picks up the re-queued run), the
25    /// delay step is replayed as completed via the replay mechanism.
26    ///
27    /// # Errors
28    ///
29    /// Returns [`EngineError::DelaySleeping`] to suspend the run.
30    ///
31    /// # Examples
32    ///
33    /// ```no_run
34    /// use ironflow_engine::context::WorkflowContext;
35    /// use ironflow_engine::config::delay::DelayConfig;
36    /// use ironflow_engine::error::EngineError;
37    ///
38    /// # async fn example(ctx: &mut WorkflowContext) -> Result<(), EngineError> {
39    /// ctx.delay("cooldown", DelayConfig::from_secs(300)).await?;
40    /// # Ok(())
41    /// # }
42    /// ```
43    pub async fn delay(&mut self, name: &str, config: DelayConfig) -> Result<(), EngineError> {
44        let position = self.next_position();
45
46        if let Some(existing) = self.replay_steps().get(&position)
47            && existing.kind == StepKind::Custom("delay".to_string())
48            && existing.status.state == StepStatus::Completed
49        {
50            self.set_last_step_ids(vec![existing.id]);
51            info!(
52                run_id = %self.run_id(),
53                step = %name,
54                position,
55                "delay step replayed (already completed)"
56            );
57            return Ok(());
58        }
59
60        let trace_id = step_trace_id(self.run_id(), name, position);
61        let step = self
62            .store()
63            .create_step(NewStep {
64                run_id: self.run_id(),
65                trace_id,
66                name: name.to_string(),
67                kind: StepKind::Custom("delay".to_string()),
68                position,
69                input: Some(serde_json::to_value(&config)?),
70                is_error_handler: false,
71            })
72            .await?;
73
74        let now = Utc::now();
75        self.start_step(step.id, now).await?;
76
77        if config.is_zero() {
78            self.store()
79                .update_step(
80                    step.id,
81                    StepUpdate {
82                        status: Some(StepStatus::Completed),
83                        completed_at: Some(now),
84                        ..StepUpdate::default()
85                    },
86                )
87                .await?;
88            self.set_last_step_ids(vec![step.id]);
89            info!(run_id = %self.run_id(), step = %name, "delay(0) completed immediately");
90            return Ok(());
91        }
92
93        let wake_at = now + Duration::seconds(config.duration_secs() as i64);
94
95        self.store()
96            .update_step(
97                step.id,
98                StepUpdate {
99                    status: Some(StepStatus::Completed),
100                    output: Some(json!({"wake_at": wake_at.to_rfc3339()})),
101                    completed_at: Some(Utc::now()),
102                    ..StepUpdate::default()
103                },
104            )
105            .await?;
106
107        self.set_last_step_ids(vec![step.id]);
108
109        Err(EngineError::DelaySleeping {
110            run_id: self.run_id(),
111            step_id: step.id,
112            wake_at,
113        })
114    }
115}
ironflow_engine/control_flow.rs

ironflow_engine/
control_flow.rs
