Skip to main content

max / makenotwork

12.9 KB · 378 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::domain::AppId;
10 use crate::topology::Topology;
11 use anyhow::Result;
12 use sqlx::SqlitePool;
13
14 /// Reconcile one product's topology into the tier and node tables.
15 ///
16 /// Every statement is scoped to `app`. Without that, syncing one product would
17 /// treat every other product's tiers as removed from config and delete them:
18 /// the stale-row sweep asks "which rows are not in this TOML", and one TOML has
19 /// never described more than one product.
20 pub async fn sync(pool: &SqlitePool, app: &AppId, topo: &Topology) -> Result<()> {
21 let mut tx = pool.begin().await?;
22
23 let want_tiers: Vec<&str> = topo.tiers.iter().map(|t| t.name.as_str()).collect();
24 let want_nodes: Vec<(&str, &str)> = topo
25 .tiers
26 .iter()
27 .flat_map(|t| {
28 t.nodes
29 .iter()
30 .map(move |n| (t.name.as_str(), n.name.as_str()))
31 })
32 .collect();
33
34 // Drop stale nodes first (FK to tiers).
35 let existing_nodes: Vec<(String, String)> =
36 sqlx::query_as("SELECT name, tier FROM nodes WHERE app = ?")
37 .bind(app)
38 .fetch_all(&mut *tx)
39 .await?;
40 for (name, tier) in existing_nodes {
41 if !want_nodes.iter().any(|(t, n)| *t == tier && *n == name) {
42 sqlx::query("DELETE FROM nodes WHERE app = ? AND name = ?")
43 .bind(app)
44 .bind(&name)
45 .execute(&mut *tx)
46 .await?;
47 }
48 }
49
50 // Drop stale tiers. tier_state rows referencing them are preserved by
51 // clearing the FK target only after a manual cleanup — for now we just
52 // refuse to delete a tier that still has tier_state with non-null version.
53 let existing_tiers: Vec<String> = sqlx::query_scalar("SELECT name FROM tiers WHERE app = ?")
54 .bind(app)
55 .fetch_all(&mut *tx)
56 .await?;
57 for t in existing_tiers {
58 if !want_tiers.contains(&t.as_str()) {
59 let in_use: Option<String> = sqlx::query_scalar(
60 "SELECT current_version FROM tier_state WHERE app = ? AND tier = ?",
61 )
62 .bind(app)
63 .bind(&t)
64 .fetch_optional(&mut *tx)
65 .await?
66 .flatten();
67 anyhow::ensure!(
68 in_use.is_none(),
69 "refusing to remove tier {t} from app `{app}`'s config: tier_state still pins a \
70 version. clean it up by hand before editing the topology.",
71 );
72 sqlx::query("DELETE FROM tier_state WHERE app = ? AND tier = ?")
73 .bind(app)
74 .bind(&t)
75 .execute(&mut *tx)
76 .await?;
77 sqlx::query("DELETE FROM tiers WHERE app = ? AND name = ?")
78 .bind(app)
79 .bind(&t)
80 .execute(&mut *tx)
81 .await?;
82 }
83 }
84
85 // Upsert tiers in declaration order; `ord` mirrors that order so the
86 // promotion sequence is queryable without re-reading the TOML.
87 for (i, t) in topo.tiers.iter().enumerate() {
88 sqlx::query(
89 "INSERT INTO tiers (app, name, ord, provisioned, canary)
90 VALUES (?, ?, ?, ?, ?)
91 ON CONFLICT(app, name) DO UPDATE SET
92 ord = excluded.ord,
93 provisioned = excluded.provisioned,
94 canary = excluded.canary",
95 )
96 .bind(app)
97 .bind(&t.name)
98 .bind(i as i64)
99 .bind(t.provisioned as i64)
100 .bind(t.canary.as_str())
101 .execute(&mut *tx)
102 .await?;
103
104 sqlx::query("INSERT OR IGNORE INTO tier_state (app, tier) VALUES (?, ?)")
105 .bind(app)
106 .bind(&t.name)
107 .execute(&mut *tx)
108 .await?;
109
110 for n in &t.nodes {
111 sqlx::query(
112 "INSERT INTO nodes (app, name, tier, ssh_target, release_root)
113 VALUES (?, ?, ?, ?, ?)
114 ON CONFLICT(app, name) DO UPDATE SET
115 tier = excluded.tier,
116 ssh_target = excluded.ssh_target,
117 release_root = excluded.release_root",
118 )
119 .bind(app)
120 .bind(&n.name)
121 .bind(&t.name)
122 .bind(&n.ssh_target)
123 .bind(&n.release_root)
124 .execute(&mut *tx)
125 .await?;
126 }
127 }
128
129 tx.commit().await?;
130 Ok(())
131 }
132
133 #[cfg(test)]
134 mod tests {
135 use super::*;
136 use crate::topology::{BackupConfig, CanaryPolicy, Gate, Node, RepoConfig, Tier, Topology};
137 use sqlx::sqlite::SqlitePoolOptions;
138
139 async fn fresh_pool() -> SqlitePool {
140 let pool = SqlitePoolOptions::new()
141 .max_connections(1)
142 .connect("sqlite::memory:")
143 .await
144 .unwrap();
145 sqlx::migrate!("./migrations").run(&pool).await.unwrap();
146 pool
147 }
148
149 fn app() -> AppId {
150 AppId::default()
151 }
152
153 fn topo(tiers: Vec<Tier>) -> Topology {
154 Topology {
155 repo: Some(RepoConfig {
156 bare_path: "/tmp/x".into(),
157 branch: "main".into(),
158 upstream: None,
159 }),
160 backup: vec![BackupConfig {
161 name: "server".into(),
162 source: "file:///tmp/b".into(),
163 local_path: "/tmp/b".into(),
164 }],
165 tiers,
166 aux_repos: Vec::new(),
167 }
168 }
169
170 fn tier(name: &str, provisioned: bool, nodes: Vec<Node>) -> Tier {
171 Tier {
172 public_url: None,
173 name: name.into(),
174 provisioned,
175 gates: vec![Gate::BootSmoke],
176 canary: CanaryPolicy::Sequential,
177 nodes,
178 }
179 }
180
181 fn node(name: &str) -> Node {
182 Node {
183 platform: None,
184 name: name.into(),
185 ssh_target: format!("deploy@{name}"),
186 release_root: "/opt/mnw".into(),
187 service_name: "makenotwork.service".into(),
188 health_url: None,
189 config_check_env_file: None,
190 actuate: crate::topology::default_actuate(),
191 observe: crate::topology::default_observe(),
192 companions: Vec::new(),
193 }
194 }
195
196 #[tokio::test]
197 async fn syncs_tiers_nodes_and_inits_tier_state() {
198 let pool = fresh_pool().await;
199 let t = topo(vec![
200 tier("host", true, vec![]),
201 tier("a", true, vec![node("testnot-1")]),
202 tier("c", false, vec![]),
203 ]);
204
205 sync(&pool, &app(), &t).await.unwrap();
206
207 let tier_names: Vec<String> = sqlx::query_scalar("SELECT name FROM tiers ORDER BY ord")
208 .fetch_all(&pool)
209 .await
210 .unwrap();
211 assert_eq!(tier_names, vec!["host", "a", "c"]);
212
213 let node_names: Vec<String> = sqlx::query_scalar("SELECT name FROM nodes")
214 .fetch_all(&pool)
215 .await
216 .unwrap();
217 assert_eq!(node_names, vec!["testnot-1"]);
218
219 let state_count: i64 = sqlx::query_scalar("SELECT COUNT(*) FROM tier_state")
220 .fetch_one(&pool)
221 .await
222 .unwrap();
223 assert_eq!(state_count, 3);
224 }
225
226 #[tokio::test]
227 async fn second_sync_is_idempotent() {
228 let pool = fresh_pool().await;
229 let t = topo(vec![
230 tier("host", true, vec![]),
231 tier("a", true, vec![node("n1")]),
232 ]);
233 sync(&pool, &app(), &t).await.unwrap();
234 sync(&pool, &app(), &t).await.unwrap();
235
236 let nodes: i64 = sqlx::query_scalar("SELECT COUNT(*) FROM nodes")
237 .fetch_one(&pool)
238 .await
239 .unwrap();
240 assert_eq!(nodes, 1);
241 let states: i64 = sqlx::query_scalar("SELECT COUNT(*) FROM tier_state")
242 .fetch_one(&pool)
243 .await
244 .unwrap();
245 assert_eq!(states, 2);
246 }
247
248 #[tokio::test]
249 async fn removing_node_from_config_drops_row() {
250 let pool = fresh_pool().await;
251 let t1 = topo(vec![tier("a", true, vec![node("n1"), node("n2")])]);
252 sync(&pool, &app(), &t1).await.unwrap();
253 let t2 = topo(vec![tier("a", true, vec![node("n1")])]);
254 sync(&pool, &app(), &t2).await.unwrap();
255
256 let nodes: Vec<String> = sqlx::query_scalar("SELECT name FROM nodes")
257 .fetch_all(&pool)
258 .await
259 .unwrap();
260 assert_eq!(nodes, vec!["n1"]);
261 }
262
263 /// Syncing one product leaves every other product's tiers and nodes alone.
264 ///
265 /// The sweep at the top of `sync` asks "which rows are not in this TOML",
266 /// and a TOML describes one product. Unscoped, syncing pom at startup would
267 /// answer "all of MNW's" and delete them — and since startup syncs each app
268 /// in turn, the last one to run would be the only one left standing.
269 #[tokio::test]
270 async fn syncing_one_app_does_not_touch_another() {
271 let pool = fresh_pool().await;
272 let mnw = AppId::new("mnw");
273 let pom = AppId::new("pom");
274 sync(
275 &pool,
276 &mnw,
277 &topo(vec![tier("host", true, vec![node("n1")])]),
278 )
279 .await
280 .unwrap();
281 sync(
282 &pool,
283 &pom,
284 &topo(vec![tier("host", true, vec![node("n2")])]),
285 )
286 .await
287 .unwrap();
288
289 // Both survive, and a tier name they share is two rows, not one.
290 let tiers: Vec<(String, String)> =
291 sqlx::query_as("SELECT app, name FROM tiers ORDER BY app")
292 .fetch_all(&pool)
293 .await
294 .unwrap();
295 assert_eq!(
296 tiers,
297 vec![
298 ("mnw".to_string(), "host".to_string()),
299 ("pom".to_string(), "host".to_string())
300 ]
301 );
302 let nodes: Vec<(String, String)> =
303 sqlx::query_as("SELECT app, name FROM nodes ORDER BY app")
304 .fetch_all(&pool)
305 .await
306 .unwrap();
307 assert_eq!(
308 nodes,
309 vec![
310 ("mnw".to_string(), "n1".to_string()),
311 ("pom".to_string(), "n2".to_string())
312 ]
313 );
314 // And one tier_state row each, not one shared.
315 let states: i64 = sqlx::query_scalar("SELECT COUNT(*) FROM tier_state")
316 .fetch_one(&pool)
317 .await
318 .unwrap();
319 assert_eq!(states, 2);
320 }
321
322 /// A tier pinned in one product does not block removing the same-named tier
323 /// from another.
324 #[tokio::test]
325 async fn a_pin_in_one_app_does_not_block_another_apps_edit() {
326 let pool = fresh_pool().await;
327 let mnw = AppId::new("mnw");
328 let pom = AppId::new("pom");
329 let two = topo(vec![tier("host", true, vec![]), tier("a", true, vec![])]);
330 sync(&pool, &mnw, &two).await.unwrap();
331 sync(&pool, &pom, &two).await.unwrap();
332
333 // MNW pins a version on tier a.
334 sqlx::query("INSERT INTO versions (app, version, git_sha, built_at, artifact_path) VALUES ('mnw','0.1.0','deadbeef','2026-05-22T00:00:00Z','/r/0.1.0')")
335 .execute(&pool).await.unwrap();
336 sqlx::query(
337 "UPDATE tier_state SET current_version = '0.1.0' WHERE app = 'mnw' AND tier = 'a'",
338 )
339 .execute(&pool)
340 .await
341 .unwrap();
342
343 // pom dropping ITS tier a is fine; MNW's pin is not pom's business.
344 sync(&pool, &pom, &topo(vec![tier("host", true, vec![])]))
345 .await
346 .unwrap();
347 // MNW dropping the same tier is still refused.
348 let err = sync(&pool, &mnw, &topo(vec![tier("host", true, vec![])]))
349 .await
350 .unwrap_err();
351 assert!(err.to_string().contains("tier_state still pins"), "{err}");
352 }
353
354 #[tokio::test]
355 async fn refuses_to_drop_tier_with_pinned_version() {
356 let pool = fresh_pool().await;
357 let t1 = topo(vec![tier("host", true, vec![]), tier("a", true, vec![])]);
358 sync(&pool, &app(), &t1).await.unwrap();
359
360 // Simulate a version being deployed on tier a.
361 sqlx::query("INSERT INTO versions (app, version, git_sha, built_at, artifact_path) VALUES ('mnw', '0.1.0', 'deadbeef', '2026-05-22T00:00:00Z', '/r/0.1.0')")
362 .execute(&pool).await.unwrap();
363 sqlx::query(
364 "UPDATE tier_state SET current_version = '0.1.0' WHERE app = 'mnw' AND tier = 'a'",
365 )
366 .execute(&pool)
367 .await
368 .unwrap();
369
370 let t2 = topo(vec![tier("host", true, vec![])]);
371 let err = sync(&pool, &app(), &t2).await.unwrap_err();
372 assert!(
373 err.to_string().contains("tier_state still pins"),
374 "got: {err}"
375 );
376 }
377 }
378