a3s_flow/model/
command.rs

1use chrono::{DateTime, Duration as ChronoDuration, Utc};
2use serde::{Deserialize, Serialize};
3use serde_json::Value;
4use std::time::Duration;
5
6use crate::error::{FlowError, Result};
7use crate::runtime_build::RuntimeBuildId;
8
9use super::{ChildOperationReference, WorkflowProgress};
10
11/// JSON payload exchanged between the engine and runtimes.
12pub type JsonValue = Value;
13
14/// Runtime family used to execute workflow code.
15#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
16#[serde(rename_all = "snake_case")]
17pub enum RuntimeKind {
18    /// TypeScript compiled to a native executable through a native toolchain.
19    NativeTs,
20    /// Host-provided Rust runtime. Useful for tests and embedded deployments.
21    RustEmbedded,
22}
23
24/// Runtime metadata stored with a run so replay can happen on another process.
25#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
26pub struct RuntimeSpec {
27    pub kind: RuntimeKind,
28    pub entrypoint: String,
29    pub export_name: String,
30}
31
32impl RuntimeSpec {
33    pub fn native_ts(entrypoint: impl Into<String>, export_name: impl Into<String>) -> Self {
34        Self {
35            kind: RuntimeKind::NativeTs,
36            entrypoint: entrypoint.into(),
37            export_name: export_name.into(),
38        }
39    }
40
41    pub fn rust_embedded(entrypoint: impl Into<String>, export_name: impl Into<String>) -> Self {
42        Self {
43            kind: RuntimeKind::RustEmbedded,
44            entrypoint: entrypoint.into(),
45            export_name: export_name.into(),
46        }
47    }
48}
49
50/// Durable workflow definition.
51#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
52pub struct WorkflowSpec {
53    pub name: String,
54    pub version: String,
55    pub runtime: RuntimeSpec,
56    /// Exact deployed runtime build required to replay this run.
57    #[serde(default, skip_serializing_if = "Option::is_none")]
58    pub runtime_build_id: Option<RuntimeBuildId>,
59}
60
61impl WorkflowSpec {
62    pub fn native_ts(
63        name: impl Into<String>,
64        version: impl Into<String>,
65        entrypoint: impl Into<String>,
66        export_name: impl Into<String>,
67    ) -> Self {
68        Self {
69            name: name.into(),
70            version: version.into(),
71            runtime: RuntimeSpec::native_ts(entrypoint, export_name),
72            runtime_build_id: None,
73        }
74    }
75
76    pub fn rust_embedded(
77        name: impl Into<String>,
78        version: impl Into<String>,
79        entrypoint: impl Into<String>,
80        export_name: impl Into<String>,
81    ) -> Self {
82        Self {
83            name: name.into(),
84            version: version.into(),
85            runtime: RuntimeSpec::rust_embedded(entrypoint, export_name),
86            runtime_build_id: None,
87        }
88    }
89
90    /// Pin new runs to the exact runtime build that can replay them.
91    pub fn with_runtime_build(mut self, runtime_build_id: RuntimeBuildId) -> Self {
92        self.runtime_build_id = Some(runtime_build_id);
93        self
94    }
95
96    pub fn validate(&self) -> Result<()> {
97        if self.name.trim().is_empty() {
98            return Err(FlowError::InvalidWorkflow(
99                "workflow name must not be empty".to_string(),
100            ));
101        }
102        if self.version.trim().is_empty() {
103            return Err(FlowError::InvalidWorkflow(
104                "workflow version must not be empty".to_string(),
105            ));
106        }
107        if self.runtime.entrypoint.trim().is_empty() {
108            return Err(FlowError::InvalidWorkflow(
109                "runtime entrypoint must not be empty".to_string(),
110            ));
111        }
112        if self.runtime.export_name.trim().is_empty() {
113            return Err(FlowError::InvalidWorkflow(
114                "runtime export_name must not be empty".to_string(),
115            ));
116        }
117        Ok(())
118    }
119}
120
121/// What the engine should do after a step exhausts its retry attempts.
122#[derive(Debug, Clone, Copy, Default, Serialize, Deserialize, PartialEq, Eq)]
123#[serde(rename_all = "snake_case")]
124pub enum StepFailureAction {
125    /// Record `step_failed`, then fail the workflow run.
126    #[default]
127    FailRun,
128    /// Record `step_failed`, then replay the workflow so it can choose a
129    /// fallback, compensation, or explicit failure command.
130    ContinueWorkflow,
131}
132
133impl StepFailureAction {
134    pub fn is_fail_run(&self) -> bool {
135        matches!(self, Self::FailRun)
136    }
137}
138
139/// Retry behavior for a step command.
140#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq)]
141pub struct RetryPolicy {
142    pub max_attempts: u32,
143    pub delay_ms: u64,
144    #[serde(default, skip_serializing_if = "StepFailureAction::is_fail_run")]
145    pub on_exhausted: StepFailureAction,
146}
147
148impl RetryPolicy {
149    pub fn none() -> Self {
150        Self {
151            max_attempts: 1,
152            delay_ms: 0,
153            on_exhausted: StepFailureAction::FailRun,
154        }
155    }
156
157    pub fn fixed(max_attempts: u32, delay: Duration) -> Self {
158        Self {
159            max_attempts: max_attempts.max(1),
160            delay_ms: delay.as_millis().min(u128::from(u64::MAX)) as u64,
161            on_exhausted: StepFailureAction::FailRun,
162        }
163    }
164
165    pub fn with_failure_action(mut self, action: StepFailureAction) -> Self {
166        self.on_exhausted = action;
167        self
168    }
169
170    pub fn continue_workflow_on_failure(self) -> Self {
171        self.with_failure_action(StepFailureAction::ContinueWorkflow)
172    }
173
174    pub(crate) fn retry_after(self, now: DateTime<Utc>) -> Result<Option<DateTime<Utc>>> {
175        if self.delay_ms == 0 {
176            return Ok(None);
177        }
178        let delay_ms = i64::try_from(self.delay_ms).map_err(|_| self.invalid_delay_error())?;
179        let delay =
180            ChronoDuration::try_milliseconds(delay_ms).ok_or_else(|| self.invalid_delay_error())?;
181        now.checked_add_signed(delay)
182            .map(Some)
183            .ok_or_else(|| self.invalid_delay_error())
184    }
185
186    fn invalid_delay_error(self) -> FlowError {
187        FlowError::InvalidTransition(format!(
188            "retry delay {}ms cannot be represented as a UTC deadline",
189            self.delay_ms
190        ))
191    }
192}
193
194impl Default for RetryPolicy {
195    fn default() -> Self {
196        Self {
197            max_attempts: 3,
198            delay_ms: 0,
199            on_exhausted: StepFailureAction::FailRun,
200        }
201    }
202}
203
204/// Command emitted by the workflow runtime after replay.
205#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
206#[serde(tag = "type", rename_all = "snake_case")]
207pub enum RuntimeCommand {
208    Complete {
209        output: JsonValue,
210    },
211    Fail {
212        error: String,
213    },
214    /// Finish a previously requested cleanup-aware cancellation.
215    Cancel,
216    /// Finish a run with a typed timeout outcome.
217    Timeout {
218        deadline: DateTime<Utc>,
219        #[serde(default, skip_serializing_if = "Option::is_none")]
220        reason: Option<String>,
221    },
222    /// Persist progress before replaying the workflow.
223    RecordProgress {
224        progress: WorkflowProgress,
225    },
226    /// Persist a parent-to-child operation reference before replaying.
227    LinkChildOperation {
228        child: ChildOperationReference,
229    },
230    ScheduleStep {
231        step_id: String,
232        step_name: String,
233        input: JsonValue,
234        #[serde(default)]
235        retry: RetryPolicy,
236    },
237    ScheduleSteps {
238        steps: Vec<StepCommand>,
239    },
240    WaitUntil {
241        wait_id: String,
242        resume_at: DateTime<Utc>,
243    },
244    CreateHook {
245        hook_id: String,
246        token: String,
247        #[serde(default)]
248        metadata: JsonValue,
249    },
250}
251
252/// Step definition returned by workflow replay.
253#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
254pub struct StepCommand {
255    pub step_id: String,
256    pub step_name: String,
257    pub input: JsonValue,
258    #[serde(default)]
259    pub retry: RetryPolicy,
260}
261
262impl StepCommand {
263    pub fn new(step_id: impl Into<String>, step_name: impl Into<String>, input: JsonValue) -> Self {
264        Self {
265            step_id: step_id.into(),
266            step_name: step_name.into(),
267            input,
268            retry: RetryPolicy::default(),
269        }
270    }
271
272    pub fn with_retry(mut self, retry: RetryPolicy) -> Self {
273        self.retry = retry;
274        self
275    }
276}
277
278impl RuntimeCommand {
279    pub fn schedule_step(
280        step_id: impl Into<String>,
281        step_name: impl Into<String>,
282        input: JsonValue,
283    ) -> Self {
284        Self::ScheduleStep {
285            step_id: step_id.into(),
286            step_name: step_name.into(),
287            input,
288            retry: RetryPolicy::default(),
289        }
290    }
291
292    pub fn schedule_steps(steps: Vec<StepCommand>) -> Self {
293        Self::ScheduleSteps { steps }
294    }
295}
a3s_flow/model/command.rs

a3s_flow/model/
command.rs
