ironflow_engine/context/steps/
delay.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, to_value};
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::context::lifecycle::check_replay_identity;
15use crate::error::EngineError;
16use crate::plan::lock_plan;
17
18impl WorkflowContext {
19    /// Execute a delay (timed pause) step.
20    ///
21    /// A zero-duration delay completes immediately. Otherwise, the
22    /// delay step is marked completed and the method returns
23    /// [`EngineError::DelaySleeping`] so the engine transitions the
24    /// run to [`Sleeping`](ironflow_store::entities::RunStatus::Sleeping).
25    ///
26    /// On resume (after the worker picks up the re-queued run), the
27    /// delay step is replayed as completed via the replay mechanism.
28    ///
29    /// # Errors
30    ///
31    /// Returns [`EngineError::DelaySleeping`] to suspend the run. Returns
32    /// [`EngineError::ReplayDivergence`] when the step recorded at this
33    /// position has a different name or kind.
34    ///
35    /// # Examples
36    ///
37    /// ```no_run
38    /// use ironflow_engine::context::WorkflowContext;
39    /// use ironflow_engine::config::delay::DelayConfig;
40    /// use ironflow_engine::error::EngineError;
41    ///
42    /// # async fn example(ctx: &mut WorkflowContext) -> Result<(), EngineError> {
43    /// ctx.delay("cooldown", DelayConfig::from_secs(300)).await?;
44    /// # Ok(())
45    /// # }
46    /// ```
47    pub async fn delay(&mut self, name: &str, config: DelayConfig) -> Result<(), EngineError> {
48        // Plan mode: record the pause without sleeping. The configured delay is
49        // the estimate when history has none -- a delay always lasts exactly
50        // as long as it was configured for.
51        if let Some(plan) = self.plan().cloned() {
52            self.position += 1;
53            let mut recorder = lock_plan(&plan);
54            recorder.seed_estimate(name, config.duration());
55            recorder.record_or_stop(
56                name,
57                StepKind::Custom("delay".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.next_position();
66
67        if let Some(existing) = self.replay_steps().get(&position) {
68            check_replay_identity(
69                existing,
70                position,
71                name,
72                &StepKind::Custom("delay".to_string()),
73            )?;
74
75            if existing.status.state == StepStatus::Completed {
76                self.set_last_step_ids(vec![existing.id]);
77                info!(
78                    run_id = %self.run_id(),
79                    step = %name,
80                    position,
81                    "delay step replayed (already completed)"
82                );
83                return Ok(());
84            }
85        }
86
87        let trace_id = step_trace_id(self.run_id(), name, position);
88        let step = self
89            .store
90            .create_step(NewStep {
91                run_id: self.run_id(),
92                trace_id,
93                name: name.to_string(),
94                kind: StepKind::Custom("delay".to_string()),
95                position,
96                input: Some(to_value(&config)?),
97                is_error_handler: false,
98            })
99            .await?;
100
101        let now = Utc::now();
102        self.start_step(step.id, now).await?;
103
104        if config.is_zero() {
105            self.store
106                .update_step(
107                    step.id,
108                    StepUpdate {
109                        status: Some(StepStatus::Completed),
110                        completed_at: Some(now),
111                        ..StepUpdate::default()
112                    },
113                )
114                .await?;
115            self.set_last_step_ids(vec![step.id]);
116            info!(run_id = %self.run_id(), step = %name, "delay(0) completed immediately");
117            return Ok(());
118        }
119
120        let wake_at = now + Duration::seconds(config.duration_secs() as i64);
121
122        self.store
123            .update_step(
124                step.id,
125                StepUpdate {
126                    status: Some(StepStatus::Completed),
127                    output: Some(json!({"wake_at": wake_at.to_rfc3339()})),
128                    completed_at: Some(Utc::now()),
129                    ..StepUpdate::default()
130                },
131            )
132            .await?;
133
134        self.set_last_step_ids(vec![step.id]);
135
136        Err(EngineError::DelaySleeping {
137            run_id: self.run_id(),
138            step_id: step.id,
139            wake_at,
140        })
141    }
142}
ironflow_engine/context/steps/delay.rs

ironflow_engine/context/steps/
delay.rs
