a3s_flow/engine/
runs.rs

1use std::collections::BTreeSet;
2
3use uuid::Uuid;
4
5use crate::error::{FlowError, Result};
6use crate::model::{project_run, validate_run_id, FlowEvent, WorkflowRunStatus, WorkflowSpec};
7
8use super::validation::{ensure_same_start, is_event_conflict};
9use super::FlowEngine;
10
11impl FlowEngine {
12    /// Start a workflow run and drive it until completion or suspension.
13    pub async fn start(&self, spec: WorkflowSpec, input: serde_json::Value) -> Result<String> {
14        let run_id = Uuid::new_v4().to_string();
15        self.start_with_id(run_id, spec, input).await
16    }
17
18    /// Start a workflow run using a caller-provided durable run id.
19    ///
20    /// Reusing the same `run_id` with the same workflow spec and input is
21    /// idempotent. A fully terminal execution is acknowledged without runtime
22    /// build admission; an active leaf still requires its pinned build before
23    /// replay. Reusing the id with different spec or input returns a conflict.
24    pub async fn start_with_id(
25        &self,
26        run_id: impl Into<String>,
27        spec: WorkflowSpec,
28        input: serde_json::Value,
29    ) -> Result<String> {
30        let run_id = run_id.into();
31        self.ensure_run_started(&run_id, &spec, &input).await?;
32        self.drive(&run_id).await?;
33        Ok(run_id)
34    }
35
36    pub(super) async fn terminate_run(&self, run_id: &str, event: FlowEvent) -> Result<()> {
37        self.terminate_run_with_context(run_id, event, 0, &BTreeSet::new())
38            .await
39    }
40
41    pub(super) async fn terminate_run_with_context(
42        &self,
43        run_id: &str,
44        event: FlowEvent,
45        child_depth: usize,
46        ancestry: &BTreeSet<String>,
47    ) -> Result<()> {
48        for _ in 0..self.max_replay_iterations {
49            let snapshot = self.ensure_continuation_leaf(run_id).await?;
50            if ancestry.contains(&snapshot.run_id) {
51                return Err(FlowError::ChildWorkflowCycle(snapshot.run_id));
52            }
53            if snapshot.status.is_terminal() {
54                return Ok(());
55            }
56            let mut active_ancestry = ancestry.clone();
57            active_ancestry.insert(snapshot.run_id.clone());
58            if matches!(
59                self.terminate_child_workflows(
60                    &snapshot,
61                    terminal_reason(&event),
62                    child_depth,
63                    &active_ancestry,
64                )
65                .await?,
66                super::child_workflows::ChildReconciliation::Replay
67            ) {
68                continue;
69            }
70            match self
71                .record_event_at(&snapshot.run_id, snapshot.last_sequence, event.clone())
72                .await
73            {
74                Ok(_) => return Ok(()),
75                Err(error) if is_event_conflict(&error) => continue,
76                Err(error) => return Err(error),
77            }
78        }
79        Err(FlowError::ReplayLimitExceeded(self.max_replay_iterations))
80    }
81
82    pub(super) async fn ensure_run_started(
83        &self,
84        run_id: &str,
85        spec: &WorkflowSpec,
86        input: &serde_json::Value,
87    ) -> Result<()> {
88        self.ensure_run_started_with_admission(run_id, spec, input, true)
89            .await
90    }
91
92    pub(super) async fn ensure_run_started_with_admission(
93        &self,
94        run_id: &str,
95        spec: &WorkflowSpec,
96        input: &serde_json::Value,
97        require_runtime_build: bool,
98    ) -> Result<()> {
99        spec.validate()?;
100        validate_run_id(run_id)?;
101
102        for _ in 0..self.max_replay_iterations {
103            match self.store.list(run_id).await {
104                Ok(history) => {
105                    let snapshot = project_run(run_id, &history)?;
106                    ensure_same_start(run_id, &snapshot, spec, input)?;
107                    if snapshot.status != WorkflowRunStatus::Pending {
108                        return Ok(());
109                    }
110                    // Starting a pending root enables replay, so keep this
111                    // lifecycle write behind the root's build admission.
112                    if require_runtime_build {
113                        self.ensure_runtime_build_available(run_id, spec)?;
114                    }
115                    match self
116                        .record_event_at(run_id, snapshot.last_sequence, FlowEvent::RunStarted)
117                        .await
118                    {
119                        Ok(_) => return Ok(()),
120                        Err(error) if is_event_conflict(&error) => continue,
121                        Err(error) => return Err(error),
122                    }
123                }
124                Err(FlowError::RunNotFound(_)) => {
125                    // New root authority must not be persisted by a worker
126                    // that cannot replay the requested build.
127                    if require_runtime_build {
128                        self.ensure_runtime_build_available(run_id, spec)?;
129                    }
130                    match self
131                        .record_event_at(
132                            run_id,
133                            0,
134                            FlowEvent::RunCreated {
135                                spec: spec.clone(),
136                                input: input.clone(),
137                            },
138                        )
139                        .await
140                    {
141                        Ok(_) => continue,
142                        Err(error) if is_event_conflict(&error) => continue,
143                        Err(error) => return Err(error),
144                    }
145                }
146                Err(error) => return Err(error),
147            }
148        }
149
150        Err(FlowError::ReplayLimitExceeded(self.max_replay_iterations))
151    }
152}
153
154fn terminal_reason(event: &FlowEvent) -> Option<String> {
155    match event {
156        FlowEvent::RunCancelled { reason }
157        | FlowEvent::RunTimedOut { reason, .. }
158        | FlowEvent::RunHostShutdown { reason } => reason.clone(),
159        FlowEvent::RunFailed { error } | FlowEvent::RunRetryExhausted { error, .. } => {
160            Some(error.clone())
161        }
162        _ => None,
163    }
164}
a3s_flow/engine/runs.rs

a3s_flow/engine/
runs.rs
