Skip to main content

max / makenotwork

153.7 KB · 3805 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::classify;
7 use crate::config::AppConfig;
8 use crate::domain::{AppId, GateKind, GateRunId, TierId, Version};
9 use crate::events::{self, Event, EventTx};
10 use crate::outcome::{GateBlocker, GateFailure, GateOutcome, LogRef, PassNote};
11 use crate::topology::Gate;
12 use anyhow::{Context, Result};
13 use chrono::Utc;
14 use ops_core::live_log::LiveLog;
15 use ops_core::remote::LogSink; // brings `LiveLog::write_chunk` (the sink trait) into scope
16 use sqlx::SqlitePool;
17 use std::collections::HashMap;
18 use std::path::Path;
19 use std::path::PathBuf;
20 use std::sync::Arc;
21 use tokio::io::AsyncReadExt;
22 use tokio::process::Command;
23
24 /// The gate live-log callback: emit each chunk as a `GateLogChunk` event so the
25 /// TUI sees the tail stream in real time. `ops_core::live_log::LiveLog` owns the
26 /// disk append and the per-run sequence counter; this closure is the one
27 /// tool-specific bit (Sando previously carried a whole `live_log.rs` copy that
28 /// hardcoded exactly this emit).
29 fn gate_chunk_cb(events: EventTx, run_id: GateRunId) -> ops_core::live_log::ChunkCallback {
30 Box::new(move |seq, text| {
31 events::emit(
32 &events,
33 Event::GateLogChunk {
34 run_id,
35 seq,
36 text: text.to_owned(),
37 },
38 );
39 })
40 }
41
42 pub struct GateCtx {
43 pub pool: SqlitePool,
44 pub cfg: Arc<AppConfig>,
45 pub tier: TierId,
46 pub version: Version,
47 /// The checkout this run's artifact was built from, when there is one.
48 ///
49 /// `None` for an accepted artifact: it was built elsewhere and Sando has no
50 /// source tree for it. That is the boundary made visible (wiki
51 /// [[sando-bento-boundary]]) — artifact-scoped gates belong to the builder,
52 /// so a gate that reads source is one Sando should refuse to run here rather
53 /// than resolve against a path that does not exist.
54 pub worktree: Option<PathBuf>,
55 /// The published, content-addressed bundle this run is about, when it has
56 /// been published yet. `migration_dry_run` prefers it over the worktree, so
57 /// what it proves is inside the digest rather than beside it.
58 pub bundle: Option<PathBuf>,
59 pub events: EventTx,
60 /// Nodes the `node_health` post-deploy gate probes. Empty for build-time
61 /// gate runs on the host (where `node_health` never appears); filled at
62 /// promote time with each freshly-deployed node and its executor.
63 pub nodes: Vec<NodeProbe>,
64 /// The `build_runs.id` this gate run vouches for — the artifact identity
65 /// (wiki [[release-artifact-identity]]). Recorded on every `gate_runs` row so
66 /// promote can resolve the artifact through the evidence for a specific build,
67 /// not through a version string that a later rebuild can silently reuse.
68 /// `None` for legacy/pre-identity runs and gate unit tests.
69 pub build_id: Option<i64>,
70 /// Where each `[[aux_repo]]` is checked out, by topology name. Aux repos sit
71 /// beside the per-sha worktree rather than under it, so a `test_target` that
72 /// names one cannot be resolved against `worktree` alone. Filled from the
73 /// topology at build time; empty at promote time, where the only gate that
74 /// runs is `node_health` and there is no checkout at all.
75 pub aux_dirs: HashMap<String, PathBuf>,
76 }
77
78 impl GateCtx {
79 /// Absolute directory a `test_target` runs in: under the worktree, or under
80 /// the named aux repo's checkout.
81 ///
82 /// An `aux_repo` naming nothing this run knows about resolves to `None`
83 /// rather than to a wrong path. Callers treat that as "not present in this
84 /// run" and skip, the same as a target missing from an older sha —
85 /// `--check-config` is what stops a genuine typo from reaching here
86 /// (`Topology::ensure_test_target_aux_repos_exist`).
87 pub fn target_dir(&self, target: &crate::config::TestTarget) -> Option<PathBuf> {
88 match target.aux_repo.as_deref() {
89 None => Some(self.worktree.as_ref()?.join(&target.dir)),
90 Some(name) => Some(self.aux_dirs.get(name)?.join(&target.dir)),
91 }
92 }
93
94 /// The checkout, or a typed refusal for a gate that cannot work without one.
95 ///
96 /// Every caller of this is a gate whose evidence is about the *artifact*
97 /// rather than about the artifact in an environment, which the boundary
98 /// assigns to the builder. Reaching this arm means a tier asked Sando to
99 /// re-run a builder's gate against a bundle it was handed, and the honest
100 /// answer is to say so rather than to pass on having run nothing.
101 pub fn worktree_for(&self, gate: GateKind) -> std::result::Result<&Path, GateOutcome> {
102 self.worktree.as_deref().ok_or_else(|| {
103 GateOutcome::failed(GateFailure::NeedsSource {
104 gate,
105 artifact: self.bundle.as_ref().map_or_else(
106 || "an artifact built elsewhere".into(),
107 |b| b.display().to_string(),
108 ),
109 })
110 })
111 }
112
113 /// Where a `migration_check` finds its migrations.
114 ///
115 /// The bundle wins when it carries them. That is the point of shipping
116 /// migrations as a `release_contents` entry: it puts them inside the digest,
117 /// so the dry run proves something about the bytes that ship rather than
118 /// about a checkout that happens to sit next to them. The worktree is the
119 /// fallback for a build whose config has not opted in yet, and for an
120 /// accepted artifact there is no fallback at all — if the builder did not
121 /// bundle its migrations, Sando cannot dry-run them and says so.
122 pub fn migrations_dir(&self, dir: &Path) -> Option<PathBuf> {
123 if let Some(bundle) = &self.bundle {
124 let in_bundle = bundle.join(dir);
125 if in_bundle.is_dir() {
126 return Some(in_bundle);
127 }
128 }
129 let in_worktree = self.worktree.as_ref()?.join(dir);
130 in_worktree.is_dir().then_some(in_worktree)
131 }
132 }
133
134 /// One node the `node_health` gate verifies: its id, the systemd unit to
135 /// confirm active after the restart, an optional HTTP readiness URL, and the
136 /// executor that reaches it (the same transport the deploy used).
137 pub struct NodeProbe {
138 pub node: crate::domain::NodeId,
139 pub service: String,
140 pub health_url: Option<String>,
141 pub executor: Arc<dyn ops_exec::Executor>,
142 }
143
144 /// Run a single gate end-to-end: insert the in-flight row, execute the gate,
145 /// update the row with the outcome. Returns the outcome for the caller.
146 pub async fn run(ctx: &GateCtx, gate: &Gate) -> Result<GateOutcome> {
147 let kind = gate.kind();
148 let started_at = Utc::now().to_rfc3339();
149
150 let id: i64 = sqlx::query_scalar(
151 "INSERT INTO gate_runs (app, version, tier, gate_kind, started_at, build_id)
152 VALUES (?, ?, ?, ?, ?, ?)
153 RETURNING id",
154 )
155 .bind(&ctx.cfg.id)
156 .bind(&ctx.version)
157 .bind(&ctx.tier)
158 .bind(kind)
159 .bind(&started_at)
160 .bind(ctx.build_id)
161 .fetch_one(&ctx.pool)
162 .await?;
163 let run_id = GateRunId(id);
164
165 tracing::info!(
166 run_id = %run_id, tier = %ctx.tier, version = %ctx.version, gate = %kind,
167 "gate start",
168 );
169 events::emit(
170 &ctx.events,
171 Event::GateStart {
172 run_id,
173 tier: ctx.tier.clone(),
174 version: ctx.version.clone(),
175 gate: kind,
176 },
177 );
178
179 let outcome = match gate {
180 // cargo_test bounds its own run internally (it kills the specific child).
181 Gate::CargoTest => cargo_test(ctx, run_id).await,
182 // hardening_test bounds its own run internally, same as cargo_test.
183 Gate::HardeningTest => hardening_test(ctx, run_id).await,
184 // Each bounds itself the same way cargo_test does: one deadline across
185 // every target, so N crates cannot multiply the ceiling by N.
186 Gate::Clippy => clippy(ctx, run_id).await,
187 Gate::Fmt => fmt_check(ctx, run_id).await,
188 Gate::CargoAudit => supply_chain(ctx, run_id, GateKind::CargoAudit).await,
189 Gate::CargoDeny => supply_chain(ctx, run_id, GateKind::CargoDeny).await,
190 // migration_dry_run's psql restore + sqlx migrate could wedge; bound the
191 // whole gate here. Its bash restore sets kill_on_drop, so a timeout-drop
192 // doesn't orphan it.
193 Gate::MigrationDryRun => {
194 let ceiling = std::time::Duration::from_secs(ctx.cfg.gate_timeout_secs);
195 match tokio::time::timeout(ceiling, migration_dry_run(ctx, run_id)).await {
196 Ok(res) => res,
197 Err(_elapsed) => Ok(GateOutcome::failed(GateFailure::Timeout {
198 gate: GateKind::MigrationDryRun,
199 after_s: ctx.cfg.gate_timeout_secs as u32,
200 })
201 .with_log_ref(LogRef::new(&ctx.version, GateKind::MigrationDryRun))),
202 }
203 }
204 // code_smoke boots the real binary (migrate-from-scratch + seed + serve),
205 // any step of which could wedge; bound the whole gate here. Both child
206 // processes set kill_on_drop, so a timeout-drop can't orphan them. A
207 // timeout leaves the throwaway DB behind; the next run's createdb drops
208 // it first (DROP IF EXISTS), same as migration_dry_run's scratch reset.
209 Gate::CodeSmoke => {
210 let ceiling = std::time::Duration::from_secs(ctx.cfg.gate_timeout_secs);
211 match tokio::time::timeout(ceiling, code_smoke(ctx, run_id)).await {
212 Ok(res) => res,
213 Err(_elapsed) => Ok(GateOutcome::failed(GateFailure::Timeout {
214 gate: GateKind::CodeSmoke,
215 after_s: ctx.cfg.gate_timeout_secs as u32,
216 })
217 .with_log_ref(LogRef::new(&ctx.version, GateKind::CodeSmoke))),
218 }
219 }
220 Gate::BootSmoke => boot_smoke(ctx, run_id).await,
221 Gate::NodeHealth => node_health(ctx).await,
222 Gate::BurnIn { hours } => burn_in(ctx, *hours).await,
223 Gate::ManualConfirm => manual_confirm(ctx).await,
224 };
225
226 let outcome = outcome.unwrap_or_else(|e| {
227 GateOutcome::failed(GateFailure::Unclassified {
228 legacy_detail: Some(format!("gate runner errored: {e}")),
229 })
230 });
231
232 let outcome_json = serde_json::to_string(&outcome)
233 .unwrap_or_else(|e| format!("{{\"_serialize_error\":{e:?}}}"));
234 sqlx::query(
235 "UPDATE gate_runs
236 SET finished_at = ?, status = ?, outcome_json = ?, log_ref = ?
237 WHERE id = ?",
238 )
239 .bind(Utc::now().to_rfc3339())
240 .bind(outcome.status_str())
241 .bind(&outcome_json)
242 .bind(outcome.log_ref.as_ref().map(super::outcome::LogRef::as_str))
243 .bind(id)
244 .execute(&ctx.pool)
245 .await?;
246
247 tracing::info!(
248 tier = %ctx.tier, version = %ctx.version, gate = %kind,
249 status = outcome.status_str(), "gate done",
250 );
251 events::emit(
252 &ctx.events,
253 Event::GateDone {
254 run_id,
255 tier: ctx.tier.clone(),
256 version: ctx.version.clone(),
257 gate: kind,
258 outcome: outcome.clone(),
259 },
260 );
261
262 Ok(outcome)
263 }
264
265 /// Run every gate in order and return the kinds that did not pass (empty means
266 /// green). We deliberately do NOT short-circuit on first failure — every gate's
267 /// outcome is recorded in `gate_runs`, which is the operator's only visibility
268 /// into pipeline health. Hiding later gates because an earlier one failed makes
269 /// diagnosis worse.
270 ///
271 /// Returning the failing kinds rather than a bare bool is what lets the promote
272 /// path name them in the tier's `partial_reason` and in the error it returns to
273 /// the operator, instead of a generic "something was red".
274 pub async fn run_all(ctx: &GateCtx, gates: &[Gate]) -> Result<Vec<GateKind>> {
275 let mut failed = Vec::new();
276 for g in gates {
277 let o = run(ctx, g).await?;
278 if !o.is_passed() {
279 failed.push(g.kind());
280 }
281 }
282 Ok(failed)
283 }
284
285 // ---- individual gate runners ----
286
287 /// Run every configured `test_target`'s suite, in order, under one gate.
288 ///
289 /// This used to be hardcoded to `worktree/server`, which meant every other
290 /// crate in the repo shipped ungated — including `mnw-cli`, which is built as a
291 /// companion and installed onto prod-1 in the same promote. The targets are now
292 /// configured (`[[test_target]]` in the daemon config), defaulting to the
293 /// historical single `server` entry.
294 ///
295 /// The whole set shares one `gate_runs` row and one log file: from the
296 /// pipeline's point of view "the tests" either pass or don't. The first failing
297 /// target ends the gate, since a red suite blocks the promote regardless of what
298 /// the remaining crates would have said, and running them would only delay the
299 /// operator's answer. Its name is carried in the failure so the summary points
300 /// at the crate, not just the test.
301 async fn cargo_test(ctx: &GateCtx, run_id: GateRunId) -> Result<GateOutcome> {
302 let log_path = gate_log_path(ctx, GateKind::CargoTest);
303 let log_ref = LogRef::new(&ctx.version, GateKind::CargoTest);
304
305 // Best-effort: drop our own role's stale `mnw_test_*` databases (the
306 // template + any per-test clones orphaned by a previously-killed run)
307 // before the suite, so they can't accumulate or collide. Foreign-owned
308 // leftovers are left alone — the harness now namespaces its template per
309 // role, so they no longer wedge the gate.
310 if let Some(scratch_url) = ctx.cfg.scratch_db_url.as_deref() {
311 clean_stale_test_dbs(scratch_url).await;
312 }
313
314 let started = std::time::Instant::now();
315 // One ceiling for the whole gate, not per target: the point is to bound how
316 // long a hung suite can block the pipeline, and N targets each allowed the
317 // full timeout would multiply that by N.
318 let deadline = started + std::time::Duration::from_secs(ctx.cfg.gate_timeout_secs);
319 let mut ran = 0usize;
320
321 for target in &ctx.cfg.test_targets {
322 let label = target.label();
323 // A target absent from this sha is skipped, not fatal: sando has to be
324 // able to build older shas (bisect, rollback rebuild) from a config that
325 // describes the tip. The zero-targets-ran check below is what stops this
326 // from quietly turning the gate into a no-op.
327 let Some(dir) = ctx
328 .target_dir(target)
329 .filter(|d| d.join("Cargo.toml").is_file())
330 else {
331 tracing::warn!(
332 target = %label, version = %ctx.version,
333 "test_target has no Cargo.toml in this run; skipping",
334 );
335 continue;
336 };
337 let features: Vec<&str> = target.features.iter().map(String::as_str).collect();
338
339 // Fast pre-gate: compile the test targets WITHOUT running them. This
340 // builds the exact artifacts the full run needs (so the subsequent run
341 // reuses the cache — no wasted work), but fails in ~minutes with the
342 // real `error[Ennnn]: ...` on a test-only-target compile break. That
343 // class (a field missing in a `#[cfg(test)]`-only binary like `load`)
344 // otherwise compiles fine under the build step and only blows up here,
345 // after a full build, as an opaque mass test failure.
346 let banner = format!("\n==== test_target: {label} ====\n");
347 append_to_log(&log_path, banner.as_bytes()).await;
348 let mut pre = match cargo_test_command(ctx, &dir, target, &features, &["--no-run"]).spawn()
349 {
350 Ok(c) => c,
351 Err(e) => {
352 return Ok(GateOutcome::failed(GateFailure::SpawnFailed {
353 message: format!("{label}: {e}"),
354 })
355 .with_log_ref(log_ref));
356 }
357 };
358 let (pre_out, pre_err, pre_status) = match run_to_deadline_for(
359 &mut pre,
360 ctx,
361 run_id,
362 log_path.clone(),
363 deadline,
364 started,
365 GateKind::CargoTest,
366 )
367 .await?
368 {
369 Ok(v) => v,
370 Err(timeout) => return Ok(timeout.with_log_ref(log_ref)),
371 };
372 if !pre_status.success() {
373 let failure = classify::classify_compile_error(&pre_out, &pre_err);
374 return Ok(GateOutcome::failed(name_target(failure, &target.dir)).with_log_ref(log_ref));
375 }
376
377 // Full run: the test binaries are already built above, so cargo's
378 // up-to-date check skips compilation and this just runs the tests.
379 let mut child = match cargo_test_command(ctx, &dir, target, &features, &[]).spawn() {
380 Ok(c) => c,
381 Err(e) => {
382 return Ok(GateOutcome::failed(GateFailure::SpawnFailed {
383 message: format!("{label}: {e}"),
384 })
385 .with_log_ref(log_ref));
386 }
387 };
388 let (stdout_buf, stderr_buf, status) = match run_to_deadline_for(
389 &mut child,
390 ctx,
391 run_id,
392 log_path.clone(),
393 deadline,
394 started,
395 GateKind::CargoTest,
396 )
397 .await?
398 {
399 Ok(v) => v,
400 Err(timeout) => return Ok(timeout.with_log_ref(log_ref)),
401 };
402 if !status.success() {
403 let failure = classify::classify_cargo_test(&stdout_buf, &stderr_buf);
404 return Ok(GateOutcome::failed(name_target(failure, &target.dir)).with_log_ref(log_ref));
405 }
406 ran += 1;
407 }
408
409 // Every configured target was missing from the worktree. Exiting green here
410 // would report "tests passed" having run none of them.
411 if ran == 0 {
412 return Ok(GateOutcome::failed(GateFailure::Unclassified {
413 legacy_detail: Some(format!(
414 "cargo_test ran no targets: none of the {} configured test_target dir(s) \
415 exist in this worktree",
416 ctx.cfg.test_targets.len(),
417 )),
418 })
419 .with_log_ref(log_ref));
420 }
421
422 let duration_s = started.elapsed().as_secs() as u32;
423 Ok(GateOutcome::passed(PassNote::TestsPassed { duration_s }).with_log_ref(log_ref))
424 }
425
426 /// Stream a child to the live log, bounded by the gate-wide `deadline`. `Ok(Err(_))`
427 /// is the timeout outcome (child killed); `Err(_)` is an IO error on the stream.
428 #[allow(clippy::type_complexity)]
429 async fn run_to_deadline_for(
430 child: &mut tokio::process::Child,
431 ctx: &GateCtx,
432 run_id: GateRunId,
433 log_path: PathBuf,
434 deadline: std::time::Instant,
435 started: std::time::Instant,
436 kind: GateKind,
437 ) -> Result<std::result::Result<(Vec<u8>, Vec<u8>, std::process::ExitStatus), GateOutcome>> {
438 let remaining = deadline.saturating_duration_since(std::time::Instant::now());
439 let stream = stream_child_to_live_log(child, ctx.events.clone(), run_id, log_path);
440 match tokio::time::timeout(remaining, stream).await {
441 Ok(res) => Ok(Ok(res?)),
442 Err(_elapsed) => {
443 child.start_kill().ok();
444 let _ = child.wait().await;
445 Ok(Err(GateOutcome::failed(GateFailure::Timeout {
446 gate: kind,
447 after_s: started.elapsed().as_secs() as u32,
448 })))
449 }
450 }
451 }
452
453 /// Prefix a test/compile failure's headline with the crate it came from, so a
454 /// red gate across many targets says *which* crate broke. Other failure kinds
455 /// are single-target by construction and pass through untouched.
456 fn name_target(failure: GateFailure, dir: &std::path::Path) -> GateFailure {
457 let at = dir.display();
458 match failure {
459 GateFailure::CargoTest {
460 failed_count,
461 first_failed,
462 first_panic,
463 } => GateFailure::CargoTest {
464 failed_count,
465 first_failed: Some(match first_failed {
466 Some(name) => format!("{at}: {name}"),
467 None => at.to_string(),
468 }),
469 first_panic,
470 },
471 GateFailure::CompileError {
472 error_count,
473 first_error,
474 } => GateFailure::CompileError {
475 error_count,
476 first_error: Some(match first_error {
477 Some(e) => format!("{at}: {e}"),
478 None => at.to_string(),
479 }),
480 },
481 other => other,
482 }
483 }
484
485 /// Append raw bytes to a gate log outside the child-streaming path (target
486 /// banners). Best-effort, same as `LiveLog`: a broken log dir never turns a
487 /// passing gate red.
488 async fn append_to_log(path: &std::path::Path, bytes: &[u8]) {
489 use tokio::io::AsyncWriteExt;
490 if let Some(parent) = path.parent()
491 && tokio::fs::create_dir_all(parent).await.is_err()
492 {
493 return;
494 }
495 if let Ok(mut f) = tokio::fs::OpenOptions::new()
496 .create(true)
497 .append(true)
498 .open(path)
499 .await
500 {
501 let _ = f.write_all(bytes).await;
502 }
503 }
504
505 /// `cargo clippy --all-targets -- -D warnings` over every configured
506 /// `test_target`.
507 ///
508 /// Before this gate, `-D warnings` was enforced in exactly one place —
509 /// `server/deploy/run-ci.sh`, which died with the astra pipeline — so lint drift
510 /// reached prod ungated. Two crates even carried comments referring to "the
511 /// -D warnings CI gate" for a gate that did not exist.
512 async fn clippy(ctx: &GateCtx, run_id: GateRunId) -> Result<GateOutcome> {
513 lint_over_targets(ctx, run_id, GateKind::Clippy, |target, features| {
514 let mut args = vec!["clippy".to_string(), "--all-targets".to_string()];
515 if target.all_features {
516 args.push("--all-features".to_string());
517 } else if !features.is_empty() {
518 args.push("--features".to_string());
519 args.push(features.join(","));
520 }
521 // Everything after `--` goes to rustc, which is where -D lives.
522 args.push("--".to_string());
523 args.push("-D".to_string());
524 args.push("warnings".to_string());
525 args
526 })
527 .await
528 }
529
530 /// `cargo fmt --check` over every configured `test_target`.
531 ///
532 /// No `rustfmt.toml` exists anywhere in the tree, so this is plain rustfmt
533 /// defaults. Cheap: no compilation, just a parse.
534 async fn fmt_check(ctx: &GateCtx, run_id: GateRunId) -> Result<GateOutcome> {
535 lint_over_targets(ctx, run_id, GateKind::Fmt, |_target, _features| {
536 vec!["fmt".to_string(), "--check".to_string()]
537 })
538 .await
539 }
540
541 /// `cargo audit` / `cargo deny check`, over the crates that carry the matching
542 /// config file.
543 ///
544 /// Config-gated on purpose. Both tools are only meaningful against a triaged
545 /// posture: four crates in this repo fail `cargo audit` today purely because
546 /// they have no `.cargo/audit.toml` recording which transitive advisories have
547 /// been reviewed and accepted. Running them everywhere would make the gate
548 /// permanently and uninformatively red. Dropping the config file into a crate
549 /// is what opts it in.
550 async fn supply_chain(ctx: &GateCtx, run_id: GateRunId, kind: GateKind) -> Result<GateOutcome> {
551 let (config_rel, args): (&str, Vec<String>) = match kind {
552 GateKind::CargoAudit => (".cargo/audit.toml", vec!["audit".into()]),
553 GateKind::CargoDeny => ("deny.toml", vec!["deny".into(), "check".into()]),
554 other => unreachable!("supply_chain called for {other:?}"),
555 };
556 run_over_targets(ctx, run_id, kind, |_target, target_dir| {
557 target_dir.join(config_rel).is_file().then(|| args.clone())
558 })
559 .await
560 }
561
562 /// Shared driver for the lint gates: run `cargo <args>` in every configured
563 /// `test_target` that exists in this worktree.
564 async fn lint_over_targets(
565 ctx: &GateCtx,
566 run_id: GateRunId,
567 kind: GateKind,
568 build_args: impl Fn(&crate::config::TestTarget, &[String]) -> Vec<String>,
569 ) -> Result<GateOutcome> {
570 // The target is handed in directly. This used to reverse-look-it-up by
571 // comparing `worktree.join(dir)` against the resolved path, which silently
572 // stopped matching for anything resolved anywhere else — an aux repo, say.
573 run_over_targets(ctx, run_id, kind, move |t, _dir| {
574 Some(build_args(t, &t.features))
575 })
576 .await
577 }
578
579 /// Run one cargo invocation per configured `test_target`, under a single
580 /// gate-wide deadline. `args_for` returns `None` to skip a target (used by the
581 /// supply-chain gates, which only apply where their config file lives).
582 ///
583 /// Shares `cargo_test`'s conventions: per-target log banners, stop at the first
584 /// failure with the crate named, skip targets absent from the worktree, and fail
585 /// closed if that leaves nothing to run.
586 async fn run_over_targets(
587 ctx: &GateCtx,
588 run_id: GateRunId,
589 kind: GateKind,
590 args_for: impl Fn(&crate::config::TestTarget, &std::path::Path) -> Option<Vec<String>>,
591 ) -> Result<GateOutcome> {
592 let log_path = gate_log_path(ctx, kind);
593 let log_ref = LogRef::new(&ctx.version, kind);
594 let started = std::time::Instant::now();
595 let deadline = started + std::time::Duration::from_secs(ctx.cfg.gate_timeout_secs);
596 let mut ran = 0usize;
597
598 for target in &ctx.cfg.test_targets {
599 let label = target.label();
600 let Some(dir) = ctx
601 .target_dir(target)
602 .filter(|d| d.join("Cargo.toml").is_file())
603 else {
604 tracing::warn!(
605 gate = kind.as_str(), target = %label,
606 "target has no Cargo.toml in this run; skipping",
607 );
608 continue;
609 };
610 let Some(args) = args_for(target, &dir) else {
611 continue;
612 };
613
614 append_to_log(
615 &log_path,
616 format!("\n==== {}: {label} ====\n", kind.as_str()).as_bytes(),
617 )
618 .await;
619
620 let mut cmd = Command::new("cargo");
621 cmd.args(&args)
622 .current_dir(&dir)
623 .stdout(std::process::Stdio::piped())
624 .stderr(std::process::Stdio::piped())
625 .kill_on_drop(true);
626 if let Some(t) = ctx.cfg.cargo_target_dir.as_deref() {
627 cmd.env("CARGO_TARGET_DIR", t);
628 }
629 // clippy type-checks, so it needs the same sqlx online-mode env the
630 // build and cargo_test steps get. fmt/audit/deny never touch the DB.
631 if kind == GateKind::Clippy
632 && let Some(url) = ctx
633 .cfg
634 .scratch_db_url
635 .as_deref()
636 .filter(|_| target.scratch_db)
637 {
638 cmd.env("DATABASE_URL", url);
639 cmd.env(
640 "TEST_DATABASE_URL",
641 url.split_once('?').map_or(url, |(b, _)| b),
642 );
643 }
644
645 let mut child = match cmd.spawn() {
646 Ok(c) => c,
647 Err(e) => {
648 return Ok(GateOutcome::failed(GateFailure::SpawnFailed {
649 message: format!("{label}: {e}"),
650 })
651 .with_log_ref(log_ref));
652 }
653 };
654 let (stdout_buf, stderr_buf, status) = match run_to_deadline_for(
655 &mut child,
656 ctx,
657 run_id,
658 log_path.clone(),
659 deadline,
660 started,
661 kind,
662 )
663 .await?
664 {
665 Ok(v) => v,
666 Err(timeout) => return Ok(timeout.with_log_ref(log_ref)),
667 };
668 if !status.success() {
669 let failure = match kind {
670 // clippy speaks rustc diagnostics, so the compile-error
671 // classifier extracts the real `error: ...` headline.
672 GateKind::Clippy => classify::classify_compile_error(&stdout_buf, &stderr_buf),
673 _ => GateFailure::Unclassified {
674 legacy_detail: Some(first_meaningful_line(&stdout_buf, &stderr_buf)),
675 },
676 };
677 return Ok(GateOutcome::failed(name_target(failure, &target.dir)).with_log_ref(log_ref));
678 }
679 ran += 1;
680 }
681
682 if ran == 0 {
683 return Ok(GateOutcome::failed(GateFailure::Unclassified {
684 legacy_detail: Some(format!(
685 "{} ran nothing: no configured target in this worktree qualified",
686 kind.as_str(),
687 )),
688 })
689 .with_log_ref(log_ref));
690 }
691 Ok(GateOutcome::passed(PassNote::TestsPassed {
692 duration_s: started.elapsed().as_secs() as u32,
693 })
694 .with_log_ref(log_ref))
695 }
696
697 /// First line that looks like a diagnostic, for gates whose tools have no
698 /// dedicated classifier (`cargo audit`, `cargo deny`). Falls back to a generic
699 /// note rather than an empty string.
700 fn first_meaningful_line(stdout: &[u8], stderr: &[u8]) -> String {
701 for buf in [stderr, stdout] {
702 let text = String::from_utf8_lossy(buf);
703 if let Some(line) = text
704 .lines()
705 .map(str::trim)
706 .find(|l| l.starts_with("error") || l.contains("vulnerabilit") || l.contains("FAILED"))
707 {
708 return line.chars().take(200).collect();
709 }
710 }
711 "tool reported failure; see the gate log".into()
712 }
713
714 /// The tests `cargo_test` cannot reach, run against production constants.
715 ///
716 /// `cargo_test` builds with `--features fast-tests`, which relaxes
717 /// `AUTH_RATE_LIMIT_BURST` 5 → 20, `SANDBOX_RATE_LIMIT_MS` 30s → 10ms, and
718 /// argon2 from 46 MiB/t=2 to 8 MiB/t=1 — and the rate-limiting suite is
719 /// `#[cfg_attr(feature = "fast-tests", ignore)]`d on top of that, because a
720 /// bucket refilling at 100/sec never depletes under parallel test threads. The
721 /// net effect was that Sando's only code gate silently skipped every test of
722 /// the auth hardening it most needs to protect.
723 ///
724 /// So: no features, `--test-threads=1` (these tests key on a shared per-IP
725 /// bucket and must not interleave), and a name filter rather than the whole
726 /// suite — the rest of the suite is tuned for `fast-tests` and would only go
727 /// slow and flaky here. The filter is a substring match, so it catches
728 /// `..._rate_limit_...` and `..._rate_limited` alike.
729 ///
730 /// This costs a second compile of the lib + integration binary (a different
731 /// feature set is a different cfg, so no artifact sharing with `cargo_test`).
732 /// That is the price of the coverage; the filter keeps the *run* to seconds.
733 async fn hardening_test(ctx: &GateCtx, run_id: GateRunId) -> Result<GateOutcome> {
734 let server_dir = match ctx.worktree_for(GateKind::HardeningTest) {
735 Ok(w) => w.join("server"),
736 Err(outcome) => return Ok(outcome),
737 };
738 // No features is the whole point; the scratch DB is needed because the
739 // server's sqlx macros type-check against it. Unlike cargo_test, this gate
740 // is deliberately not driven by `test_targets`: it targets one specific
741 // suite in one specific crate, not "the repo's tests".
742 let target = crate::config::TestTarget {
743 dir: std::path::PathBuf::from("server"),
744 aux_repo: None,
745 features: Vec::new(),
746 all_features: false,
747 scratch_db: true,
748 };
749 let log_path = gate_log_path(ctx, GateKind::HardeningTest);
750 let log_ref = LogRef::new(&ctx.version, GateKind::HardeningTest);
751
752 if let Some(scratch_url) = ctx.cfg.scratch_db_url.as_deref() {
753 clean_stale_test_dbs(scratch_url).await;
754 }
755
756 let started = std::time::Instant::now();
757
758 // Same two-step shape as cargo_test: compile first so a test-target break
759 // reports as a compile error rather than an opaque mass test failure.
760 let mut pre = match cargo_test_command(
761 ctx,
762 &server_dir,
763 &target,
764 &[],
765 &["--no-run", "--test", "integration"],
766 )
767 .spawn()
768 {
769 Ok(c) => c,
770 Err(e) => {
771 return Ok(GateOutcome::failed(GateFailure::SpawnFailed {
772 message: e.to_string(),
773 })
774 .with_log_ref(log_ref));
775 }
776 };
777 let (pre_out, pre_err, pre_status) =
778 stream_child_to_live_log(&mut pre, ctx.events.clone(), run_id, log_path.clone()).await?;
779 if !pre_status.success() {
780 let failure = classify::classify_compile_error(&pre_out, &pre_err);
781 return Ok(GateOutcome::failed(failure).with_log_ref(log_ref));
782 }
783
784 let mut child = match cargo_test_command(
785 ctx,
786 &server_dir,
787 &target,
788 &[],
789 &[
790 "--test",
791 "integration",
792 "--",
793 "--test-threads=1",
794 "rate_limit",
795 ],
796 )
797 .spawn()
798 {
799 Ok(c) => c,
800 Err(e) => {
801 return Ok(GateOutcome::failed(GateFailure::SpawnFailed {
802 message: e.to_string(),
803 })
804 .with_log_ref(log_ref));
805 }
806 };
807 let ceiling = std::time::Duration::from_secs(ctx.cfg.gate_timeout_secs);
808 let stream = stream_child_to_live_log(&mut child, ctx.events.clone(), run_id, log_path);
809 let (stdout_buf, stderr_buf, status) = match tokio::time::timeout(ceiling, stream).await {
810 Ok(res) => res?,
811 Err(_elapsed) => {
812 child.start_kill().ok();
813 let _ = child.wait().await;
814 return Ok(GateOutcome::failed(GateFailure::Timeout {
815 gate: GateKind::HardeningTest,
816 after_s: started.elapsed().as_secs() as u32,
817 })
818 .with_log_ref(log_ref));
819 }
820 };
821 let duration_s = started.elapsed().as_secs() as u32;
822 if status.success() {
823 // A filter that matches nothing exits 0, which would make this gate a
824 // green no-op the day someone renames the tests. Fail closed instead.
825 if tests_run(&stdout_buf) == 0 {
826 return Ok(GateOutcome::failed(GateFailure::Unclassified {
827 legacy_detail: Some(
828 "hardening_test ran 0 tests: the `rate_limit` filter matched nothing. \
829 The suite was renamed or moved — this gate is proving nothing."
830 .into(),
831 ),
832 })
833 .with_log_ref(log_ref));
834 }
835 Ok(GateOutcome::passed(PassNote::TestsPassed { duration_s }).with_log_ref(log_ref))
836 } else {
837 let failure = classify::classify_cargo_test(&stdout_buf, &stderr_buf);
838 Ok(GateOutcome::failed(failure).with_log_ref(log_ref))
839 }
840 }
841
842 /// Count the tests libtest reports as run, from its `test result: ok. N passed`
843 /// summary line. Returns 0 when no summary is present, which is itself the
844 /// "nothing ran" case the caller fails on.
845 fn tests_run(stdout: &[u8]) -> u32 {
846 String::from_utf8_lossy(stdout)
847 .lines()
848 .filter_map(|l| l.trim().strip_prefix("test result:"))
849 .filter_map(|rest| rest.split_once(" passed"))
850 .filter_map(|(head, _)| {
851 head.rsplit(' ')
852 .find(|t| !t.is_empty())?
853 .parse::<u32>()
854 .ok()
855 })
856 .sum()
857 }
858
859 /// Configure (but don't spawn) `cargo test --release [--features <features>]
860 /// <extra>` in `dir`, wired to the scratch DB. Shared by the `--no-run`
861 /// pre-gate compile and the full test run so both go through one env setup,
862 /// and by both test gates so they can't drift apart on env.
863 ///
864 /// `cargo_test` passes `fast-tests`, matching what the retired astra CI did:
865 /// it relaxes the auth rate-limit burst (5 → 20) and argon2 cost so
866 /// signup-heavy + lockout workflow tests complete without hitting Governor
867 /// before the hand-rolled lockout check (documented at
868 /// `server/src/constants.rs:87`). `hardening_test` passes no features, which is
869 /// the whole point of that gate — see `GateKind::HardeningTest`.
870 fn cargo_test_command(
871 ctx: &GateCtx,
872 dir: &std::path::Path,
873 target: &crate::config::TestTarget,
874 features: &[&str],
875 extra: &[&str],
876 ) -> Command {
877 let mut cmd = Command::new("cargo");
878 cmd.args(["test", "--release"]);
879 if target.all_features {
880 cmd.arg("--all-features");
881 } else if !features.is_empty() {
882 cmd.args(["--features", &features.join(",")]);
883 }
884 cmd.args(extra)
885 .current_dir(dir)
886 .stdout(std::process::Stdio::piped())
887 .stderr(std::process::Stdio::piped())
888 .kill_on_drop(true);
889 // Share the build step's target dir so the test compile reuses its
890 // artifacts (and the `--no-run` precompile reuses them again). Must match
891 // `build.rs` or the gate would clean-compile the whole tree a second time.
892 if let Some(target) = ctx.cfg.cargo_target_dir.as_deref() {
893 cmd.env("CARGO_TARGET_DIR", target);
894 }
895 // Same online-mode rationale as the build step: sqlx query macros need a
896 // live DB to type-check against. The scratch DB is left in migrated state
897 // by the preceding build, so we can reuse it here.
898 //
899 // Opt-in per target: setting DATABASE_URL switches sqlx OUT of offline mode,
900 // so a crate that ships `.sqlx` query data would stop using it and try to
901 // type-check against a database that has none of its tables.
902 if let Some(scratch_url) = ctx
903 .cfg
904 .scratch_db_url
905 .as_deref()
906 .filter(|_| target.scratch_db)
907 {
908 cmd.env("DATABASE_URL", scratch_url);
909 // The server test harness (tests/harness/db.rs) parses TEST_DATABASE_URL
910 // with rfind('/'), which mangles URLs whose query string contains '/'
911 // (e.g. `?host=/var/run/postgresql`). Strip the query — libpq defaults
912 // to /var/run/postgresql on Debian/Ubuntu when host is unspecified.
913 let test_url = scratch_url
914 .split_once('?')
915 .map_or(scratch_url, |(base, _)| base);
916 cmd.env("TEST_DATABASE_URL", test_url);
917 }
918 cmd
919 }
920
921 async fn migration_dry_run(ctx: &GateCtx, run_id: GateRunId) -> Result<GateOutcome> {
922 let log = GateLog::open(ctx, run_id, GateKind::MigrationDryRun).await;
923 let outcome = migration_dry_run_inner(ctx, &log).await;
924 log.close().await;
925 outcome.map(|o| o.with_log_ref(LogRef::new(&ctx.version, GateKind::MigrationDryRun)))
926 }
927
928 /// The staged interior of [`migration_dry_run`], writing every step through the
929 /// gate's live log. The caller owns the sink so it can flush it on every exit
930 /// path, and attaches the `log_ref` once instead of at each return.
931 /// Runs one configured check per database, in config order, and stops at the
932 /// first that does not pass — a red gate is a red gate, and continuing would
933 /// bury it under a second restore's output.
934 ///
935 /// The server's check runs against `scratch_db_url` itself and is deliberately
936 /// last-writer for it: `cargo_test` reuses that database in migrated state, so
937 /// every other check must name its own `scratch_db` (enforced at config load).
938 async fn migration_dry_run_inner(ctx: &GateCtx, log: &GateLog) -> Result<GateOutcome> {
939 let Some(scratch_url) = ctx.cfg.scratch_db_url.as_deref() else {
940 log.line("scratch_db_url unset in daemon config\n").await;
941 return Ok(GateOutcome::blocked(GateBlocker::ScratchDbUrlUnset));
942 };
943
944 let mut checked = Vec::new();
945 let mut primary_backup_path = String::new();
946 for check in &ctx.cfg.migration_checks {
947 let label = check.dir.display().to_string();
948 log.line(&format!("==== migration_check: {label} ====\n"))
949 .await;
950 match run_migration_check(ctx, log, scratch_url, check).await? {
951 CheckResult::Passed { backup_path } => {
952 if primary_backup_path.is_empty() {
953 primary_backup_path = backup_path;
954 }
955 checked.push(label);
956 }
957 CheckResult::Stopped(outcome) => return Ok(outcome),
958 }
959 }
960
961 log.line(&format!(
962 "all {} migration check(s) passed: {}",
963 checked.len(),
964 checked.join(", ")
965 ))
966 .await;
967 Ok(GateOutcome::passed(PassNote::Migrated {
968 backup_path: primary_backup_path,
969 checks: checked,
970 }))
971 }
972
973 /// One check's verdict: it passed (against `backup_path`), or it produced the
974 /// outcome the whole gate reports.
975 enum CheckResult {
976 Passed { backup_path: String },
977 Stopped(GateOutcome),
978 }
979
980 /// Restore one database's dump into its scratch DB and run its migrations on top.
981 async fn run_migration_check(
982 ctx: &GateCtx,
983 log: &GateLog,
984 scratch_url: &str,
985 check: &crate::config::MigrationCheck,
986 ) -> Result<CheckResult> {
987 let label = check.dir.display().to_string();
988
989 let backup: Option<(String, String)> = sqlx::query_as(
990 "SELECT local_path, fetched_at FROM backups
991 WHERE app = ? AND name = ? ORDER BY id DESC LIMIT 1",
992 )
993 .bind(&ctx.cfg.id)
994 .bind(&check.backup)
995 .fetch_optional(&ctx.pool)
996 .await?;
997 let Some((backup_path, fetched_at)) = backup else {
998 log.line(&format!(
999 "no {} backup fetched; call /backup/fetch first\n",
1000 check.backup
1001 ))
1002 .await;
1003 return Ok(CheckResult::Stopped(GateOutcome::blocked(
1004 GateBlocker::NoBackupAvailable {
1005 check: label,
1006 backup: check.backup.clone(),
1007 },
1008 )));
1009 };
1010
1011 // Presence is not freshness. A fetch that quietly stopped working leaves this
1012 // row in place, and restoring it dry-runs the migrations against a schema prod
1013 // has moved past — green, and worthless. Block on age instead. An unparsable
1014 // timestamp is treated as stale: this row is daemon-written RFC 3339, so a
1015 // value that will not parse means something is wrong, and failing closed on a
1016 // freshness check is the whole point.
1017 let age_hours = chrono::DateTime::parse_from_rfc3339(&fetched_at).map_or(i64::MAX, |t| {
1018 (Utc::now() - t.with_timezone(&Utc)).num_hours()
1019 });
1020 let max_age_hours = ctx.cfg.backup_max_age_hours;
1021 if age_hours > i64::from(max_age_hours) {
1022 let msg = format!(
1023 "backup {backup_path} was fetched {fetched_at} ({age_hours}h ago, max \
1024 {max_age_hours}h); re-run /backup/fetch\n"
1025 );
1026 log.line(&msg).await;
1027 return Ok(CheckResult::Stopped(GateOutcome::blocked(
1028 GateBlocker::BackupStale {
1029 age_hours,
1030 max_age_hours,
1031 check: label,
1032 },
1033 )));
1034 }
1035
1036 // A check with its own `scratch_db` gets that database created here rather
1037 // than by a host bootstrap step: adding a `[[migration_check]]` should not
1038 // silently depend on someone having remembered to `createdb` on the Sando
1039 // host, which is exactly the class of footgun this gate exists to remove.
1040 // DROP + CREATE also makes the database sando-owned, so the PG15+ public
1041 // schema grants `reset_scratch` applies next are the owner's to give.
1042 let db_url = match check.scratch_db.as_deref() {
1043 None => scratch_url.to_string(),
1044 Some(dbname) => {
1045 let maintenance_url = pg_url_with_dbname(scratch_url, "postgres");
1046 log.line(&format!("---- create scratch db {dbname} ----\n"))
1047 .await;
1048 if let Err(e) = pg_create_db(&maintenance_url, dbname).await {
1049 let msg = format!("{label}: creating scratch db {dbname}: {e}");
1050 log.line(&msg).await;
1051 return Ok(CheckResult::Stopped(GateOutcome::failed(
1052 GateFailure::RestoreFailed { reason: msg },
1053 )));
1054 }
1055 pg_url_with_dbname(scratch_url, dbname)
1056 }
1057 };
1058
1059 let owner_role = check
1060 .owner_role
1061 .as_deref()
1062 .unwrap_or(&ctx.cfg.scratch_owner_role);
1063 log.line("---- reset_scratch ----\n").await;
1064 if let Err(e) = reset_scratch(&db_url, owner_role).await {
1065 let msg = format!("{label}: scratch reset: {e}");
1066 log.line(&msg).await;
1067 return Ok(CheckResult::Stopped(GateOutcome::failed(
1068 GateFailure::RestoreFailed { reason: msg },
1069 )));
1070 }
1071 log.line(&format!("---- restore_dump ({backup_path}) ----\n"))
1072 .await;
1073 if let Err(e) = restore_dump(&db_url, &backup_path, log).await {
1074 let msg = format!("{label}: restore: {e}");
1075 log.line(&msg).await;
1076 return Ok(CheckResult::Stopped(GateOutcome::failed(
1077 GateFailure::RestoreFailed { reason: msg },
1078 )));
1079 }
1080
1081 let Some(migrations_dir) = ctx.migrations_dir(&check.dir) else {
1082 // Neither the bundle nor a checkout holds them. For an accepted
1083 // artifact that means the builder did not ship its migrations, and a
1084 // dry run over nothing would report green having proved nothing.
1085 let msg = format!(
1086 "{label}: no migrations at {} in the bundle or a checkout",
1087 check.dir.display()
1088 );
1089 log.line(&msg).await;
1090 return Ok(CheckResult::Stopped(GateOutcome::failed(
1091 GateFailure::RestoreFailed { reason: msg },
1092 )));
1093 };
1094 log.line("---- run_migrator ----\n").await;
1095 match run_migrator(&db_url, &migrations_dir).await {
1096 Ok(()) => {
1097 log.line(&format!("{label}: restored {backup_path} + migrated\n"))
1098 .await;
1099 Ok(CheckResult::Passed { backup_path })
1100 }
1101 Err(e) => {
1102 let err_s = format!("{label}: {e}");
1103 log.line(&err_s).await;
1104 Ok(CheckResult::Stopped(GateOutcome::failed(
1105 classify::classify_migration_error(&err_s, None),
1106 )))
1107 }
1108 }
1109 }
1110
1111 pub(crate) async fn reset_scratch(db_url: &str, owner_role: &str) -> Result<()> {
1112 use sqlx::Executor;
1113 use sqlx::postgres::PgPoolOptions;
1114 let pool = PgPoolOptions::new()
1115 .max_connections(1)
1116 .connect(db_url)
1117 .await?;
1118 // `owner_role` is validated `[A-Za-z0-9_]+` at config load, so interpolating
1119 // it into DDL is sound. It still goes through `format('%I')` inside the DO
1120 // block for the quoting Postgres expects on an identifier.
1121 let sql = format!(
1122 r#"
1123 DO $$
1124 DECLARE s text;
1125 BEGIN
1126 -- The dump restores objects owned by the prod role and re-grants to
1127 -- it (`ALTER ... OWNER TO {owner_role}`), which errors if the role
1128 -- is absent — superuser does not imply the role exists. Create it
1129 -- NOLOGIN: the scratch DB needs the role as an *owner* only, never
1130 -- as a connecting identity. Idempotent, so a re-reset is a no-op.
1131 IF NOT EXISTS (SELECT 1 FROM pg_roles WHERE rolname = '{owner_role}') THEN
1132 EXECUTE format('CREATE ROLE %I NOLOGIN', '{owner_role}');
1133 END IF;
1134
1135 -- Drop every non-system schema, not just public — migrations create
1136 -- custom schemas (e.g. tower_sessions) that survive `DROP SCHEMA
1137 -- public CASCADE` and then collide on the next migration run.
1138 FOR s IN
1139 SELECT nspname FROM pg_namespace
1140 WHERE nspname NOT LIKE 'pg_%'
1141 AND nspname NOT IN ('information_schema')
1142 LOOP
1143 EXECUTE format('DROP SCHEMA IF EXISTS %I CASCADE', s);
1144 END LOOP;
1145 EXECUTE 'CREATE SCHEMA public';
1146 -- Restore the pre-PG15 public-schema default on the throwaway
1147 -- scratch DB. Without this, the freshly-created public is owned by
1148 -- the connecting role (sando) with no grant to anyone else, so a
1149 -- migration's FK/trigger check that Postgres runs as a *restored*
1150 -- prod-owned table's owner ({owner_role} from the backup dump)
1151 -- fails with "permission denied for schema public". Granting to
1152 -- PUBLIC is role-agnostic and safe here — this DB is disposable and
1153 -- exists only to dry-run migrations.
1154 EXECUTE 'GRANT USAGE, CREATE ON SCHEMA public TO PUBLIC';
1155 -- PG15+: the new owner needs CREATE on public in its own right, not
1156 -- only via PUBLIC, for the restore's owner-scoped DDL.
1157 EXECUTE format('GRANT USAGE, CREATE ON SCHEMA public TO %I', '{owner_role}');
1158 END $$;
1159 "#
1160 );
1161 pool.execute(sqlx::raw_sql(sqlx::AssertSqlSafe(sql)))
1162 .await?;
1163 pool.close().await;
1164 Ok(())
1165 }
1166
1167 /// Startup assertion for the scratch cluster: the gates reset it, seed an owner
1168 /// role into it, and drop leftover test databases in it — none of which a plain
1169 /// unprivileged role can do. Historically these preconditions were satisfied by
1170 /// hand on fw13 (`ALTER ROLE sando SUPERUSER`, a hand-created `makenotwork`
1171 /// role) and nothing recorded that, so a rebuild elsewhere would fail one gate
1172 /// at a time with an opaque permissions error. Assert once, at boot, loudly.
1173 ///
1174 /// Not part of `--check-config`: that path is pure by design (no DB, no
1175 /// network), and a green there must mean "this build understands its config",
1176 /// not "the cluster is reachable".
1177 pub async fn preflight_scratch_privileges(db_url: &str) -> Result<()> {
1178 use sqlx::postgres::PgPoolOptions;
1179 let pool = PgPoolOptions::new()
1180 .max_connections(1)
1181 .connect(db_url)
1182 .await
1183 .context("connecting to scratch_db_url for the startup privilege check")?;
1184 let (is_super, can_signal): (bool, bool) = sqlx::query_as(
1185 "SELECT rolsuper, pg_catalog.pg_has_role(current_user, 'pg_signal_backend', 'USAGE')
1186 FROM pg_roles WHERE rolname = current_user",
1187 )
1188 .fetch_one(&pool)
1189 .await?;
1190 pool.close().await;
1191 anyhow::ensure!(
1192 is_super || can_signal,
1193 "the scratch_db_url role has neither SUPERUSER nor pg_signal_backend; migration_dry_run \
1194 and cargo_test cannot reset the scratch DB or clear stale test databases. Grant one:\n \
1195 ALTER ROLE <role> SUPERUSER; -- what fw13 uses\n \
1196 GRANT pg_signal_backend TO <role>; -- narrower: terminate only, cannot drop \
1197 foreign-owned databases",
1198 );
1199 if !is_super {
1200 tracing::warn!(
1201 "scratch role has pg_signal_backend but not SUPERUSER: stale test databases owned by \
1202 another role cannot be dropped, and the scratch owner role cannot be created if absent"
1203 );
1204 }
1205 Ok(())
1206 }
1207
1208 /// Best-effort cleanup of stale per-test database clones (`mnw_test_<uuid>`)
1209 /// left behind by a previously-killed `cargo_test` run.
1210 ///
1211 /// Drops **foreign-owned leftovers too**, which is why the daemon asserts
1212 /// SUPERUSER at startup (`preflight_scratch_privileges`): `DROP DATABASE`
1213 /// requires ownership or superuser, and the `WITH (FORCE)` terminate requires
1214 /// superuser or `pg_signal_backend`. Without both, orphans from a run under a
1215 /// different role accumulate and degrade the gate — the failure this cleanup
1216 /// exists to prevent.
1217 ///
1218 /// OPERATIONAL HAZARD: fw13 runs one Postgres cluster shared with local `cargo
1219 /// test` as `max`, so a gate firing mid-local-test will force-drop that run's
1220 /// databases out from under it. That collision is known and tracked separately
1221 /// (give the gate its own cluster); until then, do not run local tests on fw13
1222 /// while a Sando gate is live.
1223 ///
1224 /// Deliberately **excludes the template** (`mnw_test_template_*`): the harness
1225 /// reuses it across runs when it's migration-current (skipping a full
1226 /// drop+migrate), so dropping it here would force a needless rebuild every
1227 /// gate run. Templates are bounded (one per role) and never accumulate, so
1228 /// leaving them is free. Never returns an error: a cleanup miss must not turn a
1229 /// deploy red.
1230 async fn clean_stale_test_dbs(db_url: &str) {
1231 use sqlx::Executor;
1232 use sqlx::postgres::PgPoolOptions;
1233 let pool = match PgPoolOptions::new()
1234 .max_connections(1)
1235 .connect(db_url)
1236 .await
1237 {
1238 Ok(p) => p,
1239 Err(e) => {
1240 tracing::warn!(error = %e, "stale test-db cleanup: could not connect; skipping");
1241 return;
1242 }
1243 };
1244 // Every per-test clone, whoever owns it. The ownership filter this used to
1245 // carry is what let foreign-owned orphans pile up; superuser (asserted at
1246 // startup) makes them droppable.
1247 let names: Vec<(String,)> = sqlx::query_as(
1248 "SELECT datname FROM pg_database
1249 WHERE datname LIKE 'mnw_test_%'
1250 AND datname NOT LIKE '%template%'",
1251 )
1252 .fetch_all(&pool)
1253 .await
1254 .unwrap_or_default();
1255 let count = names.len();
1256 for (name,) in names {
1257 // `name` comes straight from pg_database; quoting it is sufficient.
1258 if let Err(e) = pool
1259 .execute(sqlx::raw_sql(sqlx::AssertSqlSafe(format!(
1260 "DROP DATABASE IF EXISTS \"{name}\" WITH (FORCE)"
1261 ))))
1262 .await
1263 {
1264 tracing::warn!(error = %e, db = %name, "stale test-db cleanup: drop failed");
1265 }
1266 }
1267 if count > 0 {
1268 tracing::info!(
1269 count,
1270 "stale test-db cleanup: dropped leftover mnw_test_* databases"
1271 );
1272 }
1273 pool.close().await;
1274 }
1275
1276 /// Build the restore shell line. Two pipelines we accept:
1277 /// *.sql -> psql -v ON_ERROR_STOP=1 $url < dump
1278 /// *.sql.gz -> set -o pipefail; gunzip -c dump | psql -v ON_ERROR_STOP=1 $url
1279 ///
1280 /// Two safety flags are load-bearing (CF4):
1281 /// - `ON_ERROR_STOP=1`: without it, psql exits 0 even when individual statements
1282 /// error, so a partial/corrupt restore would *pass* the gate.
1283 /// - `set -o pipefail`: without it a shell pipeline reports only the last
1284 /// command's status, so a `gunzip` failure on a truncated archive is masked by
1285 /// psql's exit. pipefail is a bash builtin (not POSIX sh), so the runner uses
1286 /// `bash -c`.
1287 fn restore_shell(db_url: &str, dump: &str) -> String {
1288 if std::path::Path::new(dump)
1289 .extension()
1290 .is_some_and(|ext| ext.eq_ignore_ascii_case("gz"))
1291 {
1292 format!(
1293 "set -o pipefail; gunzip -c {q} | psql -v ON_ERROR_STOP=1 {url}",
1294 q = shell_escape(dump),
1295 url = shell_escape(db_url),
1296 )
1297 } else {
1298 format!(
1299 "psql -v ON_ERROR_STOP=1 {url} < {q}",
1300 url = shell_escape(db_url),
1301 q = shell_escape(dump),
1302 )
1303 }
1304 }
1305
1306 async fn restore_dump(db_url: &str, dump: &str, log: &GateLog) -> Result<()> {
1307 // Split the password out of the URL and hand it to psql via PGPASSWORD, so it
1308 // never lands in argv (visible in /proc/<pid>/cmdline to any local user).
1309 // The sanitized URL — user/host/db, no secret — goes on the command line.
1310 let (sanitized, password) = split_pg_password(db_url);
1311 let shell = restore_shell(&sanitized, dump);
1312 // `bash` (not `sh`): `set -o pipefail` is a bash builtin. The restore runs
1313 // locally on the Sando host (fw13), which has bash.
1314 let mut cmd = Command::new("bash");
1315 cmd.arg("-c").arg(&shell);
1316 // kill_on_drop so the gate's wall-clock ceiling (dispatcher-level timeout on
1317 // migration_dry_run) can't orphan a wedged psql restore.
1318 cmd.kill_on_drop(true);
1319 if let Some(pw) = password {
1320 cmd.env("PGPASSWORD", pw);
1321 }
1322 // Streamed, not `.output()`: a prod-sized restore runs for minutes, and
1323 // psql's progress is the only thing an operator has to watch during it.
1324 let (_stdout, stderr, status) = log.run(&mut cmd).await?;
1325 anyhow::ensure!(
1326 status.success(),
1327 "restore failed: {}",
1328 String::from_utf8_lossy(&stderr),
1329 );
1330 Ok(())
1331 }
1332
1333 /// Split a `postgres://user:password@host/db` URL into its password-free form and
1334 /// the (percent-decoded) password. Returns the URL unchanged with `None` when
1335 /// there is no userinfo password. psql reads the password from `PGPASSWORD`, so
1336 /// keeping it off the command line removes the /proc exposure.
1337 fn split_pg_password(db_url: &str) -> (String, Option<String>) {
1338 let Some(after) = db_url.find("://").map(|i| i + 3) else {
1339 return (db_url.to_string(), None);
1340 };
1341 // The authority ends at the first '/', '?' or '#'; the password (if any) is
1342 // between the first ':' and the '@' within the userinfo of that authority.
1343 let authority_end = db_url[after..]
1344 .find(['/', '?', '#'])
1345 .map_or(db_url.len(), |i| after + i);
1346 let Some(at) = db_url[after..authority_end].find('@').map(|i| after + i) else {
1347 return (db_url.to_string(), None);
1348 };
1349 let userinfo = &db_url[after..at];
1350 let Some(colon) = userinfo.find(':') else {
1351 return (db_url.to_string(), None);
1352 };
1353 let password = percent_decode(&userinfo[colon + 1..]);
1354 let sanitized = format!(
1355 "{}{}{}",
1356 &db_url[..after],
1357 &userinfo[..colon],
1358 &db_url[at..]
1359 );
1360 (sanitized, Some(password))
1361 }
1362
1363 /// Minimal `%XX` percent-decode for a URL userinfo component. Non-escape bytes
1364 /// pass through; a malformed escape is left literal.
1365 fn percent_decode(s: &str) -> String {
1366 let b = s.as_bytes();
1367 let mut out = Vec::with_capacity(b.len());
1368 let mut i = 0;
1369 while i < b.len() {
1370 if b[i] == b'%'
1371 && i + 2 < b.len()
1372 && let (Some(h), Some(l)) = (hex_val(b[i + 1]), hex_val(b[i + 2]))
1373 {
1374 out.push((h << 4) | l);
1375 i += 3;
1376 } else {
1377 out.push(b[i]);
1378 i += 1;
1379 }
1380 }
1381 String::from_utf8_lossy(&out).into_owned()
1382 }
1383
1384 fn hex_val(c: u8) -> Option<u8> {
1385 match c {
1386 b'0'..=b'9' => Some(c - b'0'),
1387 b'a'..=b'f' => Some(c - b'a' + 10),
1388 b'A'..=b'F' => Some(c - b'A' + 10),
1389 _ => None,
1390 }
1391 }
1392
1393 pub(crate) async fn run_migrator(db_url: &str, dir: &std::path::Path) -> Result<()> {
1394 use sqlx::postgres::PgPoolOptions;
1395 let pool = PgPoolOptions::new()
1396 .max_connections(1)
1397 .connect(db_url)
1398 .await?;
1399 let migrator = sqlx::migrate::Migrator::new(dir).await?;
1400 migrator.run(&pool).await?;
1401 pool.close().await;
1402 Ok(())
1403 }
1404
1405 fn shell_escape(s: &str) -> String {
1406 format!("'{}'", s.replace('\'', "'\\''"))
1407 }
1408
1409 // ---- code_smoke ----
1410
1411 /// A 32+ char throwaway signing secret for the `code_smoke` boot. The server
1412 /// only enforces length (>= 32) outside production, and code_smoke's loopback
1413 /// `HOST_URL` keeps it in dev mode, so the value is irrelevant beyond that.
1414 const CODE_SMOKE_SIGNING_SECRET: &str = "sando-code-smoke-dummy-signing-secret-0000000000";
1415
1416 /// Seconds to wait for the real server to come up and serve `GET /health`
1417 /// during `code_smoke`. Longer than `boot_smoke`'s 3s: this boots the *full*
1418 /// server (config, pool, session store, webauthn, doc load, app build), not the
1419 /// minimal no-DB smoke server. The whole gate is also bounded by
1420 /// `gate_timeout_secs` at the dispatcher.
1421 const CODE_SMOKE_READY_SECS: u64 = 30;
1422
1423 /// `code_smoke` — the first host gate. Boots the freshly-built binary against a
1424 /// throwaway *empty* DB it migrates from scratch and seeds the example catalog
1425 /// into, then proves the real server serves `GET /health` against that
1426 /// nonempty DB. Fast and infra-light (one local Postgres, no prod-dump restore,
1427 /// no scratch-role reset, no external services), so a green here proves the
1428 /// code is sound and isolates a later `cargo_test`/`migration_dry_run` red as an
1429 /// environment problem rather than a code one.
1430 ///
1431 /// Reuses existing binary entrypoints, so the server needs no smoke-specific
1432 /// mode: `<bin> --seed-examples` loads config, connects, migrates from scratch,
1433 /// seeds, and exits; a plain `<bin>` then serves the real app. Both run with CWD
1434 /// at the server crate root so `docs/business/assumptions.toml` + `site-docs/`
1435 /// resolve (a missing assumptions file panics real startup), and with a loopback
1436 /// `HOST_URL` so config stays in dev mode (no CDN/S3/signing-secret prod
1437 /// enforcement). The seed's host allowlist already admits `127.0.0.1`, and the
1438 /// fresh DB trivially satisfies its no-real-users guard.
1439 async fn code_smoke(ctx: &GateCtx, run_id: GateRunId) -> Result<GateOutcome> {
1440 let log = GateLog::open(ctx, run_id, GateKind::CodeSmoke).await;
1441 let outcome = code_smoke_inner(ctx, &log).await;
1442 log.close().await;
1443 outcome.map(|o| o.with_log_ref(LogRef::new(&ctx.version, GateKind::CodeSmoke)))
1444 }
1445
1446 /// The staged interior of [`code_smoke`], writing every step through the gate's
1447 /// live log. Same split as [`migration_dry_run_inner`]: the caller owns the sink
1448 /// and attaches the `log_ref`.
1449 async fn code_smoke_inner(ctx: &GateCtx, log: &GateLog) -> Result<GateOutcome> {
1450 let Some(scratch_url) = ctx.cfg.scratch_db_url.as_deref() else {
1451 log.line("scratch_db_url unset in daemon config\n").await;
1452 return Ok(GateOutcome::blocked(GateBlocker::ScratchDbUrlUnset));
1453 };
1454
1455 // The staged binary (set by build_and_run_host before gating). code_smoke
1456 // runs first among the host gates, but staging precedes all gating, so the
1457 // artifact path is already recorded.
1458 let bin: Option<(String,)> =
1459 sqlx::query_as("SELECT artifact_path FROM versions WHERE app = ? AND version = ?")
1460 .bind(&ctx.cfg.id)
1461 .bind(&ctx.version)
1462 .fetch_optional(&ctx.pool)
1463 .await?;
1464 let Some((bin,)) = bin else {
1465 return Ok(GateOutcome::blocked(GateBlocker::ArtifactMissing {
1466 version: ctx.version.clone(),
1467 }));
1468 };
1469
1470 // Frontend builds, before anything else: they need no DB and no staged
1471 // binary, and a `tsc` error is the one failure the Rust build deliberately
1472 // swallows (both MNW build scripts emit `cargo::warning` and succeed against
1473 // a stale `static/dist/`). Failing here is what stops the deploy rsyncing
1474 // the previous build's bundle.
1475 if let Some(outcome) = code_smoke_frontends(ctx, log).await {
1476 return Ok(outcome);
1477 }
1478
1479 // Docs integrity, first and cheapest: run the staged binary's DB-free
1480 // `MNW_CHECK_DOCS` mode before creating the throwaway DB. A broken internal
1481 // docs link (a `[..](x.md)` resolving to a slug no page serves) fails here
1482 // in well under a second instead of after a full migrate+seed+boot, and a
1483 // rotted link never reaches prod as a live 404. Collisions are reported by
1484 // the check but do not fail it; only broken links do.
1485 if let Some(outcome) = code_smoke_docs_check(ctx, &bin, log).await {
1486 return Ok(outcome);
1487 }
1488
1489 let dbname = code_smoke_db_name(&ctx.version);
1490 let maintenance_url = pg_url_with_dbname(scratch_url, "postgres");
1491 let throwaway_url = pg_url_with_dbname(scratch_url, &dbname);
1492
1493 // Create the throwaway DB (dropping any stale one from a killed prior run).
1494 log.line(&format!("---- createdb {dbname} ----\n")).await;
1495 if let Err(e) = pg_create_db(&maintenance_url, &dbname).await {
1496 let reason = format!("createdb {dbname}: {e}");
1497 log.line(&reason).await;
1498 return Ok(GateOutcome::failed(GateFailure::CodeSmokeSetup { reason }));
1499 }
1500
1501 // Everything past createdb must drop the DB on the way out, pass or fail.
1502 let outcome = code_smoke_body(ctx, &bin, &throwaway_url, log).await;
1503
1504 log.line(&format!("\n---- dropdb {dbname} ----\n")).await;
1505 if let Err(e) = pg_drop_db(&maintenance_url, &dbname).await {
1506 // A teardown miss must not turn a passing gate red — log it and move on.
1507 // The next run's createdb drops it first anyway.
1508 tracing::warn!(error = %e, db = %dbname, "code_smoke: dropdb failed; next run will reclaim it");
1509 log.line(&format!("dropdb warning (non-fatal): {e}")).await;
1510 }
1511
1512 Ok(outcome)
1513 }
1514
1515 /// Compile every configured `frontend_build` in the worktree.
1516 ///
1517 /// Returns `Some(failed)` on the first project that does not build; `None` when
1518 /// all of them do (or none are configured). Output streams to `log` either way.
1519 ///
1520 /// `npm ci` runs only when `node_modules` is absent. Usually it is not: the app
1521 /// build script installed it during the `cargo build` that produced the artifact
1522 /// this gate is about to smoke, so the common path here is just `npm run build`
1523 /// against a warm install — seconds. The install branch covers the gate running
1524 /// against a worktree whose build script was skipped or failed at `npm ci`, and
1525 /// it is as fatal as a compile failure, because the alternative is compiling
1526 /// against whatever some earlier sha installed.
1527 ///
1528 /// Unlike the app build scripts, nothing here is best-effort. That asymmetry is
1529 /// the point of the gate.
1530 async fn code_smoke_frontends(ctx: &GateCtx, log: &GateLog) -> Option<GateOutcome> {
1531 if ctx.cfg.frontend_builds.is_empty() {
1532 return None;
1533 }
1534 let worktree = match ctx.worktree_for(GateKind::CodeSmoke) {
1535 Ok(w) => w.to_path_buf(),
1536 Err(outcome) => return Some(outcome),
1537 };
1538 for fe in &ctx.cfg.frontend_builds {
1539 let dir = worktree.join(&fe.dir);
1540 let label = fe.dir.display().to_string();
1541 log.line(&format!("---- frontend build ({label}) ----\n"))
1542 .await;
1543
1544 if !dir.is_dir() {
1545 // An older sha predating the frontend, mid-bisect. Skipping keeps
1546 // sando able to rebuild history; the log says so out loud.
1547 log.line(&format!("{label} absent from this worktree; skipping\n"))
1548 .await;
1549 continue;
1550 }
1551
1552 if !dir.join("node_modules").is_dir()
1553 && let Some(outcome) = run_npm(&dir, &label, &["ci"], "npm ci", ctx, log).await
1554 {
1555 return Some(outcome);
1556 }
1557
1558 if let Some(outcome) = run_npm(
1559 &dir,
1560 &label,
1561 &["run", &fe.script],
1562 &format!("npm run {}", fe.script),
1563 ctx,
1564 log,
1565 )
1566 .await
1567 {
1568 return Some(outcome);
1569 }
1570 }
1571 None
1572 }
1573
1574 /// One `npm` invocation for [`code_smoke_frontends`], bounded by the gate
1575 /// timeout so a wedged install cannot hold the whole pipeline (the enclosing
1576 /// `code_smoke` ceiling would catch it eventually, but this attributes the
1577 /// failure to the project that hung).
1578 async fn run_npm(
1579 dir: &std::path::Path,
1580 label: &str,
1581 args: &[&str],
1582 what: &str,
1583 ctx: &GateCtx,
1584 log: &GateLog,
1585 ) -> Option<GateOutcome> {
1586 log.line(&format!("$ {what}\n")).await;
1587 let mut cmd = tokio::process::Command::new("npm");
1588 cmd.args(args).current_dir(dir).kill_on_drop(true);
1589 let ceiling = std::time::Duration::from_secs(ctx.cfg.gate_timeout_secs);
1590 // On the timeout branch the whole `run` future is dropped, which drops the
1591 // child; `kill_on_drop` is what turns that into an actual kill.
1592 let status = match tokio::time::timeout(ceiling, log.run(&mut cmd)).await {
1593 Ok(Ok((_stdout, _stderr, status))) => status,
1594 Ok(Err(e)) => {
1595 // A missing `npm` lands here. Fatal, not skipped: a build host
1596 // without Node cannot produce the bundle the release serves, and
1597 // silently passing is how the stale bundle shipped in the first place.
1598 log.line(&format!("{what} could not be spawned: {e}\n"))
1599 .await;
1600 return Some(GateOutcome::failed(GateFailure::SpawnFailed {
1601 message: format!("{what} in {label}: {e}"),
1602 }));
1603 }
1604 Err(_elapsed) => {
1605 log.line(&format!(
1606 "{what} timed out after {}s\n",
1607 ctx.cfg.gate_timeout_secs
1608 ))
1609 .await;
1610 return Some(GateOutcome::failed(GateFailure::CodeSmokeFrontend {
1611 dir: label.to_string(),
1612 exit_code: None,
1613 }));
1614 }
1615 };
1616 if status.success() {
1617 return None;
1618 }
1619 Some(GateOutcome::failed(GateFailure::CodeSmokeFrontend {
1620 dir: label.to_string(),
1621 exit_code: status.code(),
1622 }))
1623 }
1624
1625 /// Run the staged binary's DB-free docs integrity check (`MNW_CHECK_DOCS=1`).
1626 ///
1627 /// Returns `Some(failed)` if the check reports broken links, cannot be spawned,
1628 /// or overruns its ceiling; `None` when the docs are clean. Output streams to
1629 /// `log` either way. The 60s ceiling backstops the case where the staged binary
1630 /// predates the flag and would fall through to a normal (DB-needing) boot and
1631 /// hang.
1632 async fn code_smoke_docs_check(ctx: &GateCtx, bin: &str, log: &GateLog) -> Option<GateOutcome> {
1633 let server_dir = match ctx.worktree_for(GateKind::CodeSmoke) {
1634 Ok(w) => w.join("server"),
1635 Err(outcome) => return Some(outcome),
1636 };
1637 log.line("---- docs check (MNW_CHECK_DOCS) ----\n").await;
1638 let mut cmd = tokio::process::Command::new(bin);
1639 cmd.env("MNW_CHECK_DOCS", "1")
1640 .current_dir(&server_dir)
1641 .kill_on_drop(true);
1642 let (stdout, _stderr, status) =
1643 match tokio::time::timeout(std::time::Duration::from_mins(1), log.run(&mut cmd)).await {
1644 Ok(Ok(out)) => out,
1645 Ok(Err(e)) => {
1646 log.line(&format!("docs check spawn failed: {e}\n")).await;
1647 return Some(GateOutcome::failed(GateFailure::SpawnFailed {
1648 message: e.to_string(),
1649 }));
1650 }
1651 Err(_elapsed) => {
1652 let reason =
1653 "docs check timed out after 60s (staged binary may predate MNW_CHECK_DOCS)"
1654 .to_string();
1655 log.line(&format!("{reason}\n")).await;
1656 return Some(GateOutcome::failed(GateFailure::CodeSmokeSetup { reason }));
1657 }
1658 };
1659 if status.success() {
1660 return None;
1661 }
1662 Some(GateOutcome::failed(GateFailure::CodeSmokeDocs {
1663 broken: parse_check_docs_broken_count(&stdout),
1664 }))
1665 }
1666
1667 /// Best-effort parse of the broken-link count from the `MNW_CHECK_DOCS` sentinel
1668 /// line (`MNW_CHECK_DOCS: N broken link(s)`). Returns 0 if absent — the failure
1669 /// still stands, only the summary count is unknown.
1670 fn parse_check_docs_broken_count(stdout: &[u8]) -> u32 {
1671 let text = String::from_utf8_lossy(stdout);
1672 for line in text.lines() {
1673 if let Some(rest) = line.strip_prefix("MNW_CHECK_DOCS:") {
1674 for tok in rest.split_whitespace() {
1675 if let Ok(n) = tok.parse::<u32>() {
1676 return n;
1677 }
1678 }
1679 }
1680 }
1681 0
1682 }
1683
1684 /// The createdb-to-dropdb interior of `code_smoke`: migrate+seed, then boot and
1685 /// probe. Returns the outcome without a `log_ref` (the caller attaches it after
1686 /// teardown). Never returns `Err` — spawn/child failures map to typed outcomes.
1687 async fn code_smoke_body(ctx: &GateCtx, bin: &str, db_url: &str, log: &GateLog) -> GateOutcome {
1688 let server_dir = match ctx.worktree_for(GateKind::CodeSmoke) {
1689 Ok(w) => w.join("server"),
1690 Err(outcome) => return outcome,
1691 };
1692
1693 // Phase 1: migrate-from-scratch + seed. `--seed-examples` loads config,
1694 // connects, runs migrations against the empty DB, seeds the catalog, exits.
1695 // A non-zero exit here is the "code is unsound" signal (broken migration,
1696 // seed error, or config-load failure).
1697 log.line("---- migrate + seed (--seed-examples) ----\n")
1698 .await;
1699 let mut seed_cmd = tokio::process::Command::new(bin);
1700 seed_cmd.arg("--seed-examples").current_dir(&server_dir);
1701 code_smoke_env(&mut seed_cmd, ctx, db_url);
1702 seed_cmd.env("ALLOW_EXAMPLE_SEED", "1").kill_on_drop(true);
1703 let seed_status = match log.run(&mut seed_cmd).await {
1704 Ok((_stdout, _stderr, status)) => status,
1705 Err(e) => {
1706 return GateOutcome::failed(GateFailure::SpawnFailed {
1707 message: e.to_string(),
1708 });
1709 }
1710 };
1711 if !seed_status.success() {
1712 return GateOutcome::failed(GateFailure::CodeSmokeSeed {
1713 exit_code: seed_status.code(),
1714 });
1715 }
1716
1717 // Phase 2: boot the real server against the now-migrated + seeded DB and
1718 // assert both startup signals: it logs `listening` (emitted just before the
1719 // socket bind) AND serves GET /health with a 200. The full stdout/stderr is
1720 // persisted for the operator either way.
1721 log.line("\n---- boot + probe /health ----\n").await;
1722 let mut serve_cmd = tokio::process::Command::new(bin);
1723 serve_cmd.current_dir(&server_dir);
1724 code_smoke_env(&mut serve_cmd, ctx, db_url);
1725 serve_cmd
1726 .stdout(std::process::Stdio::piped())
1727 .stderr(std::process::Stdio::piped())
1728 .kill_on_drop(true);
1729 let mut child = match serve_cmd.spawn() {
1730 Ok(c) => c,
1731 Err(e) => {
1732 return GateOutcome::failed(GateFailure::SpawnFailed {
1733 message: e.to_string(),
1734 });
1735 }
1736 };
1737
1738 // Stream stdout/stderr into the gate log (and out as chunk events) while
1739 // the probe loop runs below; the tasks finish when the pipes close (child
1740 // exits or is killed). The buffers they return are what the `listening`
1741 // assertion reads.
1742 let (stdout_task, stderr_task) = log.drain_pipes(&mut child);
1743
1744 let probe_timeout = std::time::Duration::from_millis(500);
1745 let started = std::time::Instant::now();
1746 let window = std::time::Duration::from_secs(CODE_SMOKE_READY_SECS);
1747 let mut probe_ok_after: Option<u32> = None;
1748 let mut last_probe_err = "never responded".to_string();
1749 let mut early_exit = None;
1750 while started.elapsed() < window {
1751 if let Ok(Some(status)) = child.try_wait() {
1752 early_exit = Some(status);
1753 break;
1754 }
1755 match tokio::time::timeout(probe_timeout, probe_health(ctx.cfg.code_smoke_port)).await {
1756 Ok(Ok(())) => {
1757 probe_ok_after = Some(started.elapsed().as_millis() as u32);
1758 break;
1759 }
1760 Ok(Err(e)) => last_probe_err = e,
1761 Err(_) => last_probe_err = "probe timed out".to_string(),
1762 }
1763 tokio::time::sleep(std::time::Duration::from_millis(250)).await;
1764 }
1765
1766 let exit = match early_exit {
1767 Some(status) => Some(status),
1768 None => {
1769 let e = child.try_wait().ok().flatten();
1770 if e.is_none() {
1771 let _ = child.kill().await;
1772 }
1773 e
1774 }
1775 };
1776 // Keep the serve run's own output so the `listening`-log assertion checks it
1777 // (not the seed run's, which exits before binding). The bytes are already in
1778 // the gate log; these buffers exist only for the assertion.
1779 let serve_stdout = stdout_task.await.unwrap_or_default();
1780 let serve_stderr = stderr_task.await.unwrap_or_default();
1781 let logged_listening =
1782 bytes_contain(&serve_stdout, b"listening") || bytes_contain(&serve_stderr, b"listening");
1783
1784 match (exit, probe_ok_after) {
1785 // Exited on its own within the window — panic / config error / bind fail.
1786 (Some(status), _) => GateOutcome::failed(classify::classify_boot_smoke(status.code())),
1787 // Stayed up and served /health. Assert both required startup signals:
1788 // the `listening` bind log AND the /health 200.
1789 (None, Some(after_ms)) if logged_listening => {
1790 GateOutcome::passed(PassNote::HealthyProbe { after_ms })
1791 }
1792 // Served /health but the `listening` log never appeared.
1793 (None, Some(_)) => GateOutcome::failed(GateFailure::CodeSmokeNoListeningLog),
1794 // Stayed up but never served /health — started, not ready.
1795 (None, None) => GateOutcome::failed(GateFailure::BootHealthProbeFailed {
1796 last_error: last_probe_err,
1797 }),
1798 }
1799 }
1800
1801 /// Substring search over raw bytes (the server's log output), for the
1802 /// `code_smoke` startup-log assertion. Avoids a lossy UTF-8 conversion of the
1803 /// whole buffer just to run `str::contains`.
1804 fn bytes_contain(haystack: &[u8], needle: &[u8]) -> bool {
1805 if needle.is_empty() || haystack.len() < needle.len() {
1806 return needle.is_empty();
1807 }
1808 haystack.windows(needle.len()).any(|w| w == needle)
1809 }
1810
1811 /// Apply the minimal env every `code_smoke` invocation shares: point the binary
1812 /// at the throwaway DB, force loopback (dev-mode config, no prod enforcement),
1813 /// hand it a dummy signing secret, and disable file scanning (no AV/YARA on the
1814 /// build host). The worktree is a clean git checkout, so no stray `.env` shadows
1815 /// these (and dotenvy never overrides already-set vars).
1816 ///
1817 /// This list has to carry EVERY var the server's `Config::from_env` treats as
1818 /// mandatory, because `code_smoke` is the only gate that reaches that function
1819 /// at all: `boot_smoke` and the docs check both short-circuit in `main` before
1820 /// config is loaded. So when the server makes a new var required, this is where
1821 /// it has to be answered, and nothing connects the two lists automatically.
1822 ///
1823 /// That has already bitten once. `CDN_BASE_URL` became mandatory on 2026-07-30
1824 /// (server 3e3b1d15, closing a cover URL that could expire), 13 days after this
1825 /// function was written, and `code_smoke` failed with `MissingCdnBaseUrl` the
1826 /// first time it ran on the host — which was 2026-08-03, because the gate was
1827 /// configured in the repo and absent from `/etc/sando` in between.
1828 ///
1829 /// The values are deliberately throwaway. The gate asks whether this code can
1830 /// migrate, seed, boot and serve; whether a given deployment's env is complete
1831 /// is the `config_check_env_file` guard's job, on the node, against that node's
1832 /// real env file.
1833 fn code_smoke_env(cmd: &mut tokio::process::Command, ctx: &GateCtx, db_url: &str) {
1834 // `localhost`, not `127.0.0.1`, and the distinction is load-bearing. The
1835 // server derives its WebAuthn relying-party id from HOST_URL's host, and
1836 // `WebauthnBuilder::new` validates that id against `Url::domain()` — which
1837 // is `None` for an IP literal, so an origin of `http://127.0.0.1:<port>`
1838 // fails with WebauthnError::Configuration before the server ever binds. It
1839 // is a real ceiling on the gate, not a preference: no IP-literal origin can
1840 // boot this binary. `localhost` is a domain, and matches the derived rp_id.
1841 //
1842 // HOST stays 127.0.0.1: that is the bind address, and the gate probes the
1843 // loopback address directly, so only the advertised origin changes.
1844 let origin = format!("http://localhost:{}", ctx.cfg.code_smoke_port);
1845 cmd.env("DATABASE_URL", db_url)
1846 .env("HOST", "127.0.0.1")
1847 .env("PORT", ctx.cfg.code_smoke_port.to_string())
1848 .env("HOST_URL", &origin)
1849 // Required unconditionally by Config::from_env. Pointing it at the smoke
1850 // server's own origin keeps every rendered media URL resolvable within
1851 // the gate; no request is ever made to it.
1852 .env("CDN_BASE_URL", &origin)
1853 .env("SIGNING_SECRET", CODE_SMOKE_SIGNING_SECRET)
1854 .env("SCAN_ENABLED", "false")
1855 .env("INSECURE_COOKIES", "1");
1856 }
1857
1858 /// The throwaway smoke DB name for a version: `sando_code_smoke_<version>` with
1859 /// every non-alphanumeric char folded to `_` and lowercased, capped at Postgres'
1860 /// 63-byte identifier limit. Sanitized to `[a-z0-9_]` so it's safe to quote into
1861 /// DDL. Deterministic per version, so a stale DB from a killed run is reclaimed
1862 /// by the next run's `DROP DATABASE IF EXISTS` rather than accumulating.
1863 fn code_smoke_db_name(version: &Version) -> String {
1864 let mut name = String::from("sando_code_smoke_");
1865 for c in version.to_string().chars() {
1866 name.push(if c.is_ascii_alphanumeric() {
1867 c.to_ascii_lowercase()
1868 } else {
1869 '_'
1870 });
1871 }
1872 name.truncate(63);
1873 name
1874 }
1875
1876 /// Rewrite a `postgres://` URL to point at database `dbname`, preserving scheme,
1877 /// userinfo, host/port, and any query (e.g. the socket `?host=/var/run/postgresql`
1878 /// form) + fragment. Used to derive the maintenance connection (`postgres`) and
1879 /// the throwaway smoke DB URL from the configured `scratch_db_url`.
1880 fn pg_url_with_dbname(url: &str, dbname: &str) -> String {
1881 let Some(after_scheme) = url.find("://").map(|i| i + 3) else {
1882 return url.to_string();
1883 };
1884 let rest = &url[after_scheme..];
1885 // Authority ends at the first '/', '?' or '#'; whatever follows is the
1886 // path (the old dbname) plus an optional query/fragment we must keep.
1887 let auth_end = rest.find(['/', '?', '#']).unwrap_or(rest.len());
1888 let authority = &rest[..auth_end];
1889 let tail = &rest[auth_end..];
1890 let query_and_frag = match tail.find(['?', '#']) {
1891 Some(i) => &tail[i..],
1892 None => "",
1893 };
1894 format!(
1895 "{}{}/{}{}",
1896 &url[..after_scheme],
1897 authority,
1898 dbname,
1899 query_and_frag
1900 )
1901 }
1902
1903 /// Create the throwaway smoke DB on the cluster `maintenance_url` points at,
1904 /// dropping any stale one first. `dbname` is sanitized to `[a-z0-9_]` by
1905 /// `code_smoke_db_name`, so quoting it is sufficient. `CREATE DATABASE` cannot
1906 /// run inside a transaction, so these go through the simple-query protocol (a
1907 /// raw `&str` execute), matching `reset_scratch`.
1908 async fn pg_create_db(maintenance_url: &str, dbname: &str) -> Result<()> {
1909 use sqlx::Executor;
1910 use sqlx::postgres::PgPoolOptions;
1911 let pool = PgPoolOptions::new()
1912 .max_connections(1)
1913 .connect(maintenance_url)
1914 .await?;
1915 pool.execute(sqlx::raw_sql(sqlx::AssertSqlSafe(format!(
1916 "DROP DATABASE IF EXISTS \"{dbname}\" WITH (FORCE)"
1917 ))))
1918 .await?;
1919 pool.execute(sqlx::raw_sql(sqlx::AssertSqlSafe(format!(
1920 "CREATE DATABASE \"{dbname}\""
1921 ))))
1922 .await?;
1923 pool.close().await;
1924 Ok(())
1925 }
1926
1927 /// Drop the throwaway smoke DB, forcing off any lingering connection (the killed
1928 /// server's pool). Best-effort at the call site — a failure is logged, not fatal.
1929 async fn pg_drop_db(maintenance_url: &str, dbname: &str) -> Result<()> {
1930 use sqlx::Executor;
1931 use sqlx::postgres::PgPoolOptions;
1932 let pool = PgPoolOptions::new()
1933 .max_connections(1)
1934 .connect(maintenance_url)
1935 .await?;
1936 pool.execute(sqlx::raw_sql(sqlx::AssertSqlSafe(format!(
1937 "DROP DATABASE IF EXISTS \"{dbname}\" WITH (FORCE)"
1938 ))))
1939 .await?;
1940 pool.close().await;
1941 Ok(())
1942 }
1943
1944 async fn boot_smoke(ctx: &GateCtx, run_id: GateRunId) -> Result<GateOutcome> {
1945 let bin: Option<(String,)> =
1946 sqlx::query_as("SELECT artifact_path FROM versions WHERE app = ? AND version = ?")
1947 .bind(&ctx.cfg.id)
1948 .bind(&ctx.version)
1949 .fetch_optional(&ctx.pool)
1950 .await?;
1951 let Some((bin,)) = bin else {
1952 return Ok(GateOutcome::blocked(GateBlocker::ArtifactMissing {
1953 version: ctx.version.clone(),
1954 }));
1955 };
1956
1957 // Readiness smoke: start the binary and confirm it serves `GET /health`
1958 // within the window, not merely that the process stays up. Panics in main,
1959 // missing config, and port-bind failures still surface as an early exit; a
1960 // process that comes up but never serves /health is now its own failure.
1961 //
1962 // The server requires DATABASE_URL or it panics on config load before
1963 // we can observe anything. We point it at the scratch DB (already
1964 // migrated by the build step and refreshed by migration_dry_run if
1965 // that gate ran first). SCAN_ENABLED=false skips loading YARA rules
1966 // from /opt/makenotwork/yara-rules which doesn't exist on the build
1967 // host. SANDO_BOOT_SMOKE_PORT tells the smoke server which loopback port
1968 // to bind so we know where to probe. Other config has sane optional defaults.
1969 let mut cmd = tokio::process::Command::new(&bin);
1970 cmd.env("SANDO_BOOT_SMOKE", "1")
1971 .env("SANDO_BOOT_SMOKE_PORT", ctx.cfg.boot_smoke_port.to_string())
1972 .env("SCAN_ENABLED", "false")
1973 .stdout(std::process::Stdio::piped())
1974 .stderr(std::process::Stdio::piped())
1975 .kill_on_drop(true);
1976 if let Some(scratch_url) = ctx.cfg.scratch_db_url.as_deref() {
1977 cmd.env("DATABASE_URL", scratch_url);
1978 }
1979 let log_path = gate_log_path(ctx, GateKind::BootSmoke);
1980 let log_ref = LogRef::new(&ctx.version, GateKind::BootSmoke);
1981 let mut child = match cmd.spawn() {
1982 Ok(c) => c,
1983 Err(e) => {
1984 // Spawn failures get a one-off log line via LiveLog so the
1985 // on-disk file still exists for `GET /logs/...`.
1986 let mut log = LiveLog::open(log_path, gate_chunk_cb(ctx.events.clone(), run_id)).await;
1987 log.write_chunk(format!("spawn: {e}\n").as_bytes()).await;
1988 log.close().await;
1989 return Ok(GateOutcome::failed(GateFailure::SpawnFailed {
1990 message: e.to_string(),
1991 })
1992 .with_log_ref(log_ref));
1993 }
1994 };
1995
1996 // The boot smoke window is 3s. Drain stdout/stderr concurrently through
1997 // a shared LiveLog sink so the operator sees panics/log lines stream in
1998 // real time before the kill, AND the on-disk log gets the full byte
1999 // stream for post-mortem reads. The drainers exit when their pipe
2000 // closes — which happens when the child exits naturally or after kill.
2001 let log = std::sync::Arc::new(tokio::sync::Mutex::new(
2002 LiveLog::open(log_path, gate_chunk_cb(ctx.events.clone(), run_id)).await,
2003 ));
2004 let stdout_task = tokio::spawn(stream_into_log(child.stdout.take(), log.clone()));
2005 let stderr_task = tokio::spawn(stream_into_log(child.stderr.take(), log.clone()));
2006
2007 // Poll readiness across the 3s window instead of a flat sleep: GET /health
2008 // must return 2xx. A crash mid-window short-circuits to the exit-code
2009 // failure path (try_wait below); a process that stays up but never serves
2010 // /health is a distinct readiness failure.
2011 let probe_timeout = std::time::Duration::from_millis(500);
2012 let started = std::time::Instant::now();
2013 let window = std::time::Duration::from_secs(3);
2014 let mut probe_ok_after: Option<u32> = None;
2015 let mut last_probe_err = "never responded".to_string();
2016 let mut early_exit = None;
2017 while started.elapsed() < window {
2018 if let Some(status) = child.try_wait()? {
2019 early_exit = Some(status);
2020 break;
2021 }
2022 match tokio::time::timeout(probe_timeout, probe_health(ctx.cfg.boot_smoke_port)).await {
2023 Ok(Ok(())) => {
2024 probe_ok_after = Some(started.elapsed().as_millis() as u32);
2025 break;
2026 }
2027 Ok(Err(e)) => last_probe_err = e,
2028 Err(_) => last_probe_err = "probe timed out".to_string(),
2029 }
2030 tokio::time::sleep(std::time::Duration::from_millis(150)).await;
2031 }
2032
2033 // Stop the child unless it already exited, then drain the log tasks.
2034 let exit = match early_exit {
2035 Some(status) => Some(status),
2036 None => {
2037 let e = child.try_wait()?;
2038 if e.is_none() {
2039 let _ = child.kill().await;
2040 }
2041 e
2042 }
2043 };
2044 // The streamed bytes already landed in the live log and the on-disk file for
2045 // the post-mortem reader. Drain the join handles to avoid hangs.
2046 let _ = stdout_task.await;
2047 let _ = stderr_task.await;
2048 // Unique owner of the Arc at this point (both tasks dropped their clones).
2049 if let Ok(mutex) = std::sync::Arc::try_unwrap(log) {
2050 mutex.into_inner().close().await;
2051 }
2052
2053 match (exit, probe_ok_after) {
2054 // Exited on its own within the window — a crash/panic/bind failure.
2055 (Some(status), _) => {
2056 let failure = classify::classify_boot_smoke(status.code());
2057 Ok(GateOutcome::failed(failure).with_log_ref(log_ref))
2058 }
2059 // Stayed up and served /health — readiness proven.
2060 (None, Some(after_ms)) => {
2061 Ok(GateOutcome::passed(PassNote::HealthyProbe { after_ms }).with_log_ref(log_ref))
2062 }
2063 // Stayed up but never served /health — started, not ready.
2064 (None, None) => Ok(GateOutcome::failed(GateFailure::BootHealthProbeFailed {
2065 last_error: last_probe_err,
2066 })
2067 .with_log_ref(log_ref)),
2068 }
2069 }
2070
2071 /// One readiness probe of the boot-smoke server: connect to `127.0.0.1:port`
2072 /// and `GET /health`, returning `Ok(())` only on a `200`. A hand-rolled HTTP/1.0
2073 /// request over a raw `TcpStream` keeps the outbound probe dependency-free
2074 /// (reqwest is dev-only); the smoke server serves the one route over axum, which
2075 /// speaks 1.0. `Err` carries a short reason for the operator's failure note. The
2076 /// caller wraps each call in a timeout.
2077 async fn probe_health(port: u16) -> std::result::Result<(), String> {
2078 use tokio::io::{AsyncReadExt, AsyncWriteExt};
2079 let mut stream = tokio::net::TcpStream::connect((std::net::Ipv4Addr::LOCALHOST, port))
2080 .await
2081 .map_err(|e| format!("connect: {e}"))?;
2082 stream
2083 .write_all(b"GET /health HTTP/1.0\r\nHost: localhost\r\nConnection: close\r\n\r\n")
2084 .await
2085 .map_err(|e| format!("write: {e}"))?;
2086 let mut buf = Vec::new();
2087 stream
2088 .read_to_end(&mut buf)
2089 .await
2090 .map_err(|e| format!("read: {e}"))?;
2091 let text = String::from_utf8_lossy(&buf);
2092 let status_line = text.lines().next().unwrap_or("");
2093 if status_line.contains(" 200 ") {
2094 Ok(())
2095 } else {
2096 Err(format!("unexpected status line: {status_line:?}"))
2097 }
2098 }
2099
2100 /// Sink that drops streamed output. `node_health` keeps the probe's stderr from
2101 /// its `RunOutput` for the failure note, so the live byte stream isn't needed.
2102 struct DiscardSink;
2103 #[async_trait::async_trait]
2104 impl ops_exec::LogSink for DiscardSink {
2105 async fn write_chunk(&mut self, _bytes: &[u8]) {}
2106 }
2107
2108 /// `node_health` — the post-deploy gate that proves the *deployed nodes* are
2109 /// serving, recording one outcome per (tier, version) that the next promote
2110 /// checks. Distinct from `boot_smoke`, which boots the staged artifact on the
2111 /// build host: this probes each node over the same executor the deploy used, so
2112 /// a node that took a corrupt artifact, wrong-arch binary, or failed restart is
2113 /// caught here rather than waved through (Run-2 SERIOUS-3). Fails closed: any
2114 /// unhealthy node fails the gate, and an empty node set is `Blocked`.
2115 async fn node_health(ctx: &GateCtx) -> Result<GateOutcome> {
2116 if ctx.nodes.is_empty() {
2117 return Ok(GateOutcome::blocked(GateBlocker::NoNodesToProbe));
2118 }
2119 for probe in &ctx.nodes {
2120 if let Err(detail) = probe_node(probe).await {
2121 return Ok(GateOutcome::failed(GateFailure::NodeUnhealthy {
2122 node: probe.node.to_string(),
2123 detail,
2124 }));
2125 }
2126 }
2127 Ok(GateOutcome::passed(PassNote::NodesHealthy {
2128 nodes: ctx.nodes.len() as u32,
2129 }))
2130 }
2131
2132 /// Probe one node over its executor: confirm the unit is active post-restart
2133 /// and, when a `health_url` is configured, that it serves a 2xx. Retries across
2134 /// ~10s because the service may still be restarting / warming. Runs under the
2135 /// read-only `Observe(Health)` capability (every Sando node grants it), so the
2136 /// probe needs no deploy authority. `Ok(())` = healthy; `Err(detail)` carries a
2137 /// short reason for the gate's failure note.
2138 async fn probe_node(probe: &NodeProbe) -> std::result::Result<(), String> {
2139 use ops_exec::{Action, ObserveKind, Step, sh_quote};
2140 let svc = sh_quote(&probe.service);
2141 let url = probe
2142 .health_url
2143 .as_deref()
2144 .map_or_else(|| "''".to_string(), sh_quote);
2145 // One executor round-trip with the retry loop on the node: is-active, then
2146 // (if a url is set) curl it for a 2xx. Exit 0 only when both hold.
2147 let script = format!(
2148 "svc={svc}; url={url}; \
2149 for _ in $(seq 1 10); do \
2150 if systemctl is-active --quiet \"$svc\"; then \
2151 if [ -z \"$url\" ] || curl -fsS --max-time 5 \"$url\" >/dev/null 2>&1; then exit 0; fi; \
2152 fi; \
2153 sleep 1; \
2154 done; \
2155 echo 'service not active or health url not 2xx after retries' >&2; exit 1"
2156 );
2157 let step = Step::shell(Action::Observe(ObserveKind::Health), script);
2158 let mut sink = DiscardSink;
2159 let out = probe
2160 .executor
2161 .run_streaming(&step, &mut sink)
2162 .await
2163 .map_err(|e| format!("probe spawn: {e}"))?;
2164 if out.status.success() {
2165 Ok(())
2166 } else {
2167 let code = out
2168 .status
2169 .code()
2170 .map_or_else(|| "signal".to_string(), |c| c.to_string());
2171 let stderr: String = String::from_utf8_lossy(&out.stderr)
2172 .chars()
2173 .take(200)
2174 .collect();
2175 Err(format!("exit {code}: {stderr}"))
2176 }
2177 }
2178
2179 /// Drain `stream` into the shared `LiveLog` (which forwards each chunk to
2180 /// the on-disk log file AND broadcasts a `GateLogChunk` event), and return
2181 /// the concatenated bytes so the classifier can still operate on the full
2182 /// output post-hoc.
2183 async fn stream_into_log<R>(
2184 stream: Option<R>,
2185 log: std::sync::Arc<tokio::sync::Mutex<LiveLog>>,
2186 ) -> Vec<u8>
2187 where
2188 R: tokio::io::AsyncRead + Unpin + Send + 'static,
2189 {
2190 let mut total = Vec::new();
2191 let Some(mut s) = stream else { return total };
2192 let mut buf = [0u8; 4096];
2193 loop {
2194 match s.read(&mut buf).await {
2195 Ok(0) => break,
2196 Err(_) => break,
2197 Ok(n) => {
2198 total.extend_from_slice(&buf[..n]);
2199 log.lock().await.write_chunk(&buf[..n]).await;
2200 }
2201 }
2202 }
2203 total
2204 }
2205
2206 /// Spawn a child, drain its stdout/stderr through a `LiveLog`, return the
2207 /// combined buffers and exit status. Shared by `cargo_test` (no deadline)
2208 /// and ad-hoc callers — `boot_smoke` rolls its own variant because of its
2209 /// 3s kill window.
2210 async fn stream_child_to_live_log(
2211 child: &mut tokio::process::Child,
2212 events: EventTx,
2213 run_id: GateRunId,
2214 log_path: PathBuf,
2215 ) -> Result<(Vec<u8>, Vec<u8>, std::process::ExitStatus)> {
2216 let log = GateLog::new(LiveLog::open(log_path, gate_chunk_cb(events, run_id)).await);
2217 let out = log.stream_child(child).await;
2218 log.close().await;
2219 out
2220 }
2221
2222 /// One gate's live log, held across every step of a multi-step gate.
2223 ///
2224 /// Single-child gates call [`stream_child_to_live_log`] and are done. The
2225 /// staged gates (`code_smoke`, `migration_dry_run`) instead run several
2226 /// children plus banner lines between them, and they hold one `GateLog` across
2227 /// the lot: a single sink means chunk sequence numbers stay monotonic for the
2228 /// whole gate, and the on-disk log reads in the order things actually happened
2229 /// rather than as stdout-then-stderr assembled at the end.
2230 ///
2231 /// Every write is best-effort in the same way `LiveLog` is: a log directory
2232 /// that cannot be written degrades to callback-only and never turns a passing
2233 /// gate red.
2234 struct GateLog {
2235 sink: Arc<tokio::sync::Mutex<LiveLog>>,
2236 }
2237
2238 impl GateLog {
2239 fn new(sink: LiveLog) -> Self {
2240 Self {
2241 sink: Arc::new(tokio::sync::Mutex::new(sink)),
2242 }
2243 }
2244
2245 /// Open the sink for `gate`'s log file, streaming to the TUI under `run_id`.
2246 async fn open(ctx: &GateCtx, run_id: GateRunId, gate: GateKind) -> Self {
2247 Self::new(
2248 LiveLog::open(
2249 gate_log_path(ctx, gate),
2250 gate_chunk_cb(ctx.events.clone(), run_id),
2251 )
2252 .await,
2253 )
2254 }
2255
2256 /// Emit a banner (or any line the gate itself produces) through the same
2257 /// sink the children stream to, so it lands in sequence with their output.
2258 async fn write(&self, bytes: &[u8]) {
2259 self.sink.lock().await.write_chunk(bytes).await;
2260 }
2261
2262 /// Same, for the common `format!`-a-line case.
2263 async fn line(&self, s: &str) {
2264 self.write(s.as_bytes()).await;
2265 }
2266
2267 /// Spawn `cmd` with both pipes captured, stream them into the sink as they
2268 /// arrive, and return the buffers plus the exit status. The buffers are
2269 /// what the classifiers still operate on post-hoc.
2270 async fn run(
2271 &self,
2272 cmd: &mut Command,
2273 ) -> std::io::Result<(Vec<u8>, Vec<u8>, std::process::ExitStatus)> {
2274 cmd.stdout(std::process::Stdio::piped())
2275 .stderr(std::process::Stdio::piped());
2276 let mut child = cmd.spawn()?;
2277 self.stream_child(&mut child)
2278 .await
2279 .map_err(std::io::Error::other)
2280 }
2281
2282 /// Drain an already-spawned child's pipes into the sink and wait for it.
2283 async fn stream_child(
2284 &self,
2285 child: &mut tokio::process::Child,
2286 ) -> Result<(Vec<u8>, Vec<u8>, std::process::ExitStatus)> {
2287 let (stdout_task, stderr_task) = self.drain_pipes(child);
2288 let status = child.wait().await?;
2289 let stdout_buf = stdout_task.await.unwrap_or_default();
2290 let stderr_buf = stderr_task.await.unwrap_or_default();
2291 Ok((stdout_buf, stderr_buf, status))
2292 }
2293
2294 /// Start draining a child's pipes into the sink *without* waiting on the
2295 /// child. For `code_smoke`'s serve phase, which probes `/health` while the
2296 /// server is still up. The caller must await both handles for the buffers
2297 /// (and before [`Self::close`], so the flush isn't skipped).
2298 #[allow(clippy::type_complexity)]
2299 fn drain_pipes(
2300 &self,
2301 child: &mut tokio::process::Child,
2302 ) -> (
2303 tokio::task::JoinHandle<Vec<u8>>,
2304 tokio::task::JoinHandle<Vec<u8>>,
2305 ) {
2306 (
2307 tokio::spawn(stream_into_log(child.stdout.take(), self.sink.clone())),
2308 tokio::spawn(stream_into_log(child.stderr.take(), self.sink.clone())),
2309 )
2310 }
2311
2312 /// Flush the file. A still-outstanding streaming task (only possible if the
2313 /// gate returned without awaiting it) leaves the `Arc` shared, in which case
2314 /// the flush is skipped — `LiveLog` writes unbuffered to the OS either way,
2315 /// so nothing already written is lost.
2316 async fn close(self) {
2317 if let Ok(mutex) = Arc::try_unwrap(self.sink) {
2318 mutex.into_inner().close().await;
2319 }
2320 }
2321 }
2322
2323 fn gate_log_path(ctx: &GateCtx, gate: GateKind) -> PathBuf {
2324 ctx.cfg
2325 .logs_root
2326 .join(ctx.version.to_string())
2327 .join(format!("{}.log", gate.as_str()))
2328 }
2329
2330 /// Live check: has `tier`'s burn-in window of `hours` elapsed since its clock
2331 /// (`tier_state.burn_in_started_at`, started by a promote onto the tier)? Used
2332 /// by the promote-time gate check (`unsatisfied_gates`) so a stale `blocked`
2333 /// row never masks an elapsed — or not-yet-elapsed — window. The `burn_in` gate
2334 /// runner below wraps the same state with a richer outcome for `/state`.
2335 pub async fn burn_in_satisfied(
2336 pool: &SqlitePool,
2337 app: &AppId,
2338 tier: &TierId,
2339 hours: u32,
2340 ) -> Result<bool> {
2341 let started: Option<String> =
2342 sqlx::query_scalar("SELECT burn_in_started_at FROM tier_state WHERE app = ? AND tier = ?")
2343 .bind(app)
2344 .bind(tier)
2345 .fetch_optional(pool)
2346 .await?
2347 .flatten();
2348 let Some(started) = started else {
2349 return Ok(false);
2350 };
2351 let started = chrono::DateTime::parse_from_rfc3339(&started)?.with_timezone(&Utc);
2352 Ok(Utc::now() - started >= chrono::Duration::hours(hours as i64))
2353 }
2354
2355 async fn burn_in(ctx: &GateCtx, hours: u32) -> Result<GateOutcome> {
2356 // Check tier_state.burn_in_started_at on this tier; pass if enough time
2357 // has elapsed. The clock is started by /promote when a version lands on
2358 // the burn-in tier.
2359 let started: Option<String> =
2360 sqlx::query_scalar("SELECT burn_in_started_at FROM tier_state WHERE app = ? AND tier = ?")
2361 .bind(&ctx.cfg.id)
2362 .bind(&ctx.tier)
2363 .fetch_optional(&ctx.pool)
2364 .await?
2365 .flatten();
2366 let Some(started) = started else {
2367 return Ok(GateOutcome::blocked(GateBlocker::BurnInClockNotStarted));
2368 };
2369 let started = chrono::DateTime::parse_from_rfc3339(&started)?.with_timezone(&Utc);
2370 let elapsed = Utc::now() - started;
2371 let needed = chrono::Duration::hours(hours as i64);
2372 if elapsed >= needed {
2373 Ok(GateOutcome::passed(PassNote::BurnInElapsed {
2374 hours: elapsed.num_hours() as u32,
2375 }))
2376 } else {
2377 let remaining = (needed - elapsed).num_hours().max(0) as u32;
2378 Ok(GateOutcome::blocked(GateBlocker::BurnInRemaining {
2379 hours_remaining: remaining,
2380 hours_total: hours,
2381 }))
2382 }
2383 }
2384
2385 async fn manual_confirm(ctx: &GateCtx) -> Result<GateOutcome> {
2386 // Pass iff a row in gate_runs exists with status='passed' for this
2387 // (tier, version, manual_confirm) that was inserted out-of-band by an
2388 // operator action. Since the harness inserts the in-flight row itself,
2389 // look for a prior confirmation row.
2390 let prior_at: Option<String> = sqlx::query_scalar(
2391 "SELECT finished_at FROM gate_runs
2392 WHERE app = ? AND tier = ? AND version = ? AND gate_kind = 'manual_confirm'
2393 AND status = 'passed'
2394 ORDER BY id DESC LIMIT 1",
2395 )
2396 .bind(&ctx.cfg.id)
2397 .bind(&ctx.tier)
2398 .bind(&ctx.version)
2399 .fetch_optional(&ctx.pool)
2400 .await?;
2401 match prior_at {
2402 Some(at_str) => {
2403 let at = chrono::DateTime::parse_from_rfc3339(&at_str)
2404 .map_or_else(|_| Utc::now(), |d| d.with_timezone(&Utc));
2405 Ok(GateOutcome::passed(PassNote::OperatorConfirmed { at }))
2406 }
2407 None => Ok(GateOutcome::blocked(
2408 GateBlocker::AwaitingOperatorConfirmation,
2409 )),
2410 }
2411 }
2412
2413 #[cfg(test)]
2414 mod tests {
2415 use super::*;
2416 use crate::events;
2417 use sqlx::sqlite::SqlitePoolOptions;
2418
2419 fn target(dir: &str) -> crate::config::TestTarget {
2420 crate::config::TestTarget {
2421 dir: std::path::PathBuf::from(dir),
2422 aux_repo: None,
2423 features: Vec::new(),
2424 all_features: false,
2425 scratch_db: false,
2426 }
2427 }
2428
2429 #[test]
2430 fn parse_check_docs_broken_count_reads_the_sentinel() {
2431 // The failing sentinel carries the count.
2432 assert_eq!(
2433 parse_check_docs_broken_count(b"some log\nMNW_CHECK_DOCS: 3 broken link(s)\n"),
2434 3
2435 );
2436 // Count with a preceding log line still parses (first bare int wins).
2437 assert_eq!(
2438 parse_check_docs_broken_count(
2439 b" broken link: a -> b\nMNW_CHECK_DOCS: 1 broken link(s)\n"
2440 ),
2441 1
2442 );
2443 // The parser only runs on failure; the "ok" sentinel is never fed to it,
2444 // and its "(2" token is not a bare int, so it yields 0 harmlessly.
2445 assert_eq!(
2446 parse_check_docs_broken_count(b"MNW_CHECK_DOCS: ok (2 collision(s) reported)\n"),
2447 0
2448 );
2449 // Absent sentinel -> 0; the failure still stands, only the count is lost.
2450 assert_eq!(parse_check_docs_broken_count(b"unrelated output"), 0);
2451 }
2452
2453 #[test]
2454 fn name_target_points_a_test_failure_at_its_crate() {
2455 // With one target the crate was implicit; with fifteen the operator
2456 // needs the summary to say which one broke.
2457 let f = name_target(
2458 GateFailure::CargoTest {
2459 failed_count: 3,
2460 first_failed: Some("workflows::sync::round_trip".into()),
2461 first_panic: None,
2462 },
2463 std::path::Path::new("shared/synckit-client"),
2464 );
2465 assert_eq!(
2466 f.summary(),
2467 "3 test(s) failed; first: shared/synckit-client: workflows::sync::round_trip",
2468 );
2469 }
2470
2471 #[test]
2472 fn name_target_points_a_compile_failure_at_its_crate() {
2473 let f = name_target(
2474 GateFailure::CompileError {
2475 error_count: 1,
2476 first_error: Some("error[E0063]".into()),
2477 },
2478 std::path::Path::new("mnw-cli"),
2479 );
2480 assert_eq!(
2481 f.summary(),
2482 "compile failed (1 error(s)); first: mnw-cli: error[E0063]"
2483 );
2484 }
2485
2486 #[test]
2487 fn name_target_names_the_crate_even_without_a_test_name() {
2488 let f = name_target(
2489 GateFailure::CargoTest {
2490 failed_count: 2,
2491 first_failed: None,
2492 first_panic: None,
2493 },
2494 std::path::Path::new("pom"),
2495 );
2496 assert_eq!(f.summary(), "2 test(s) failed; first: pom");
2497 }
2498
2499 #[test]
2500 fn name_target_leaves_unrelated_failures_alone() {
2501 let f = name_target(
2502 GateFailure::SpawnFailed {
2503 message: "no cargo".into(),
2504 },
2505 std::path::Path::new("pom"),
2506 );
2507 assert!(matches!(f, GateFailure::SpawnFailed { .. }));
2508 }
2509
2510 /// `target()` above, but resolved against an aux repo's checkout.
2511 fn aux_target(dir: &str, repo: &str) -> crate::config::TestTarget {
2512 crate::config::TestTarget {
2513 aux_repo: Some(repo.to_string()),
2514 ..target(dir)
2515 }
2516 }
2517
2518 /// True when the URL's host parses as a domain rather than an IP literal,
2519 /// which is the distinction `Url::domain()` draws and WebAuthn depends on.
2520 fn url_host_is_a_domain(url: &str) -> bool {
2521 let after = url.split("://").nth(1).unwrap_or("");
2522 let host = after.split(['/', '?', '#']).next().unwrap_or("");
2523 let host = host.rsplit('@').next().unwrap_or(host);
2524 let host = if let Some(rest) = host.strip_prefix('[') {
2525 rest.split(']').next().unwrap_or("")
2526 } else {
2527 host.split(':').next().unwrap_or("")
2528 };
2529 !host.is_empty() && host.parse::<std::net::IpAddr>().is_err()
2530 }
2531
2532 fn resolving_ctx(worktree: &str, aux: &[(&str, &str)]) -> GateCtx {
2533 GateCtx {
2534 pool: SqlitePool::connect_lazy("sqlite::memory:").unwrap(),
2535 cfg: std::sync::Arc::new(crate::config::AppConfig::for_tests()),
2536 tier: TierId::new("host"),
2537 version: "0.1.0".parse().unwrap(),
2538 worktree: Some(PathBuf::from(worktree)),
2539 bundle: None,
2540 events: events::channel(),
2541 nodes: Vec::new(),
2542 build_id: None,
2543 aux_dirs: aux
2544 .iter()
2545 .map(|(n, d)| ((*n).to_string(), PathBuf::from(d)))
2546 .collect(),
2547 }
2548 }
2549
2550 #[tokio::test]
2551 async fn a_plain_target_resolves_under_the_worktree() {
2552 let ctx = resolving_ctx("/w/abc123", &[]);
2553 assert_eq!(
2554 ctx.target_dir(&target("shared/tagtree")),
2555 Some(PathBuf::from("/w/abc123/shared/tagtree")),
2556 );
2557 }
2558
2559 #[tokio::test]
2560 async fn an_aux_target_resolves_beside_the_worktree_not_under_it() {
2561 // The whole point: `Libraries/docengine` is a sibling of the per-sha
2562 // worktree, so a worktree-relative path can never reach it.
2563 let ctx = resolving_ctx("/w/abc123", &[("docengine", "/w/Libraries/docengine")]);
2564 assert_eq!(
2565 ctx.target_dir(&aux_target("", "docengine")),
2566 Some(PathBuf::from("/w/Libraries/docengine")),
2567 );
2568 // A subdirectory of an aux repo resolves under its checkout.
2569 assert_eq!(
2570 ctx.target_dir(&aux_target("crates/inner", "docengine")),
2571 Some(PathBuf::from("/w/Libraries/docengine/crates/inner")),
2572 );
2573 }
2574
2575 #[tokio::test]
2576 async fn an_aux_target_with_no_checkout_this_run_resolves_to_nothing() {
2577 // Promote-time gates carry no aux dirs. Resolving to a wrong path (say,
2578 // the worktree) would run the gate against whatever happened to sit
2579 // there; `None` makes the caller skip, and --check-config is what
2580 // catches a real typo.
2581 let ctx = resolving_ctx("/w/abc123", &[]);
2582 assert_eq!(ctx.target_dir(&aux_target("", "docengine")), None);
2583 }
2584
2585 #[test]
2586 fn labels_name_the_repo_an_aux_target_lives_in() {
2587 assert_eq!(target("server").label(), "server");
2588 assert_eq!(aux_target("", "docengine").label(), "docengine (aux)");
2589 assert_eq!(
2590 aux_target("crates/inner", "docengine").label(),
2591 "docengine/crates/inner (aux)",
2592 );
2593 }
2594
2595 /// A `GateCtx` over `worktree` with the given frontend projects configured.
2596 /// No DB, no artifact — `code_smoke_frontends` touches neither.
2597 async fn frontend_ctx(worktree: &std::path::Path, dirs: &[&str]) -> GateCtx {
2598 let mut cfg = crate::config::AppConfig::for_tests();
2599 cfg.frontend_builds = dirs
2600 .iter()
2601 .map(|d| crate::config::FrontendBuild {
2602 dir: PathBuf::from(d),
2603 script: "build".into(),
2604 })
2605 .collect();
2606 cfg.logs_root = worktree.join("logs");
2607 GateCtx {
2608 pool: SqlitePoolOptions::new()
2609 .max_connections(1)
2610 .connect("sqlite::memory:")
2611 .await
2612 .unwrap(),
2613 cfg: std::sync::Arc::new(cfg),
2614 tier: TierId::new("host"),
2615 version: "0.1.0".parse().unwrap(),
2616 worktree: Some(worktree.to_path_buf()),
2617 bundle: None,
2618 events: events::channel(),
2619 nodes: Vec::new(),
2620 build_id: None,
2621 aux_dirs: HashMap::new(),
2622 }
2623 }
2624
2625 /// A `GateCtx` for the `migration_dry_run` freshness checks: a migrated
2626 /// in-memory pool (so `backups` exists) and a scratch URL set, so the gate
2627 /// reaches the backup lookup instead of bailing on config. Nothing here
2628 /// touches postgres — every assertion below blocks before `reset_scratch`.
2629 async fn dry_run_ctx(worktree: &std::path::Path, max_age_hours: u32) -> GateCtx {
2630 let mut cfg = crate::config::AppConfig::for_tests();
2631 cfg.scratch_db_url = Some("postgres:///sando_scratch".into());
2632 cfg.backup_max_age_hours = max_age_hours;
2633 cfg.logs_root = worktree.join("logs");
2634 let pool = SqlitePoolOptions::new()
2635 .max_connections(1)
2636 .connect("sqlite::memory:")
2637 .await
2638 .unwrap();
2639 crate::db::migrate(&pool).await.unwrap();
2640 GateCtx {
2641 pool,
2642 cfg: std::sync::Arc::new(cfg),
2643 tier: TierId::new("host"),
2644 version: "0.1.0".parse().unwrap(),
2645 worktree: Some(worktree.to_path_buf()),
2646 bundle: None,
2647 events: events::channel(),
2648 nodes: Vec::new(),
2649 build_id: None,
2650 aux_dirs: HashMap::new(),
2651 }
2652 }
2653
2654 /// Record a `server` backup row fetched `hours_ago`, as `/backup/fetch` would.
2655 async fn seed_backup(ctx: &GateCtx, hours_ago: i64) {
2656 seed_named_backup(ctx, "server", hours_ago).await;
2657 }
2658
2659 /// Record a backup row for one named dump.
2660 async fn seed_named_backup(ctx: &GateCtx, name: &str, hours_ago: i64) {
2661 let at = (Utc::now() - chrono::Duration::hours(hours_ago)).to_rfc3339();
2662 sqlx::query(
2663 "INSERT INTO backups (name, fetched_at, source, local_path, byte_size)
2664 VALUES (?, ?, 'file:///x.sql.gz', '/tmp/sando-test-backup.sql.gz', 1000000)",
2665 )
2666 .bind(name)
2667 .bind(at)
2668 .execute(&ctx.pool)
2669 .await
2670 .unwrap();
2671 }
2672
2673 /// The multithreaded check, as `sando-daemon.toml` configures it.
2674 fn mt_check() -> crate::config::MigrationCheck {
2675 crate::config::MigrationCheck {
2676 dir: std::path::PathBuf::from("multithreaded/migrations"),
2677 backup: "multithreaded".into(),
2678 scratch_db: Some("sando_scratch_mt".into()),
2679 owner_role: Some("multithreaded".into()),
2680 }
2681 }
2682
2683 /// Re-point a `dry_run_ctx` at one check, keeping its pool and scratch URL.
2684 fn with_check(ctx: &mut GateCtx, check: crate::config::MigrationCheck) {
2685 let mut cfg = crate::config::AppConfig::for_tests();
2686 cfg.scratch_db_url = ctx.cfg.scratch_db_url.clone();
2687 cfg.backup_max_age_hours = ctx.cfg.backup_max_age_hours;
2688 cfg.logs_root = ctx.cfg.logs_root.clone();
2689 cfg.migration_checks = vec![check];
2690 ctx.cfg = std::sync::Arc::new(cfg);
2691 }
2692
2693 #[tokio::test]
2694 async fn migration_dry_run_blocks_when_a_checks_own_dump_was_never_fetched() {
2695 // The hazard the check list exists for: multithreaded applies its own
2696 // migrations at boot against its own database, so the server's dump must
2697 // never stand in for it. A fetched `server` row with no `multithreaded`
2698 // row is exactly that substitution, and it has to block.
2699 let tmp = tempfile::tempdir().unwrap();
2700 let mut ctx = dry_run_ctx(tmp.path(), 48).await;
2701 with_check(&mut ctx, mt_check());
2702 seed_named_backup(&ctx, "server", 1).await;
2703 let log = GateLog::open(&ctx, GateRunId(0), GateKind::MigrationDryRun).await;
2704
2705 let outcome = migration_dry_run_inner(&ctx, &log).await.unwrap();
2706 log.close().await;
2707
2708 let crate::outcome::GateStatus::Blocked { blocker } = outcome.status else {
2709 panic!("a missing multithreaded dump must block");
2710 };
2711 let GateBlocker::NoBackupAvailable { check, backup } = blocker else {
2712 panic!("expected NoBackupAvailable, got {blocker:?}");
2713 };
2714 assert_eq!(backup, "multithreaded", "names the dump that is missing");
2715 assert!(
2716 check.contains("multithreaded/migrations"),
2717 "names the check that wanted it, got {check}"
2718 );
2719 }
2720
2721 #[tokio::test]
2722 async fn migration_dry_run_freshness_is_per_dump() {
2723 // A fresh server dump must not make a 45-day-old multithreaded dump look
2724 // current: the clock is per-database, or the second check inherits the
2725 // first's freshness and the gate is theatre.
2726 let tmp = tempfile::tempdir().unwrap();
2727 let mut ctx = dry_run_ctx(tmp.path(), 48).await;
2728 with_check(&mut ctx, mt_check());
2729 seed_named_backup(&ctx, "server", 1).await;
2730 seed_named_backup(&ctx, "multithreaded", 24 * 45).await;
2731 let log = GateLog::open(&ctx, GateRunId(0), GateKind::MigrationDryRun).await;
2732
2733 let outcome = migration_dry_run_inner(&ctx, &log).await.unwrap();
2734 log.close().await;
2735
2736 let crate::outcome::GateStatus::Blocked { blocker } = outcome.status else {
2737 panic!("a 45-day-old multithreaded dump must block");
2738 };
2739 let GateBlocker::BackupStale { check, .. } = blocker else {
2740 panic!("expected BackupStale, got {blocker:?}");
2741 };
2742 assert!(
2743 check.contains("multithreaded/migrations"),
2744 "names the check whose dump is stale, got {check}"
2745 );
2746 }
2747
2748 #[tokio::test]
2749 async fn migration_dry_run_blocks_on_a_stale_backup() {
2750 // The failure this closes: the gate used to check only that a backups row
2751 // existed, so a fetch that silently stopped working left it green against
2752 // an ever-older schema. Sando ran 45 days that way.
2753 let tmp = tempfile::tempdir().unwrap();
2754 let ctx = dry_run_ctx(tmp.path(), 48).await;
2755 seed_backup(&ctx, 24 * 45).await;
2756 let log = GateLog::open(&ctx, GateRunId(0), GateKind::MigrationDryRun).await;
2757
2758 let outcome = migration_dry_run_inner(&ctx, &log).await.unwrap();
2759 log.close().await;
2760
2761 let crate::outcome::GateStatus::Blocked { blocker } = outcome.status else {
2762 panic!("a 45-day-old backup must block");
2763 };
2764 let GateBlocker::BackupStale {
2765 age_hours,
2766 max_age_hours,
2767 ..
2768 } = blocker
2769 else {
2770 panic!("expected BackupStale, got {blocker:?}");
2771 };
2772 assert_eq!(max_age_hours, 48);
2773 assert!(
2774 age_hours >= 24 * 45,
2775 "reports the real age, got {age_hours}"
2776 );
2777 }
2778
2779 #[tokio::test]
2780 async fn migration_dry_run_accepts_a_fresh_backup() {
2781 // The other side of the boundary: a backup inside the window must not be
2782 // blocked on freshness. It fails later (there is no such dump on disk),
2783 // which is exactly the proof the age check let it through.
2784 let tmp = tempfile::tempdir().unwrap();
2785 let ctx = dry_run_ctx(tmp.path(), 48).await;
2786 seed_backup(&ctx, 6).await;
2787 let log = GateLog::open(&ctx, GateRunId(0), GateKind::MigrationDryRun).await;
2788
2789 let outcome = migration_dry_run_inner(&ctx, &log).await.unwrap();
2790 log.close().await;
2791
2792 assert!(
2793 !matches!(
2794 outcome.status,
2795 crate::outcome::GateStatus::Blocked {
2796 blocker: GateBlocker::BackupStale { .. }
2797 }
2798 ),
2799 "a 6h-old backup is fresh, got {:?}",
2800 outcome.status,
2801 );
2802 }
2803
2804 #[tokio::test]
2805 async fn migration_dry_run_treats_an_unparsable_fetched_at_as_stale() {
2806 // Fail closed: `fetched_at` is daemon-written RFC 3339, so a value that
2807 // will not parse means the row is untrustworthy — and a freshness check
2808 // that shrugs at a timestamp it cannot read is not a freshness check.
2809 let tmp = tempfile::tempdir().unwrap();
2810 let ctx = dry_run_ctx(tmp.path(), 48).await;
2811 sqlx::query(
2812 "INSERT INTO backups (fetched_at, source, local_path, byte_size)
2813 VALUES ('not-a-timestamp', 'file:///x.sql.gz', '/tmp/x.sql.gz', 1000000)",
2814 )
2815 .execute(&ctx.pool)
2816 .await
2817 .unwrap();
2818 let log = GateLog::open(&ctx, GateRunId(0), GateKind::MigrationDryRun).await;
2819
2820 let outcome = migration_dry_run_inner(&ctx, &log).await.unwrap();
2821 log.close().await;
2822
2823 assert!(
2824 matches!(
2825 outcome.status,
2826 crate::outcome::GateStatus::Blocked {
2827 blocker: GateBlocker::BackupStale { .. }
2828 }
2829 ),
2830 "an unreadable fetched_at must block, got {:?}",
2831 outcome.status,
2832 );
2833 }
2834
2835 /// A `code_smoke` live log over `ctx.cfg.logs_root`, for the helpers that
2836 /// take one. `GateRunId(0)` never matches a real row; nothing reads the
2837 /// chunk events in these tests.
2838 async fn test_gate_log(ctx: &GateCtx) -> GateLog {
2839 GateLog::open(ctx, GateRunId(0), GateKind::CodeSmoke).await
2840 }
2841
2842 /// Close `log` (flushing it) and read back what it wrote on disk.
2843 async fn read_gate_log(ctx: &GateCtx, log: GateLog) -> String {
2844 log.close().await;
2845 tokio::fs::read_to_string(gate_log_path(ctx, GateKind::CodeSmoke))
2846 .await
2847 .expect("the gate log must exist on disk")
2848 }
2849
2850 /// Write a minimal npm project at `worktree/<dir>` whose `build` script
2851 /// exits with `exit_code`. Pre-creates `node_modules` so the gate skips
2852 /// `npm ci` — these tests are about the build step, not the network.
2853 fn fake_npm_project(worktree: &std::path::Path, dir: &str, exit_code: u8) {
2854 let root = worktree.join(dir);
2855 std::fs::create_dir_all(root.join("node_modules")).unwrap();
2856 std::fs::write(
2857 root.join("package.json"),
2858 format!(
2859 r#"{{"name":"fake","version":"0.0.0","private":true,
2860 "scripts":{{"build":"exit {exit_code}"}}}}"#
2861 ),
2862 )
2863 .unwrap();
2864 }
2865
2866 #[tokio::test]
2867 async fn frontend_gate_fails_on_a_build_error_and_names_the_project() {
2868 // The whole point of the gate: the app build scripts downgrade this to a
2869 // cargo::warning, so if it passes here nothing stops a stale bundle.
2870 let tmp = tempfile::tempdir().unwrap();
2871 fake_npm_project(tmp.path(), "server/frontend", 0);
2872 fake_npm_project(tmp.path(), "multithreaded/frontend", 2);
2873 let ctx = frontend_ctx(tmp.path(), &["server/frontend", "multithreaded/frontend"]).await;
2874
2875 let log = test_gate_log(&ctx).await;
2876 let outcome = code_smoke_frontends(&ctx, &log)
2877 .await
2878 .expect("a failing tsc must fail the gate");
2879 let crate::outcome::GateStatus::Failed { failure } = &outcome.status else {
2880 panic!("expected a failure, got {:?}", outcome.status)
2881 };
2882 assert!(
2883 matches!(
2884 failure,
2885 GateFailure::CodeSmokeFrontend { dir, exit_code: Some(2) }
2886 if dir == "multithreaded/frontend"
2887 ),
2888 "got: {failure:?}"
2889 );
2890 // The passing project ran first; its output belongs in the log too.
2891 let text = read_gate_log(&ctx, log).await;
2892 assert!(text.contains("server/frontend"), "log: {text}");
2893 }
2894
2895 #[tokio::test]
2896 async fn frontend_gate_passes_when_every_project_builds() {
2897 let tmp = tempfile::tempdir().unwrap();
2898 fake_npm_project(tmp.path(), "server/frontend", 0);
2899 let ctx = frontend_ctx(tmp.path(), &["server/frontend"]).await;
2900 let log = test_gate_log(&ctx).await;
2901 assert!(
2902 code_smoke_frontends(&ctx, &log).await.is_none(),
2903 "a clean build must not fail the gate"
2904 );
2905 }
2906
2907 /// The point of routing `code_smoke` through `LiveLog`: its output reaches
2908 /// the operator *while* the gate runs, as `GateLogChunk` events, instead of
2909 /// appearing all at once when the gate finishes.
2910 #[tokio::test]
2911 async fn code_smoke_streams_chunks_as_it_runs() {
2912 let tmp = tempfile::tempdir().unwrap();
2913 fake_npm_project(tmp.path(), "server/frontend", 0);
2914 let ctx = frontend_ctx(tmp.path(), &["server/frontend"]).await;
2915 let mut rx = ctx.events.subscribe_logs();
2916
2917 let log = GateLog::open(&ctx, GateRunId(7), GateKind::CodeSmoke).await;
2918 assert!(code_smoke_frontends(&ctx, &log).await.is_none());
2919 log.close().await;
2920
2921 let mut chunks = Vec::new();
2922 while let Ok(envelope) = rx.try_recv() {
2923 if let Event::GateLogChunk { run_id, seq, text } = envelope.event {
2924 assert_eq!(run_id, GateRunId(7));
2925 chunks.push((seq, text));
2926 }
2927 }
2928 assert!(!chunks.is_empty(), "no chunk ever reached the bus");
2929 // Sequence numbers are per-run and monotonic across every step of the
2930 // gate, which is why the whole gate shares one sink.
2931 let seqs: Vec<u32> = chunks.iter().map(|(seq, _)| *seq).collect();
2932 assert!(
2933 seqs.windows(2).all(|w| w[0] < w[1]),
2934 "chunk seq must be monotonic, got {seqs:?}"
2935 );
2936 let joined: String = chunks.into_iter().map(|(_, text)| text).collect();
2937 assert!(joined.contains("server/frontend"), "chunks: {joined}");
2938 }
2939
2940 #[tokio::test]
2941 async fn frontend_gate_skips_a_project_absent_from_the_worktree() {
2942 // Rebuilding an older sha that predates the frontend must stay possible.
2943 let tmp = tempfile::tempdir().unwrap();
2944 let ctx = frontend_ctx(tmp.path(), &["multithreaded/frontend"]).await;
2945 let log = test_gate_log(&ctx).await;
2946 assert!(code_smoke_frontends(&ctx, &log).await.is_none());
2947 assert!(
2948 read_gate_log(&ctx, log).await.contains("skipping"),
2949 "the skip must be visible in the log"
2950 );
2951 }
2952
2953 #[tokio::test]
2954 async fn cargo_test_fails_closed_when_no_target_exists_in_the_worktree() {
2955 // A worktree missing every configured crate must not report "tests
2956 // passed" having run none. Uses an empty tempdir as the worktree, so
2957 // no cargo process is ever spawned.
2958 let tmp = tempfile::tempdir().unwrap();
2959 let mut cfg = crate::config::AppConfig::for_tests();
2960 cfg.test_targets = vec![target("server"), target("mnw-cli")];
2961 cfg.logs_root = tmp.path().join("logs");
2962 let pool = SqlitePoolOptions::new()
2963 .max_connections(1)
2964 .connect("sqlite::memory:")
2965 .await
2966 .unwrap();
2967 let ctx = GateCtx {
2968 pool,
2969 cfg: std::sync::Arc::new(cfg),
2970 tier: TierId::new("host"),
2971 version: "0.1.0".parse().unwrap(),
2972 worktree: Some(tmp.path().to_path_buf()),
2973 bundle: None,
2974 events: events::channel(),
2975 nodes: Vec::new(),
2976 build_id: None,
2977 aux_dirs: HashMap::new(),
2978 };
2979 let out = cargo_test(&ctx, GateRunId(1)).await.unwrap();
2980 assert_eq!(
2981 out.status_str(),
2982 "failed",
2983 "green here would be a silent no-op gate"
2984 );
2985 let crate::outcome::GateStatus::Failed { failure } = &out.status else {
2986 panic!("expected a failure")
2987 };
2988 assert!(
2989 failure.summary().contains("ran no targets"),
2990 "got: {}",
2991 failure.summary()
2992 );
2993 }
2994
2995 #[tokio::test]
2996 async fn scratch_db_env_is_opt_in_per_target() {
2997 // Exporting DATABASE_URL knocks sqlx out of offline mode, so a crate
2998 // shipping .sqlx data must not see it.
2999 let mut cfg = crate::config::AppConfig::for_tests();
3000 cfg.scratch_db_url = Some("postgres://sando@127.0.0.1/sando_scratch".into());
3001 let ctx = GateCtx {
3002 pool: SqlitePoolOptions::new()
3003 .max_connections(1)
3004 .connect_lazy("sqlite::memory:")
3005 .unwrap(),
3006 cfg: std::sync::Arc::new(cfg),
3007 tier: TierId::new("host"),
3008 version: "0.1.0".parse().unwrap(),
3009 worktree: Some(std::path::PathBuf::from("/tmp/wt")),
3010 bundle: None,
3011 events: events::channel(),
3012 nodes: Vec::new(),
3013 build_id: None,
3014 aux_dirs: HashMap::new(),
3015 };
3016 let dir = std::path::Path::new("/tmp/wt/x");
3017
3018 let off = cargo_test_command(&ctx, dir, &target("x"), &[], &[]);
3019 let has_db = |c: &Command| {
3020 c.as_std()
3021 .get_envs()
3022 .any(|(k, v)| k == "DATABASE_URL" && v.is_some())
3023 };
3024 assert!(!has_db(&off), "scratch_db defaults off");
3025
3026 let mut on_target = target("x");
3027 on_target.scratch_db = true;
3028 assert!(
3029 has_db(&cargo_test_command(&ctx, dir, &on_target, &[], &[])),
3030 "opt-in exports it"
3031 );
3032 }
3033
3034 #[tokio::test]
3035 async fn all_features_replaces_the_feature_list() {
3036 let ctx = GateCtx {
3037 pool: SqlitePoolOptions::new()
3038 .max_connections(1)
3039 .connect_lazy("sqlite::memory:")
3040 .unwrap(),
3041 cfg: std::sync::Arc::new(crate::config::AppConfig::for_tests()),
3042 tier: TierId::new("host"),
3043 version: "0.1.0".parse().unwrap(),
3044 worktree: Some(std::path::PathBuf::from("/tmp/wt")),
3045 bundle: None,
3046 events: events::channel(),
3047 nodes: Vec::new(),
3048 build_id: None,
3049 aux_dirs: HashMap::new(),
3050 };
3051 let mut t = target("shared/ops-exec");
3052 t.all_features = true;
3053 let cmd = cargo_test_command(&ctx, std::path::Path::new("/tmp/wt"), &t, &[], &[]);
3054 let args: Vec<_> = cmd
3055 .as_std()
3056 .get_args()
3057 .map(|a| a.to_string_lossy().into_owned())
3058 .collect();
3059 assert!(args.iter().any(|a| a == "--all-features"), "got: {args:?}");
3060 assert!(!args.iter().any(|a| a == "--features"), "got: {args:?}");
3061 }
3062
3063 #[test]
3064 fn every_gate_kind_round_trips_through_its_wire_string() {
3065 // gate_kind is a TEXT column and a WS event field; as_str and FromStr
3066 // disagreeing would make a gate's evidence unreadable by
3067 // unsatisfied_gates, which fails the promote closed with no explanation.
3068 for k in [
3069 GateKind::CargoTest,
3070 GateKind::HardeningTest,
3071 GateKind::Clippy,
3072 GateKind::Fmt,
3073 GateKind::CargoAudit,
3074 GateKind::CargoDeny,
3075 GateKind::MigrationDryRun,
3076 GateKind::CodeSmoke,
3077 GateKind::BootSmoke,
3078 GateKind::NodeHealth,
3079 GateKind::BurnIn,
3080 GateKind::ManualConfirm,
3081 ] {
3082 assert_eq!(
3083 k.as_str().parse::<GateKind>().unwrap(),
3084 k,
3085 "round trip for {k:?}"
3086 );
3087 }
3088 }
3089
3090 #[test]
3091 fn first_meaningful_line_prefers_the_diagnostic() {
3092 let stderr = b" Updating crates.io index\nerror: 1 vulnerability found!\n";
3093 assert_eq!(
3094 first_meaningful_line(b"", stderr),
3095 "error: 1 vulnerability found!"
3096 );
3097 }
3098
3099 #[test]
3100 fn first_meaningful_line_finds_a_deny_verdict() {
3101 let out = b"advisories FAILED, bans ok, licenses FAILED, sources ok\n";
3102 assert!(first_meaningful_line(out, b"").contains("FAILED"));
3103 }
3104
3105 #[test]
3106 fn first_meaningful_line_falls_back_rather_than_returning_empty() {
3107 assert!(first_meaningful_line(b"", b"").contains("see the gate log"));
3108 }
3109
3110 #[tokio::test]
3111 async fn supply_chain_gates_skip_crates_without_their_config() {
3112 // Four crates in this repo fail `cargo audit` purely for want of a
3113 // triaged .cargo/audit.toml. Running it there would make the gate
3114 // permanently red, so a target only qualifies once it carries the file.
3115 // The worktree here has a Cargo.toml but no audit config, so nothing
3116 // qualifies and the gate fails closed rather than passing over zero work.
3117 let tmp = tempfile::tempdir().unwrap();
3118 let crate_dir = tmp.path().join("server");
3119 std::fs::create_dir_all(&crate_dir).unwrap();
3120 std::fs::write(crate_dir.join("Cargo.toml"), "[package]\nname = \"x\"\n").unwrap();
3121
3122 let mut cfg = crate::config::AppConfig::for_tests();
3123 cfg.test_targets = vec![target("server")];
3124 cfg.logs_root = tmp.path().join("logs");
3125 let ctx = GateCtx {
3126 pool: SqlitePoolOptions::new()
3127 .max_connections(1)
3128 .connect("sqlite::memory:")
3129 .await
3130 .unwrap(),
3131 cfg: std::sync::Arc::new(cfg),
3132 tier: TierId::new("host"),
3133 version: "0.1.0".parse().unwrap(),
3134 worktree: Some(tmp.path().to_path_buf()),
3135 bundle: None,
3136 events: events::channel(),
3137 nodes: Vec::new(),
3138 build_id: None,
3139 aux_dirs: HashMap::new(),
3140 };
3141 let out = supply_chain(&ctx, GateRunId(1), GateKind::CargoAudit)
3142 .await
3143 .unwrap();
3144 assert_eq!(out.status_str(), "failed");
3145 let crate::outcome::GateStatus::Failed { failure } = &out.status else {
3146 panic!("expected a failure")
3147 };
3148 assert!(
3149 failure.summary().contains("ran nothing"),
3150 "got: {}",
3151 failure.summary()
3152 );
3153
3154 // Drop the config in and the same target now qualifies.
3155 std::fs::create_dir_all(crate_dir.join(".cargo")).unwrap();
3156 std::fs::write(crate_dir.join(".cargo/audit.toml"), "[advisories]\n").unwrap();
3157 // The fixture crate is not a real cargo project, so the tool itself
3158 // still errors — but on its own terms, not with "ran nothing". That
3159 // distinction is the thing under test: the target was attempted.
3160 let out = supply_chain(&ctx, GateRunId(2), GateKind::CargoAudit)
3161 .await
3162 .unwrap();
3163 if let crate::outcome::GateStatus::Failed { failure } = &out.status {
3164 assert!(
3165 !failure.summary().contains("ran nothing"),
3166 "a target carrying the config must be attempted, not skipped; got: {}",
3167 failure.summary(),
3168 );
3169 }
3170 }
3171
3172 #[test]
3173 fn tests_run_reads_the_libtest_summary() {
3174 let out = b"running 6 tests\ntest auth_rate_limit_triggers_on_burst ... ok\n\n\
3175 test result: ok. 6 passed; 0 failed; 0 ignored; 0 measured; 118 filtered out\n";
3176 assert_eq!(tests_run(out), 6);
3177 }
3178
3179 #[test]
3180 fn tests_run_sums_across_test_binaries() {
3181 let out = b"test result: ok. 6 passed; 0 failed\ntest result: ok. 2 passed; 0 failed\n";
3182 assert_eq!(tests_run(out), 8);
3183 }
3184
3185 #[test]
3186 fn tests_run_is_zero_when_the_filter_matched_nothing() {
3187 // The case hardening_test fails closed on: a filter matching no tests
3188 // exits 0, so a rename would otherwise make the gate a green no-op.
3189 let out = b"running 0 tests\n\n\
3190 test result: ok. 0 passed; 0 failed; 0 ignored; 0 measured; 124 filtered out\n";
3191 assert_eq!(tests_run(out), 0);
3192 }
3193
3194 #[test]
3195 fn tests_run_is_zero_without_a_summary_line() {
3196 assert_eq!(tests_run(b"error: could not compile `makenotwork`\n"), 0);
3197 }
3198
3199 #[tokio::test]
3200 async fn hardening_test_command_carries_no_features() {
3201 // The entire point of the gate: production constants, which means no
3202 // `fast-tests`. A stray feature here silently restores the blind spot.
3203 let ctx = GateCtx {
3204 pool: SqlitePoolOptions::new()
3205 .max_connections(1)
3206 .connect_lazy("sqlite::memory:")
3207 .unwrap(),
3208 cfg: std::sync::Arc::new(crate::config::AppConfig::for_tests()),
3209 tier: TierId::new("host"),
3210 version: "0.1.0".parse().unwrap(),
3211 worktree: Some(std::path::PathBuf::from("/tmp/wt")),
3212 bundle: None,
3213 events: events::channel(),
3214 nodes: Vec::new(),
3215 build_id: None,
3216 aux_dirs: HashMap::new(),
3217 };
3218 let plain = crate::config::TestTarget {
3219 dir: std::path::PathBuf::from("server"),
3220 aux_repo: None,
3221 features: Vec::new(),
3222 all_features: false,
3223 scratch_db: true,
3224 };
3225 let cmd = cargo_test_command(
3226 &ctx,
3227 std::path::Path::new("/tmp/wt/server"),
3228 &plain,
3229 &[],
3230 &["--test", "integration"],
3231 );
3232 let args: Vec<_> = cmd
3233 .as_std()
3234 .get_args()
3235 .map(|a| a.to_string_lossy().into_owned())
3236 .collect();
3237 assert!(
3238 !args.iter().any(|a| a == "--features"),
3239 "hardening_test must pass no features: {args:?}"
3240 );
3241 assert!(
3242 !args.iter().any(|a| a.contains("fast-tests")),
3243 "got: {args:?}"
3244 );
3245
3246 let fast_target = crate::config::TestTarget {
3247 dir: std::path::PathBuf::from("server"),
3248 aux_repo: None,
3249 features: vec!["fast-tests".into()],
3250 all_features: false,
3251 scratch_db: true,
3252 };
3253 let fast = cargo_test_command(
3254 &ctx,
3255 std::path::Path::new("/tmp/wt/server"),
3256 &fast_target,
3257 &["fast-tests"],
3258 &[],
3259 );
3260 let fast_args: Vec<_> = fast
3261 .as_std()
3262 .get_args()
3263 .map(|a| a.to_string_lossy().into_owned())
3264 .collect();
3265 assert!(
3266 fast_args
3267 .windows(2)
3268 .any(|w| w == ["--features", "fast-tests"]),
3269 "got: {fast_args:?}"
3270 );
3271 }
3272
3273 /// Spawn a one-shot loopback server that answers the first connection with
3274 /// `status_line` + a tiny body, then closes. Returns the bound port.
3275 async fn oneshot_http(status_line: &'static str) -> u16 {
3276 use tokio::io::{AsyncReadExt, AsyncWriteExt};
3277 let listener = tokio::net::TcpListener::bind((std::net::Ipv4Addr::LOCALHOST, 0))
3278 .await
3279 .unwrap();
3280 let port = listener.local_addr().unwrap().port();
3281 tokio::spawn(async move {
3282 if let Ok((mut sock, _)) = listener.accept().await {
3283 let mut scratch = [0u8; 1024];
3284 let _ = sock.read(&mut scratch).await; // drain the request line
3285 let resp =
3286 format!("{status_line}\r\nContent-Length: 2\r\nConnection: close\r\n\r\nok");
3287 let _ = sock.write_all(resp.as_bytes()).await;
3288 }
3289 });
3290 port
3291 }
3292
3293 #[tokio::test]
3294 async fn probe_health_ok_on_200() {
3295 let port = oneshot_http("HTTP/1.1 200 OK").await;
3296 assert!(probe_health(port).await.is_ok());
3297 }
3298
3299 #[tokio::test]
3300 async fn probe_health_err_on_non_200() {
3301 let port = oneshot_http("HTTP/1.1 503 Service Unavailable").await;
3302 let err = probe_health(port).await.unwrap_err();
3303 assert!(err.contains("status line"), "{err}");
3304 }
3305
3306 #[tokio::test]
3307 async fn probe_health_err_on_connection_refused() {
3308 // Bind then drop to get an almost-certainly-free port nothing listens on.
3309 let port = {
3310 let l = tokio::net::TcpListener::bind((std::net::Ipv4Addr::LOCALHOST, 0))
3311 .await
3312 .unwrap();
3313 l.local_addr().unwrap().port()
3314 };
3315 let err = probe_health(port).await.unwrap_err();
3316 assert!(err.contains("connect"), "{err}");
3317 }
3318
3319 /// burn_in returns a typed Blocked when the clock isn't started; the
3320 /// runner persists status='blocked' + outcome_json (the json carries
3321 /// blocker.kind = 'burn_in_clock_not_started').
3322 #[tokio::test]
3323 async fn burn_in_blocked_persists_typed_outcome() {
3324 let pool = SqlitePoolOptions::new()
3325 .max_connections(1)
3326 .connect("sqlite::memory:")
3327 .await
3328 .unwrap();
3329 crate::db::migrate(&pool).await.unwrap();
3330 // Topology sync expects a tier row before gate_runs can reference it.
3331 sqlx::query("INSERT INTO tiers (name, ord, provisioned, canary) VALUES ('host', 0, 0, 'sequential')")
3332 .execute(&pool).await.unwrap();
3333 sqlx::query("INSERT INTO tier_state (tier) VALUES ('host')")
3334 .execute(&pool)
3335 .await
3336 .unwrap();
3337 // versions FK target.
3338 sqlx::query("INSERT INTO versions (version, git_sha, built_at, artifact_path) VALUES ('0.1.0', 'abc1234', '2026-01-01T00:00:00Z', '/tmp/x')")
3339 .execute(&pool).await.unwrap();
3340
3341 let cfg = std::sync::Arc::new(crate::config::AppConfig::for_tests());
3342 let ctx = GateCtx {
3343 pool: pool.clone(),
3344 cfg,
3345 tier: TierId::new("host"),
3346 version: "0.1.0".parse().unwrap(),
3347 worktree: Some(std::path::PathBuf::from("/tmp/unused")),
3348 bundle: None,
3349 events: events::channel(),
3350 nodes: Vec::new(),
3351 build_id: None,
3352 aux_dirs: HashMap::new(),
3353 };
3354 let out = run(&ctx, &Gate::BurnIn { hours: 24 }).await.unwrap();
3355 assert_eq!(out.status_str(), "blocked");
3356 assert!(!out.is_passed());
3357
3358 // Read the persisted row.
3359 let row: (Option<String>, Option<String>) =
3360 sqlx::query_as("SELECT status, outcome_json FROM gate_runs ORDER BY id DESC LIMIT 1")
3361 .fetch_one(&pool)
3362 .await
3363 .unwrap();
3364 assert_eq!(row.0.as_deref(), Some("blocked"), "typed status");
3365 let json: serde_json::Value = serde_json::from_str(row.1.as_deref().unwrap()).unwrap();
3366 assert_eq!(json["status"]["kind"], "blocked");
3367 assert_eq!(
3368 json["status"]["blocker"]["kind"],
3369 "burn_in_clock_not_started"
3370 );
3371 }
3372
3373 /// node_health fails closed when there are no nodes to probe: a serving tier
3374 /// should always carry nodes, so an empty set is a misconfiguration that must
3375 /// block promotion, not pass it.
3376 #[tokio::test]
3377 async fn node_health_blocks_with_no_nodes() {
3378 let pool = SqlitePoolOptions::new()
3379 .max_connections(1)
3380 .connect("sqlite::memory:")
3381 .await
3382 .unwrap();
3383 crate::db::migrate(&pool).await.unwrap();
3384 sqlx::query(
3385 "INSERT INTO tiers (name, ord, provisioned, canary) VALUES ('b', 2, 1, 'sequential')",
3386 )
3387 .execute(&pool)
3388 .await
3389 .unwrap();
3390 sqlx::query("INSERT INTO tier_state (tier) VALUES ('b')")
3391 .execute(&pool)
3392 .await
3393 .unwrap();
3394 sqlx::query("INSERT INTO versions (version, git_sha, built_at, artifact_path) VALUES ('0.1.0', 'abc1234', '2026-01-01T00:00:00Z', '/tmp/x')")
3395 .execute(&pool).await.unwrap();
3396
3397 let cfg = std::sync::Arc::new(crate::config::AppConfig::for_tests());
3398 let ctx = GateCtx {
3399 pool: pool.clone(),
3400 cfg,
3401 tier: TierId::new("b"),
3402 version: "0.1.0".parse().unwrap(),
3403 worktree: None,
3404 bundle: None,
3405 events: events::channel(),
3406 nodes: Vec::new(), // no nodes -> fail closed
3407 build_id: None,
3408 aux_dirs: HashMap::new(),
3409 };
3410 let out = run(&ctx, &Gate::NodeHealth).await.unwrap();
3411 assert_eq!(out.status_str(), "blocked");
3412 assert!(!out.is_passed());
3413 let row: (Option<String>, Option<String>) =
3414 sqlx::query_as("SELECT status, outcome_json FROM gate_runs ORDER BY id DESC LIMIT 1")
3415 .fetch_one(&pool)
3416 .await
3417 .unwrap();
3418 let json: serde_json::Value = serde_json::from_str(row.1.as_deref().unwrap()).unwrap();
3419 assert_eq!(json["status"]["blocker"]["kind"], "no_nodes_to_probe");
3420 }
3421
3422 /// reset_scratch must drop every non-system schema, not just `public` —
3423 /// otherwise migrations that create custom schemas (e.g. tower_sessions)
3424 /// collide on the next run. This regressed once (Phase 0) and the fix is
3425 /// load-bearing for migration_dry_run.
3426 ///
3427 /// Gated on `SANDO_TEST_PG_URL` so it only runs where postgres is
3428 /// available. Set `SANDO_TEST_PG_URL=postgres:///sando_scratch?host=/var/run/postgresql`
3429 /// (or similar) before `cargo test`.
3430 #[tokio::test]
3431 async fn reset_scratch_drops_all_non_system_schemas() {
3432 let Ok(url) = std::env::var("SANDO_TEST_PG_URL") else {
3433 eprintln!("skipping: SANDO_TEST_PG_URL not set");
3434 return;
3435 };
3436 use sqlx::Executor;
3437 use sqlx::postgres::PgPoolOptions;
3438
3439 let pool = PgPoolOptions::new()
3440 .max_connections(1)
3441 .connect(&url)
3442 .await
3443 .unwrap();
3444 // Plant two non-system schemas + a table in each.
3445 pool.execute(
3446 "DROP SCHEMA IF EXISTS foo CASCADE; CREATE SCHEMA foo; CREATE TABLE foo.t (i int);",
3447 )
3448 .await
3449 .unwrap();
3450 pool.execute("DROP SCHEMA IF EXISTS tower_sessions CASCADE; CREATE SCHEMA tower_sessions; CREATE TABLE tower_sessions.session (id text);")
3451 .await.unwrap();
3452 pool.close().await;
3453
3454 reset_scratch(&url, "makenotwork")
3455 .await
3456 .expect("reset_scratch");
3457
3458 let pool = PgPoolOptions::new()
3459 .max_connections(1)
3460 .connect(&url)
3461 .await
3462 .unwrap();
3463 let rows: Vec<(String,)> = sqlx::query_as(
3464 "SELECT nspname FROM pg_namespace WHERE nspname NOT LIKE 'pg_%' AND nspname <> 'information_schema'",
3465 )
3466 .fetch_all(&pool)
3467 .await
3468 .unwrap();
3469 let names: Vec<String> = rows.into_iter().map(|(s,)| s).collect();
3470 // After reset, only `public` should remain among non-system schemas.
3471 assert_eq!(names, vec!["public".to_string()], "got: {names:?}");
3472 pool.close().await;
3473 }
3474
3475 /// reset_scratch must leave the dump's owner role existing and able to
3476 /// create in `public`, because a prod `pg_dump` carries `ALTER ... OWNER TO
3477 /// <role>` for every object. This was satisfied by a hand-created NOLOGIN
3478 /// role on fw13; nothing recorded it, so any other box failed
3479 /// migration_dry_run at the restore with "role does not exist".
3480 ///
3481 /// Uses a throwaway role name so it can prove the *creation* path rather
3482 /// than passing on fw13's pre-existing `makenotwork`. Same
3483 /// `SANDO_TEST_PG_URL` gate as above; needs a superuser connection.
3484 #[tokio::test]
3485 async fn reset_scratch_seeds_the_dump_owner_role_when_absent() {
3486 let Ok(url) = std::env::var("SANDO_TEST_PG_URL") else {
3487 eprintln!("skipping: SANDO_TEST_PG_URL not set");
3488 return;
3489 };
3490 use sqlx::Executor;
3491 use sqlx::postgres::PgPoolOptions;
3492
3493 let role = "sando_test_owner_probe";
3494 // `DROP ROLE` refuses while the role still holds the grants reset_scratch
3495 // gave it, so drop what it owns first. Idempotent, and a no-op when the
3496 // role is absent (the usual case on a first run).
3497 let drop_role = format!(
3498 "DO $$ BEGIN
3499 IF EXISTS (SELECT 1 FROM pg_roles WHERE rolname = '{role}') THEN
3500 EXECUTE 'DROP OWNED BY {role}';
3501 EXECUTE 'DROP ROLE {role}';
3502 END IF;
3503 END $$;"
3504 );
3505
3506 let pool = PgPoolOptions::new()
3507 .max_connections(1)
3508 .connect(&url)
3509 .await
3510 .unwrap();
3511 pool.execute(sqlx::raw_sql(sqlx::AssertSqlSafe(drop_role.clone())))
3512 .await
3513 .unwrap();
3514 pool.close().await;
3515
3516 reset_scratch(&url, role)
3517 .await
3518 .expect("reset_scratch creates the owner role");
3519
3520 let pool = PgPoolOptions::new()
3521 .max_connections(1)
3522 .connect(&url)
3523 .await
3524 .unwrap();
3525 let (exists, can_login): (bool, bool) =
3526 sqlx::query_as("SELECT true, rolcanlogin FROM pg_roles WHERE rolname = $1")
3527 .bind(role)
3528 .fetch_one(&pool)
3529 .await
3530 .expect("owner role exists after reset");
3531 assert!(exists);
3532 assert!(
3533 !can_login,
3534 "the owner role is an owner only, never a login identity"
3535 );
3536
3537 // The restore's owner-scoped DDL needs CREATE on public in the role's
3538 // own right (PG15+ dropped the implicit grant).
3539 let (has_create,): (bool,) =
3540 sqlx::query_as("SELECT pg_catalog.has_schema_privilege($1, 'public', 'CREATE')")
3541 .bind(role)
3542 .fetch_one(&pool)
3543 .await
3544 .unwrap();
3545 assert!(has_create, "owner role must be able to create in public");
3546
3547 // Idempotent: a second reset must not error on the now-existing role.
3548 pool.close().await;
3549 reset_scratch(&url, role)
3550 .await
3551 .expect("reset_scratch is idempotent");
3552
3553 let pool = PgPoolOptions::new()
3554 .max_connections(1)
3555 .connect(&url)
3556 .await
3557 .unwrap();
3558 pool.execute(sqlx::raw_sql(sqlx::AssertSqlSafe(drop_role)))
3559 .await
3560 .unwrap();
3561 pool.close().await;
3562 }
3563
3564 /// The preflight must pass against a privileged scratch connection. Guards
3565 /// the catalog query itself: a wrong column or a `current_user` that matches
3566 /// no `pg_roles` row would make `fetch_one` error (or, worse, a silently
3567 /// swapped pair would invert the check) and brick startup for everyone.
3568 /// `SANDO_TEST_PG_URL` is expected to be a superuser connection, as the
3569 /// gates require.
3570 #[tokio::test]
3571 async fn preflight_passes_on_a_privileged_scratch_connection() {
3572 let Ok(url) = std::env::var("SANDO_TEST_PG_URL") else {
3573 eprintln!("skipping: SANDO_TEST_PG_URL not set");
3574 return;
3575 };
3576 preflight_scratch_privileges(&url)
3577 .await
3578 .expect("a superuser scratch connection must satisfy the preflight");
3579 }
3580
3581 /// CF4: the restore pipeline must carry `ON_ERROR_STOP=1` (so psql fails on
3582 /// a bad statement instead of exiting 0 on a partial restore) and, for a
3583 /// gzip source, `set -o pipefail` (so a `gunzip` failure on a truncated
3584 /// archive isn't masked by psql's exit). Pure string check — no postgres.
3585 #[test]
3586 fn restore_shell_has_error_stop_and_pipefail() {
3587 let gz = restore_shell("postgres:///scratch", "/srv/sando/backups/latest.sql.gz");
3588 assert!(gz.contains("ON_ERROR_STOP=1"), "gz: {gz}");
3589 assert!(gz.contains("set -o pipefail"), "gz: {gz}");
3590 assert!(gz.contains("gunzip -c"), "gz: {gz}");
3591
3592 let plain = restore_shell("postgres:///scratch", "/srv/sando/backups/dump.sql");
3593 assert!(plain.contains("ON_ERROR_STOP=1"), "plain: {plain}");
3594 // No pipeline for a plain .sql, so pipefail is unnecessary there.
3595 assert!(!plain.contains("gunzip"), "plain: {plain}");
3596 // The db url is single-quote escaped in both forms.
3597 assert!(plain.contains("'postgres:///scratch'"), "plain: {plain}");
3598 }
3599
3600 #[test]
3601 fn split_pg_password_extracts_and_sanitizes() {
3602 // Password lifted out of the URL; the sanitized form keeps user/host/db.
3603 let (url, pw) = split_pg_password("postgres://sando:s3cret@db.host:5432/scratch");
3604 assert_eq!(url, "postgres://sando@db.host:5432/scratch");
3605 assert_eq!(pw.as_deref(), Some("s3cret"));
3606 // Percent-encoded password is decoded for PGPASSWORD.
3607 let (url, pw) = split_pg_password("postgresql://u:p%40ss%2Fword@h/d");
3608 assert_eq!(url, "postgresql://u@h/d");
3609 assert_eq!(pw.as_deref(), Some("p@ss/word"));
3610 }
3611
3612 #[test]
3613 fn split_pg_password_noop_without_password() {
3614 // No userinfo password -> unchanged, None. (A ':' after the '@', e.g. a
3615 // port, must not be mistaken for the password delimiter.)
3616 assert_eq!(
3617 split_pg_password("postgres:///scratch"),
3618 ("postgres:///scratch".to_string(), None),
3619 );
3620 assert_eq!(
3621 split_pg_password("postgres://sando@db.host:5432/scratch"),
3622 ("postgres://sando@db.host:5432/scratch".to_string(), None),
3623 );
3624 }
3625
3626 #[test]
3627 fn percent_decode_handles_escapes_and_malformed() {
3628 assert_eq!(percent_decode("plain"), "plain");
3629 assert_eq!(percent_decode("a%2Fb"), "a/b");
3630 // A malformed trailing escape is left literal, not dropped.
3631 assert_eq!(percent_decode("ab%2"), "ab%2");
3632 assert_eq!(percent_decode("ab%zz"), "ab%zz");
3633 }
3634
3635 #[test]
3636 fn pg_url_with_dbname_rewrites_the_database() {
3637 // user:pass@host:port/db?query — swap db, keep everything else.
3638 assert_eq!(
3639 pg_url_with_dbname(
3640 "postgres://sando:pw@db.host:5432/sando_scratch?sslmode=require",
3641 "postgres"
3642 ),
3643 "postgres://sando:pw@db.host:5432/postgres?sslmode=require",
3644 );
3645 // Socket form: the query carries `host=/var/run/postgresql` and must survive.
3646 assert_eq!(
3647 pg_url_with_dbname(
3648 "postgres:///sando_scratch?host=/var/run/postgresql",
3649 "sando_code_smoke_0_9_6"
3650 ),
3651 "postgres:///sando_code_smoke_0_9_6?host=/var/run/postgresql",
3652 );
3653 // Plain host/db, no query.
3654 assert_eq!(
3655 pg_url_with_dbname("postgres://localhost/scratch", "postgres"),
3656 "postgres://localhost/scratch".replace("scratch", "postgres"),
3657 );
3658 // No authority, no query (loopback socket, default db path).
3659 assert_eq!(
3660 pg_url_with_dbname("postgres:///scratch", "postgres"),
3661 "postgres:///postgres",
3662 );
3663 }
3664
3665 #[test]
3666 fn bytes_contain_matches_listening_in_log_output() {
3667 // JSON release log carries the message field verbatim.
3668 assert!(bytes_contain(
3669 br#"{"timestamp":"...","level":"INFO","fields":{"message":"listening","addr":"127.0.0.1:18182"}}"#,
3670 b"listening",
3671 ));
3672 // Human-format dev log.
3673 assert!(bytes_contain(
3674 b"2026-07-17 INFO makenotwork: listening addr=127.0.0.1:18182",
3675 b"listening"
3676 ));
3677 assert!(!bytes_contain(
3678 b"migrations complete; seeding catalog",
3679 b"listening"
3680 ));
3681 assert!(!bytes_contain(b"", b"listening"));
3682 }
3683
3684 #[tokio::test]
3685 async fn code_smoke_env_supplies_every_mandatory_server_var() {
3686 // code_smoke is the only gate that reaches the server's Config::from_env
3687 // (boot_smoke and the docs check short-circuit before it), so anything
3688 // that function requires has to be answered here. CDN_BASE_URL became
3689 // mandatory 13 days after this env was written and went unnoticed until
3690 // the gate first ran on the host; this test is what makes the next one
3691 // fail here instead of in a promote.
3692 let ctx = resolving_ctx("/w/abc", &[]);
3693 let mut cmd = tokio::process::Command::new("true");
3694 code_smoke_env(&mut cmd, &ctx, "postgres:///throwaway");
3695 let set: std::collections::HashMap<String, String> = cmd
3696 .as_std()
3697 .get_envs()
3698 .filter_map(|(k, v)| Some((k.to_str()?.to_string(), v?.to_str()?.to_string())))
3699 .collect();
3700 for key in [
3701 "DATABASE_URL",
3702 "HOST",
3703 "PORT",
3704 "HOST_URL",
3705 "CDN_BASE_URL",
3706 "SIGNING_SECRET",
3707 ] {
3708 assert!(set.contains_key(key), "code_smoke_env must set {key}");
3709 assert!(!set[key].is_empty(), "{key} must not be empty");
3710 }
3711 // Loopback, so Config::from_env's is_production branch stays false and
3712 // the gate never trips MissingPublicBucket for want of an S3 bucket.
3713 assert!(set["HOST_URL"].starts_with("http://localhost"));
3714 // Not an IP literal: the server derives its WebAuthn rp_id from this
3715 // host, and WebauthnBuilder rejects an origin whose Url::domain() is
3716 // None, which is every IP address. An IP here cannot boot the server.
3717 assert!(
3718 url_host_is_a_domain(&set["HOST_URL"]),
3719 "HOST_URL host must be a domain, not an IP literal: {}",
3720 set["HOST_URL"],
3721 );
3722 assert_eq!(set["HOST"], "127.0.0.1");
3723 // The signing secret has to clear the server's 32-char floor, or the
3724 // gate fails with WeakSigningSecret instead of testing anything.
3725 assert!(set["SIGNING_SECRET"].len() >= 32);
3726 }
3727
3728 #[test]
3729 fn code_smoke_db_name_sanitizes_and_caps() {
3730 assert_eq!(
3731 code_smoke_db_name(&"0.9.6".parse().unwrap()),
3732 "sando_code_smoke_0_9_6"
3733 );
3734 // Pre-release/build metadata folds to underscores; result stays [a-z0-9_].
3735 let n = code_smoke_db_name(&"1.0.0-rc.1+build".parse().unwrap());
3736 assert_eq!(n, "sando_code_smoke_1_0_0_rc_1_build");
3737 assert!(
3738 n.bytes()
3739 .all(|b| b.is_ascii_lowercase() || b.is_ascii_digit() || b == b'_')
3740 );
3741 assert!(n.len() <= 63);
3742 }
3743
3744 /// code_smoke is Blocked (not Failed) when the daemon has no scratch_db_url:
3745 /// there's no cluster to create the throwaway DB in, and that's an operator
3746 /// precondition, rendered yellow — the same shape as migration_dry_run.
3747 #[tokio::test]
3748 async fn code_smoke_blocks_without_scratch_db_url() {
3749 let pool = SqlitePoolOptions::new()
3750 .max_connections(1)
3751 .connect("sqlite::memory:")
3752 .await
3753 .unwrap();
3754 crate::db::migrate(&pool).await.unwrap();
3755 sqlx::query("INSERT INTO tiers (name, ord, provisioned, canary) VALUES ('host', 0, 1, 'sequential')")
3756 .execute(&pool).await.unwrap();
3757 sqlx::query("INSERT INTO tier_state (tier) VALUES ('host')")
3758 .execute(&pool)
3759 .await
3760 .unwrap();
3761 sqlx::query("INSERT INTO versions (version, git_sha, built_at, artifact_path) VALUES ('0.1.0', 'abc1234', '2026-01-01T00:00:00Z', '/tmp/x')")
3762 .execute(&pool).await.unwrap();
3763
3764 let cfg = std::sync::Arc::new(crate::config::AppConfig::for_tests()); // scratch_db_url: None
3765 let ctx = GateCtx {
3766 pool: pool.clone(),
3767 cfg,
3768 tier: TierId::new("host"),
3769 version: "0.1.0".parse().unwrap(),
3770 worktree: Some(std::path::PathBuf::from("/tmp/unused")),
3771 bundle: None,
3772 events: events::channel(),
3773 nodes: Vec::new(),
3774 build_id: None,
3775 aux_dirs: HashMap::new(),
3776 };
3777 let out = run(&ctx, &Gate::CodeSmoke).await.unwrap();
3778 assert_eq!(out.status_str(), "blocked");
3779 assert!(!out.is_passed());
3780 let row: (Option<String>, Option<String>) =
3781 sqlx::query_as("SELECT status, outcome_json FROM gate_runs ORDER BY id DESC LIMIT 1")
3782 .fetch_one(&pool)
3783 .await
3784 .unwrap();
3785 assert_eq!(row.0.as_deref(), Some("blocked"));
3786 let json: serde_json::Value = serde_json::from_str(row.1.as_deref().unwrap()).unwrap();
3787 assert_eq!(json["status"]["blocker"]["kind"], "scratch_db_url_unset");
3788 }
3789
3790 /// Sanity: applying MNW migrations from a *non-existent* dir errors,
3791 /// rather than silently no-op'ing. Cheap pure check, no postgres needed
3792 /// (the sqlx::Migrator::new constructor itself reads the dir).
3793 #[tokio::test]
3794 async fn run_migrator_errors_on_missing_dir() {
3795 // The first thing run_migrator does is `Migrator::new(dir)`, which
3796 // needs a real dir to read migration files from.
3797 let res = run_migrator(
3798 "postgres:///does-not-matter",
3799 std::path::Path::new("/nonexistent/sando-test-migrations"),
3800 )
3801 .await;
3802 assert!(res.is_err());
3803 }
3804 }
3805