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

a3s_flow/model/
command.rs
