Skip to main content

max / makenotwork

23.3 KB · 575 lines History Blame Raw
1 //! Startup crash-recovery reconcile of `deploys` against `tier_state`.
2 //!
3 //! A promote lands a build on a tier's nodes and *then* advances tier_state:
4 //! [`crate::routes`] inserts the per-node `deploys` row (outcome `ok`) inside the
5 //! canary loop, and only after every node succeeds does [`crate::runs::advance_tier`]
6 //! record the new version. If sandod dies between those two writes — OOM, a
7 //! `systemctl restart`, a self-update SIGKILL — the node is serving the new
8 //! version but tier_state still names the old one, and `partial_reason` is NULL
9 //! because no failure path executed. `/state` then shows a clean tier on the
10 //! wrong version, and a later `/rollback` walks back to `previous_version`, which
11 //! is now *two* versions behind what the node actually runs (audit 2026-07-21,
12 //! sando-h5-crash-mid-promote).
13 //!
14 //! [`recover_unrecorded_deploys`] runs once at startup, next to
15 //! [`crate::runs::recover_orphaned_running`], and flags any tier whose latest
16 //! successful deploy postdates the last recorded state change yet names a
17 //! different version/build than tier_state holds. The flag is a `partial_reason`,
18 //! the same channel the promote/rollback failure paths use, so `/state` and the
19 //! TUI go loud and force the operator to reconcile by hand rather than trusting a
20 //! stale-green tier.
21 //!
22 //! The comparison instant is `tier_state.advanced_at` (migration 009), set on
23 //! every advance and rollback and never nulled — unlike `burn_in_started_at`,
24 //! which rollback and `reset_burn_in` clear while the deployed identity stays
25 //! real. A clean promote sets `advanced_at` *after* its `deploys` row, so its own
26 //! landing never postdates it; a rollback sets it after the roll-off row, so a
27 //! cleanly rolled-back tier is not mistaken for an unrecorded deploy.
28 //!
29 //! The `deploys` row is written `in_progress` *before* the node is touched and
30 //! finalized to ok/failed right after, so the only sliver still uncovered is a
31 //! crash between the pre-write and the swap — where the node was never changed,
32 //! so there is nothing to reconcile. Any row still `in_progress` at startup is an
33 //! orphaned in-flight deploy: it is flagged (the swap may or may not have landed)
34 //! and settled to a terminal `failed` so the audit trail carries no eternal
35 //! in-flight row.
36
37 use anyhow::Result;
38 use chrono::{DateTime, Utc};
39 use sqlx::{Row, SqlitePool};
40
41 /// One tier's most recent deploy row, whatever its outcome.
42 struct LastDeploy {
43 version: String,
44 build_id: Option<i64>,
45 /// `'ok' | 'failed' | 'in_progress'`.
46 outcome: String,
47 /// `finished_at` when the row is terminal, else `started_at` — the instant to
48 /// weigh against the tier's last state change. `started_at` is NOT NULL.
49 instant: String,
50 }
51
52 /// Reconcile `deploys` against `tier_state` after a crash. Flags any tier whose
53 /// most recent deploy postdates the last recorded state change but disagrees with
54 /// what tier_state holds (a promote that died before advancing) or is still
55 /// in-flight (a promote interrupted mid-node). Sets `partial_reason` — only when
56 /// it is currently NULL, so a more specific promote/rollback failure reason
57 /// stands — and settles orphaned `in_progress` rows. Returns how many tiers were
58 /// flagged. Run once at startup, before serving.
59 pub async fn recover_unrecorded_deploys(pool: &SqlitePool) -> Result<u64> {
60 let tiers = sqlx::query(
61 "SELECT tier, current_version, current_build_id, advanced_at, partial_reason
62 FROM tier_state",
63 )
64 .fetch_all(pool)
65 .await?;
66
67 let mut flagged = 0u64;
68 for row in tiers {
69 let tier: String = row.get("tier");
70 // A tier already flagged partial carries a more specific reason from the
71 // path that flagged it; don't overwrite it.
72 if row.get::<Option<String>, _>("partial_reason").is_some() {
73 continue;
74 }
75 let Some(landing) = latest_deploy(pool, &tier).await? else {
76 continue;
77 };
78 let current_version: Option<String> = row.get("current_version");
79 let current_build_id: Option<i64> = row.get("current_build_id");
80 let advanced_at: Option<String> = row.get("advanced_at");
81
82 // Only a deploy that postdates the last recorded state change is suspect:
83 // a clean promote/rollback stamps `advanced_at` after its own rows.
84 if !postdates(&landing.instant, advanced_at.as_deref()) {
85 continue;
86 }
87 let reason = match landing.outcome.as_str() {
88 // Interrupted mid-node: the swap may or may not have landed, and no
89 // outcome was ever recorded. Force a look regardless of identity.
90 "in_progress" => Some(format!(
91 "crash-recovery: a deploy of {deployed} to this tier was in flight when the \
92 daemon last stopped and never recorded an outcome. The node's binary may or \
93 may not have been swapped. Verify what the tier is actually running, then \
94 re-promote or roll back deliberately.",
95 deployed = describe(&landing.version, landing.build_id),
96 )),
97 // Landed successfully but tier_state names something else — the promote
98 // died after the deploy but before `advance_tier`.
99 "ok" if !identity_matches(&landing, current_version.as_deref(), current_build_id) => {
100 Some(format!(
101 "crash-recovery: node(s) on this tier were deployed {deployed} at {when} but \
102 tier_state still records {recorded}. A promote most likely died after the \
103 deploy landed but before tier_state advanced. Verify what the tier is \
104 actually running, then re-promote or roll back deliberately — a blind \
105 /rollback would target the wrong version.",
106 deployed = describe(&landing.version, landing.build_id),
107 when = landing.instant,
108 recorded = current_version.as_deref().unwrap_or("no version"),
109 ))
110 }
111 // A terminal 'failed' row was handled by the deploy's own failure path
112 // (which sets its own partial_reason); nothing to add here.
113 _ => None,
114 };
115 let Some(reason) = reason else { continue };
116
117 // `AND partial_reason IS NULL` keeps the write a no-op if anything set a
118 // reason since the read above (startup is single-threaded, but the guard
119 // costs nothing and states the intent).
120 let res = sqlx::query(
121 "UPDATE tier_state SET partial_reason = ? WHERE tier = ? AND partial_reason IS NULL",
122 )
123 .bind(&reason)
124 .bind(&tier)
125 .execute(pool)
126 .await?;
127 if res.rows_affected() > 0 {
128 tracing::error!(tier = %tier, %reason, "startup reconcile: interrupted/unrecorded deploy on tier");
129 flagged += 1;
130 }
131 }
132
133 // Settle any orphaned in-flight deploy rows so the trail carries no eternal
134 // 'in_progress'. Done after the flag decisions above, which read that state.
135 // At startup (single-threaded, pre-serving) any 'in_progress' row is an orphan.
136 settle_orphaned_in_progress(pool).await?;
137 Ok(flagged)
138 }
139
140 /// The most recent deploy on `tier`, any outcome. Rows are ordered by `id`
141 /// (monotonic autoincrement), so this is the last one regardless of clock skew.
142 async fn latest_deploy(pool: &SqlitePool, tier: &str) -> Result<Option<LastDeploy>> {
143 let row = sqlx::query(
144 "SELECT version, build_id, outcome, started_at, finished_at FROM deploys
145 WHERE tier = ? ORDER BY id DESC LIMIT 1",
146 )
147 .bind(tier)
148 .fetch_optional(pool)
149 .await?;
150 Ok(row.map(|r| {
151 let started_at: String = r.get("started_at");
152 let finished_at: Option<String> = r.get("finished_at");
153 LastDeploy {
154 version: r.get("version"),
155 build_id: r.get("build_id"),
156 outcome: r.get("outcome"),
157 instant: finished_at.unwrap_or(started_at),
158 }
159 }))
160 }
161
162 /// Mark every `in_progress` deploy row terminal-failed with an interruption note,
163 /// mirroring [`crate::runs::recover_orphaned_running`] for build runs. Returns the
164 /// number of rows settled.
165 async fn settle_orphaned_in_progress(pool: &SqlitePool) -> Result<u64> {
166 let outcome =
167 crate::outcome::DeployOutcome::failed(crate::outcome::DeployFailureKind::Unclassified {
168 detail: "daemon restarted mid-deploy".into(),
169 });
170 let outcome_json = serde_json::to_string(&outcome)?;
171 let res = sqlx::query(
172 "UPDATE deploys SET outcome = 'failed', finished_at = ?, outcome_json = ?
173 WHERE outcome = 'in_progress'",
174 )
175 .bind(Utc::now().to_rfc3339())
176 .bind(&outcome_json)
177 .execute(pool)
178 .await?;
179 let n = res.rows_affected();
180 if n > 0 {
181 tracing::warn!(
182 settled = n,
183 "startup reconcile: settled orphaned in-flight deploy row(s)"
184 );
185 }
186 Ok(n)
187 }
188
189 /// Whether `deploy_finished_at` is later than the tier's last recorded state
190 /// change. A NULL `advanced_at` (state never advanced, yet a deploy landed) and
191 /// any unparseable timestamp both count as postdating — fail toward flagging, so
192 /// a corrupt or missing instant forces operator attention rather than hiding.
193 fn postdates(deploy_finished_at: &str, advanced_at: Option<&str>) -> bool {
194 let Some(advanced_at) = advanced_at else {
195 return true;
196 };
197 match (
198 DateTime::parse_from_rfc3339(deploy_finished_at),
199 DateTime::parse_from_rfc3339(advanced_at),
200 ) {
201 (Ok(d), Ok(a)) => d > a,
202 _ => true,
203 }
204 }
205
206 /// Whether the landed artifact is the one tier_state records. Build identity wins
207 /// when both sides carry it (a re-deploy of the same version from a different
208 /// build is still a mismatch); otherwise fall back to the version string. A NULL
209 /// `current_version` with a real landing is a mismatch.
210 fn identity_matches(
211 landing: &LastDeploy,
212 current_version: Option<&str>,
213 current_build_id: Option<i64>,
214 ) -> bool {
215 match (landing.build_id, current_build_id) {
216 (Some(landed), Some(current)) => landed == current,
217 _ => current_version == Some(landing.version.as_str()),
218 }
219 }
220
221 /// Human label for a landed artifact: the version, plus the build id when known.
222 fn describe(version: &str, build_id: Option<i64>) -> String {
223 match build_id {
224 Some(id) => format!("{version} (build {id})"),
225 None => version.to_string(),
226 }
227 }
228
229 #[cfg(test)]
230 mod tests {
231 use super::*;
232 use chrono::{Duration, Utc};
233 use sqlx::sqlite::SqlitePoolOptions;
234
235 async fn pool() -> SqlitePool {
236 let pool = SqlitePoolOptions::new()
237 .max_connections(1)
238 .connect("sqlite::memory:")
239 .await
240 .unwrap();
241 crate::db::migrate(&pool).await.unwrap();
242 pool
243 }
244
245 async fn seed_tier(pool: &SqlitePool, tier: &str) {
246 sqlx::query(
247 "INSERT INTO tiers (name, ord, provisioned, canary) VALUES (?, 0, 1, 'sequential')",
248 )
249 .bind(tier)
250 .execute(pool)
251 .await
252 .unwrap();
253 sqlx::query("INSERT INTO tier_state (tier) VALUES (?)")
254 .bind(tier)
255 .execute(pool)
256 .await
257 .unwrap();
258 }
259
260 /// Set the tier's recorded state the way an advance/rollback would.
261 async fn set_state(
262 pool: &SqlitePool,
263 tier: &str,
264 current_version: Option<&str>,
265 advanced_at: Option<&str>,
266 ) {
267 if let Some(v) = current_version {
268 sqlx::query(
269 "INSERT OR IGNORE INTO versions (version, git_sha, built_at, artifact_path)
270 VALUES (?, 'sha', '2026-07-21T00:00:00Z', '/r/x')",
271 )
272 .bind(v)
273 .execute(pool)
274 .await
275 .unwrap();
276 }
277 sqlx::query("UPDATE tier_state SET current_version = ?, advanced_at = ? WHERE tier = ?")
278 .bind(current_version)
279 .bind(advanced_at)
280 .bind(tier)
281 .execute(pool)
282 .await
283 .unwrap();
284 }
285
286 /// FK parents `deploys` needs: a `versions` row and a `nodes` row. sqlx
287 /// enables `foreign_keys` by default, so these must exist first.
288 async fn seed_deploy_fks(pool: &SqlitePool, tier: &str, version: &str) {
289 sqlx::query(
290 "INSERT OR IGNORE INTO versions (version, git_sha, built_at, artifact_path)
291 VALUES (?, 'sha', '2026-07-21T00:00:00Z', '/r/x')",
292 )
293 .bind(version)
294 .execute(pool)
295 .await
296 .unwrap();
297 sqlx::query(
298 "INSERT OR IGNORE INTO nodes (name, tier, ssh_target, release_root)
299 VALUES ('n1', ?, 'local', '/opt/mnw')",
300 )
301 .bind(tier)
302 .execute(pool)
303 .await
304 .unwrap();
305 }
306
307 /// Insert a `deploys` row, seeding its FK parents first.
308 async fn insert_deploy(
309 pool: &SqlitePool,
310 tier: &str,
311 version: &str,
312 outcome: &str,
313 finished_at: &str,
314 ) {
315 seed_deploy_fks(pool, tier, version).await;
316 sqlx::query(
317 "INSERT INTO deploys (version, tier, node, started_at, finished_at, outcome)
318 VALUES (?, ?, 'n1', ?, ?, ?)",
319 )
320 .bind(version)
321 .bind(tier)
322 .bind(finished_at)
323 .bind(finished_at)
324 .bind(outcome)
325 .execute(pool)
326 .await
327 .unwrap();
328 }
329
330 async fn partial_reason(pool: &SqlitePool, tier: &str) -> Option<String> {
331 sqlx::query_scalar("SELECT partial_reason FROM tier_state WHERE tier = ?")
332 .bind(tier)
333 .fetch_one(pool)
334 .await
335 .unwrap()
336 }
337
338 #[tokio::test]
339 async fn clean_promote_is_not_flagged() {
340 // deploy lands, THEN tier_state advances: advanced_at postdates the row.
341 let pool = pool().await;
342 seed_tier(&pool, "b").await;
343 let deployed = Utc::now();
344 insert_deploy(&pool, "b", "0.10.15", "ok", &deployed.to_rfc3339()).await;
345 set_state(
346 &pool,
347 "b",
348 Some("0.10.15"),
349 Some(&(deployed + Duration::seconds(1)).to_rfc3339()),
350 )
351 .await;
352
353 assert_eq!(recover_unrecorded_deploys(&pool).await.unwrap(), 0);
354 assert!(partial_reason(&pool, "b").await.is_none());
355 }
356
357 #[tokio::test]
358 async fn crash_mid_promote_is_flagged() {
359 // tier_state still on the OLD version; an 'ok' deploy of the NEW version
360 // postdates the last advance — the exact SIGKILL-before-advance window.
361 let pool = pool().await;
362 seed_tier(&pool, "b").await;
363 let advanced = Utc::now() - Duration::hours(1);
364 set_state(&pool, "b", Some("0.10.14"), Some(&advanced.to_rfc3339())).await;
365 insert_deploy(&pool, "b", "0.10.15", "ok", &Utc::now().to_rfc3339()).await;
366
367 assert_eq!(recover_unrecorded_deploys(&pool).await.unwrap(), 1);
368 let reason = partial_reason(&pool, "b").await.expect("flagged");
369 assert!(reason.contains("0.10.15"), "names the deployed version");
370 assert!(reason.contains("0.10.14"), "names the recorded version");
371 assert!(reason.contains("crash-recovery"));
372 }
373
374 #[tokio::test]
375 async fn rolled_back_tier_is_not_flagged() {
376 // Promote to 0.10.15 landed (ok deploy), then a rollback moved tier_state
377 // to 0.10.14 and stamped advanced_at AFTER the roll-off deploy. The stale
378 // deploy row must not read as unrecorded.
379 let pool = pool().await;
380 seed_tier(&pool, "b").await;
381 let deployed = Utc::now() - Duration::minutes(5);
382 insert_deploy(&pool, "b", "0.10.15", "ok", &deployed.to_rfc3339()).await;
383 set_state(&pool, "b", Some("0.10.14"), Some(&Utc::now().to_rfc3339())).await;
384
385 assert_eq!(recover_unrecorded_deploys(&pool).await.unwrap(), 0);
386 assert!(partial_reason(&pool, "b").await.is_none());
387 }
388
389 #[tokio::test]
390 async fn an_already_partial_tier_keeps_its_reason() {
391 let pool = pool().await;
392 seed_tier(&pool, "b").await;
393 let advanced = Utc::now() - Duration::hours(1);
394 set_state(&pool, "b", Some("0.10.14"), Some(&advanced.to_rfc3339())).await;
395 insert_deploy(&pool, "b", "0.10.15", "ok", &Utc::now().to_rfc3339()).await;
396 sqlx::query(
397 "UPDATE tier_state SET partial_reason = 'canary rollback incomplete' WHERE tier = 'b'",
398 )
399 .execute(&pool)
400 .await
401 .unwrap();
402
403 assert_eq!(recover_unrecorded_deploys(&pool).await.unwrap(), 0);
404 assert_eq!(
405 partial_reason(&pool, "b").await.as_deref(),
406 Some("canary rollback incomplete"),
407 "the specific reason is not clobbered",
408 );
409 }
410
411 #[tokio::test]
412 async fn a_failed_deploy_does_not_flag() {
413 // The latest deploy is a failure (its own path handles partial state);
414 // there is no successful landing that tier_state failed to record.
415 let pool = pool().await;
416 seed_tier(&pool, "b").await;
417 let advanced = Utc::now() - Duration::hours(1);
418 set_state(&pool, "b", Some("0.10.14"), Some(&advanced.to_rfc3339())).await;
419 insert_deploy(&pool, "b", "0.10.15", "failed", &Utc::now().to_rfc3339()).await;
420
421 assert_eq!(recover_unrecorded_deploys(&pool).await.unwrap(), 0);
422 assert!(partial_reason(&pool, "b").await.is_none());
423 }
424
425 #[tokio::test]
426 async fn a_tier_with_no_deploys_is_skipped() {
427 let pool = pool().await;
428 seed_tier(&pool, "b").await;
429 set_state(&pool, "b", None, None).await;
430 assert_eq!(recover_unrecorded_deploys(&pool).await.unwrap(), 0);
431 }
432
433 #[tokio::test]
434 async fn crash_on_first_ever_promote_is_flagged() {
435 // Never advanced (advanced_at NULL, current_version NULL) but a node was
436 // deployed: prod is running something tier_state has no record of.
437 let pool = pool().await;
438 seed_tier(&pool, "b").await;
439 insert_deploy(&pool, "b", "0.10.15", "ok", &Utc::now().to_rfc3339()).await;
440
441 assert_eq!(recover_unrecorded_deploys(&pool).await.unwrap(), 1);
442 assert!(
443 partial_reason(&pool, "b")
444 .await
445 .unwrap()
446 .contains("0.10.15")
447 );
448 }
449
450 #[tokio::test]
451 async fn build_id_mismatch_is_flagged_even_when_version_matches() {
452 // Same version string, different build — a re-deploy landed a build the
453 // tier never recorded. Build identity is the tie-breaker.
454 let pool = pool().await;
455 seed_tier(&pool, "b").await;
456 seed_deploy_fks(&pool, "b", "0.10.15").await;
457 // build_runs 7 and 9, referenced by tier_state.current_build_id and
458 // deploys.build_id respectively.
459 for id in [7, 9] {
460 sqlx::query(
461 "INSERT INTO build_runs (id, sha, phase, result, started_at)
462 VALUES (?, 'sha', 'done', 'passed', '2026-07-21T00:00:00Z')",
463 )
464 .bind(id)
465 .execute(&pool)
466 .await
467 .unwrap();
468 }
469 let advanced = Utc::now() - Duration::hours(1);
470 // tier_state: version 0.10.15, build 7.
471 sqlx::query(
472 "UPDATE tier_state SET current_version = '0.10.15', current_build_id = 7, advanced_at = ? WHERE tier = 'b'",
473 )
474 .bind(advanced.to_rfc3339())
475 .execute(&pool)
476 .await
477 .unwrap();
478 // deploy: same version 0.10.15 but build 9, postdating the advance.
479 sqlx::query(
480 "INSERT INTO deploys (version, tier, node, started_at, finished_at, outcome, build_id)
481 VALUES ('0.10.15', 'b', 'n1', ?, ?, 'ok', 9)",
482 )
483 .bind(Utc::now().to_rfc3339())
484 .bind(Utc::now().to_rfc3339())
485 .execute(&pool)
486 .await
487 .unwrap();
488
489 assert_eq!(recover_unrecorded_deploys(&pool).await.unwrap(), 1);
490 }
491
492 /// Insert an in-flight deploy row: `outcome = 'in_progress'`, `finished_at`
493 /// NULL, `started_at` set — the shape the promote loop writes before touching
494 /// a node.
495 async fn insert_in_progress(pool: &SqlitePool, tier: &str, version: &str, started_at: &str) {
496 seed_deploy_fks(pool, tier, version).await;
497 sqlx::query(
498 "INSERT INTO deploys (version, tier, node, started_at, outcome)
499 VALUES (?, ?, 'n1', ?, 'in_progress')",
500 )
501 .bind(version)
502 .bind(tier)
503 .bind(started_at)
504 .execute(pool)
505 .await
506 .unwrap();
507 }
508
509 async fn outcome_of_latest(pool: &SqlitePool, tier: &str) -> String {
510 sqlx::query_scalar("SELECT outcome FROM deploys WHERE tier = ? ORDER BY id DESC LIMIT 1")
511 .bind(tier)
512 .fetch_one(pool)
513 .await
514 .unwrap()
515 }
516
517 #[tokio::test]
518 async fn orphaned_in_progress_deploy_is_flagged_and_settled() {
519 // A deploy was in flight (row written before the swap) when the daemon
520 // died: flag the tier and settle the row so it isn't in_progress forever.
521 let pool = pool().await;
522 seed_tier(&pool, "b").await;
523 let advanced = Utc::now() - Duration::hours(1);
524 set_state(&pool, "b", Some("0.10.14"), Some(&advanced.to_rfc3339())).await;
525 insert_in_progress(&pool, "b", "0.10.15", &Utc::now().to_rfc3339()).await;
526
527 assert_eq!(recover_unrecorded_deploys(&pool).await.unwrap(), 1);
528 let reason = partial_reason(&pool, "b").await.expect("flagged");
529 assert!(
530 reason.contains("in flight"),
531 "names the interruption: {reason}"
532 );
533 assert!(reason.contains("0.10.15"));
534 // The orphan row is settled to a terminal outcome.
535 assert_eq!(outcome_of_latest(&pool, "b").await, "failed");
536 }
537
538 #[tokio::test]
539 async fn in_progress_row_predating_last_advance_is_only_settled_not_flagged() {
540 // An old in_progress row from before the tier's last clean advance is
541 // stale bookkeeping, not an unrecorded deploy: settle it, don't flag.
542 let pool = pool().await;
543 seed_tier(&pool, "b").await;
544 let old = Utc::now() - Duration::hours(2);
545 insert_in_progress(&pool, "b", "0.10.13", &old.to_rfc3339()).await;
546 // A later clean promote advanced the tier past that row.
547 let deployed = Utc::now() - Duration::minutes(10);
548 insert_deploy(&pool, "b", "0.10.14", "ok", &deployed.to_rfc3339()).await;
549 set_state(&pool, "b", Some("0.10.14"), Some(&Utc::now().to_rfc3339())).await;
550
551 assert_eq!(recover_unrecorded_deploys(&pool).await.unwrap(), 0);
552 assert!(partial_reason(&pool, "b").await.is_none());
553 // But the stale in_progress row is still settled.
554 let leftover: i64 =
555 sqlx::query_scalar("SELECT COUNT(*) FROM deploys WHERE outcome = 'in_progress'")
556 .fetch_one(&pool)
557 .await
558 .unwrap();
559 assert_eq!(leftover, 0, "no in_progress rows survive");
560 }
561
562 #[tokio::test]
563 async fn is_idempotent_after_flagging() {
564 let pool = pool().await;
565 seed_tier(&pool, "b").await;
566 let advanced = Utc::now() - Duration::hours(1);
567 set_state(&pool, "b", Some("0.10.14"), Some(&advanced.to_rfc3339())).await;
568 insert_deploy(&pool, "b", "0.10.15", "ok", &Utc::now().to_rfc3339()).await;
569
570 assert_eq!(recover_unrecorded_deploys(&pool).await.unwrap(), 1);
571 // Second pass: the reason is already set, so nothing new is flagged.
572 assert_eq!(recover_unrecorded_deploys(&pool).await.unwrap(), 0);
573 }
574 }
575