use crate::types::QedStep;
use std::collections::{BTreeSet, HashSet};
pub const DEFAULT_MAX_PARALLEL: usize = 4;
#[derive(Debug, Clone, PartialEq, Eq, thiserror::Error)]
pub enum DagError {
#[error(
"step `{step}`: `needs` names unknown step `{missing}` — \
it must match another step's `name` in this pipeline (a matrix step \
is matched by its un-suffixed name)"
)]
UnknownNeed { step: String, missing: String },
#[error("step `{0}`: `needs` names the step itself")]
SelfDependency(String),
#[error(
"steps `{0}` form a dependency cycle — no step in the cycle can ever \
become ready"
)]
Cycle(String),
#[error(
"step name `{0}` is used by more than one step and is referenced by a \
`needs` — rename one, or the edge is ambiguous"
)]
AmbiguousName(String),
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum Missing {
Reject,
Satisfied,
}
pub fn name_matches(step_name: &str, need: &str) -> bool {
step_name == need
|| step_name
.strip_prefix(need)
.is_some_and(|rest| rest.starts_with(" ["))
}
pub fn predecessors(steps: &[QedStep], missing: Missing) -> Result<Vec<Vec<usize>>, DagError> {
let mut out: Vec<Vec<usize>> = Vec::with_capacity(steps.len());
for (i, step) in steps.iter().enumerate() {
let Some(needs) = step.needs.as_ref() else {
out.push(if i == 0 { Vec::new() } else { vec![i - 1] });
continue;
};
let mut preds: BTreeSet<usize> = BTreeSet::new();
for need in needs {
let matched: Vec<usize> = steps
.iter()
.enumerate()
.filter(|(_, s)| name_matches(&s.name, need))
.map(|(j, _)| j)
.collect();
if matched.contains(&i) {
return Err(DagError::SelfDependency(step.name.clone()));
}
if matched.len() > 1 && matched.iter().filter(|&&j| steps[j].name == *need).count() > 1
{
return Err(DagError::AmbiguousName(need.clone()));
}
if matched.is_empty() {
match missing {
Missing::Reject => {
return Err(DagError::UnknownNeed {
step: step.name.clone(),
missing: need.clone(),
})
}
Missing::Satisfied => continue,
}
}
preds.extend(matched);
}
out.push(preds.into_iter().collect());
}
Ok(out)
}
pub fn waves(steps: &[QedStep], missing: Missing) -> Result<Vec<Vec<usize>>, DagError> {
let preds = predecessors(steps, missing)?;
let mut done: Vec<bool> = vec![false; steps.len()];
let mut remaining = steps.len();
let mut out: Vec<Vec<usize>> = Vec::new();
while remaining > 0 {
let wave: Vec<usize> = (0..steps.len())
.filter(|&i| !done[i] && preds[i].iter().all(|&p| done[p]))
.collect();
if wave.is_empty() {
let stuck: Vec<&str> = (0..steps.len())
.filter(|&i| !done[i])
.map(|i| steps[i].name.as_str())
.collect();
return Err(DagError::Cycle(stuck.join(", ")));
}
for &i in &wave {
done[i] = true;
}
remaining -= wave.len();
out.push(wave);
}
Ok(out)
}
pub fn is_explicit(steps: &[QedStep]) -> bool {
steps.iter().any(|s| s.needs.is_some())
}
pub fn dependents(preds: &[Vec<usize>], root: usize) -> HashSet<usize> {
let mut seen: HashSet<usize> = HashSet::new();
let mut frontier = vec![root];
while let Some(cur) = frontier.pop() {
for (i, p) in preds.iter().enumerate() {
if p.contains(&cur) && seen.insert(i) {
frontier.push(i);
}
}
}
seen
}
pub fn legs(preds: &[Vec<usize>]) -> Vec<usize> {
let mut leg: Vec<usize> = (0..preds.len()).collect();
fn find(leg: &mut [usize], mut i: usize) -> usize {
while leg[i] != i {
leg[i] = leg[leg[i]];
i = leg[i];
}
i
}
for (i, ps) in preds.iter().enumerate() {
for &p in ps {
let (a, b) = (find(&mut leg, i), find(&mut leg, p));
let (lo, hi) = if a < b { (a, b) } else { (b, a) };
leg[hi] = lo;
}
}
(0..preds.len()).map(|i| find(&mut leg, i)).collect()
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn legs_are_weakly_connected_components() {
assert_eq!(legs(&[vec![], vec![0], vec![1]]), vec![0, 0, 0]);
assert_eq!(legs(&[vec![], vec![0], vec![], vec![2]]), vec![0, 0, 2, 2]);
assert_eq!(legs(&[vec![], vec![], vec![1, 0]]), vec![0, 0, 0]);
assert_eq!(legs(&[vec![2], vec![2], vec![]]), vec![0, 0, 0]);
}
fn s(name: &str, needs: Option<&[&str]>) -> QedStep {
QedStep {
name: name.to_string(),
argv: vec!["true".into()],
needs: needs.map(|n| n.iter().map(|x| x.to_string()).collect()),
..Default::default()
}
}
#[test]
fn absent_needs_is_the_implicit_serial_chain() {
let steps = vec![s("a", None), s("b", None), s("c", None)];
let preds = predecessors(&steps, Missing::Reject).unwrap();
assert_eq!(preds, vec![vec![], vec![0], vec![1]]);
assert_eq!(
waves(&steps, Missing::Reject).unwrap(),
vec![vec![0], vec![1], vec![2]]
);
}
#[test]
fn empty_needs_is_a_root_not_an_absent_needs() {
let steps = vec![
s("setup", Some(&[])),
s("left", Some(&["setup"])),
s("right", Some(&["setup"])),
s("join", Some(&["left", "right"])),
];
assert_eq!(
waves(&steps, Missing::Reject).unwrap(),
vec![vec![0], vec![1, 2], vec![3]]
);
}
#[test]
fn two_independent_roots_share_the_first_wave() {
let steps = vec![s("a", Some(&[])), s("b", Some(&[]))];
assert_eq!(waves(&steps, Missing::Reject).unwrap(), vec![vec![0, 1]]);
}
#[test]
fn a_step_after_an_explicit_one_still_chains_implicitly() {
let steps = vec![s("a", Some(&[])), s("b", Some(&["a"])), s("tail", None)];
let preds = predecessors(&steps, Missing::Reject).unwrap();
assert_eq!(preds[2], vec![1]);
}
#[test]
fn needs_matches_every_matrix_row_of_the_named_step() {
let steps = vec![
s("build [target=x86]", Some(&[])),
s("build [target=arm]", Some(&[])),
s("join", Some(&["build"])),
];
let preds = predecessors(&steps, Missing::Reject).unwrap();
assert_eq!(preds[2], vec![0, 1], "the join waits for every row");
}
#[test]
fn matrix_prefix_match_requires_the_space_bracket() {
let steps = vec![s("build-extra", Some(&[])), s("join", Some(&["build"]))];
let err = predecessors(&steps, Missing::Reject).unwrap_err();
assert_eq!(
err,
DagError::UnknownNeed { step: "join".into(), missing: "build".into() }
);
}
#[test]
fn unknown_need_is_rejected_or_dropped_by_policy() {
let steps = vec![s("publish", Some(&["build"]))];
assert!(matches!(
predecessors(&steps, Missing::Reject),
Err(DagError::UnknownNeed { .. })
));
assert_eq!(
predecessors(&steps, Missing::Satisfied).unwrap(),
vec![Vec::<usize>::new()]
);
}
#[test]
fn self_dependency_is_rejected() {
let steps = vec![s("a", Some(&["a"]))];
assert_eq!(
predecessors(&steps, Missing::Reject).unwrap_err(),
DagError::SelfDependency("a".into())
);
}
#[test]
fn a_cycle_names_the_steps_stuck_in_it() {
let steps = vec![s("a", Some(&["b"])), s("b", Some(&["a"]))];
let err = waves(&steps, Missing::Reject).unwrap_err();
assert_eq!(err, DagError::Cycle("a, b".into()));
}
#[test]
fn duplicate_names_are_ambiguous_only_when_referenced() {
let dup = vec![s("x", None), s("x", None), s("y", None)];
assert!(predecessors(&dup, Missing::Reject).is_ok());
let referenced = vec![s("x", Some(&[])), s("x", Some(&[])), s("y", Some(&["x"]))];
assert_eq!(
predecessors(&referenced, Missing::Reject).unwrap_err(),
DagError::AmbiguousName("x".into())
);
}
#[test]
fn forward_reference_is_allowed_when_it_closes_no_cycle() {
let steps = vec![s("first", Some(&["second"])), s("second", Some(&[]))];
assert_eq!(waves(&steps, Missing::Reject).unwrap(), vec![vec![1], vec![0]]);
}
#[test]
fn dependents_is_transitive_and_excludes_the_root() {
let steps = vec![
s("a", Some(&[])),
s("b", Some(&["a"])),
s("c", Some(&["b"])),
s("island", Some(&[])),
];
let preds = predecessors(&steps, Missing::Reject).unwrap();
let d = dependents(&preds, 0);
assert_eq!(d, HashSet::from([1, 2]));
assert!(dependents(&preds, 3).is_empty());
}
#[test]
fn is_explicit_distinguishes_a_dag_from_the_legacy_chain() {
assert!(!is_explicit(&[s("a", None), s("b", None)]));
assert!(is_explicit(&[s("a", None), s("b", Some(&[]))]));
}
}