Skip to main content

max / makenotwork

12.8 KB · 377 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 name: name.into(),
173 provisioned,
174 gates: vec![Gate::BootSmoke],
175 canary: CanaryPolicy::Sequential,
176 nodes,
177 }
178 }
179
180 fn node(name: &str) -> Node {
181 Node {
182 platform: None,
183 name: name.into(),
184 ssh_target: format!("deploy@{name}"),
185 release_root: "/opt/mnw".into(),
186 service_name: "makenotwork.service".into(),
187 health_url: None,
188 config_check_env_file: None,
189 actuate: crate::topology::default_actuate(),
190 observe: crate::topology::default_observe(),
191 companions: Vec::new(),
192 }
193 }
194
195 #[tokio::test]
196 async fn syncs_tiers_nodes_and_inits_tier_state() {
197 let pool = fresh_pool().await;
198 let t = topo(vec![
199 tier("host", true, vec![]),
200 tier("a", true, vec![node("testnot-1")]),
201 tier("c", false, vec![]),
202 ]);
203
204 sync(&pool, &app(), &t).await.unwrap();
205
206 let tier_names: Vec<String> = sqlx::query_scalar("SELECT name FROM tiers ORDER BY ord")
207 .fetch_all(&pool)
208 .await
209 .unwrap();
210 assert_eq!(tier_names, vec!["host", "a", "c"]);
211
212 let node_names: Vec<String> = sqlx::query_scalar("SELECT name FROM nodes")
213 .fetch_all(&pool)
214 .await
215 .unwrap();
216 assert_eq!(node_names, vec!["testnot-1"]);
217
218 let state_count: i64 = sqlx::query_scalar("SELECT COUNT(*) FROM tier_state")
219 .fetch_one(&pool)
220 .await
221 .unwrap();
222 assert_eq!(state_count, 3);
223 }
224
225 #[tokio::test]
226 async fn second_sync_is_idempotent() {
227 let pool = fresh_pool().await;
228 let t = topo(vec![
229 tier("host", true, vec![]),
230 tier("a", true, vec![node("n1")]),
231 ]);
232 sync(&pool, &app(), &t).await.unwrap();
233 sync(&pool, &app(), &t).await.unwrap();
234
235 let nodes: i64 = sqlx::query_scalar("SELECT COUNT(*) FROM nodes")
236 .fetch_one(&pool)
237 .await
238 .unwrap();
239 assert_eq!(nodes, 1);
240 let states: i64 = sqlx::query_scalar("SELECT COUNT(*) FROM tier_state")
241 .fetch_one(&pool)
242 .await
243 .unwrap();
244 assert_eq!(states, 2);
245 }
246
247 #[tokio::test]
248 async fn removing_node_from_config_drops_row() {
249 let pool = fresh_pool().await;
250 let t1 = topo(vec![tier("a", true, vec![node("n1"), node("n2")])]);
251 sync(&pool, &app(), &t1).await.unwrap();
252 let t2 = topo(vec![tier("a", true, vec![node("n1")])]);
253 sync(&pool, &app(), &t2).await.unwrap();
254
255 let nodes: Vec<String> = sqlx::query_scalar("SELECT name FROM nodes")
256 .fetch_all(&pool)
257 .await
258 .unwrap();
259 assert_eq!(nodes, vec!["n1"]);
260 }
261
262 /// Syncing one product leaves every other product's tiers and nodes alone.
263 ///
264 /// The sweep at the top of `sync` asks "which rows are not in this TOML",
265 /// and a TOML describes one product. Unscoped, syncing pom at startup would
266 /// answer "all of MNW's" and delete them — and since startup syncs each app
267 /// in turn, the last one to run would be the only one left standing.
268 #[tokio::test]
269 async fn syncing_one_app_does_not_touch_another() {
270 let pool = fresh_pool().await;
271 let mnw = AppId::new("mnw");
272 let pom = AppId::new("pom");
273 sync(
274 &pool,
275 &mnw,
276 &topo(vec![tier("host", true, vec![node("n1")])]),
277 )
278 .await
279 .unwrap();
280 sync(
281 &pool,
282 &pom,
283 &topo(vec![tier("host", true, vec![node("n2")])]),
284 )
285 .await
286 .unwrap();
287
288 // Both survive, and a tier name they share is two rows, not one.
289 let tiers: Vec<(String, String)> =
290 sqlx::query_as("SELECT app, name FROM tiers ORDER BY app")
291 .fetch_all(&pool)
292 .await
293 .unwrap();
294 assert_eq!(
295 tiers,
296 vec![
297 ("mnw".to_string(), "host".to_string()),
298 ("pom".to_string(), "host".to_string())
299 ]
300 );
301 let nodes: Vec<(String, String)> =
302 sqlx::query_as("SELECT app, name FROM nodes ORDER BY app")
303 .fetch_all(&pool)
304 .await
305 .unwrap();
306 assert_eq!(
307 nodes,
308 vec![
309 ("mnw".to_string(), "n1".to_string()),
310 ("pom".to_string(), "n2".to_string())
311 ]
312 );
313 // And one tier_state row each, not one shared.
314 let states: i64 = sqlx::query_scalar("SELECT COUNT(*) FROM tier_state")
315 .fetch_one(&pool)
316 .await
317 .unwrap();
318 assert_eq!(states, 2);
319 }
320
321 /// A tier pinned in one product does not block removing the same-named tier
322 /// from another.
323 #[tokio::test]
324 async fn a_pin_in_one_app_does_not_block_another_apps_edit() {
325 let pool = fresh_pool().await;
326 let mnw = AppId::new("mnw");
327 let pom = AppId::new("pom");
328 let two = topo(vec![tier("host", true, vec![]), tier("a", true, vec![])]);
329 sync(&pool, &mnw, &two).await.unwrap();
330 sync(&pool, &pom, &two).await.unwrap();
331
332 // MNW pins a version on tier a.
333 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')")
334 .execute(&pool).await.unwrap();
335 sqlx::query(
336 "UPDATE tier_state SET current_version = '0.1.0' WHERE app = 'mnw' AND tier = 'a'",
337 )
338 .execute(&pool)
339 .await
340 .unwrap();
341
342 // pom dropping ITS tier a is fine; MNW's pin is not pom's business.
343 sync(&pool, &pom, &topo(vec![tier("host", true, vec![])]))
344 .await
345 .unwrap();
346 // MNW dropping the same tier is still refused.
347 let err = sync(&pool, &mnw, &topo(vec![tier("host", true, vec![])]))
348 .await
349 .unwrap_err();
350 assert!(err.to_string().contains("tier_state still pins"), "{err}");
351 }
352
353 #[tokio::test]
354 async fn refuses_to_drop_tier_with_pinned_version() {
355 let pool = fresh_pool().await;
356 let t1 = topo(vec![tier("host", true, vec![]), tier("a", true, vec![])]);
357 sync(&pool, &app(), &t1).await.unwrap();
358
359 // Simulate a version being deployed on tier a.
360 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')")
361 .execute(&pool).await.unwrap();
362 sqlx::query(
363 "UPDATE tier_state SET current_version = '0.1.0' WHERE app = 'mnw' AND tier = 'a'",
364 )
365 .execute(&pool)
366 .await
367 .unwrap();
368
369 let t2 = topo(vec![tier("host", true, vec![])]);
370 let err = sync(&pool, &app(), &t2).await.unwrap_err();
371 assert!(
372 err.to_string().contains("tier_state still pins"),
373 "got: {err}"
374 );
375 }
376 }
377