Skip to main content

max / makenotwork

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