arig 0.2.1

Polyglot service orchestrator
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154

use crate::config::ArigConfig;
use std::collections::{HashMap, HashSet, VecDeque};

/// Returns service names in topological order (respecting depends_on).
/// Services with no dependencies come first. Services in the same "wave"
/// can be started in parallel.
pub fn toposort(config: &ArigConfig) -> anyhow::Result<Vec<Vec<String>>> {
    let names: HashSet<&str> = config.services.keys().map(|s| s.as_str()).collect();

    // Validate that all depends_on references exist
    for (name, service) in &config.services {
        for dep in &service.depends_on {
            if !names.contains(dep.as_str()) {
                anyhow::bail!("service '{name}' depends on unknown service '{dep}'");
            }
        }
    }

    // Build in-degree map
    let mut in_degree: HashMap<&str, usize> = HashMap::new();
    let mut dependents: HashMap<&str, Vec<&str>> = HashMap::new();
    for name in &names {
        in_degree.insert(name, 0);
    }
    for (name, service) in &config.services {
        in_degree.insert(name.as_str(), service.depends_on.len());
        for dep in &service.depends_on {
            dependents
                .entry(dep.as_str())
                .or_default()
                .push(name.as_str());
        }
    }

    // Kahn's algorithm, collecting by wave
    let mut waves: Vec<Vec<String>> = Vec::new();
    let mut queue: VecDeque<&str> = VecDeque::new();

    for (name, &deg) in &in_degree {
        if deg == 0 {
            queue.push_back(name);
        }
    }

    let mut visited = 0;
    while !queue.is_empty() {
        let wave: Vec<&str> = queue.drain(..).collect();
        visited += wave.len();

        for &name in &wave {
            if let Some(deps) = dependents.get(name) {
                for &dep in deps {
                    let deg = in_degree.get_mut(dep).unwrap();
                    *deg -= 1;
                    if *deg == 0 {
                        queue.push_back(dep);
                    }
                }
            }
        }

        waves.push(wave.into_iter().map(String::from).collect());
    }

    if visited != names.len() {
        anyhow::bail!("circular dependency detected");
    }

    Ok(waves)
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::config::{ArigConfig, ServiceConfig};
    use std::collections::HashMap;

    fn svc(command: &str, depends_on: Vec<&str>) -> ServiceConfig {
        ServiceConfig {
            command: command.to_string(),
            service_type: Default::default(),
            working_dir: None,
            env: HashMap::new(),
            depends_on: depends_on.into_iter().map(String::from).collect(),
            ready: None,
        }
    }

    #[test]
    fn no_dependencies() {
        let config = ArigConfig {
            services: HashMap::from([
                ("a".into(), svc("echo a", vec![])),
                ("b".into(), svc("echo b", vec![])),
            ]),
        };
        let waves = toposort(&config).unwrap();
        assert_eq!(waves.len(), 1);
        assert_eq!(waves[0].len(), 2);
    }

    #[test]
    fn linear_chain() {
        let config = ArigConfig {
            services: HashMap::from([
                ("db".into(), svc("echo db", vec![])),
                ("api".into(), svc("echo api", vec!["db"])),
                ("web".into(), svc("echo web", vec!["api"])),
            ]),
        };
        let waves = toposort(&config).unwrap();
        assert_eq!(waves.len(), 3);
        assert_eq!(waves[0], vec!["db"]);
        assert_eq!(waves[1], vec!["api"]);
        assert_eq!(waves[2], vec!["web"]);
    }

    #[test]
    fn diamond() {
        let config = ArigConfig {
            services: HashMap::from([
                ("db".into(), svc("echo db", vec![])),
                ("cache".into(), svc("echo cache", vec!["db"])),
                ("api".into(), svc("echo api", vec!["db"])),
                ("web".into(), svc("echo web", vec!["cache", "api"])),
            ]),
        };
        let waves = toposort(&config).unwrap();
        assert_eq!(waves.len(), 3);
        assert_eq!(waves[0], vec!["db"]);
        assert!(waves[1].contains(&"cache".to_string()));
        assert!(waves[1].contains(&"api".to_string()));
        assert_eq!(waves[2], vec!["web"]);
    }

    #[test]
    fn cycle_detected() {
        let config = ArigConfig {
            services: HashMap::from([
                ("a".into(), svc("echo a", vec!["b"])),
                ("b".into(), svc("echo b", vec!["a"])),
            ]),
        };
        assert!(toposort(&config).is_err());
    }

    #[test]
    fn unknown_dependency() {
        let config = ArigConfig {
            services: HashMap::from([("a".into(), svc("echo a", vec!["nonexistent"]))]),
        };
        assert!(toposort(&config).is_err());
    }
}