Skip to main content

max / makenotwork

8.6 KB · 260 lines History Blame Raw
1 //! Reconcile sando.toml into SQLite at startup.
2 //!
3 //! Tiers and nodes are config-driven; mutable per-tier state (current version,
4 //! burn-in clock) lives in tier_state and must survive across syncs. Stale
5 //! rows (tier or node removed from the TOML) are deleted, but tier_state for
6 //! a removed tier is preserved silently — the FK is cleared by deleting the
7 //! parent last. If you actually need to forget a retired tier, do it by hand.
8
9 use crate::topology::Topology;
10 use anyhow::Result;
11 use sqlx::SqlitePool;
12
13 pub async fn sync(pool: &SqlitePool, topo: &Topology) -> Result<()> {
14 let mut tx = pool.begin().await?;
15
16 let want_tiers: Vec<&str> = topo.tiers.iter().map(|t| t.name.as_str()).collect();
17 let want_nodes: Vec<(&str, &str)> = topo
18 .tiers
19 .iter()
20 .flat_map(|t| {
21 t.nodes
22 .iter()
23 .map(move |n| (t.name.as_str(), n.name.as_str()))
24 })
25 .collect();
26
27 // Drop stale nodes first (FK to tiers).
28 let existing_nodes: Vec<(String, String)> = sqlx::query_as("SELECT name, tier FROM nodes")
29 .fetch_all(&mut *tx)
30 .await?;
31 for (name, tier) in existing_nodes {
32 if !want_nodes.iter().any(|(t, n)| *t == tier && *n == name) {
33 sqlx::query("DELETE FROM nodes WHERE name = ?")
34 .bind(&name)
35 .execute(&mut *tx)
36 .await?;
37 }
38 }
39
40 // Drop stale tiers. tier_state rows referencing them are preserved by
41 // clearing the FK target only after a manual cleanup — for now we just
42 // refuse to delete a tier that still has tier_state with non-null version.
43 let existing_tiers: Vec<String> = sqlx::query_scalar("SELECT name FROM tiers")
44 .fetch_all(&mut *tx)
45 .await?;
46 for t in existing_tiers {
47 if !want_tiers.contains(&t.as_str()) {
48 let in_use: Option<String> =
49 sqlx::query_scalar("SELECT current_version FROM tier_state WHERE tier = ?")
50 .bind(&t)
51 .fetch_optional(&mut *tx)
52 .await?
53 .flatten();
54 anyhow::ensure!(
55 in_use.is_none(),
56 "refusing to remove tier {t} from config: tier_state still pins a version. \
57 clean it up by hand before editing sando.toml.",
58 );
59 sqlx::query("DELETE FROM tier_state WHERE tier = ?")
60 .bind(&t)
61 .execute(&mut *tx)
62 .await?;
63 sqlx::query("DELETE FROM tiers WHERE name = ?")
64 .bind(&t)
65 .execute(&mut *tx)
66 .await?;
67 }
68 }
69
70 // Upsert tiers in declaration order; `ord` mirrors that order so the
71 // promotion sequence is queryable without re-reading the TOML.
72 for (i, t) in topo.tiers.iter().enumerate() {
73 sqlx::query(
74 "INSERT INTO tiers (name, ord, provisioned, canary)
75 VALUES (?, ?, ?, ?)
76 ON CONFLICT(name) DO UPDATE SET
77 ord = excluded.ord,
78 provisioned = excluded.provisioned,
79 canary = excluded.canary",
80 )
81 .bind(&t.name)
82 .bind(i as i64)
83 .bind(t.provisioned as i64)
84 .bind(t.canary.as_str())
85 .execute(&mut *tx)
86 .await?;
87
88 sqlx::query("INSERT OR IGNORE INTO tier_state (tier) VALUES (?)")
89 .bind(&t.name)
90 .execute(&mut *tx)
91 .await?;
92
93 for n in &t.nodes {
94 sqlx::query(
95 "INSERT INTO nodes (name, tier, ssh_target, release_root)
96 VALUES (?, ?, ?, ?)
97 ON CONFLICT(name) DO UPDATE SET
98 tier = excluded.tier,
99 ssh_target = excluded.ssh_target,
100 release_root = excluded.release_root",
101 )
102 .bind(&n.name)
103 .bind(&t.name)
104 .bind(&n.ssh_target)
105 .bind(&n.release_root)
106 .execute(&mut *tx)
107 .await?;
108 }
109 }
110
111 tx.commit().await?;
112 Ok(())
113 }
114
115 #[cfg(test)]
116 mod tests {
117 use super::*;
118 use crate::topology::{BackupConfig, CanaryPolicy, Gate, Node, RepoConfig, Tier, Topology};
119 use sqlx::sqlite::SqlitePoolOptions;
120
121 async fn fresh_pool() -> SqlitePool {
122 let pool = SqlitePoolOptions::new()
123 .max_connections(1)
124 .connect("sqlite::memory:")
125 .await
126 .unwrap();
127 sqlx::migrate!("./migrations").run(&pool).await.unwrap();
128 pool
129 }
130
131 fn topo(tiers: Vec<Tier>) -> Topology {
132 Topology {
133 repo: RepoConfig {
134 bare_path: "/tmp/x".into(),
135 branch: "main".into(),
136 upstream: None,
137 },
138 backup: BackupConfig {
139 source: "file:///tmp/b".into(),
140 local_path: "/tmp/b".into(),
141 },
142 tiers,
143 aux_repos: Vec::new(),
144 }
145 }
146
147 fn tier(name: &str, provisioned: bool, nodes: Vec<Node>) -> Tier {
148 Tier {
149 name: name.into(),
150 provisioned,
151 gates: vec![Gate::BootSmoke],
152 canary: CanaryPolicy::Sequential,
153 nodes,
154 }
155 }
156
157 fn node(name: &str) -> Node {
158 Node {
159 name: name.into(),
160 ssh_target: format!("deploy@{name}"),
161 release_root: "/opt/mnw".into(),
162 service_name: "makenotwork.service".into(),
163 health_url: None,
164 config_check_env_file: None,
165 actuate: crate::topology::default_actuate(),
166 observe: crate::topology::default_observe(),
167 companions: Vec::new(),
168 }
169 }
170
171 #[tokio::test]
172 async fn syncs_tiers_nodes_and_inits_tier_state() {
173 let pool = fresh_pool().await;
174 let t = topo(vec![
175 tier("host", true, vec![]),
176 tier("a", true, vec![node("testnot-1")]),
177 tier("c", false, vec![]),
178 ]);
179
180 sync(&pool, &t).await.unwrap();
181
182 let tier_names: Vec<String> = sqlx::query_scalar("SELECT name FROM tiers ORDER BY ord")
183 .fetch_all(&pool)
184 .await
185 .unwrap();
186 assert_eq!(tier_names, vec!["host", "a", "c"]);
187
188 let node_names: Vec<String> = sqlx::query_scalar("SELECT name FROM nodes")
189 .fetch_all(&pool)
190 .await
191 .unwrap();
192 assert_eq!(node_names, vec!["testnot-1"]);
193
194 let state_count: i64 = sqlx::query_scalar("SELECT COUNT(*) FROM tier_state")
195 .fetch_one(&pool)
196 .await
197 .unwrap();
198 assert_eq!(state_count, 3);
199 }
200
201 #[tokio::test]
202 async fn second_sync_is_idempotent() {
203 let pool = fresh_pool().await;
204 let t = topo(vec![
205 tier("host", true, vec![]),
206 tier("a", true, vec![node("n1")]),
207 ]);
208 sync(&pool, &t).await.unwrap();
209 sync(&pool, &t).await.unwrap();
210
211 let nodes: i64 = sqlx::query_scalar("SELECT COUNT(*) FROM nodes")
212 .fetch_one(&pool)
213 .await
214 .unwrap();
215 assert_eq!(nodes, 1);
216 let states: i64 = sqlx::query_scalar("SELECT COUNT(*) FROM tier_state")
217 .fetch_one(&pool)
218 .await
219 .unwrap();
220 assert_eq!(states, 2);
221 }
222
223 #[tokio::test]
224 async fn removing_node_from_config_drops_row() {
225 let pool = fresh_pool().await;
226 let t1 = topo(vec![tier("a", true, vec![node("n1"), node("n2")])]);
227 sync(&pool, &t1).await.unwrap();
228 let t2 = topo(vec![tier("a", true, vec![node("n1")])]);
229 sync(&pool, &t2).await.unwrap();
230
231 let nodes: Vec<String> = sqlx::query_scalar("SELECT name FROM nodes")
232 .fetch_all(&pool)
233 .await
234 .unwrap();
235 assert_eq!(nodes, vec!["n1"]);
236 }
237
238 #[tokio::test]
239 async fn refuses_to_drop_tier_with_pinned_version() {
240 let pool = fresh_pool().await;
241 let t1 = topo(vec![tier("host", true, vec![]), tier("a", true, vec![])]);
242 sync(&pool, &t1).await.unwrap();
243
244 // Simulate a version being deployed on tier a.
245 sqlx::query("INSERT INTO versions (version, git_sha, built_at, artifact_path) VALUES ('0.1.0', 'deadbeef', '2026-05-22T00:00:00Z', '/r/0.1.0')")
246 .execute(&pool).await.unwrap();
247 sqlx::query("UPDATE tier_state SET current_version = '0.1.0' WHERE tier = 'a'")
248 .execute(&pool)
249 .await
250 .unwrap();
251
252 let t2 = topo(vec![tier("host", true, vec![])]);
253 let err = sync(&pool, &t2).await.unwrap_err();
254 assert!(
255 err.to_string().contains("tier_state still pins"),
256 "got: {err}"
257 );
258 }
259 }
260