Skip to main content

max / makenotwork

19.6 KB · 533 lines History Blame Raw
1 //! Build-run tracking: one `build_runs` row per `/rebuild`, updated as the
2 //! pipeline moves through its phases, terminating in passed/failed/aborted.
3 //!
4 //! This is the resource that makes Sando driveable headlessly. `/state` only
5 //! ever reflects the last *successful* deploy, so on a red pipeline a poller
6 //! of `/state` sees stale-green for the whole build (the 0.10.2 incident). A
7 //! `RunId` returned by `/rebuild` + `GET /runs/{id}` gives a non-TUI caller
8 //! one pollable resource tied to the build it triggered, carrying the phase,
9 //! the per-gate status, and — the highest-value bit — a `failure_summary`
10 //! (first compile error / first failed gate) so the cause is in the API, not
11 //! behind `sudo journalctl`.
12 //!
13 //! Terminal writes (`mark_passed`/`mark_failed`/`mark_aborted`) are guarded on
14 //! `result = 'building'`, so whichever site settles the run first wins: a
15 //! build-step compile error, the first red gate, or the task-level catch for
16 //! pre-build bails. Later writes are silent no-ops.
17
18 use crate::domain::{RunId, Version};
19 use anyhow::Result;
20 use chrono::Utc;
21 use serde::Serialize;
22 use sqlx::{Row, SqlitePool};
23
24 /// In-flight sub-state. Plain strings in the DB; this enum names the values so
25 /// call sites can't typo them.
26 #[derive(Debug, Clone, Copy)]
27 pub enum Phase {
28 Fetching,
29 Compiling,
30 Staging,
31 Gating,
32 }
33
34 impl Phase {
35 pub fn as_str(self) -> &'static str {
36 match self {
37 Phase::Fetching => "fetching",
38 Phase::Compiling => "compiling",
39 Phase::Staging => "staging",
40 Phase::Gating => "gating",
41 }
42 }
43 }
44
45 /// Insert a fresh `building` run for `sha` and return its id.
46 pub async fn create(pool: &SqlitePool, sha: &str) -> Result<RunId> {
47 let id: i64 = sqlx::query_scalar(
48 "INSERT INTO build_runs (sha, phase, result, started_at)
49 VALUES (?, 'queued', 'building', ?) RETURNING id",
50 )
51 .bind(sha)
52 .bind(Utc::now().to_rfc3339())
53 .fetch_one(pool)
54 .await?;
55 Ok(RunId(id))
56 }
57
58 /// Advance the in-flight phase. No-op once the run is terminal so a late
59 /// phase write can't resurrect a finished row.
60 pub async fn set_phase(pool: &SqlitePool, run_id: RunId, phase: Phase) -> Result<()> {
61 sqlx::query("UPDATE build_runs SET phase = ? WHERE id = ? AND result = 'building'")
62 .bind(phase.as_str())
63 .bind(run_id.0)
64 .execute(pool)
65 .await?;
66 Ok(())
67 }
68
69 /// Forward-advance a tier to `version` in a single atomic UPDATE. `previous_version`
70 /// is set from the row's *old* `current_version` (SQLite evaluates every RHS against
71 /// the original row), so there is no read-modify-write to lose under concurrency
72 /// (CF3) — no separate SELECT exists to race. `burn_in_started_at = now` starts the
73 /// tier's burn-in clock.
74 ///
75 /// This is the *only* forward-advance writer of `tier_state`. The host build path
76 /// and `/promote` both go through it; keeping the fetch-then-write shape out of the
77 /// codebase is the point (ultra-fuzz Run 2, S1). Callers MUST hold `deploy_lock` so
78 /// the logical advance is serialized against `/rollback`.
79 ///
80 /// Returns the raw `sqlx::Error` so the route layer can map it to its typed
81 /// `Error::Db`; anyhow callers (the build pipeline) get `?`-conversion for free.
82 /// `build_id` is the `build_runs.id` of the build landing on the tier — the
83 /// artifact identity (wiki [[release-artifact-identity]]). It advances in
84 /// lockstep with the version label: `previous_build_id` takes the old
85 /// `current_build_id` in the same self-referential UPDATE, so the build the
86 /// tier just stepped off is the rollback target. `None` for legacy/pre-identity
87 /// callers writes NULL — treated as "no recorded build" downstream, with the
88 /// version-string path as the fallback.
89 ///
90 /// `burn_in_started_at = now` starts the tier's burn-in clock. Because the
91 /// clock resets on every advance and the clock is what burn-in reads, the clock
92 /// always belongs to the build now current on the tier — this is what stops a
93 /// promote from crediting one build with another build's elapsed burn-in.
94 ///
95 /// `advanced_at = now` records when the tier's deployed identity last changed.
96 /// It tracks the burn-in clock on an advance but, unlike it, is never nulled by
97 /// rollback or reset_burn_in — so the startup reconcile can trust it as the
98 /// instant to compare a `deploys` row against (wiki [[sando-overview]], migration
99 /// 009 / [`crate::reconcile`]).
100 pub async fn advance_tier(
101 pool: &SqlitePool,
102 tier: &str,
103 version: &Version,
104 build_id: Option<i64>,
105 ) -> Result<(), sqlx::Error> {
106 let now = Utc::now().to_rfc3339();
107 sqlx::query(
108 "UPDATE tier_state
109 SET previous_version = current_version,
110 current_version = ?,
111 previous_build_id = current_build_id,
112 current_build_id = ?,
113 burn_in_started_at = ?,
114 advanced_at = ?
115 WHERE tier = ?",
116 )
117 .bind(version)
118 .bind(build_id)
119 .bind(&now)
120 .bind(&now)
121 .bind(tier)
122 .execute(pool)
123 .await?;
124 Ok(())
125 }
126
127 /// Record the version once it's been read from the worktree's Cargo.toml.
128 pub async fn set_version(pool: &SqlitePool, run_id: RunId, version: &Version) -> Result<()> {
129 sqlx::query("UPDATE build_runs SET version = ? WHERE id = ? AND result = 'building'")
130 .bind(version.to_string())
131 .bind(run_id.0)
132 .execute(pool)
133 .await?;
134 Ok(())
135 }
136
137 /// Record the build's content identity once its bundle has been staged and
138 /// hashed: the full 64-hex `bundle_digest` and the `staged_path` it lives at.
139 /// This is what promote/burn-in/retention key on once build id becomes the
140 /// identity (wiki [[release-artifact-identity]]). Guarded on `building` so a
141 /// settled run is never mutated.
142 pub async fn set_identity(
143 pool: &SqlitePool,
144 run_id: RunId,
145 bundle_digest: &str,
146 staged_path: &str,
147 ) -> Result<()> {
148 sqlx::query(
149 "UPDATE build_runs SET bundle_digest = ?, staged_path = ?
150 WHERE id = ? AND result = 'building'",
151 )
152 .bind(bundle_digest)
153 .bind(staged_path)
154 .bind(run_id.0)
155 .execute(pool)
156 .await?;
157 Ok(())
158 }
159
160 /// Settle the run green. First terminal write wins (guarded on `building`).
161 pub async fn mark_passed(pool: &SqlitePool, run_id: RunId) -> Result<()> {
162 sqlx::query(
163 "UPDATE build_runs SET result = 'passed', phase = 'done', finished_at = ?
164 WHERE id = ? AND result = 'building'",
165 )
166 .bind(Utc::now().to_rfc3339())
167 .bind(run_id.0)
168 .execute(pool)
169 .await?;
170 Ok(())
171 }
172
173 /// Settle the run red with a human-readable cause. First terminal write wins,
174 /// so the most specific failure (build compile error, first red gate) recorded
175 /// before the task-level catch is the one that sticks.
176 pub async fn mark_failed(pool: &SqlitePool, run_id: RunId, summary: &str) -> Result<()> {
177 // Bound the stored summary — it's a headline, not the log. The full output
178 // is at the gate's log_ref / journald.
179 let summary: String = summary.chars().take(600).collect();
180 sqlx::query(
181 "UPDATE build_runs SET result = 'failed', phase = 'done', failure_summary = ?, finished_at = ?
182 WHERE id = ? AND result = 'building'",
183 )
184 .bind(&summary)
185 .bind(Utc::now().to_rfc3339())
186 .bind(run_id.0)
187 .execute(pool)
188 .await?;
189 Ok(())
190 }
191
192 /// Settle the run as superseded by a newer `/rebuild`.
193 pub async fn mark_aborted(pool: &SqlitePool, run_id: RunId) -> Result<()> {
194 sqlx::query(
195 "UPDATE build_runs SET result = 'aborted', phase = 'done',
196 failure_summary = 'superseded by a newer /rebuild', finished_at = ?
197 WHERE id = ? AND result = 'building'",
198 )
199 .bind(Utc::now().to_rfc3339())
200 .bind(run_id.0)
201 .execute(pool)
202 .await?;
203 Ok(())
204 }
205
206 /// Settle any `build_runs` left `result = 'building'` by a daemon that died
207 /// mid-build (crash, OOM, `systemctl restart`, a self-update that SIGKILLed an
208 /// in-flight build). Without this they stay `'building'` forever and `/state` +
209 /// `GET /runs/{id}/wait` report an ever-growing elapsed for a build that will
210 /// never settle (the Run-2 SERIOUS-2 gap). Run once at startup before serving.
211 /// Returns the number of orphaned runs reconciled.
212 pub async fn recover_orphaned_running(pool: &SqlitePool) -> Result<u64> {
213 let res = sqlx::query(
214 "UPDATE build_runs SET result = 'aborted', phase = 'done',
215 failure_summary = 'daemon restarted mid-build', finished_at = ?
216 WHERE result = 'building'",
217 )
218 .bind(Utc::now().to_rfc3339())
219 .execute(pool)
220 .await?;
221 Ok(res.rows_affected())
222 }
223
224 /// One gate's status within a run view.
225 #[derive(Debug, Serialize)]
226 pub struct RunGateView {
227 pub kind: String,
228 /// `'passed' | 'failed' | 'blocked'` or NULL while in-flight.
229 pub status: Option<String>,
230 /// Relative path under `cfg.logs_root` for the full byte stream.
231 pub log_ref: Option<String>,
232 }
233
234 /// The `GET /runs/{id}` payload.
235 #[derive(Debug, Serialize)]
236 pub struct RunView {
237 pub run_id: i64,
238 pub sha: String,
239 pub version: Option<String>,
240 pub phase: String,
241 /// `'building' | 'passed' | 'failed' | 'aborted'`.
242 pub result: String,
243 pub started_at: String,
244 pub finished_at: Option<String>,
245 /// Headline cause when `result = 'failed'`: first compile error or first
246 /// red gate. NULL otherwise.
247 pub failure_summary: Option<String>,
248 /// Gates run on the host tier for this run's version, latest row per kind.
249 /// Empty until the run reaches a version + the gating phase.
250 pub gates: Vec<RunGateView>,
251 }
252
253 /// Load a run plus its host-tier gate statuses. `None` if the id is unknown.
254 pub async fn get(pool: &SqlitePool, run_id: RunId) -> Result<Option<RunView>> {
255 let Some(row) = sqlx::query(
256 "SELECT id, sha, version, phase, result, started_at, finished_at, failure_summary
257 FROM build_runs WHERE id = ?",
258 )
259 .bind(run_id.0)
260 .fetch_optional(pool)
261 .await?
262 else {
263 return Ok(None);
264 };
265
266 let version: Option<String> = row.get("version");
267 // Gates are keyed by (tier, version); a build run drives the `host` tier.
268 // Latest row per gate_kind, matching `/state`'s per-tier query shape.
269 let gates: Vec<RunGateView> = if let Some(ver) = version.as_deref() {
270 sqlx::query(
271 "SELECT gate_kind, status, log_ref
272 FROM gate_runs g
273 WHERE tier = 'host' AND version = ?1
274 AND id = (SELECT MAX(id) FROM gate_runs
275 WHERE tier = 'host' AND version = ?1 AND gate_kind = g.gate_kind)
276 ORDER BY gate_kind",
277 )
278 .bind(ver)
279 .fetch_all(pool)
280 .await?
281 .into_iter()
282 .map(|gr| RunGateView {
283 kind: gr.get("gate_kind"),
284 status: gr.get("status"),
285 log_ref: gr.get("log_ref"),
286 })
287 .collect()
288 } else {
289 Vec::new()
290 };
291
292 Ok(Some(RunView {
293 run_id: row.get("id"),
294 sha: row.get("sha"),
295 version,
296 phase: row.get("phase"),
297 result: row.get("result"),
298 started_at: row.get("started_at"),
299 finished_at: row.get("finished_at"),
300 failure_summary: row.get("failure_summary"),
301 gates,
302 }))
303 }
304
305 /// Compact view of the latest build run for `/state`'s liveness line.
306 #[derive(Debug, Serialize)]
307 pub struct BuildSummary {
308 pub run_id: i64,
309 pub sha: String,
310 pub version: Option<String>,
311 pub phase: String,
312 pub result: String,
313 pub failure_summary: Option<String>,
314 /// Seconds from start to finish (or to now while building). Lets a
315 /// `/state` poller show "building <ver>, phase=<x>, elapsed Ns" instead of
316 /// a version frozen at the last success for the whole ~10-min build.
317 pub elapsed_s: i64,
318 }
319
320 /// The most recent build run, for `/state`. `None` until the first `/rebuild`.
321 pub async fn latest_summary(pool: &SqlitePool) -> Result<Option<BuildSummary>> {
322 let Some(row) = sqlx::query(
323 "SELECT id, sha, version, phase, result, failure_summary, started_at, finished_at
324 FROM build_runs ORDER BY id DESC LIMIT 1",
325 )
326 .fetch_optional(pool)
327 .await?
328 else {
329 return Ok(None);
330 };
331 let started_at: String = row.get("started_at");
332 let finished_at: Option<String> = row.get("finished_at");
333 Ok(Some(BuildSummary {
334 run_id: row.get("id"),
335 sha: row.get("sha"),
336 version: row.get("version"),
337 phase: row.get("phase"),
338 result: row.get("result"),
339 failure_summary: row.get("failure_summary"),
340 elapsed_s: elapsed_seconds(&started_at, finished_at.as_deref()),
341 }))
342 }
343
344 /// Seconds between an rfc3339 `started_at` and (`finished_at` or now), clamped
345 /// at 0. A parse failure yields 0 rather than erroring the whole `/state` call.
346 fn elapsed_seconds(started_at: &str, finished_at: Option<&str>) -> i64 {
347 let Ok(start) = chrono::DateTime::parse_from_rfc3339(started_at) else {
348 return 0;
349 };
350 let end = match finished_at {
351 Some(f) => chrono::DateTime::parse_from_rfc3339(f)
352 .map_or_else(|_| Utc::now(), |d| d.with_timezone(&Utc)),
353 None => Utc::now(),
354 };
355 (end - start.with_timezone(&Utc)).num_seconds().max(0)
356 }
357
358 /// The summary of the first failed gate for `version` on the host tier, if
359 /// any — used by the build pipeline to populate `failure_summary` when
360 /// `run_all` reports a red pipeline. Reads the typed `outcome_json` so the
361 /// stored headline matches what the TUI renders.
362 pub async fn first_failed_gate_summary(pool: &SqlitePool, version: &Version) -> Option<String> {
363 let row = sqlx::query(
364 "SELECT gate_kind, outcome_json FROM gate_runs
365 WHERE tier = 'host' AND version = ? AND status = 'failed'
366 ORDER BY id ASC LIMIT 1",
367 )
368 .bind(version.to_string())
369 .fetch_optional(pool)
370 .await
371 .ok()
372 .flatten()?;
373 let kind: String = row.get("gate_kind");
374 let outcome_json: Option<String> = row.get("outcome_json");
375 let summary = outcome_json
376 .and_then(|s| serde_json::from_str::<crate::outcome::GateOutcome>(&s).ok())
377 .map_or_else(
378 || "gate failed".to_string(),
379 |o| match o.status {
380 crate::outcome::GateStatus::Failed { failure } => failure.summary(),
381 other => format!("{other:?}"),
382 },
383 );
384 Some(format!("{kind}: {summary}"))
385 }
386
387 #[cfg(test)]
388 mod tests {
389 use super::*;
390 use sqlx::sqlite::SqlitePoolOptions;
391
392 async fn pool() -> SqlitePool {
393 let pool = SqlitePoolOptions::new()
394 .max_connections(1)
395 .connect("sqlite::memory:")
396 .await
397 .unwrap();
398 crate::db::migrate(&pool).await.unwrap();
399 pool
400 }
401
402 #[tokio::test]
403 async fn create_then_get_roundtrips_building() {
404 let pool = pool().await;
405 let id = create(&pool, "abc1234").await.unwrap();
406 let v = get(&pool, id).await.unwrap().expect("run exists");
407 assert_eq!(v.sha, "abc1234");
408 assert_eq!(v.result, "building");
409 assert_eq!(v.phase, "queued");
410 assert!(v.version.is_none());
411 assert!(v.gates.is_empty());
412 assert!(v.failure_summary.is_none());
413 }
414
415 #[tokio::test]
416 async fn recover_orphaned_running_settles_building_runs() {
417 let pool = pool().await;
418 // Two in-flight runs (as if the daemon died mid-build) + one already
419 // settled, which must be left untouched.
420 let run_a = create(&pool, "aaaaaaa").await.unwrap();
421 let run_b = create(&pool, "bbbbbbb").await.unwrap();
422 let run_c = create(&pool, "ccccccc").await.unwrap();
423 mark_passed(&pool, run_c).await.unwrap();
424
425 let reconciled = recover_orphaned_running(&pool).await.unwrap();
426 assert_eq!(reconciled, 2, "both 'building' runs reconciled");
427
428 for id in [run_a, run_b] {
429 let rec = get(&pool, id).await.unwrap().unwrap();
430 assert_eq!(rec.result, "aborted");
431 assert_eq!(rec.phase, "done");
432 assert!(rec.finished_at.is_some());
433 assert_eq!(
434 rec.failure_summary.as_deref(),
435 Some("daemon restarted mid-build")
436 );
437 }
438 // The already-settled run is unchanged.
439 assert_eq!(get(&pool, run_c).await.unwrap().unwrap().result, "passed");
440
441 // Idempotent: a second pass finds nothing to do.
442 assert_eq!(recover_orphaned_running(&pool).await.unwrap(), 0);
443 }
444
445 #[tokio::test]
446 async fn phase_and_version_advance_then_pass() {
447 let pool = pool().await;
448 let id = create(&pool, "abc1234").await.unwrap();
449 set_phase(&pool, id, Phase::Compiling).await.unwrap();
450 let ver: Version = "0.10.2".parse().unwrap();
451 set_version(&pool, id, &ver).await.unwrap();
452 mark_passed(&pool, id).await.unwrap();
453
454 let v = get(&pool, id).await.unwrap().unwrap();
455 assert_eq!(v.result, "passed");
456 assert_eq!(v.phase, "done");
457 assert_eq!(v.version.as_deref(), Some("0.10.2"));
458 assert!(v.finished_at.is_some());
459 }
460
461 #[tokio::test]
462 async fn first_terminal_write_wins() {
463 let pool = pool().await;
464 let id = create(&pool, "abc1234").await.unwrap();
465 mark_failed(&pool, id, "error[E0063]: missing field user_pages_host")
466 .await
467 .unwrap();
468 // A later pass attempt (e.g. the task catch racing a build-step error)
469 // must not overwrite the recorded failure.
470 mark_passed(&pool, id).await.unwrap();
471 // And a second failure summary doesn't clobber the first.
472 mark_failed(&pool, id, "something else").await.unwrap();
473
474 let v = get(&pool, id).await.unwrap().unwrap();
475 assert_eq!(v.result, "failed");
476 assert_eq!(
477 v.failure_summary.as_deref(),
478 Some("error[E0063]: missing field user_pages_host")
479 );
480 }
481
482 #[tokio::test]
483 async fn phase_write_after_terminal_is_noop() {
484 let pool = pool().await;
485 let id = create(&pool, "abc1234").await.unwrap();
486 mark_passed(&pool, id).await.unwrap();
487 set_phase(&pool, id, Phase::Gating).await.unwrap();
488 let v = get(&pool, id).await.unwrap().unwrap();
489 assert_eq!(
490 v.phase, "done",
491 "a late phase write must not move a finished run"
492 );
493 }
494
495 #[test]
496 fn elapsed_seconds_uses_finished_when_present() {
497 // Both timestamps present → exact span, no wall-clock dependency.
498 let s = elapsed_seconds("2026-06-13T00:00:00Z", Some("2026-06-13T00:02:05Z"));
499 assert_eq!(s, 125);
500 // Unparseable start → 0, never a panic / negative.
501 assert_eq!(elapsed_seconds("not-a-date", None), 0);
502 }
503
504 #[tokio::test]
505 async fn latest_summary_reports_most_recent_run() {
506 let pool = pool().await;
507 assert!(latest_summary(&pool).await.unwrap().is_none());
508 let _old = create(&pool, "old1234").await.unwrap();
509 let new = create(&pool, "new5678").await.unwrap();
510 set_phase(&pool, new, Phase::Compiling).await.unwrap();
511 let sum = latest_summary(&pool).await.unwrap().expect("a run exists");
512 assert_eq!(sum.run_id, new.0);
513 assert_eq!(sum.sha, "new5678");
514 assert_eq!(sum.phase, "compiling");
515 assert_eq!(sum.result, "building");
516 }
517
518 #[tokio::test]
519 async fn get_unknown_id_is_none() {
520 let pool = pool().await;
521 assert!(get(&pool, RunId(999)).await.unwrap().is_none());
522 }
523
524 #[tokio::test]
525 async fn failure_summary_is_bounded() {
526 let pool = pool().await;
527 let id = create(&pool, "abc1234").await.unwrap();
528 mark_failed(&pool, id, &"x".repeat(5_000)).await.unwrap();
529 let v = get(&pool, id).await.unwrap().unwrap();
530 assert!(v.failure_summary.unwrap().len() <= 600);
531 }
532 }
533