a3s_flow/workflow_dsl/
plan.rs

1use std::collections::{BTreeMap, BTreeSet};
2
3use super::{WorkflowDag, WorkflowDslError};
4
5const MAX_WORKFLOW_DAG_NODES: usize = 10_000;
6const MAX_WORKFLOW_DAG_EDGES: usize = 100_000;
7
8/// Deterministic topological node order grouped by container scope.
9#[derive(Debug, Clone, PartialEq, Eq)]
10pub struct WorkflowDagPlan {
11    scopes: BTreeMap<Option<String>, Vec<String>>,
12}
13
14impl WorkflowDagPlan {
15    /// Returns deterministic execution order for top-level nodes.
16    pub fn top_level(&self) -> &[String] {
17        self.scopes.get(&None).map(Vec::as_slice).unwrap_or(&[])
18    }
19
20    /// Returns deterministic execution order inside one container.
21    pub fn scope(&self, parent_id: &str) -> Option<&[String]> {
22        self.scopes
23            .get(&Some(parent_id.to_owned()))
24            .map(Vec::as_slice)
25    }
26
27    /// Returns all top-level and container-scoped execution orders.
28    pub fn scopes(&self) -> &BTreeMap<Option<String>, Vec<String>> {
29        &self.scopes
30    }
31}
32
33pub(super) fn build_execution_plan(
34    graph: &WorkflowDag,
35) -> Result<WorkflowDagPlan, WorkflowDslError> {
36    if graph.nodes().is_empty() {
37        return Err(invalid_graph(
38            "an executable graph requires at least one node",
39        ));
40    }
41    if graph.nodes().len() > MAX_WORKFLOW_DAG_NODES {
42        return Err(invalid_graph(format!(
43            "node count {} exceeds {MAX_WORKFLOW_DAG_NODES}",
44            graph.nodes().len()
45        )));
46    }
47    if graph.edges().len() > MAX_WORKFLOW_DAG_EDGES {
48        return Err(invalid_graph(format!(
49            "edge count {} exceeds {MAX_WORKFLOW_DAG_EDGES}",
50            graph.edges().len()
51        )));
52    }
53
54    let mut nodes = BTreeMap::new();
55    let mut scope_nodes: BTreeMap<Option<String>, BTreeSet<String>> = BTreeMap::new();
56    for node in graph.nodes() {
57        if node.id().trim().is_empty() {
58            return Err(invalid_graph("node ID is empty"));
59        }
60        if node.node_type().trim().is_empty() {
61            return Err(invalid_graph(format!(
62                "node {:?} has no string data.type",
63                node.id()
64            )));
65        }
66        if nodes.insert(node.id(), node).is_some() {
67            return Err(invalid_graph(format!("duplicate node ID {:?}", node.id())));
68        }
69        scope_nodes
70            .entry(node.parent_id().map(str::to_owned))
71            .or_default()
72            .insert(node.id().to_owned());
73    }
74    validate_container_scopes(&nodes)?;
75
76    let mut edge_ids = BTreeSet::new();
77    let mut outgoing: BTreeMap<&str, Vec<&str>> =
78        nodes.keys().copied().map(|id| (id, Vec::new())).collect();
79    let mut indegree: BTreeMap<&str, usize> = nodes.keys().copied().map(|id| (id, 0)).collect();
80    for edge in graph.edges() {
81        if edge.id().trim().is_empty() {
82            return Err(invalid_graph("edge ID is empty"));
83        }
84        if !edge_ids.insert(edge.id()) {
85            return Err(invalid_graph(format!("duplicate edge ID {:?}", edge.id())));
86        }
87        let source = nodes.get(edge.source()).ok_or_else(|| {
88            invalid_graph(format!(
89                "edge {:?} references missing source {:?}",
90                edge.id(),
91                edge.source()
92            ))
93        })?;
94        let target = nodes.get(edge.target()).ok_or_else(|| {
95            invalid_graph(format!(
96                "edge {:?} references missing target {:?}",
97                edge.id(),
98                edge.target()
99            ))
100        })?;
101        if edge.source() == edge.target() {
102            return Err(invalid_graph(format!(
103                "edge {:?} connects a node to itself",
104                edge.id()
105            )));
106        }
107        if source.parent_id() != target.parent_id() {
108            return Err(invalid_graph(format!(
109                "edge {:?} crosses workflow DAG scopes",
110                edge.id()
111            )));
112        }
113        outgoing
114            .get_mut(edge.source())
115            .ok_or_else(|| invalid_graph("validated edge source has no adjacency state"))?
116            .push(edge.target());
117        let target_indegree = indegree
118            .get_mut(edge.target())
119            .ok_or_else(|| invalid_graph("validated edge target has no indegree state"))?;
120        *target_indegree = target_indegree
121            .checked_add(1)
122            .ok_or_else(|| invalid_graph("workflow DAG indegree overflowed"))?;
123    }
124
125    for targets in outgoing.values_mut() {
126        targets.sort_unstable();
127    }
128    let mut plans = BTreeMap::new();
129    for (scope, ids) in scope_nodes {
130        let mut scoped_indegree = BTreeMap::new();
131        for id in &ids {
132            let count = indegree
133                .get(id.as_str())
134                .copied()
135                .ok_or_else(|| invalid_graph(format!("node {id:?} has no indegree state")))?;
136            scoped_indegree.insert(id.as_str(), count);
137        }
138        let mut ready = scoped_indegree
139            .iter()
140            .filter_map(|(id, count)| (*count == 0).then_some(*id))
141            .collect::<BTreeSet<_>>();
142        let mut order = Vec::with_capacity(ids.len());
143        while let Some(id) = ready.pop_first() {
144            order.push(id.to_owned());
145            let targets = outgoing
146                .get(id)
147                .ok_or_else(|| invalid_graph(format!("node {id:?} has no adjacency state")))?;
148            for target in targets {
149                let count = scoped_indegree.get_mut(target).ok_or_else(|| {
150                    invalid_graph(format!(
151                        "same-scope target {target:?} has no indegree state"
152                    ))
153                })?;
154                *count = count
155                    .checked_sub(1)
156                    .ok_or_else(|| invalid_graph("workflow DAG indegree underflowed"))?;
157                if *count == 0 {
158                    ready.insert(target);
159                }
160            }
161        }
162        if order.len() != ids.len() {
163            return Err(invalid_graph(match scope.as_deref() {
164                Some(parent_id) => format!("scope {parent_id:?} contains a cycle"),
165                None => "top-level graph contains a cycle".to_owned(),
166            }));
167        }
168        plans.insert(scope, order);
169    }
170
171    Ok(WorkflowDagPlan { scopes: plans })
172}
173
174fn validate_container_scopes(
175    nodes: &BTreeMap<&str, &super::WorkflowDagNode>,
176) -> Result<(), WorkflowDslError> {
177    for node in nodes.values() {
178        let Some(parent_id) = node.parent_id() else {
179            continue;
180        };
181        let parent = nodes.get(parent_id).ok_or_else(|| {
182            invalid_graph(format!(
183                "node {:?} references missing parent {parent_id:?}",
184                node.id()
185            ))
186        })?;
187        let parent_type = parent.node_type();
188        if !matches!(parent_type, "iteration" | "loop") {
189            return Err(invalid_graph(format!(
190                "node {:?} parent {parent_id:?} is not an iteration or loop",
191                node.id()
192            )));
193        }
194        match (parent_type, node.node_type()) {
195            ("iteration", "loop-start") => {
196                return Err(invalid_graph(format!(
197                    "iteration {parent_id:?} requires an iteration-start child, not {:?}",
198                    node.id()
199                )))
200            }
201            ("loop", "iteration-start") => {
202                return Err(invalid_graph(format!(
203                    "loop {parent_id:?} requires a loop-start child, not {:?}",
204                    node.id()
205                )))
206            }
207            _ => {}
208        }
209    }
210
211    for node in nodes.values() {
212        let expected_start_type = match node.node_type() {
213            "iteration" => "iteration-start",
214            "loop" => "loop-start",
215            _ => continue,
216        };
217        let start_node_id = node
218            .data()
219            .get("start_node_id")
220            .and_then(serde_json::Value::as_str)
221            .filter(|id| !id.trim().is_empty())
222            .ok_or_else(|| {
223                invalid_graph(format!(
224                    "{} container {:?} has no string data.start_node_id",
225                    node.node_type(),
226                    node.id()
227                ))
228            })?;
229        let start = nodes.get(start_node_id).ok_or_else(|| {
230            invalid_graph(format!(
231                "{} container {:?} references missing start node {start_node_id:?}",
232                node.node_type(),
233                node.id()
234            ))
235        })?;
236        if start.parent_id() != Some(node.id()) || start.node_type() != expected_start_type {
237            return Err(invalid_graph(format!(
238                "{} container {:?} requires an {expected_start_type} child referenced by data.start_node_id",
239                node.node_type(),
240                node.id()
241            )));
242        }
243        if !nodes.values().any(|candidate| {
244            candidate.parent_id() == Some(node.id()) && candidate.id() != start_node_id
245        }) {
246            return Err(invalid_graph(format!(
247                "{} container {:?} has no executable child",
248                node.node_type(),
249                node.id()
250            )));
251        }
252    }
253
254    for node in nodes.values() {
255        let mut current = node;
256        let mut ancestors = BTreeSet::new();
257        while let Some(parent_id) = current.parent_id() {
258            if !ancestors.insert(parent_id) {
259                return Err(invalid_graph(format!(
260                    "node {:?} has a cycle in its parentId chain",
261                    node.id()
262                )));
263            }
264            current = nodes.get(parent_id).ok_or_else(|| {
265                invalid_graph(format!(
266                    "node {:?} references missing parent {parent_id:?}",
267                    current.id()
268                ))
269            })?;
270        }
271    }
272    Ok(())
273}
274
275fn invalid_graph(message: impl Into<String>) -> WorkflowDslError {
276    WorkflowDslError::InvalidGraph {
277        message: message.into(),
278    }
279}
a3s_flow/workflow_dsl/plan.rs

a3s_flow/workflow_dsl/
plan.rs
