a3s_flow/model/
command.rs

1use chrono::{DateTime, Duration as ChronoDuration, Utc};
2use serde::{Deserialize, Serialize};
3use serde_json::Value;
4use sha2::{Digest, Sha256};
5use std::collections::BTreeSet;
6use std::time::Duration;
7
8use crate::error::{FlowError, Result};
9use crate::runtime_build::RuntimeBuildId;
10
11use super::patch::deserialize_patch_markers;
12use super::{
13    ChildOperationReference, ChildWorkflowCancellationPolicy, WorkflowPatchId, WorkflowProgress,
14    MAX_WORKFLOW_PATCH_MARKERS,
15};
16
17/// JSON payload exchanged between the engine and runtimes.
18pub type JsonValue = Value;
19
20/// Runtime family used to execute workflow code.
21#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
22#[non_exhaustive]
23#[serde(rename_all = "snake_case")]
24pub enum RuntimeKind {
25    /// TypeScript compiled to a native executable through a native toolchain.
26    NativeTs,
27    /// Host-provided Rust runtime. Useful for tests and embedded deployments.
28    RustEmbedded,
29}
30
31/// Runtime metadata stored with a run so replay can happen on another process.
32#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
33#[non_exhaustive]
34pub struct RuntimeSpec {
35    /// Runtime family responsible for executing the entrypoint.
36    pub kind: RuntimeKind,
37    /// Runtime-specific module or executable entrypoint.
38    pub entrypoint: String,
39    /// Exported workflow function within the entrypoint.
40    pub export_name: String,
41}
42
43impl RuntimeSpec {
44    /// Creates metadata for a natively compiled TypeScript runtime.
45    pub fn native_ts(entrypoint: impl Into<String>, export_name: impl Into<String>) -> Self {
46        Self {
47            kind: RuntimeKind::NativeTs,
48            entrypoint: entrypoint.into(),
49            export_name: export_name.into(),
50        }
51    }
52
53    /// Creates metadata for a host-provided embedded Rust runtime.
54    pub fn rust_embedded(entrypoint: impl Into<String>, export_name: impl Into<String>) -> Self {
55        Self {
56            kind: RuntimeKind::RustEmbedded,
57            entrypoint: entrypoint.into(),
58            export_name: export_name.into(),
59        }
60    }
61}
62
63/// Durable workflow definition.
64#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
65#[non_exhaustive]
66pub struct WorkflowSpec {
67    /// Stable workflow type name used for registration and inspection.
68    pub name: String,
69    /// Application-defined workflow definition version.
70    pub version: String,
71    /// Runtime entrypoint used to replay the workflow.
72    pub runtime: RuntimeSpec,
73    /// Exact deployed runtime build required to replay this run.
74    #[serde(default, skip_serializing_if = "Option::is_none")]
75    pub runtime_build_id: Option<RuntimeBuildId>,
76    /// Replay-safe code changes enabled for this run at creation time.
77    #[serde(
78        default,
79        deserialize_with = "deserialize_patch_markers",
80        skip_serializing_if = "BTreeSet::is_empty"
81    )]
82    pub patch_markers: BTreeSet<WorkflowPatchId>,
83    /// Named asynchronous message contracts accepted by this workflow.
84    #[serde(default, skip_serializing_if = "BTreeSet::is_empty")]
85    pub signal_names: BTreeSet<String>,
86}
87
88impl WorkflowSpec {
89    /// Creates a workflow definition backed by native TypeScript.
90    pub fn native_ts(
91        name: impl Into<String>,
92        version: impl Into<String>,
93        entrypoint: impl Into<String>,
94        export_name: impl Into<String>,
95    ) -> Self {
96        Self {
97            name: name.into(),
98            version: version.into(),
99            runtime: RuntimeSpec::native_ts(entrypoint, export_name),
100            runtime_build_id: None,
101            patch_markers: BTreeSet::new(),
102            signal_names: BTreeSet::new(),
103        }
104    }
105
106    /// Creates a workflow definition backed by an embedded Rust runtime.
107    pub fn rust_embedded(
108        name: impl Into<String>,
109        version: impl Into<String>,
110        entrypoint: impl Into<String>,
111        export_name: impl Into<String>,
112    ) -> Self {
113        Self {
114            name: name.into(),
115            version: version.into(),
116            runtime: RuntimeSpec::rust_embedded(entrypoint, export_name),
117            runtime_build_id: None,
118            patch_markers: BTreeSet::new(),
119            signal_names: BTreeSet::new(),
120        }
121    }
122
123    /// Pin new runs to the exact runtime build that can replay them.
124    pub fn with_runtime_build(mut self, runtime_build_id: RuntimeBuildId) -> Self {
125        self.runtime_build_id = Some(runtime_build_id);
126        self
127    }
128
129    /// Enable a replay-safe code path for every run created from this spec.
130    ///
131    /// The complete marker set is persisted atomically inside `run_created`.
132    /// Reusing an existing run ID with a different set is rejected as a start
133    /// conflict instead of changing the behavior of an in-flight history.
134    pub fn with_patch_marker(mut self, patch_id: WorkflowPatchId) -> Self {
135        self.patch_markers.insert(patch_id);
136        self
137    }
138
139    /// Return whether this immutable run definition contains `patch_id`.
140    pub fn has_patch_marker(&self, patch_id: &str) -> bool {
141        self.patch_markers.contains(patch_id)
142    }
143
144    /// Declare one named asynchronous signal contract for this workflow.
145    pub fn with_signal(mut self, signal_name: impl Into<String>) -> Self {
146        self.signal_names.insert(signal_name.into());
147        self
148    }
149
150    /// Return whether this immutable workflow definition accepts `signal_name`.
151    pub fn accepts_signal(&self, signal_name: &str) -> bool {
152        self.signal_names.contains(signal_name)
153    }
154
155    /// Validates identifiers, runtime metadata, patch limits, and signal names.
156    pub fn validate(&self) -> Result<()> {
157        if self.name.trim().is_empty() {
158            return Err(FlowError::InvalidWorkflow(
159                "workflow name must not be empty".to_string(),
160            ));
161        }
162        if self.version.trim().is_empty() {
163            return Err(FlowError::InvalidWorkflow(
164                "workflow version must not be empty".to_string(),
165            ));
166        }
167        if self.runtime.entrypoint.trim().is_empty() {
168            return Err(FlowError::InvalidWorkflow(
169                "runtime entrypoint must not be empty".to_string(),
170            ));
171        }
172        if self.runtime.export_name.trim().is_empty() {
173            return Err(FlowError::InvalidWorkflow(
174                "runtime export_name must not be empty".to_string(),
175            ));
176        }
177        if self.patch_markers.len() > MAX_WORKFLOW_PATCH_MARKERS {
178            return Err(FlowError::InvalidWorkflow(format!(
179                "workflow patch marker count {} exceeds {MAX_WORKFLOW_PATCH_MARKERS}",
180                self.patch_markers.len()
181            )));
182        }
183        for signal_name in &self.signal_names {
184            if signal_name.trim().is_empty() {
185                return Err(FlowError::InvalidWorkflow(
186                    "workflow signal name must not be empty".to_string(),
187                ));
188            }
189        }
190        Ok(())
191    }
192}
193
194/// What the engine should do after a step exhausts its retry attempts.
195#[derive(Debug, Clone, Copy, Default, Serialize, Deserialize, PartialEq, Eq)]
196#[non_exhaustive]
197#[serde(rename_all = "snake_case")]
198pub enum StepFailureAction {
199    /// Record `step_failed`, then fail the workflow run.
200    #[default]
201    FailRun,
202    /// Record `step_failed`, then replay the workflow so it can choose a
203    /// fallback, compensation, or explicit failure command.
204    ContinueWorkflow,
205}
206
207impl StepFailureAction {
208    /// Returns `true` when retry exhaustion must fail the workflow run.
209    pub fn is_fail_run(&self) -> bool {
210        matches!(self, Self::FailRun)
211    }
212}
213
214/// Delay progression used between durable step attempts.
215#[derive(Debug, Clone, Copy, Default, Serialize, Deserialize, PartialEq, Eq)]
216#[non_exhaustive]
217#[serde(rename_all = "snake_case")]
218pub enum RetryBackoff {
219    /// Reuse the same delay after every failed attempt.
220    #[default]
221    Fixed,
222    /// Double the delay cap after each failed attempt and select a stable,
223    /// full-jitter delay from the run, step, and attempt identities.
224    Exponential,
225}
226
227impl RetryBackoff {
228    fn is_fixed(&self) -> bool {
229        matches!(self, Self::Fixed)
230    }
231}
232
233fn is_zero(value: &u64) -> bool {
234    *value == 0
235}
236
237/// Retry behavior for a step command.
238#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq)]
239#[non_exhaustive]
240pub struct RetryPolicy {
241    /// Maximum number of attempts, including the first execution.
242    pub max_attempts: u32,
243    /// Fixed delay or initial exponential delay in milliseconds.
244    pub delay_ms: u64,
245    /// Delay progression. Omitted fixed policies preserve the pre-v1 history
246    /// encoding exactly.
247    #[serde(default, skip_serializing_if = "RetryBackoff::is_fixed")]
248    pub backoff: RetryBackoff,
249    /// Maximum exponential delay in milliseconds. Zero is canonical for a
250    /// fixed policy.
251    #[serde(default, skip_serializing_if = "is_zero")]
252    pub max_delay_ms: u64,
253    /// Action taken after the final permitted attempt fails.
254    #[serde(default, skip_serializing_if = "StepFailureAction::is_fail_run")]
255    pub on_exhausted: StepFailureAction,
256}
257
258impl RetryPolicy {
259    /// Creates a policy that permits exactly one attempt.
260    pub fn none() -> Self {
261        Self {
262            max_attempts: 1,
263            delay_ms: 0,
264            backoff: RetryBackoff::Fixed,
265            max_delay_ms: 0,
266            on_exhausted: StepFailureAction::FailRun,
267        }
268    }
269
270    /// Creates a fixed-delay retry policy.
271    ///
272    /// `max_attempts` is clamped to at least one and delays larger than
273    /// [`u64::MAX`] milliseconds are saturated.
274    pub fn fixed(max_attempts: u32, delay: Duration) -> Self {
275        Self {
276            max_attempts: max_attempts.max(1),
277            delay_ms: delay.as_millis().min(u128::from(u64::MAX)) as u64,
278            backoff: RetryBackoff::Fixed,
279            max_delay_ms: 0,
280            on_exhausted: StepFailureAction::FailRun,
281        }
282    }
283
284    /// Creates a capped exponential policy with deterministic full jitter.
285    ///
286    /// The first delay is clamped to at least one millisecond, the maximum is
287    /// clamped to at least the first delay, and every delay is selected from
288    /// `1..=current_cap` using the immutable run, step, and failed-attempt
289    /// identities. Restarts therefore retain the same backoff decision without
290    /// coordinating random state.
291    pub fn exponential(
292        max_attempts: u32,
293        initial_delay: Duration,
294        maximum_delay: Duration,
295    ) -> Self {
296        let delay_ms = initial_delay.as_millis().min(u128::from(u64::MAX)).max(1) as u64;
297        let max_delay_ms =
298            (maximum_delay.as_millis().min(u128::from(u64::MAX)) as u64).max(delay_ms);
299        Self {
300            max_attempts: max_attempts.max(1),
301            delay_ms,
302            backoff: RetryBackoff::Exponential,
303            max_delay_ms,
304            on_exhausted: StepFailureAction::FailRun,
305        }
306    }
307
308    /// Sets the action taken after retry exhaustion.
309    pub fn with_failure_action(mut self, action: StepFailureAction) -> Self {
310        self.on_exhausted = action;
311        self
312    }
313
314    /// Configures replay to continue after the final step failure.
315    pub fn continue_workflow_on_failure(self) -> Self {
316        self.with_failure_action(StepFailureAction::ContinueWorkflow)
317    }
318
319    pub(crate) fn retry_after(self, now: DateTime<Utc>) -> Result<Option<DateTime<Utc>>> {
320        let delay_ms = self.maximum_delay_ms()?;
321        self.deadline_after(now, delay_ms)
322    }
323
324    pub(crate) fn retry_after_for_step(
325        self,
326        now: DateTime<Utc>,
327        failed_attempt: u32,
328        run_id: &str,
329        step_id: &str,
330    ) -> Result<Option<DateTime<Utc>>> {
331        let delay_ms = self.delay_for_step(failed_attempt, run_id, step_id)?;
332        self.deadline_after(now, delay_ms)
333    }
334
335    fn maximum_delay_ms(self) -> Result<u64> {
336        match self.backoff {
337            RetryBackoff::Fixed if self.max_delay_ms == 0 => Ok(self.delay_ms),
338            RetryBackoff::Fixed => Err(FlowError::InvalidTransition(
339                "fixed retry policy cannot define max_delay_ms".to_string(),
340            )),
341            RetryBackoff::Exponential
342                if self.delay_ms > 0 && self.max_delay_ms >= self.delay_ms =>
343            {
344                Ok(self.max_delay_ms)
345            }
346            RetryBackoff::Exponential => Err(FlowError::InvalidTransition(
347                "exponential retry delays must satisfy 1 <= delay_ms <= max_delay_ms".to_string(),
348            )),
349        }
350    }
351
352    fn delay_for_step(self, failed_attempt: u32, run_id: &str, step_id: &str) -> Result<u64> {
353        self.maximum_delay_ms()?;
354        match self.backoff {
355            RetryBackoff::Fixed => Ok(self.delay_ms),
356            RetryBackoff::Exponential => {
357                let exponent = failed_attempt.saturating_sub(1).min(63);
358                let multiplier = 1_u64.checked_shl(exponent).unwrap_or(u64::MAX);
359                let cap = self
360                    .delay_ms
361                    .saturating_mul(multiplier)
362                    .min(self.max_delay_ms);
363                Ok(deterministic_full_jitter(
364                    cap,
365                    run_id,
366                    step_id,
367                    failed_attempt,
368                ))
369            }
370        }
371    }
372
373    fn deadline_after(self, now: DateTime<Utc>, delay_ms: u64) -> Result<Option<DateTime<Utc>>> {
374        if delay_ms == 0 {
375            return Ok(None);
376        }
377        let delay_ms = i64::try_from(delay_ms).map_err(|_| self.invalid_delay_error(delay_ms))?;
378        let delay = ChronoDuration::try_milliseconds(delay_ms)
379            .ok_or_else(|| self.invalid_delay_error(delay_ms as u64))?;
380        now.checked_add_signed(delay)
381            .map(Some)
382            .ok_or_else(|| self.invalid_delay_error(delay_ms as u64))
383    }
384
385    fn invalid_delay_error(self, delay_ms: u64) -> FlowError {
386        FlowError::InvalidTransition(format!(
387            "retry delay {delay_ms}ms cannot be represented as a UTC deadline"
388        ))
389    }
390}
391
392fn deterministic_full_jitter(cap: u64, run_id: &str, step_id: &str, failed_attempt: u32) -> u64 {
393    debug_assert!(cap > 0);
394    let mut hasher = Sha256::new();
395    hasher.update(b"a3s-flow.retry-jitter.v1");
396    hash_retry_part(&mut hasher, run_id.as_bytes());
397    hash_retry_part(&mut hasher, step_id.as_bytes());
398    hasher.update(failed_attempt.to_be_bytes());
399    let digest = hasher.finalize();
400    let sample = digest
401        .iter()
402        .take(8)
403        .fold(0_u64, |value, byte| (value << 8) | u64::from(*byte));
404    1 + sample % cap
405}
406
407fn hash_retry_part(hasher: &mut Sha256, bytes: &[u8]) {
408    hasher.update((bytes.len() as u64).to_be_bytes());
409    hasher.update(bytes);
410}
411
412impl Default for RetryPolicy {
413    fn default() -> Self {
414        Self {
415            max_attempts: 3,
416            delay_ms: 0,
417            backoff: RetryBackoff::Fixed,
418            max_delay_ms: 0,
419            on_exhausted: StepFailureAction::FailRun,
420        }
421    }
422}
423
424#[cfg(test)]
425mod retry_policy_tests {
426    use super::*;
427
428    #[test]
429    fn exponential_delay_is_identity_stable_and_capped_per_attempt() {
430        let policy =
431            RetryPolicy::exponential(8, Duration::from_millis(100), Duration::from_millis(400));
432
433        for (attempt, cap) in [(1, 100), (2, 200), (3, 400), (20, 400)] {
434            let first = policy.delay_for_step(attempt, "run-1", "step-1").unwrap();
435            let replay = policy.delay_for_step(attempt, "run-1", "step-1").unwrap();
436            assert_eq!(first, replay);
437            assert!((1..=cap).contains(&first));
438        }
439
440        assert_ne!(
441            policy.delay_for_step(3, "run-1", "step-1").unwrap(),
442            policy.delay_for_step(3, "run-2", "step-1").unwrap()
443        );
444        assert_ne!(
445            policy.delay_for_step(3, "run-1", "step-1").unwrap(),
446            policy.delay_for_step(3, "run-1", "step-2").unwrap()
447        );
448    }
449
450    #[test]
451    fn exponential_constructor_clamps_to_a_valid_positive_range() {
452        assert_eq!(
453            RetryPolicy::exponential(0, Duration::ZERO, Duration::ZERO),
454            RetryPolicy {
455                max_attempts: 1,
456                delay_ms: 1,
457                backoff: RetryBackoff::Exponential,
458                max_delay_ms: 1,
459                on_exhausted: StepFailureAction::FailRun,
460            }
461        );
462    }
463}
464
465/// Maximum number of first-class child workflows in one durable batch.
466///
467/// Larger fan-outs must be split into replay-stable batches. This bounds the
468/// number of child workflow executions one parent drive may activate at once.
469pub const MAX_CHILD_WORKFLOW_BATCH_SIZE: usize = 64;
470
471/// First-class child workflow definition returned as part of a durable batch.
472#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
473#[non_exhaustive]
474pub struct ChildWorkflowCommand {
475    /// Replay-stable parent-local child identifier.
476    pub child_id: String,
477    /// Workflow definition used to create the child.
478    pub spec: WorkflowSpec,
479    /// Initial JSON input supplied to the child.
480    pub input: JsonValue,
481    /// Policy applied when the parent is cancelled or terminated.
482    #[serde(default)]
483    pub cancellation_policy: ChildWorkflowCancellationPolicy,
484}
485
486impl ChildWorkflowCommand {
487    /// Create a child definition with the default cancellation policy.
488    pub fn new(child_id: impl Into<String>, spec: WorkflowSpec, input: JsonValue) -> Self {
489        Self {
490            child_id: child_id.into(),
491            spec,
492            input,
493            cancellation_policy: ChildWorkflowCancellationPolicy::default(),
494        }
495    }
496
497    /// Replace the policy applied when the parent stops.
498    pub fn with_cancellation_policy(
499        mut self,
500        cancellation_policy: ChildWorkflowCancellationPolicy,
501    ) -> Self {
502        self.cancellation_policy = cancellation_policy;
503        self
504    }
505}
506
507/// Command emitted by the workflow runtime after replay.
508#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
509#[non_exhaustive]
510#[serde(tag = "type", rename_all = "snake_case")]
511pub enum RuntimeCommand {
512    /// Completes the workflow run successfully.
513    Complete {
514        /// Final JSON value returned by the workflow.
515        output: JsonValue,
516    },
517    /// Fails the workflow run with an application error.
518    Fail {
519        /// Human-readable failure description.
520        error: String,
521    },
522    /// Finish a previously requested cleanup-aware cancellation.
523    Cancel,
524    /// Finish a run with a typed timeout outcome.
525    Timeout {
526        /// UTC deadline that caused the timeout.
527        deadline: DateTime<Utc>,
528        /// Optional context for the timeout decision.
529        #[serde(default, skip_serializing_if = "Option::is_none")]
530        reason: Option<String>,
531    },
532    /// Close this history segment and start a successor with the same spec.
533    ContinueAsNew {
534        /// Initial JSON input for the successor run.
535        input: JsonValue,
536    },
537    /// Persist progress before replaying the workflow.
538    RecordProgress {
539        /// Progress value exposed through inspection and observation APIs.
540        progress: WorkflowProgress,
541    },
542    /// Persist a parent-to-child operation reference before replaying.
543    LinkChildOperation {
544        /// Stable reference to the externally managed child operation.
545        child: ChildOperationReference,
546    },
547    /// Start or await a first-class child workflow with a stable parent-local id.
548    StartChildWorkflow {
549        /// Replay-stable parent-local child identifier.
550        child_id: String,
551        /// Workflow definition used to create the child.
552        spec: WorkflowSpec,
553        /// Initial JSON input supplied to the child.
554        input: JsonValue,
555        /// Policy applied when the parent is cancelled or terminated.
556        #[serde(default)]
557        cancellation_policy: ChildWorkflowCancellationPolicy,
558    },
559    /// Schedules one durable step or awaits its recorded outcome.
560    ScheduleStep {
561        /// Replay-stable identity of the step.
562        step_id: String,
563        /// Registered step implementation name.
564        step_name: String,
565        /// JSON input supplied to the step.
566        input: JsonValue,
567        /// Retry behavior pinned when the step is created.
568        #[serde(default)]
569        retry: RetryPolicy,
570    },
571    /// Atomically schedules a batch of durable steps.
572    ScheduleSteps {
573        /// Step definitions in deterministic scheduling order.
574        steps: Vec<StepCommand>,
575    },
576    /// Suspends replay until a UTC deadline becomes ready.
577    WaitUntil {
578        /// Replay-stable identity of the timer wait.
579        wait_id: String,
580        /// UTC time at which the wait becomes ready.
581        resume_at: DateTime<Utc>,
582    },
583    /// Creates an externally completable hook.
584    CreateHook {
585        /// Replay-stable identity of the hook.
586        hook_id: String,
587        /// Secret bearer token required to deliver the hook.
588        token: String,
589        /// Application metadata persisted with the hook.
590        #[serde(default)]
591        metadata: JsonValue,
592    },
593    /// Suspend until the next unconsumed signal with `signal_name` is paired
594    /// with this stable wait identity.
595    WaitForSignal {
596        /// Replay-stable identity of the signal wait.
597        wait_id: String,
598        /// Declared signal contract accepted by the wait.
599        signal_name: String,
600    },
601    /// Request a bounded batch before any first-class child starts.
602    StartChildWorkflows {
603        /// Child definitions in deterministic request order.
604        children: Vec<ChildWorkflowCommand>,
605    },
606}
607
608/// Step definition returned by workflow replay.
609#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
610#[non_exhaustive]
611pub struct StepCommand {
612    /// Replay-stable identity of the step.
613    pub step_id: String,
614    /// Registered step implementation name.
615    pub step_name: String,
616    /// JSON input supplied to the step.
617    pub input: JsonValue,
618    /// Retry behavior pinned when the step is created.
619    #[serde(default)]
620    pub retry: RetryPolicy,
621}
622
623impl StepCommand {
624    /// Creates a step definition with the default retry policy.
625    pub fn new(step_id: impl Into<String>, step_name: impl Into<String>, input: JsonValue) -> Self {
626        Self {
627            step_id: step_id.into(),
628            step_name: step_name.into(),
629            input,
630            retry: RetryPolicy::default(),
631        }
632    }
633
634    /// Replaces the step's retry policy.
635    pub fn with_retry(mut self, retry: RetryPolicy) -> Self {
636        self.retry = retry;
637        self
638    }
639}
640
641impl RuntimeCommand {
642    /// Creates a single-step schedule command with the default retry policy.
643    pub fn schedule_step(
644        step_id: impl Into<String>,
645        step_name: impl Into<String>,
646        input: JsonValue,
647    ) -> Self {
648        Self::ScheduleStep {
649            step_id: step_id.into(),
650            step_name: step_name.into(),
651            input,
652            retry: RetryPolicy::default(),
653        }
654    }
655
656    /// Creates an atomic batch scheduling command.
657    pub fn schedule_steps(steps: Vec<StepCommand>) -> Self {
658        Self::ScheduleSteps { steps }
659    }
660
661    /// Create a bounded batch child-workflow command.
662    pub fn start_child_workflows(children: Vec<ChildWorkflowCommand>) -> Self {
663        Self::StartChildWorkflows { children }
664    }
665}
a3s_flow/model/command.rs

a3s_flow/model/
command.rs
