a3s_flow/model/
command.rs

1use chrono::{DateTime, Utc};
2use serde::{Deserialize, Serialize};
3use serde_json::Value;
4use std::time::Duration;
5
6use crate::error::{FlowError, Result};
7
8/// JSON payload exchanged between the engine and runtimes.
9pub type JsonValue = Value;
10
11/// Runtime family used to execute workflow code.
12#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
13#[serde(rename_all = "snake_case")]
14pub enum RuntimeKind {
15    /// TypeScript compiled to a native executable through a native toolchain.
16    NativeTs,
17    /// Host-provided Rust runtime. Useful for tests and embedded deployments.
18    RustEmbedded,
19}
20
21/// Runtime metadata stored with a run so replay can happen on another process.
22#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
23pub struct RuntimeSpec {
24    pub kind: RuntimeKind,
25    pub entrypoint: String,
26    pub export_name: String,
27}
28
29impl RuntimeSpec {
30    pub fn native_ts(entrypoint: impl Into<String>, export_name: impl Into<String>) -> Self {
31        Self {
32            kind: RuntimeKind::NativeTs,
33            entrypoint: entrypoint.into(),
34            export_name: export_name.into(),
35        }
36    }
37
38    pub fn rust_embedded(entrypoint: impl Into<String>, export_name: impl Into<String>) -> Self {
39        Self {
40            kind: RuntimeKind::RustEmbedded,
41            entrypoint: entrypoint.into(),
42            export_name: export_name.into(),
43        }
44    }
45}
46
47/// Durable workflow definition.
48#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
49pub struct WorkflowSpec {
50    pub name: String,
51    pub version: String,
52    pub runtime: RuntimeSpec,
53}
54
55impl WorkflowSpec {
56    pub fn native_ts(
57        name: impl Into<String>,
58        version: impl Into<String>,
59        entrypoint: impl Into<String>,
60        export_name: impl Into<String>,
61    ) -> Self {
62        Self {
63            name: name.into(),
64            version: version.into(),
65            runtime: RuntimeSpec::native_ts(entrypoint, export_name),
66        }
67    }
68
69    pub fn rust_embedded(
70        name: impl Into<String>,
71        version: impl Into<String>,
72        entrypoint: impl Into<String>,
73        export_name: impl Into<String>,
74    ) -> Self {
75        Self {
76            name: name.into(),
77            version: version.into(),
78            runtime: RuntimeSpec::rust_embedded(entrypoint, export_name),
79        }
80    }
81
82    pub fn validate(&self) -> Result<()> {
83        if self.name.trim().is_empty() {
84            return Err(FlowError::InvalidWorkflow(
85                "workflow name must not be empty".to_string(),
86            ));
87        }
88        if self.version.trim().is_empty() {
89            return Err(FlowError::InvalidWorkflow(
90                "workflow version must not be empty".to_string(),
91            ));
92        }
93        if self.runtime.entrypoint.trim().is_empty() {
94            return Err(FlowError::InvalidWorkflow(
95                "runtime entrypoint must not be empty".to_string(),
96            ));
97        }
98        if self.runtime.export_name.trim().is_empty() {
99            return Err(FlowError::InvalidWorkflow(
100                "runtime export_name must not be empty".to_string(),
101            ));
102        }
103        Ok(())
104    }
105}
106
107/// What the engine should do after a step exhausts its retry attempts.
108#[derive(Debug, Clone, Copy, Default, Serialize, Deserialize, PartialEq, Eq)]
109#[serde(rename_all = "snake_case")]
110pub enum StepFailureAction {
111    /// Record `step_failed`, then fail the workflow run.
112    #[default]
113    FailRun,
114    /// Record `step_failed`, then replay the workflow so it can choose a
115    /// fallback, compensation, or explicit failure command.
116    ContinueWorkflow,
117}
118
119impl StepFailureAction {
120    pub fn is_fail_run(&self) -> bool {
121        matches!(self, Self::FailRun)
122    }
123}
124
125/// Retry behavior for a step command.
126#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq)]
127pub struct RetryPolicy {
128    pub max_attempts: u32,
129    pub delay_ms: u64,
130    #[serde(default, skip_serializing_if = "StepFailureAction::is_fail_run")]
131    pub on_exhausted: StepFailureAction,
132}
133
134impl RetryPolicy {
135    pub fn none() -> Self {
136        Self {
137            max_attempts: 1,
138            delay_ms: 0,
139            on_exhausted: StepFailureAction::FailRun,
140        }
141    }
142
143    pub fn fixed(max_attempts: u32, delay: Duration) -> Self {
144        Self {
145            max_attempts: max_attempts.max(1),
146            delay_ms: delay.as_millis().min(u128::from(u64::MAX)) as u64,
147            on_exhausted: StepFailureAction::FailRun,
148        }
149    }
150
151    pub fn with_failure_action(mut self, action: StepFailureAction) -> Self {
152        self.on_exhausted = action;
153        self
154    }
155
156    pub fn continue_workflow_on_failure(self) -> Self {
157        self.with_failure_action(StepFailureAction::ContinueWorkflow)
158    }
159}
160
161impl Default for RetryPolicy {
162    fn default() -> Self {
163        Self {
164            max_attempts: 3,
165            delay_ms: 0,
166            on_exhausted: StepFailureAction::FailRun,
167        }
168    }
169}
170
171/// Command emitted by the workflow runtime after replay.
172#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
173#[serde(tag = "type", rename_all = "snake_case")]
174pub enum RuntimeCommand {
175    Complete {
176        output: JsonValue,
177    },
178    Fail {
179        error: String,
180    },
181    ScheduleStep {
182        step_id: String,
183        step_name: String,
184        input: JsonValue,
185        #[serde(default)]
186        retry: RetryPolicy,
187    },
188    ScheduleSteps {
189        steps: Vec<StepCommand>,
190    },
191    WaitUntil {
192        wait_id: String,
193        resume_at: DateTime<Utc>,
194    },
195    CreateHook {
196        hook_id: String,
197        token: String,
198        #[serde(default)]
199        metadata: JsonValue,
200    },
201}
202
203/// Step definition returned by workflow replay.
204#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
205pub struct StepCommand {
206    pub step_id: String,
207    pub step_name: String,
208    pub input: JsonValue,
209    #[serde(default)]
210    pub retry: RetryPolicy,
211}
212
213impl StepCommand {
214    pub fn new(step_id: impl Into<String>, step_name: impl Into<String>, input: JsonValue) -> Self {
215        Self {
216            step_id: step_id.into(),
217            step_name: step_name.into(),
218            input,
219            retry: RetryPolicy::default(),
220        }
221    }
222
223    pub fn with_retry(mut self, retry: RetryPolicy) -> Self {
224        self.retry = retry;
225        self
226    }
227}
228
229impl RuntimeCommand {
230    pub fn schedule_step(
231        step_id: impl Into<String>,
232        step_name: impl Into<String>,
233        input: JsonValue,
234    ) -> Self {
235        Self::ScheduleStep {
236            step_id: step_id.into(),
237            step_name: step_name.into(),
238            input,
239            retry: RetryPolicy::default(),
240        }
241    }
242
243    pub fn schedule_steps(steps: Vec<StepCommand>) -> Self {
244        Self::ScheduleSteps { steps }
245    }
246}
a3s_flow/model/command.rs

a3s_flow/model/
command.rs
