Skip to main content

max / makenotwork

9.4 KB · 282 lines History Blame Raw
1 //! Gate execution. Each gate kind has a runner that produces a pass/fail
2 //! outcome plus an optional detail string (typically a stderr tail or a
3 //! human-readable reason). Outcomes are persisted to `gate_runs` so /state
4 //! and the TUI can show them.
5
6 use crate::config::Config;
7 use crate::topology::Gate;
8 use anyhow::Result;
9 use chrono::Utc;
10 use sqlx::SqlitePool;
11 use std::path::PathBuf;
12 use std::sync::Arc;
13 use tokio::process::Command;
14
15 pub struct GateCtx {
16 pub pool: SqlitePool,
17 pub cfg: Arc<Config>,
18 pub tier: String,
19 pub version: String,
20 pub worktree: PathBuf,
21 }
22
23 #[derive(Debug, Clone)]
24 pub struct GateOutcome {
25 pub passed: bool,
26 pub detail: Option<String>,
27 }
28
29 /// Run a single gate end-to-end: insert the in-flight row, execute the gate,
30 /// update the row with the outcome. Returns the outcome for the caller.
31 pub async fn run(ctx: &GateCtx, gate: &Gate) -> Result<GateOutcome> {
32 let kind = kind_str(gate);
33 let started_at = Utc::now().to_rfc3339();
34
35 let id: i64 = sqlx::query_scalar(
36 "INSERT INTO gate_runs (version, tier, gate_kind, started_at) VALUES (?, ?, ?, ?)
37 RETURNING id",
38 )
39 .bind(&ctx.version)
40 .bind(&ctx.tier)
41 .bind(kind)
42 .bind(&started_at)
43 .fetch_one(&ctx.pool)
44 .await?;
45
46 tracing::info!(tier = %ctx.tier, version = %ctx.version, gate = kind, "gate start");
47
48 let outcome = match gate {
49 Gate::CargoTest => cargo_test(ctx).await,
50 Gate::MigrationDryRun => migration_dry_run(ctx).await,
51 Gate::BootSmoke => boot_smoke(ctx).await,
52 Gate::BurnIn { hours } => burn_in(ctx, *hours).await,
53 Gate::ManualConfirm => manual_confirm(ctx).await,
54 };
55
56 let outcome = outcome.unwrap_or_else(|e| GateOutcome {
57 passed: false,
58 detail: Some(format!("gate runner errored: {e}")),
59 });
60
61 sqlx::query(
62 "UPDATE gate_runs SET finished_at = ?, passed = ?, detail = ? WHERE id = ?",
63 )
64 .bind(Utc::now().to_rfc3339())
65 .bind(outcome.passed as i64)
66 .bind(outcome.detail.as_deref())
67 .bind(id)
68 .execute(&ctx.pool)
69 .await?;
70
71 tracing::info!(
72 tier = %ctx.tier, version = %ctx.version, gate = kind,
73 passed = outcome.passed, "gate done",
74 );
75
76 Ok(outcome)
77 }
78
79 /// Run a sequence of gates; stops on the first failure (no point running the
80 /// rest if a prerequisite failed). Returns true iff every gate passed.
81 pub async fn run_all(ctx: &GateCtx, gates: &[Gate]) -> Result<bool> {
82 for g in gates {
83 let o = run(ctx, g).await?;
84 if !o.passed {
85 return Ok(false);
86 }
87 }
88 Ok(true)
89 }
90
91 fn kind_str(g: &Gate) -> &'static str {
92 match g {
93 Gate::CargoTest => "cargo_test",
94 Gate::MigrationDryRun => "migration_dry_run",
95 Gate::BootSmoke => "boot_smoke",
96 Gate::BurnIn { .. } => "burn_in",
97 Gate::ManualConfirm => "manual_confirm",
98 }
99 }
100
101 // ---- individual gate runners ----
102
103 async fn cargo_test(ctx: &GateCtx) -> Result<GateOutcome> {
104 let server_dir = ctx.worktree.join("server");
105 let out = Command::new("cargo")
106 .args(["test", "--release"])
107 .current_dir(&server_dir)
108 .output()
109 .await?;
110 Ok(GateOutcome {
111 passed: out.status.success(),
112 detail: Some(tail(&out.stderr, 4_000)),
113 })
114 }
115
116 async fn migration_dry_run(ctx: &GateCtx) -> Result<GateOutcome> {
117 let Some(db_url) = ctx.cfg.scratch_db_url.as_deref() else {
118 return Ok(GateOutcome {
119 passed: false,
120 detail: Some("scratch_db_url unset in daemon config".into()),
121 });
122 };
123
124 let backup: Option<(String,)> = sqlx::query_as(
125 "SELECT local_path FROM backups ORDER BY id DESC LIMIT 1",
126 )
127 .fetch_optional(&ctx.pool)
128 .await?;
129 let Some((backup_path,)) = backup else {
130 return Ok(GateOutcome {
131 passed: false,
132 detail: Some("no backup fetched; call /backup/fetch first".into()),
133 });
134 };
135
136 // Reset the scratch DB: drop schema public, restore dump, run migrations.
137 if let Err(e) = reset_scratch(db_url).await {
138 return Ok(GateOutcome { passed: false, detail: Some(format!("scratch reset: {e}")) });
139 }
140 if let Err(e) = restore_dump(db_url, &backup_path).await {
141 return Ok(GateOutcome { passed: false, detail: Some(format!("restore: {e}")) });
142 }
143
144 let migrations_dir = ctx.worktree.join("server").join("migrations");
145 match run_migrator(db_url, &migrations_dir).await {
146 Ok(()) => Ok(GateOutcome { passed: true, detail: Some(format!("restored {backup_path} + migrated")) }),
147 Err(e) => Ok(GateOutcome { passed: false, detail: Some(tail(e.to_string().as_bytes(), 4_000)) }),
148 }
149 }
150
151 async fn reset_scratch(db_url: &str) -> Result<()> {
152 use sqlx::postgres::PgPoolOptions;
153 use sqlx::Executor;
154 let pool = PgPoolOptions::new().max_connections(1).connect(db_url).await?;
155 pool.execute("DROP SCHEMA IF EXISTS public CASCADE; CREATE SCHEMA public;")
156 .await?;
157 pool.close().await;
158 Ok(())
159 }
160
161 async fn restore_dump(db_url: &str, dump: &str) -> Result<()> {
162 // Two pipelines we accept:
163 // *.sql -> psql $url < dump
164 // *.sql.gz -> gunzip -c dump | psql $url
165 let is_gz = dump.ends_with(".gz");
166 let shell = if is_gz {
167 format!("gunzip -c {q} | psql {url}", q = shell_escape(dump), url = shell_escape(db_url))
168 } else {
169 format!("psql {url} < {q}", q = shell_escape(dump), url = shell_escape(db_url))
170 };
171 let out = Command::new("sh").arg("-c").arg(&shell).output().await?;
172 anyhow::ensure!(
173 out.status.success(),
174 "restore failed: {}",
175 String::from_utf8_lossy(&out.stderr),
176 );
177 Ok(())
178 }
179
180 async fn run_migrator(db_url: &str, dir: &std::path::Path) -> Result<()> {
181 use sqlx::postgres::PgPoolOptions;
182 let pool = PgPoolOptions::new().max_connections(1).connect(db_url).await?;
183 let migrator = sqlx::migrate::Migrator::new(dir).await?;
184 migrator.run(&pool).await?;
185 pool.close().await;
186 Ok(())
187 }
188
189 fn shell_escape(s: &str) -> String {
190 format!("'{}'", s.replace('\'', "'\\''"))
191 }
192
193 async fn boot_smoke(ctx: &GateCtx) -> Result<GateOutcome> {
194 let bin: Option<(String,)> = sqlx::query_as(
195 "SELECT artifact_path FROM versions WHERE version = ?",
196 )
197 .bind(&ctx.version)
198 .fetch_optional(&ctx.pool)
199 .await?;
200 let Some((bin,)) = bin else {
201 return Ok(GateOutcome { passed: false, detail: Some("no artifact for version".into()) });
202 };
203
204 // Lowest-bar smoke: start the binary and verify it stays up for a few
205 // seconds without exiting. Panics in main, missing config, port-bind
206 // failures show up here. Anything more ambitious (probing /healthz on a
207 // real port) needs server config we don't generically know.
208 let mut child = match tokio::process::Command::new(&bin)
209 .env("SANDO_BOOT_SMOKE", "1")
210 .kill_on_drop(true)
211 .spawn()
212 {
213 Ok(c) => c,
214 Err(e) => return Ok(GateOutcome { passed: false, detail: Some(format!("spawn: {e}")) }),
215 };
216
217 tokio::time::sleep(std::time::Duration::from_secs(3)).await;
218
219 match child.try_wait()? {
220 Some(status) => Ok(GateOutcome {
221 passed: false,
222 detail: Some(format!("binary exited early: {status}")),
223 }),
224 None => {
225 let _ = child.kill().await;
226 Ok(GateOutcome { passed: true, detail: Some("stayed up for 3s".into()) })
227 }
228 }
229 }
230
231 async fn burn_in(ctx: &GateCtx, hours: u32) -> Result<GateOutcome> {
232 // Check tier_state.burn_in_started_at on this tier; pass if enough time
233 // has elapsed. The clock is started by /promote when a version lands on
234 // the burn-in tier.
235 let started: Option<String> = sqlx::query_scalar(
236 "SELECT burn_in_started_at FROM tier_state WHERE tier = ?",
237 )
238 .bind(&ctx.tier)
239 .fetch_optional(&ctx.pool)
240 .await?
241 .flatten();
242 let Some(started) = started else {
243 return Ok(GateOutcome { passed: false, detail: Some("burn-in clock not started".into()) });
244 };
245 let started = chrono::DateTime::parse_from_rfc3339(&started)?.with_timezone(&Utc);
246 let elapsed = Utc::now() - started;
247 let needed = chrono::Duration::hours(hours as i64);
248 if elapsed >= needed {
249 Ok(GateOutcome { passed: true, detail: Some(format!("{} hours elapsed", elapsed.num_hours())) })
250 } else {
251 let remaining = needed - elapsed;
252 Ok(GateOutcome {
253 passed: false,
254 detail: Some(format!("{} hours remaining of {hours}", remaining.num_hours())),
255 })
256 }
257 }
258
259 async fn manual_confirm(ctx: &GateCtx) -> Result<GateOutcome> {
260 // Pass iff a row in gate_runs exists with passed=1 for this (tier, version, manual_confirm)
261 // that was inserted out-of-band by an operator action. Since the harness inserts the
262 // in-flight row itself, look for a prior confirmation row.
263 let prior: Option<i64> = sqlx::query_scalar(
264 "SELECT COUNT(*) FROM gate_runs
265 WHERE tier = ? AND version = ? AND gate_kind = 'manual_confirm' AND passed = 1",
266 )
267 .bind(&ctx.tier)
268 .bind(&ctx.version)
269 .fetch_optional(&ctx.pool)
270 .await?;
271 let passed = prior.unwrap_or(0) > 0;
272 Ok(GateOutcome {
273 passed,
274 detail: if passed { None } else { Some("waiting on operator confirmation".into()) },
275 })
276 }
277
278 fn tail(buf: &[u8], max: usize) -> String {
279 let s = String::from_utf8_lossy(buf);
280 if s.len() <= max { s.into_owned() } else { format!("...{}", &s[s.len() - max..]) }
281 }
282