//! Gate execution. Each gate kind has a runner that produces a pass/fail //! outcome plus an optional detail string (typically a stderr tail or a //! human-readable reason). Outcomes are persisted to `gate_runs` so /state //! and the TUI can show them. use crate::classify; use crate::config::AppConfig; use crate::domain::{AppId, GateKind, GateRunId, TierId, Version}; use crate::events::{self, Event, EventTx}; use crate::outcome::{GateBlocker, GateFailure, GateOutcome, LogRef, PassNote}; use crate::topology::Gate; use anyhow::{Context, Result}; use chrono::Utc; use ops_core::live_log::LiveLog; use ops_core::remote::LogSink; // brings `LiveLog::write_chunk` (the sink trait) into scope use sqlx::SqlitePool; use std::collections::HashMap; use std::path::Path; use std::path::PathBuf; use std::sync::Arc; use tokio::io::AsyncReadExt; use tokio::process::Command; /// The gate live-log callback: emit each chunk as a `GateLogChunk` event so the /// TUI sees the tail stream in real time. `ops_core::live_log::LiveLog` owns the /// disk append and the per-run sequence counter; this closure is the one /// tool-specific bit (Sando previously carried a whole `live_log.rs` copy that /// hardcoded exactly this emit). fn gate_chunk_cb(events: EventTx, run_id: GateRunId) -> ops_core::live_log::ChunkCallback { Box::new(move |seq, text| { events::emit( &events, Event::GateLogChunk { run_id, seq, text: text.to_owned(), }, ); }) } pub struct GateCtx { pub pool: SqlitePool, pub cfg: Arc, pub tier: TierId, pub version: Version, /// The checkout this run's artifact was built from, when there is one. /// /// `None` for an accepted artifact: it was built elsewhere and Sando has no /// source tree for it. That is the boundary made visible (wiki /// [[sando-bento-boundary]]) — artifact-scoped gates belong to the builder, /// so a gate that reads source is one Sando should refuse to run here rather /// than resolve against a path that does not exist. pub worktree: Option, /// The published, content-addressed bundle this run is about, when it has /// been published yet. `migration_dry_run` prefers it over the worktree, so /// what it proves is inside the digest rather than beside it. pub bundle: Option, pub events: EventTx, /// Nodes the `node_health` post-deploy gate probes. Empty for build-time /// gate runs on the host (where `node_health` never appears); filled at /// promote time with each freshly-deployed node and its executor. pub nodes: Vec, /// The `build_runs.id` this gate run vouches for — the artifact identity /// (wiki [[release-artifact-identity]]). Recorded on every `gate_runs` row so /// promote can resolve the artifact through the evidence for a specific build, /// not through a version string that a later rebuild can silently reuse. /// `None` for legacy/pre-identity runs and gate unit tests. pub build_id: Option, /// Where each `[[aux_repo]]` is checked out, by topology name. Aux repos sit /// beside the per-sha worktree rather than under it, so a `test_target` that /// names one cannot be resolved against `worktree` alone. Filled from the /// topology at build time; empty at promote time, where the only gate that /// runs is `node_health` and there is no checkout at all. pub aux_dirs: HashMap, } impl GateCtx { /// Absolute directory a `test_target` runs in: under the worktree, or under /// the named aux repo's checkout. /// /// An `aux_repo` naming nothing this run knows about resolves to `None` /// rather than to a wrong path. Callers treat that as "not present in this /// run" and skip, the same as a target missing from an older sha — /// `--check-config` is what stops a genuine typo from reaching here /// (`Topology::ensure_test_target_aux_repos_exist`). pub fn target_dir(&self, target: &crate::config::TestTarget) -> Option { match target.aux_repo.as_deref() { None => Some(self.worktree.as_ref()?.join(&target.dir)), Some(name) => Some(self.aux_dirs.get(name)?.join(&target.dir)), } } /// The checkout, or a typed refusal for a gate that cannot work without one. /// /// Every caller of this is a gate whose evidence is about the *artifact* /// rather than about the artifact in an environment, which the boundary /// assigns to the builder. Reaching this arm means a tier asked Sando to /// re-run a builder's gate against a bundle it was handed, and the honest /// answer is to say so rather than to pass on having run nothing. pub fn worktree_for(&self, gate: GateKind) -> std::result::Result<&Path, GateOutcome> { self.worktree.as_deref().ok_or_else(|| { GateOutcome::failed(GateFailure::NeedsSource { gate, artifact: self.bundle.as_ref().map_or_else( || "an artifact built elsewhere".into(), |b| b.display().to_string(), ), }) }) } /// Where a `migration_check` finds its migrations. /// /// The bundle wins when it carries them. That is the point of shipping /// migrations as a `release_contents` entry: it puts them inside the digest, /// so the dry run proves something about the bytes that ship rather than /// about a checkout that happens to sit next to them. The worktree is the /// fallback for a build whose config has not opted in yet, and for an /// accepted artifact there is no fallback at all — if the builder did not /// bundle its migrations, Sando cannot dry-run them and says so. pub fn migrations_dir(&self, dir: &Path) -> Option { if let Some(bundle) = &self.bundle { let in_bundle = bundle.join(dir); if in_bundle.is_dir() { return Some(in_bundle); } } let in_worktree = self.worktree.as_ref()?.join(dir); in_worktree.is_dir().then_some(in_worktree) } } /// One node the `node_health` gate verifies: its id, the systemd unit to /// confirm active after the restart, an optional HTTP readiness URL, and the /// executor that reaches it (the same transport the deploy used). pub struct NodeProbe { pub node: crate::domain::NodeId, pub service: String, pub health_url: Option, pub executor: Arc, } /// Run a single gate end-to-end: insert the in-flight row, execute the gate, /// update the row with the outcome. Returns the outcome for the caller. pub async fn run(ctx: &GateCtx, gate: &Gate) -> Result { let kind = gate.kind(); let started_at = Utc::now().to_rfc3339(); let id: i64 = sqlx::query_scalar( "INSERT INTO gate_runs (app, version, tier, gate_kind, started_at, build_id) VALUES (?, ?, ?, ?, ?, ?) RETURNING id", ) .bind(&ctx.cfg.id) .bind(&ctx.version) .bind(&ctx.tier) .bind(kind) .bind(&started_at) .bind(ctx.build_id) .fetch_one(&ctx.pool) .await?; let run_id = GateRunId(id); tracing::info!( run_id = %run_id, tier = %ctx.tier, version = %ctx.version, gate = %kind, "gate start", ); events::emit( &ctx.events, Event::GateStart { run_id, tier: ctx.tier.clone(), version: ctx.version.clone(), gate: kind, }, ); let outcome = match gate { // cargo_test bounds its own run internally (it kills the specific child). Gate::CargoTest => cargo_test(ctx, run_id).await, // hardening_test bounds its own run internally, same as cargo_test. Gate::HardeningTest => hardening_test(ctx, run_id).await, // Each bounds itself the same way cargo_test does: one deadline across // every target, so N crates cannot multiply the ceiling by N. Gate::Clippy => clippy(ctx, run_id).await, Gate::Fmt => fmt_check(ctx, run_id).await, Gate::CargoAudit => supply_chain(ctx, run_id, GateKind::CargoAudit).await, Gate::CargoDeny => supply_chain(ctx, run_id, GateKind::CargoDeny).await, // migration_dry_run's psql restore + sqlx migrate could wedge; bound the // whole gate here. Its bash restore sets kill_on_drop, so a timeout-drop // doesn't orphan it. Gate::MigrationDryRun => { let ceiling = std::time::Duration::from_secs(ctx.cfg.gate_timeout_secs); match tokio::time::timeout(ceiling, migration_dry_run(ctx, run_id)).await { Ok(res) => res, Err(_elapsed) => Ok(GateOutcome::failed(GateFailure::Timeout { gate: GateKind::MigrationDryRun, after_s: ctx.cfg.gate_timeout_secs as u32, }) .with_log_ref(LogRef::new(&ctx.version, GateKind::MigrationDryRun))), } } // code_smoke boots the real binary (migrate-from-scratch + seed + serve), // any step of which could wedge; bound the whole gate here. Both child // processes set kill_on_drop, so a timeout-drop can't orphan them. A // timeout leaves the throwaway DB behind; the next run's createdb drops // it first (DROP IF EXISTS), same as migration_dry_run's scratch reset. Gate::CodeSmoke => { let ceiling = std::time::Duration::from_secs(ctx.cfg.gate_timeout_secs); match tokio::time::timeout(ceiling, code_smoke(ctx, run_id)).await { Ok(res) => res, Err(_elapsed) => Ok(GateOutcome::failed(GateFailure::Timeout { gate: GateKind::CodeSmoke, after_s: ctx.cfg.gate_timeout_secs as u32, }) .with_log_ref(LogRef::new(&ctx.version, GateKind::CodeSmoke))), } } Gate::BootSmoke => boot_smoke(ctx, run_id).await, Gate::NodeHealth => node_health(ctx).await, Gate::BurnIn { hours } => burn_in(ctx, *hours).await, Gate::ManualConfirm => manual_confirm(ctx).await, }; let outcome = outcome.unwrap_or_else(|e| { GateOutcome::failed(GateFailure::Unclassified { legacy_detail: Some(format!("gate runner errored: {e}")), }) }); let outcome_json = serde_json::to_string(&outcome) .unwrap_or_else(|e| format!("{{\"_serialize_error\":{e:?}}}")); sqlx::query( "UPDATE gate_runs SET finished_at = ?, status = ?, outcome_json = ?, log_ref = ? WHERE id = ?", ) .bind(Utc::now().to_rfc3339()) .bind(outcome.status_str()) .bind(&outcome_json) .bind(outcome.log_ref.as_ref().map(super::outcome::LogRef::as_str)) .bind(id) .execute(&ctx.pool) .await?; tracing::info!( tier = %ctx.tier, version = %ctx.version, gate = %kind, status = outcome.status_str(), "gate done", ); events::emit( &ctx.events, Event::GateDone { run_id, tier: ctx.tier.clone(), version: ctx.version.clone(), gate: kind, outcome: outcome.clone(), }, ); Ok(outcome) } /// Run every gate in order and return the kinds that did not pass (empty means /// green). We deliberately do NOT short-circuit on first failure — every gate's /// outcome is recorded in `gate_runs`, which is the operator's only visibility /// into pipeline health. Hiding later gates because an earlier one failed makes /// diagnosis worse. /// /// Returning the failing kinds rather than a bare bool is what lets the promote /// path name them in the tier's `partial_reason` and in the error it returns to /// the operator, instead of a generic "something was red". pub async fn run_all(ctx: &GateCtx, gates: &[Gate]) -> Result> { let mut failed = Vec::new(); for g in gates { let o = run(ctx, g).await?; if !o.is_passed() { failed.push(g.kind()); } } Ok(failed) } // ---- individual gate runners ---- /// Run every configured `test_target`'s suite, in order, under one gate. /// /// This used to be hardcoded to `worktree/server`, which meant every other /// crate in the repo shipped ungated — including `mnw-cli`, which is built as a /// companion and installed onto prod-1 in the same promote. The targets are now /// configured (`[[test_target]]` in the daemon config), defaulting to the /// historical single `server` entry. /// /// The whole set shares one `gate_runs` row and one log file: from the /// pipeline's point of view "the tests" either pass or don't. The first failing /// target ends the gate, since a red suite blocks the promote regardless of what /// the remaining crates would have said, and running them would only delay the /// operator's answer. Its name is carried in the failure so the summary points /// at the crate, not just the test. async fn cargo_test(ctx: &GateCtx, run_id: GateRunId) -> Result { let log_path = gate_log_path(ctx, GateKind::CargoTest); let log_ref = LogRef::new(&ctx.version, GateKind::CargoTest); // Best-effort: drop our own role's stale `mnw_test_*` databases (the // template + any per-test clones orphaned by a previously-killed run) // before the suite, so they can't accumulate or collide. Foreign-owned // leftovers are left alone — the harness now namespaces its template per // role, so they no longer wedge the gate. if let Some(scratch_url) = ctx.cfg.scratch_db_url.as_deref() { clean_stale_test_dbs(scratch_url).await; } let started = std::time::Instant::now(); // One ceiling for the whole gate, not per target: the point is to bound how // long a hung suite can block the pipeline, and N targets each allowed the // full timeout would multiply that by N. let deadline = started + std::time::Duration::from_secs(ctx.cfg.gate_timeout_secs); let mut ran = 0usize; for target in &ctx.cfg.test_targets { let label = target.label(); // A target absent from this sha is skipped, not fatal: sando has to be // able to build older shas (bisect, rollback rebuild) from a config that // describes the tip. The zero-targets-ran check below is what stops this // from quietly turning the gate into a no-op. let Some(dir) = ctx .target_dir(target) .filter(|d| d.join("Cargo.toml").is_file()) else { tracing::warn!( target = %label, version = %ctx.version, "test_target has no Cargo.toml in this run; skipping", ); continue; }; let features: Vec<&str> = target.features.iter().map(String::as_str).collect(); // Fast pre-gate: compile the test targets WITHOUT running them. This // builds the exact artifacts the full run needs (so the subsequent run // reuses the cache — no wasted work), but fails in ~minutes with the // real `error[Ennnn]: ...` on a test-only-target compile break. That // class (a field missing in a `#[cfg(test)]`-only binary like `load`) // otherwise compiles fine under the build step and only blows up here, // after a full build, as an opaque mass test failure. let banner = format!("\n==== test_target: {label} ====\n"); append_to_log(&log_path, banner.as_bytes()).await; let mut pre = match cargo_test_command(ctx, &dir, target, &features, &["--no-run"]).spawn() { Ok(c) => c, Err(e) => { return Ok(GateOutcome::failed(GateFailure::SpawnFailed { message: format!("{label}: {e}"), }) .with_log_ref(log_ref)); } }; let (pre_out, pre_err, pre_status) = match run_to_deadline_for( &mut pre, ctx, run_id, log_path.clone(), deadline, started, GateKind::CargoTest, ) .await? { Ok(v) => v, Err(timeout) => return Ok(timeout.with_log_ref(log_ref)), }; if !pre_status.success() { let failure = classify::classify_compile_error(&pre_out, &pre_err); return Ok(GateOutcome::failed(name_target(failure, &target.dir)).with_log_ref(log_ref)); } // Full run: the test binaries are already built above, so cargo's // up-to-date check skips compilation and this just runs the tests. let mut child = match cargo_test_command(ctx, &dir, target, &features, &[]).spawn() { Ok(c) => c, Err(e) => { return Ok(GateOutcome::failed(GateFailure::SpawnFailed { message: format!("{label}: {e}"), }) .with_log_ref(log_ref)); } }; let (stdout_buf, stderr_buf, status) = match run_to_deadline_for( &mut child, ctx, run_id, log_path.clone(), deadline, started, GateKind::CargoTest, ) .await? { Ok(v) => v, Err(timeout) => return Ok(timeout.with_log_ref(log_ref)), }; if !status.success() { let failure = classify::classify_cargo_test(&stdout_buf, &stderr_buf); return Ok(GateOutcome::failed(name_target(failure, &target.dir)).with_log_ref(log_ref)); } ran += 1; } // Every configured target was missing from the worktree. Exiting green here // would report "tests passed" having run none of them. if ran == 0 { return Ok(GateOutcome::failed(GateFailure::Unclassified { legacy_detail: Some(format!( "cargo_test ran no targets: none of the {} configured test_target dir(s) \ exist in this worktree", ctx.cfg.test_targets.len(), )), }) .with_log_ref(log_ref)); } let duration_s = started.elapsed().as_secs() as u32; Ok(GateOutcome::passed(PassNote::TestsPassed { duration_s }).with_log_ref(log_ref)) } /// Stream a child to the live log, bounded by the gate-wide `deadline`. `Ok(Err(_))` /// is the timeout outcome (child killed); `Err(_)` is an IO error on the stream. #[allow(clippy::type_complexity)] async fn run_to_deadline_for( child: &mut tokio::process::Child, ctx: &GateCtx, run_id: GateRunId, log_path: PathBuf, deadline: std::time::Instant, started: std::time::Instant, kind: GateKind, ) -> Result, Vec, std::process::ExitStatus), GateOutcome>> { let remaining = deadline.saturating_duration_since(std::time::Instant::now()); let stream = stream_child_to_live_log(child, ctx.events.clone(), run_id, log_path); match tokio::time::timeout(remaining, stream).await { Ok(res) => Ok(Ok(res?)), Err(_elapsed) => { child.start_kill().ok(); let _ = child.wait().await; Ok(Err(GateOutcome::failed(GateFailure::Timeout { gate: kind, after_s: started.elapsed().as_secs() as u32, }))) } } } /// Prefix a test/compile failure's headline with the crate it came from, so a /// red gate across many targets says *which* crate broke. Other failure kinds /// are single-target by construction and pass through untouched. fn name_target(failure: GateFailure, dir: &std::path::Path) -> GateFailure { let at = dir.display(); match failure { GateFailure::CargoTest { failed_count, first_failed, first_panic, } => GateFailure::CargoTest { failed_count, first_failed: Some(match first_failed { Some(name) => format!("{at}: {name}"), None => at.to_string(), }), first_panic, }, GateFailure::CompileError { error_count, first_error, } => GateFailure::CompileError { error_count, first_error: Some(match first_error { Some(e) => format!("{at}: {e}"), None => at.to_string(), }), }, other => other, } } /// Append raw bytes to a gate log outside the child-streaming path (target /// banners). Best-effort, same as `LiveLog`: a broken log dir never turns a /// passing gate red. async fn append_to_log(path: &std::path::Path, bytes: &[u8]) { use tokio::io::AsyncWriteExt; if let Some(parent) = path.parent() && tokio::fs::create_dir_all(parent).await.is_err() { return; } if let Ok(mut f) = tokio::fs::OpenOptions::new() .create(true) .append(true) .open(path) .await { let _ = f.write_all(bytes).await; } } /// `cargo clippy --all-targets -- -D warnings` over every configured /// `test_target`. /// /// Before this gate, `-D warnings` was enforced in exactly one place — /// `server/deploy/run-ci.sh`, which died with the astra pipeline — so lint drift /// reached prod ungated. Two crates even carried comments referring to "the /// -D warnings CI gate" for a gate that did not exist. async fn clippy(ctx: &GateCtx, run_id: GateRunId) -> Result { lint_over_targets(ctx, run_id, GateKind::Clippy, |target, features| { let mut args = vec!["clippy".to_string(), "--all-targets".to_string()]; if target.all_features { args.push("--all-features".to_string()); } else if !features.is_empty() { args.push("--features".to_string()); args.push(features.join(",")); } // Everything after `--` goes to rustc, which is where -D lives. args.push("--".to_string()); args.push("-D".to_string()); args.push("warnings".to_string()); args }) .await } /// `cargo fmt --check` over every configured `test_target`. /// /// No `rustfmt.toml` exists anywhere in the tree, so this is plain rustfmt /// defaults. Cheap: no compilation, just a parse. async fn fmt_check(ctx: &GateCtx, run_id: GateRunId) -> Result { lint_over_targets(ctx, run_id, GateKind::Fmt, |_target, _features| { vec!["fmt".to_string(), "--check".to_string()] }) .await } /// `cargo audit` / `cargo deny check`, over the crates that carry the matching /// config file. /// /// Config-gated on purpose. Both tools are only meaningful against a triaged /// posture: four crates in this repo fail `cargo audit` today purely because /// they have no `.cargo/audit.toml` recording which transitive advisories have /// been reviewed and accepted. Running them everywhere would make the gate /// permanently and uninformatively red. Dropping the config file into a crate /// is what opts it in. async fn supply_chain(ctx: &GateCtx, run_id: GateRunId, kind: GateKind) -> Result { let (config_rel, args): (&str, Vec) = match kind { GateKind::CargoAudit => (".cargo/audit.toml", vec!["audit".into()]), GateKind::CargoDeny => ("deny.toml", vec!["deny".into(), "check".into()]), other => unreachable!("supply_chain called for {other:?}"), }; run_over_targets(ctx, run_id, kind, |_target, target_dir| { target_dir.join(config_rel).is_file().then(|| args.clone()) }) .await } /// Shared driver for the lint gates: run `cargo ` in every configured /// `test_target` that exists in this worktree. async fn lint_over_targets( ctx: &GateCtx, run_id: GateRunId, kind: GateKind, build_args: impl Fn(&crate::config::TestTarget, &[String]) -> Vec, ) -> Result { // The target is handed in directly. This used to reverse-look-it-up by // comparing `worktree.join(dir)` against the resolved path, which silently // stopped matching for anything resolved anywhere else — an aux repo, say. run_over_targets(ctx, run_id, kind, move |t, _dir| { Some(build_args(t, &t.features)) }) .await } /// Run one cargo invocation per configured `test_target`, under a single /// gate-wide deadline. `args_for` returns `None` to skip a target (used by the /// supply-chain gates, which only apply where their config file lives). /// /// Shares `cargo_test`'s conventions: per-target log banners, stop at the first /// failure with the crate named, skip targets absent from the worktree, and fail /// closed if that leaves nothing to run. async fn run_over_targets( ctx: &GateCtx, run_id: GateRunId, kind: GateKind, args_for: impl Fn(&crate::config::TestTarget, &std::path::Path) -> Option>, ) -> Result { let log_path = gate_log_path(ctx, kind); let log_ref = LogRef::new(&ctx.version, kind); let started = std::time::Instant::now(); let deadline = started + std::time::Duration::from_secs(ctx.cfg.gate_timeout_secs); let mut ran = 0usize; for target in &ctx.cfg.test_targets { let label = target.label(); let Some(dir) = ctx .target_dir(target) .filter(|d| d.join("Cargo.toml").is_file()) else { tracing::warn!( gate = kind.as_str(), target = %label, "target has no Cargo.toml in this run; skipping", ); continue; }; let Some(args) = args_for(target, &dir) else { continue; }; append_to_log( &log_path, format!("\n==== {}: {label} ====\n", kind.as_str()).as_bytes(), ) .await; let mut cmd = Command::new("cargo"); cmd.args(&args) .current_dir(&dir) .stdout(std::process::Stdio::piped()) .stderr(std::process::Stdio::piped()) .kill_on_drop(true); if let Some(t) = ctx.cfg.cargo_target_dir.as_deref() { cmd.env("CARGO_TARGET_DIR", t); } // clippy type-checks, so it needs the same sqlx online-mode env the // build and cargo_test steps get. fmt/audit/deny never touch the DB. if kind == GateKind::Clippy && let Some(url) = ctx .cfg .scratch_db_url .as_deref() .filter(|_| target.scratch_db) { cmd.env("DATABASE_URL", url); cmd.env( "TEST_DATABASE_URL", url.split_once('?').map_or(url, |(b, _)| b), ); } let mut child = match cmd.spawn() { Ok(c) => c, Err(e) => { return Ok(GateOutcome::failed(GateFailure::SpawnFailed { message: format!("{label}: {e}"), }) .with_log_ref(log_ref)); } }; let (stdout_buf, stderr_buf, status) = match run_to_deadline_for( &mut child, ctx, run_id, log_path.clone(), deadline, started, kind, ) .await? { Ok(v) => v, Err(timeout) => return Ok(timeout.with_log_ref(log_ref)), }; if !status.success() { let failure = match kind { // clippy speaks rustc diagnostics, so the compile-error // classifier extracts the real `error: ...` headline. GateKind::Clippy => classify::classify_compile_error(&stdout_buf, &stderr_buf), _ => GateFailure::Unclassified { legacy_detail: Some(first_meaningful_line(&stdout_buf, &stderr_buf)), }, }; return Ok(GateOutcome::failed(name_target(failure, &target.dir)).with_log_ref(log_ref)); } ran += 1; } if ran == 0 { return Ok(GateOutcome::failed(GateFailure::Unclassified { legacy_detail: Some(format!( "{} ran nothing: no configured target in this worktree qualified", kind.as_str(), )), }) .with_log_ref(log_ref)); } Ok(GateOutcome::passed(PassNote::TestsPassed { duration_s: started.elapsed().as_secs() as u32, }) .with_log_ref(log_ref)) } /// First line that looks like a diagnostic, for gates whose tools have no /// dedicated classifier (`cargo audit`, `cargo deny`). Falls back to a generic /// note rather than an empty string. fn first_meaningful_line(stdout: &[u8], stderr: &[u8]) -> String { for buf in [stderr, stdout] { let text = String::from_utf8_lossy(buf); if let Some(line) = text .lines() .map(str::trim) .find(|l| l.starts_with("error") || l.contains("vulnerabilit") || l.contains("FAILED")) { return line.chars().take(200).collect(); } } "tool reported failure; see the gate log".into() } /// The tests `cargo_test` cannot reach, run against production constants. /// /// `cargo_test` builds with `--features fast-tests`, which relaxes /// `AUTH_RATE_LIMIT_BURST` 5 → 20, `SANDBOX_RATE_LIMIT_MS` 30s → 10ms, and /// argon2 from 46 MiB/t=2 to 8 MiB/t=1 — and the rate-limiting suite is /// `#[cfg_attr(feature = "fast-tests", ignore)]`d on top of that, because a /// bucket refilling at 100/sec never depletes under parallel test threads. The /// net effect was that Sando's only code gate silently skipped every test of /// the auth hardening it most needs to protect. /// /// So: no features, `--test-threads=1` (these tests key on a shared per-IP /// bucket and must not interleave), and a name filter rather than the whole /// suite — the rest of the suite is tuned for `fast-tests` and would only go /// slow and flaky here. The filter is a substring match, so it catches /// `..._rate_limit_...` and `..._rate_limited` alike. /// /// This costs a second compile of the lib + integration binary (a different /// feature set is a different cfg, so no artifact sharing with `cargo_test`). /// That is the price of the coverage; the filter keeps the *run* to seconds. async fn hardening_test(ctx: &GateCtx, run_id: GateRunId) -> Result { let server_dir = match ctx.worktree_for(GateKind::HardeningTest) { Ok(w) => w.join("server"), Err(outcome) => return Ok(outcome), }; // No features is the whole point; the scratch DB is needed because the // server's sqlx macros type-check against it. Unlike cargo_test, this gate // is deliberately not driven by `test_targets`: it targets one specific // suite in one specific crate, not "the repo's tests". let target = crate::config::TestTarget { dir: std::path::PathBuf::from("server"), aux_repo: None, features: Vec::new(), all_features: false, scratch_db: true, }; let log_path = gate_log_path(ctx, GateKind::HardeningTest); let log_ref = LogRef::new(&ctx.version, GateKind::HardeningTest); if let Some(scratch_url) = ctx.cfg.scratch_db_url.as_deref() { clean_stale_test_dbs(scratch_url).await; } let started = std::time::Instant::now(); // Same two-step shape as cargo_test: compile first so a test-target break // reports as a compile error rather than an opaque mass test failure. let mut pre = match cargo_test_command( ctx, &server_dir, &target, &[], &["--no-run", "--test", "integration"], ) .spawn() { Ok(c) => c, Err(e) => { return Ok(GateOutcome::failed(GateFailure::SpawnFailed { message: e.to_string(), }) .with_log_ref(log_ref)); } }; let (pre_out, pre_err, pre_status) = stream_child_to_live_log(&mut pre, ctx.events.clone(), run_id, log_path.clone()).await?; if !pre_status.success() { let failure = classify::classify_compile_error(&pre_out, &pre_err); return Ok(GateOutcome::failed(failure).with_log_ref(log_ref)); } let mut child = match cargo_test_command( ctx, &server_dir, &target, &[], &[ "--test", "integration", "--", "--test-threads=1", "rate_limit", ], ) .spawn() { Ok(c) => c, Err(e) => { return Ok(GateOutcome::failed(GateFailure::SpawnFailed { message: e.to_string(), }) .with_log_ref(log_ref)); } }; let ceiling = std::time::Duration::from_secs(ctx.cfg.gate_timeout_secs); let stream = stream_child_to_live_log(&mut child, ctx.events.clone(), run_id, log_path); let (stdout_buf, stderr_buf, status) = match tokio::time::timeout(ceiling, stream).await { Ok(res) => res?, Err(_elapsed) => { child.start_kill().ok(); let _ = child.wait().await; return Ok(GateOutcome::failed(GateFailure::Timeout { gate: GateKind::HardeningTest, after_s: started.elapsed().as_secs() as u32, }) .with_log_ref(log_ref)); } }; let duration_s = started.elapsed().as_secs() as u32; if status.success() { // A filter that matches nothing exits 0, which would make this gate a // green no-op the day someone renames the tests. Fail closed instead. if tests_run(&stdout_buf) == 0 { return Ok(GateOutcome::failed(GateFailure::Unclassified { legacy_detail: Some( "hardening_test ran 0 tests: the `rate_limit` filter matched nothing. \ The suite was renamed or moved — this gate is proving nothing." .into(), ), }) .with_log_ref(log_ref)); } Ok(GateOutcome::passed(PassNote::TestsPassed { duration_s }).with_log_ref(log_ref)) } else { let failure = classify::classify_cargo_test(&stdout_buf, &stderr_buf); Ok(GateOutcome::failed(failure).with_log_ref(log_ref)) } } /// Count the tests libtest reports as run, from its `test result: ok. N passed` /// summary line. Returns 0 when no summary is present, which is itself the /// "nothing ran" case the caller fails on. fn tests_run(stdout: &[u8]) -> u32 { String::from_utf8_lossy(stdout) .lines() .filter_map(|l| l.trim().strip_prefix("test result:")) .filter_map(|rest| rest.split_once(" passed")) .filter_map(|(head, _)| { head.rsplit(' ') .find(|t| !t.is_empty())? .parse::() .ok() }) .sum() } /// Configure (but don't spawn) `cargo test --release [--features ] /// ` in `dir`, wired to the scratch DB. Shared by the `--no-run` /// pre-gate compile and the full test run so both go through one env setup, /// and by both test gates so they can't drift apart on env. /// /// `cargo_test` passes `fast-tests`, matching what the retired astra CI did: /// it relaxes the auth rate-limit burst (5 → 20) and argon2 cost so /// signup-heavy + lockout workflow tests complete without hitting Governor /// before the hand-rolled lockout check (documented at /// `server/src/constants.rs:87`). `hardening_test` passes no features, which is /// the whole point of that gate — see `GateKind::HardeningTest`. fn cargo_test_command( ctx: &GateCtx, dir: &std::path::Path, target: &crate::config::TestTarget, features: &[&str], extra: &[&str], ) -> Command { let mut cmd = Command::new("cargo"); cmd.args(["test", "--release"]); if target.all_features { cmd.arg("--all-features"); } else if !features.is_empty() { cmd.args(["--features", &features.join(",")]); } cmd.args(extra) .current_dir(dir) .stdout(std::process::Stdio::piped()) .stderr(std::process::Stdio::piped()) .kill_on_drop(true); // Share the build step's target dir so the test compile reuses its // artifacts (and the `--no-run` precompile reuses them again). Must match // `build.rs` or the gate would clean-compile the whole tree a second time. if let Some(target) = ctx.cfg.cargo_target_dir.as_deref() { cmd.env("CARGO_TARGET_DIR", target); } // Same online-mode rationale as the build step: sqlx query macros need a // live DB to type-check against. The scratch DB is left in migrated state // by the preceding build, so we can reuse it here. // // Opt-in per target: setting DATABASE_URL switches sqlx OUT of offline mode, // so a crate that ships `.sqlx` query data would stop using it and try to // type-check against a database that has none of its tables. if let Some(scratch_url) = ctx .cfg .scratch_db_url .as_deref() .filter(|_| target.scratch_db) { cmd.env("DATABASE_URL", scratch_url); // The server test harness (tests/harness/db.rs) parses TEST_DATABASE_URL // with rfind('/'), which mangles URLs whose query string contains '/' // (e.g. `?host=/var/run/postgresql`). Strip the query — libpq defaults // to /var/run/postgresql on Debian/Ubuntu when host is unspecified. let test_url = scratch_url .split_once('?') .map_or(scratch_url, |(base, _)| base); cmd.env("TEST_DATABASE_URL", test_url); } cmd } async fn migration_dry_run(ctx: &GateCtx, run_id: GateRunId) -> Result { let log = GateLog::open(ctx, run_id, GateKind::MigrationDryRun).await; let outcome = migration_dry_run_inner(ctx, &log).await; log.close().await; outcome.map(|o| o.with_log_ref(LogRef::new(&ctx.version, GateKind::MigrationDryRun))) } /// The staged interior of [`migration_dry_run`], writing every step through the /// gate's live log. The caller owns the sink so it can flush it on every exit /// path, and attaches the `log_ref` once instead of at each return. /// Runs one configured check per database, in config order, and stops at the /// first that does not pass — a red gate is a red gate, and continuing would /// bury it under a second restore's output. /// /// The server's check runs against `scratch_db_url` itself and is deliberately /// last-writer for it: `cargo_test` reuses that database in migrated state, so /// every other check must name its own `scratch_db` (enforced at config load). async fn migration_dry_run_inner(ctx: &GateCtx, log: &GateLog) -> Result { let Some(scratch_url) = ctx.cfg.scratch_db_url.as_deref() else { log.line("scratch_db_url unset in daemon config\n").await; return Ok(GateOutcome::blocked(GateBlocker::ScratchDbUrlUnset)); }; let mut checked = Vec::new(); let mut primary_backup_path = String::new(); for check in &ctx.cfg.migration_checks { let label = check.dir.display().to_string(); log.line(&format!("==== migration_check: {label} ====\n")) .await; match run_migration_check(ctx, log, scratch_url, check).await? { CheckResult::Passed { backup_path } => { if primary_backup_path.is_empty() { primary_backup_path = backup_path; } checked.push(label); } CheckResult::Stopped(outcome) => return Ok(outcome), } } log.line(&format!( "all {} migration check(s) passed: {}", checked.len(), checked.join(", ") )) .await; Ok(GateOutcome::passed(PassNote::Migrated { backup_path: primary_backup_path, checks: checked, })) } /// One check's verdict: it passed (against `backup_path`), or it produced the /// outcome the whole gate reports. enum CheckResult { Passed { backup_path: String }, Stopped(GateOutcome), } /// Restore one database's dump into its scratch DB and run its migrations on top. async fn run_migration_check( ctx: &GateCtx, log: &GateLog, scratch_url: &str, check: &crate::config::MigrationCheck, ) -> Result { let label = check.dir.display().to_string(); let backup: Option<(String, String)> = sqlx::query_as( "SELECT local_path, fetched_at FROM backups WHERE app = ? AND name = ? ORDER BY id DESC LIMIT 1", ) .bind(&ctx.cfg.id) .bind(&check.backup) .fetch_optional(&ctx.pool) .await?; let Some((backup_path, fetched_at)) = backup else { log.line(&format!( "no {} backup fetched; call /backup/fetch first\n", check.backup )) .await; return Ok(CheckResult::Stopped(GateOutcome::blocked( GateBlocker::NoBackupAvailable { check: label, backup: check.backup.clone(), }, ))); }; // Presence is not freshness. A fetch that quietly stopped working leaves this // row in place, and restoring it dry-runs the migrations against a schema prod // has moved past — green, and worthless. Block on age instead. An unparsable // timestamp is treated as stale: this row is daemon-written RFC 3339, so a // value that will not parse means something is wrong, and failing closed on a // freshness check is the whole point. let age_hours = chrono::DateTime::parse_from_rfc3339(&fetched_at).map_or(i64::MAX, |t| { (Utc::now() - t.with_timezone(&Utc)).num_hours() }); let max_age_hours = ctx.cfg.backup_max_age_hours; if age_hours > i64::from(max_age_hours) { let msg = format!( "backup {backup_path} was fetched {fetched_at} ({age_hours}h ago, max \ {max_age_hours}h); re-run /backup/fetch\n" ); log.line(&msg).await; return Ok(CheckResult::Stopped(GateOutcome::blocked( GateBlocker::BackupStale { age_hours, max_age_hours, check: label, }, ))); } // A check with its own `scratch_db` gets that database created here rather // than by a host bootstrap step: adding a `[[migration_check]]` should not // silently depend on someone having remembered to `createdb` on the Sando // host, which is exactly the class of footgun this gate exists to remove. // DROP + CREATE also makes the database sando-owned, so the PG15+ public // schema grants `reset_scratch` applies next are the owner's to give. let db_url = match check.scratch_db.as_deref() { None => scratch_url.to_string(), Some(dbname) => { let maintenance_url = pg_url_with_dbname(scratch_url, "postgres"); log.line(&format!("---- create scratch db {dbname} ----\n")) .await; if let Err(e) = pg_create_db(&maintenance_url, dbname).await { let msg = format!("{label}: creating scratch db {dbname}: {e}"); log.line(&msg).await; return Ok(CheckResult::Stopped(GateOutcome::failed( GateFailure::RestoreFailed { reason: msg }, ))); } pg_url_with_dbname(scratch_url, dbname) } }; let owner_role = check .owner_role .as_deref() .unwrap_or(&ctx.cfg.scratch_owner_role); log.line("---- reset_scratch ----\n").await; if let Err(e) = reset_scratch(&db_url, owner_role).await { let msg = format!("{label}: scratch reset: {e}"); log.line(&msg).await; return Ok(CheckResult::Stopped(GateOutcome::failed( GateFailure::RestoreFailed { reason: msg }, ))); } log.line(&format!("---- restore_dump ({backup_path}) ----\n")) .await; if let Err(e) = restore_dump(&db_url, &backup_path, log).await { let msg = format!("{label}: restore: {e}"); log.line(&msg).await; return Ok(CheckResult::Stopped(GateOutcome::failed( GateFailure::RestoreFailed { reason: msg }, ))); } let Some(migrations_dir) = ctx.migrations_dir(&check.dir) else { // Neither the bundle nor a checkout holds them. For an accepted // artifact that means the builder did not ship its migrations, and a // dry run over nothing would report green having proved nothing. let msg = format!( "{label}: no migrations at {} in the bundle or a checkout", check.dir.display() ); log.line(&msg).await; return Ok(CheckResult::Stopped(GateOutcome::failed( GateFailure::RestoreFailed { reason: msg }, ))); }; log.line("---- run_migrator ----\n").await; match run_migrator(&db_url, &migrations_dir).await { Ok(()) => { log.line(&format!("{label}: restored {backup_path} + migrated\n")) .await; Ok(CheckResult::Passed { backup_path }) } Err(e) => { let err_s = format!("{label}: {e}"); log.line(&err_s).await; Ok(CheckResult::Stopped(GateOutcome::failed( classify::classify_migration_error(&err_s, None), ))) } } } pub(crate) async fn reset_scratch(db_url: &str, owner_role: &str) -> Result<()> { use sqlx::Executor; use sqlx::postgres::PgPoolOptions; let pool = PgPoolOptions::new() .max_connections(1) .connect(db_url) .await?; // `owner_role` is validated `[A-Za-z0-9_]+` at config load, so interpolating // it into DDL is sound. It still goes through `format('%I')` inside the DO // block for the quoting Postgres expects on an identifier. let sql = format!( r#" DO $$ DECLARE s text; BEGIN -- The dump restores objects owned by the prod role and re-grants to -- it (`ALTER ... OWNER TO {owner_role}`), which errors if the role -- is absent — superuser does not imply the role exists. Create it -- NOLOGIN: the scratch DB needs the role as an *owner* only, never -- as a connecting identity. Idempotent, so a re-reset is a no-op. IF NOT EXISTS (SELECT 1 FROM pg_roles WHERE rolname = '{owner_role}') THEN EXECUTE format('CREATE ROLE %I NOLOGIN', '{owner_role}'); END IF; -- Drop every non-system schema, not just public — migrations create -- custom schemas (e.g. tower_sessions) that survive `DROP SCHEMA -- public CASCADE` and then collide on the next migration run. FOR s IN SELECT nspname FROM pg_namespace WHERE nspname NOT LIKE 'pg_%' AND nspname NOT IN ('information_schema') LOOP EXECUTE format('DROP SCHEMA IF EXISTS %I CASCADE', s); END LOOP; EXECUTE 'CREATE SCHEMA public'; -- Restore the pre-PG15 public-schema default on the throwaway -- scratch DB. Without this, the freshly-created public is owned by -- the connecting role (sando) with no grant to anyone else, so a -- migration's FK/trigger check that Postgres runs as a *restored* -- prod-owned table's owner ({owner_role} from the backup dump) -- fails with "permission denied for schema public". Granting to -- PUBLIC is role-agnostic and safe here — this DB is disposable and -- exists only to dry-run migrations. EXECUTE 'GRANT USAGE, CREATE ON SCHEMA public TO PUBLIC'; -- PG15+: the new owner needs CREATE on public in its own right, not -- only via PUBLIC, for the restore's owner-scoped DDL. EXECUTE format('GRANT USAGE, CREATE ON SCHEMA public TO %I', '{owner_role}'); END $$; "# ); pool.execute(sqlx::raw_sql(sqlx::AssertSqlSafe(sql))) .await?; pool.close().await; Ok(()) } /// Startup assertion for the scratch cluster: the gates reset it, seed an owner /// role into it, and drop leftover test databases in it — none of which a plain /// unprivileged role can do. Historically these preconditions were satisfied by /// hand on fw13 (`ALTER ROLE sando SUPERUSER`, a hand-created `makenotwork` /// role) and nothing recorded that, so a rebuild elsewhere would fail one gate /// at a time with an opaque permissions error. Assert once, at boot, loudly. /// /// Not part of `--check-config`: that path is pure by design (no DB, no /// network), and a green there must mean "this build understands its config", /// not "the cluster is reachable". pub async fn preflight_scratch_privileges(db_url: &str) -> Result<()> { use sqlx::postgres::PgPoolOptions; let pool = PgPoolOptions::new() .max_connections(1) .connect(db_url) .await .context("connecting to scratch_db_url for the startup privilege check")?; let (is_super, can_signal): (bool, bool) = sqlx::query_as( "SELECT rolsuper, pg_catalog.pg_has_role(current_user, 'pg_signal_backend', 'USAGE') FROM pg_roles WHERE rolname = current_user", ) .fetch_one(&pool) .await?; pool.close().await; anyhow::ensure!( is_super || can_signal, "the scratch_db_url role has neither SUPERUSER nor pg_signal_backend; migration_dry_run \ and cargo_test cannot reset the scratch DB or clear stale test databases. Grant one:\n \ ALTER ROLE SUPERUSER; -- what fw13 uses\n \ GRANT pg_signal_backend TO ; -- narrower: terminate only, cannot drop \ foreign-owned databases", ); if !is_super { tracing::warn!( "scratch role has pg_signal_backend but not SUPERUSER: stale test databases owned by \ another role cannot be dropped, and the scratch owner role cannot be created if absent" ); } Ok(()) } /// Best-effort cleanup of stale per-test database clones (`mnw_test_`) /// left behind by a previously-killed `cargo_test` run. /// /// Drops **foreign-owned leftovers too**, which is why the daemon asserts /// SUPERUSER at startup (`preflight_scratch_privileges`): `DROP DATABASE` /// requires ownership or superuser, and the `WITH (FORCE)` terminate requires /// superuser or `pg_signal_backend`. Without both, orphans from a run under a /// different role accumulate and degrade the gate — the failure this cleanup /// exists to prevent. /// /// OPERATIONAL HAZARD: fw13 runs one Postgres cluster shared with local `cargo /// test` as `max`, so a gate firing mid-local-test will force-drop that run's /// databases out from under it. That collision is known and tracked separately /// (give the gate its own cluster); until then, do not run local tests on fw13 /// while a Sando gate is live. /// /// Deliberately **excludes the template** (`mnw_test_template_*`): the harness /// reuses it across runs when it's migration-current (skipping a full /// drop+migrate), so dropping it here would force a needless rebuild every /// gate run. Templates are bounded (one per role) and never accumulate, so /// leaving them is free. Never returns an error: a cleanup miss must not turn a /// deploy red. async fn clean_stale_test_dbs(db_url: &str) { use sqlx::Executor; use sqlx::postgres::PgPoolOptions; let pool = match PgPoolOptions::new() .max_connections(1) .connect(db_url) .await { Ok(p) => p, Err(e) => { tracing::warn!(error = %e, "stale test-db cleanup: could not connect; skipping"); return; } }; // Every per-test clone, whoever owns it. The ownership filter this used to // carry is what let foreign-owned orphans pile up; superuser (asserted at // startup) makes them droppable. let names: Vec<(String,)> = sqlx::query_as( "SELECT datname FROM pg_database WHERE datname LIKE 'mnw_test_%' AND datname NOT LIKE '%template%'", ) .fetch_all(&pool) .await .unwrap_or_default(); let count = names.len(); for (name,) in names { // `name` comes straight from pg_database; quoting it is sufficient. if let Err(e) = pool .execute(sqlx::raw_sql(sqlx::AssertSqlSafe(format!( "DROP DATABASE IF EXISTS \"{name}\" WITH (FORCE)" )))) .await { tracing::warn!(error = %e, db = %name, "stale test-db cleanup: drop failed"); } } if count > 0 { tracing::info!( count, "stale test-db cleanup: dropped leftover mnw_test_* databases" ); } pool.close().await; } /// Build the restore shell line. Two pipelines we accept: /// *.sql -> psql -v ON_ERROR_STOP=1 $url < dump /// *.sql.gz -> set -o pipefail; gunzip -c dump | psql -v ON_ERROR_STOP=1 $url /// /// Two safety flags are load-bearing (CF4): /// - `ON_ERROR_STOP=1`: without it, psql exits 0 even when individual statements /// error, so a partial/corrupt restore would *pass* the gate. /// - `set -o pipefail`: without it a shell pipeline reports only the last /// command's status, so a `gunzip` failure on a truncated archive is masked by /// psql's exit. pipefail is a bash builtin (not POSIX sh), so the runner uses /// `bash -c`. fn restore_shell(db_url: &str, dump: &str) -> String { if std::path::Path::new(dump) .extension() .is_some_and(|ext| ext.eq_ignore_ascii_case("gz")) { format!( "set -o pipefail; gunzip -c {q} | psql -v ON_ERROR_STOP=1 {url}", q = shell_escape(dump), url = shell_escape(db_url), ) } else { format!( "psql -v ON_ERROR_STOP=1 {url} < {q}", url = shell_escape(db_url), q = shell_escape(dump), ) } } async fn restore_dump(db_url: &str, dump: &str, log: &GateLog) -> Result<()> { // Split the password out of the URL and hand it to psql via PGPASSWORD, so it // never lands in argv (visible in /proc//cmdline to any local user). // The sanitized URL — user/host/db, no secret — goes on the command line. let (sanitized, password) = split_pg_password(db_url); let shell = restore_shell(&sanitized, dump); // `bash` (not `sh`): `set -o pipefail` is a bash builtin. The restore runs // locally on the Sando host (fw13), which has bash. let mut cmd = Command::new("bash"); cmd.arg("-c").arg(&shell); // kill_on_drop so the gate's wall-clock ceiling (dispatcher-level timeout on // migration_dry_run) can't orphan a wedged psql restore. cmd.kill_on_drop(true); if let Some(pw) = password { cmd.env("PGPASSWORD", pw); } // Streamed, not `.output()`: a prod-sized restore runs for minutes, and // psql's progress is the only thing an operator has to watch during it. let (_stdout, stderr, status) = log.run(&mut cmd).await?; anyhow::ensure!( status.success(), "restore failed: {}", String::from_utf8_lossy(&stderr), ); Ok(()) } /// Split a `postgres://user:password@host/db` URL into its password-free form and /// the (percent-decoded) password. Returns the URL unchanged with `None` when /// there is no userinfo password. psql reads the password from `PGPASSWORD`, so /// keeping it off the command line removes the /proc exposure. fn split_pg_password(db_url: &str) -> (String, Option) { let Some(after) = db_url.find("://").map(|i| i + 3) else { return (db_url.to_string(), None); }; // The authority ends at the first '/', '?' or '#'; the password (if any) is // between the first ':' and the '@' within the userinfo of that authority. let authority_end = db_url[after..] .find(['/', '?', '#']) .map_or(db_url.len(), |i| after + i); let Some(at) = db_url[after..authority_end].find('@').map(|i| after + i) else { return (db_url.to_string(), None); }; let userinfo = &db_url[after..at]; let Some(colon) = userinfo.find(':') else { return (db_url.to_string(), None); }; let password = percent_decode(&userinfo[colon + 1..]); let sanitized = format!( "{}{}{}", &db_url[..after], &userinfo[..colon], &db_url[at..] ); (sanitized, Some(password)) } /// Minimal `%XX` percent-decode for a URL userinfo component. Non-escape bytes /// pass through; a malformed escape is left literal. fn percent_decode(s: &str) -> String { let b = s.as_bytes(); let mut out = Vec::with_capacity(b.len()); let mut i = 0; while i < b.len() { if b[i] == b'%' && i + 2 < b.len() && let (Some(h), Some(l)) = (hex_val(b[i + 1]), hex_val(b[i + 2])) { out.push((h << 4) | l); i += 3; } else { out.push(b[i]); i += 1; } } String::from_utf8_lossy(&out).into_owned() } fn hex_val(c: u8) -> Option { match c { b'0'..=b'9' => Some(c - b'0'), b'a'..=b'f' => Some(c - b'a' + 10), b'A'..=b'F' => Some(c - b'A' + 10), _ => None, } } pub(crate) async fn run_migrator(db_url: &str, dir: &std::path::Path) -> Result<()> { use sqlx::postgres::PgPoolOptions; let pool = PgPoolOptions::new() .max_connections(1) .connect(db_url) .await?; let migrator = sqlx::migrate::Migrator::new(dir).await?; migrator.run(&pool).await?; pool.close().await; Ok(()) } fn shell_escape(s: &str) -> String { format!("'{}'", s.replace('\'', "'\\''")) } // ---- code_smoke ---- /// A 32+ char throwaway signing secret for the `code_smoke` boot. The server /// only enforces length (>= 32) outside production, and code_smoke's loopback /// `HOST_URL` keeps it in dev mode, so the value is irrelevant beyond that. const CODE_SMOKE_SIGNING_SECRET: &str = "sando-code-smoke-dummy-signing-secret-0000000000"; /// Seconds to wait for the real server to come up and serve `GET /health` /// during `code_smoke`. Longer than `boot_smoke`'s 3s: this boots the *full* /// server (config, pool, session store, webauthn, doc load, app build), not the /// minimal no-DB smoke server. The whole gate is also bounded by /// `gate_timeout_secs` at the dispatcher. const CODE_SMOKE_READY_SECS: u64 = 30; /// `code_smoke` — the first host gate. Boots the freshly-built binary against a /// throwaway *empty* DB it migrates from scratch and seeds the example catalog /// into, then proves the real server serves `GET /health` against that /// nonempty DB. Fast and infra-light (one local Postgres, no prod-dump restore, /// no scratch-role reset, no external services), so a green here proves the /// code is sound and isolates a later `cargo_test`/`migration_dry_run` red as an /// environment problem rather than a code one. /// /// Reuses existing binary entrypoints, so the server needs no smoke-specific /// mode: ` --seed-examples` loads config, connects, migrates from scratch, /// seeds, and exits; a plain `` then serves the real app. Both run with CWD /// at the server crate root so `docs/business/assumptions.toml` + `site-docs/` /// resolve (a missing assumptions file panics real startup), and with a loopback /// `HOST_URL` so config stays in dev mode (no CDN/S3/signing-secret prod /// enforcement). The seed's host allowlist already admits `127.0.0.1`, and the /// fresh DB trivially satisfies its no-real-users guard. async fn code_smoke(ctx: &GateCtx, run_id: GateRunId) -> Result { let log = GateLog::open(ctx, run_id, GateKind::CodeSmoke).await; let outcome = code_smoke_inner(ctx, &log).await; log.close().await; outcome.map(|o| o.with_log_ref(LogRef::new(&ctx.version, GateKind::CodeSmoke))) } /// The staged interior of [`code_smoke`], writing every step through the gate's /// live log. Same split as [`migration_dry_run_inner`]: the caller owns the sink /// and attaches the `log_ref`. async fn code_smoke_inner(ctx: &GateCtx, log: &GateLog) -> Result { let Some(scratch_url) = ctx.cfg.scratch_db_url.as_deref() else { log.line("scratch_db_url unset in daemon config\n").await; return Ok(GateOutcome::blocked(GateBlocker::ScratchDbUrlUnset)); }; // The staged binary (set by build_and_run_host before gating). code_smoke // runs first among the host gates, but staging precedes all gating, so the // artifact path is already recorded. let bin: Option<(String,)> = sqlx::query_as("SELECT artifact_path FROM versions WHERE app = ? AND version = ?") .bind(&ctx.cfg.id) .bind(&ctx.version) .fetch_optional(&ctx.pool) .await?; let Some((bin,)) = bin else { return Ok(GateOutcome::blocked(GateBlocker::ArtifactMissing { version: ctx.version.clone(), })); }; // Frontend builds, before anything else: they need no DB and no staged // binary, and a `tsc` error is the one failure the Rust build deliberately // swallows (both MNW build scripts emit `cargo::warning` and succeed against // a stale `static/dist/`). Failing here is what stops the deploy rsyncing // the previous build's bundle. if let Some(outcome) = code_smoke_frontends(ctx, log).await { return Ok(outcome); } // Docs integrity, first and cheapest: run the staged binary's DB-free // `MNW_CHECK_DOCS` mode before creating the throwaway DB. A broken internal // docs link (a `[..](x.md)` resolving to a slug no page serves) fails here // in well under a second instead of after a full migrate+seed+boot, and a // rotted link never reaches prod as a live 404. Collisions are reported by // the check but do not fail it; only broken links do. if let Some(outcome) = code_smoke_docs_check(ctx, &bin, log).await { return Ok(outcome); } let dbname = code_smoke_db_name(&ctx.version); let maintenance_url = pg_url_with_dbname(scratch_url, "postgres"); let throwaway_url = pg_url_with_dbname(scratch_url, &dbname); // Create the throwaway DB (dropping any stale one from a killed prior run). log.line(&format!("---- createdb {dbname} ----\n")).await; if let Err(e) = pg_create_db(&maintenance_url, &dbname).await { let reason = format!("createdb {dbname}: {e}"); log.line(&reason).await; return Ok(GateOutcome::failed(GateFailure::CodeSmokeSetup { reason })); } // Everything past createdb must drop the DB on the way out, pass or fail. let outcome = code_smoke_body(ctx, &bin, &throwaway_url, log).await; log.line(&format!("\n---- dropdb {dbname} ----\n")).await; if let Err(e) = pg_drop_db(&maintenance_url, &dbname).await { // A teardown miss must not turn a passing gate red — log it and move on. // The next run's createdb drops it first anyway. tracing::warn!(error = %e, db = %dbname, "code_smoke: dropdb failed; next run will reclaim it"); log.line(&format!("dropdb warning (non-fatal): {e}")).await; } Ok(outcome) } /// Compile every configured `frontend_build` in the worktree. /// /// Returns `Some(failed)` on the first project that does not build; `None` when /// all of them do (or none are configured). Output streams to `log` either way. /// /// `npm ci` runs only when `node_modules` is absent. Usually it is not: the app /// build script installed it during the `cargo build` that produced the artifact /// this gate is about to smoke, so the common path here is just `npm run build` /// against a warm install — seconds. The install branch covers the gate running /// against a worktree whose build script was skipped or failed at `npm ci`, and /// it is as fatal as a compile failure, because the alternative is compiling /// against whatever some earlier sha installed. /// /// Unlike the app build scripts, nothing here is best-effort. That asymmetry is /// the point of the gate. async fn code_smoke_frontends(ctx: &GateCtx, log: &GateLog) -> Option { if ctx.cfg.frontend_builds.is_empty() { return None; } let worktree = match ctx.worktree_for(GateKind::CodeSmoke) { Ok(w) => w.to_path_buf(), Err(outcome) => return Some(outcome), }; for fe in &ctx.cfg.frontend_builds { let dir = worktree.join(&fe.dir); let label = fe.dir.display().to_string(); log.line(&format!("---- frontend build ({label}) ----\n")) .await; if !dir.is_dir() { // An older sha predating the frontend, mid-bisect. Skipping keeps // sando able to rebuild history; the log says so out loud. log.line(&format!("{label} absent from this worktree; skipping\n")) .await; continue; } if !dir.join("node_modules").is_dir() && let Some(outcome) = run_npm(&dir, &label, &["ci"], "npm ci", ctx, log).await { return Some(outcome); } if let Some(outcome) = run_npm( &dir, &label, &["run", &fe.script], &format!("npm run {}", fe.script), ctx, log, ) .await { return Some(outcome); } } None } /// One `npm` invocation for [`code_smoke_frontends`], bounded by the gate /// timeout so a wedged install cannot hold the whole pipeline (the enclosing /// `code_smoke` ceiling would catch it eventually, but this attributes the /// failure to the project that hung). async fn run_npm( dir: &std::path::Path, label: &str, args: &[&str], what: &str, ctx: &GateCtx, log: &GateLog, ) -> Option { log.line(&format!("$ {what}\n")).await; let mut cmd = tokio::process::Command::new("npm"); cmd.args(args).current_dir(dir).kill_on_drop(true); let ceiling = std::time::Duration::from_secs(ctx.cfg.gate_timeout_secs); // On the timeout branch the whole `run` future is dropped, which drops the // child; `kill_on_drop` is what turns that into an actual kill. let status = match tokio::time::timeout(ceiling, log.run(&mut cmd)).await { Ok(Ok((_stdout, _stderr, status))) => status, Ok(Err(e)) => { // A missing `npm` lands here. Fatal, not skipped: a build host // without Node cannot produce the bundle the release serves, and // silently passing is how the stale bundle shipped in the first place. log.line(&format!("{what} could not be spawned: {e}\n")) .await; return Some(GateOutcome::failed(GateFailure::SpawnFailed { message: format!("{what} in {label}: {e}"), })); } Err(_elapsed) => { log.line(&format!( "{what} timed out after {}s\n", ctx.cfg.gate_timeout_secs )) .await; return Some(GateOutcome::failed(GateFailure::CodeSmokeFrontend { dir: label.to_string(), exit_code: None, })); } }; if status.success() { return None; } Some(GateOutcome::failed(GateFailure::CodeSmokeFrontend { dir: label.to_string(), exit_code: status.code(), })) } /// Run the staged binary's DB-free docs integrity check (`MNW_CHECK_DOCS=1`). /// /// Returns `Some(failed)` if the check reports broken links, cannot be spawned, /// or overruns its ceiling; `None` when the docs are clean. Output streams to /// `log` either way. The 60s ceiling backstops the case where the staged binary /// predates the flag and would fall through to a normal (DB-needing) boot and /// hang. async fn code_smoke_docs_check(ctx: &GateCtx, bin: &str, log: &GateLog) -> Option { let server_dir = match ctx.worktree_for(GateKind::CodeSmoke) { Ok(w) => w.join("server"), Err(outcome) => return Some(outcome), }; log.line("---- docs check (MNW_CHECK_DOCS) ----\n").await; let mut cmd = tokio::process::Command::new(bin); cmd.env("MNW_CHECK_DOCS", "1") .current_dir(&server_dir) .kill_on_drop(true); let (stdout, _stderr, status) = match tokio::time::timeout(std::time::Duration::from_mins(1), log.run(&mut cmd)).await { Ok(Ok(out)) => out, Ok(Err(e)) => { log.line(&format!("docs check spawn failed: {e}\n")).await; return Some(GateOutcome::failed(GateFailure::SpawnFailed { message: e.to_string(), })); } Err(_elapsed) => { let reason = "docs check timed out after 60s (staged binary may predate MNW_CHECK_DOCS)" .to_string(); log.line(&format!("{reason}\n")).await; return Some(GateOutcome::failed(GateFailure::CodeSmokeSetup { reason })); } }; if status.success() { return None; } Some(GateOutcome::failed(GateFailure::CodeSmokeDocs { broken: parse_check_docs_broken_count(&stdout), })) } /// Best-effort parse of the broken-link count from the `MNW_CHECK_DOCS` sentinel /// line (`MNW_CHECK_DOCS: N broken link(s)`). Returns 0 if absent — the failure /// still stands, only the summary count is unknown. fn parse_check_docs_broken_count(stdout: &[u8]) -> u32 { let text = String::from_utf8_lossy(stdout); for line in text.lines() { if let Some(rest) = line.strip_prefix("MNW_CHECK_DOCS:") { for tok in rest.split_whitespace() { if let Ok(n) = tok.parse::() { return n; } } } } 0 } /// The createdb-to-dropdb interior of `code_smoke`: migrate+seed, then boot and /// probe. Returns the outcome without a `log_ref` (the caller attaches it after /// teardown). Never returns `Err` — spawn/child failures map to typed outcomes. async fn code_smoke_body(ctx: &GateCtx, bin: &str, db_url: &str, log: &GateLog) -> GateOutcome { let server_dir = match ctx.worktree_for(GateKind::CodeSmoke) { Ok(w) => w.join("server"), Err(outcome) => return outcome, }; // Phase 1: migrate-from-scratch + seed. `--seed-examples` loads config, // connects, runs migrations against the empty DB, seeds the catalog, exits. // A non-zero exit here is the "code is unsound" signal (broken migration, // seed error, or config-load failure). log.line("---- migrate + seed (--seed-examples) ----\n") .await; let mut seed_cmd = tokio::process::Command::new(bin); seed_cmd.arg("--seed-examples").current_dir(&server_dir); code_smoke_env(&mut seed_cmd, ctx, db_url); seed_cmd.env("ALLOW_EXAMPLE_SEED", "1").kill_on_drop(true); let seed_status = match log.run(&mut seed_cmd).await { Ok((_stdout, _stderr, status)) => status, Err(e) => { return GateOutcome::failed(GateFailure::SpawnFailed { message: e.to_string(), }); } }; if !seed_status.success() { return GateOutcome::failed(GateFailure::CodeSmokeSeed { exit_code: seed_status.code(), }); } // Phase 2: boot the real server against the now-migrated + seeded DB and // assert both startup signals: it logs `listening` (emitted just before the // socket bind) AND serves GET /health with a 200. The full stdout/stderr is // persisted for the operator either way. log.line("\n---- boot + probe /health ----\n").await; let mut serve_cmd = tokio::process::Command::new(bin); serve_cmd.current_dir(&server_dir); code_smoke_env(&mut serve_cmd, ctx, db_url); serve_cmd .stdout(std::process::Stdio::piped()) .stderr(std::process::Stdio::piped()) .kill_on_drop(true); let mut child = match serve_cmd.spawn() { Ok(c) => c, Err(e) => { return GateOutcome::failed(GateFailure::SpawnFailed { message: e.to_string(), }); } }; // Stream stdout/stderr into the gate log (and out as chunk events) while // the probe loop runs below; the tasks finish when the pipes close (child // exits or is killed). The buffers they return are what the `listening` // assertion reads. let (stdout_task, stderr_task) = log.drain_pipes(&mut child); let probe_timeout = std::time::Duration::from_millis(500); let started = std::time::Instant::now(); let window = std::time::Duration::from_secs(CODE_SMOKE_READY_SECS); let mut probe_ok_after: Option = None; let mut last_probe_err = "never responded".to_string(); let mut early_exit = None; while started.elapsed() < window { if let Ok(Some(status)) = child.try_wait() { early_exit = Some(status); break; } match tokio::time::timeout(probe_timeout, probe_health(ctx.cfg.code_smoke_port)).await { Ok(Ok(())) => { probe_ok_after = Some(started.elapsed().as_millis() as u32); break; } Ok(Err(e)) => last_probe_err = e, Err(_) => last_probe_err = "probe timed out".to_string(), } tokio::time::sleep(std::time::Duration::from_millis(250)).await; } let exit = match early_exit { Some(status) => Some(status), None => { let e = child.try_wait().ok().flatten(); if e.is_none() { let _ = child.kill().await; } e } }; // Keep the serve run's own output so the `listening`-log assertion checks it // (not the seed run's, which exits before binding). The bytes are already in // the gate log; these buffers exist only for the assertion. let serve_stdout = stdout_task.await.unwrap_or_default(); let serve_stderr = stderr_task.await.unwrap_or_default(); let logged_listening = bytes_contain(&serve_stdout, b"listening") || bytes_contain(&serve_stderr, b"listening"); match (exit, probe_ok_after) { // Exited on its own within the window — panic / config error / bind fail. (Some(status), _) => GateOutcome::failed(classify::classify_boot_smoke(status.code())), // Stayed up and served /health. Assert both required startup signals: // the `listening` bind log AND the /health 200. (None, Some(after_ms)) if logged_listening => { GateOutcome::passed(PassNote::HealthyProbe { after_ms }) } // Served /health but the `listening` log never appeared. (None, Some(_)) => GateOutcome::failed(GateFailure::CodeSmokeNoListeningLog), // Stayed up but never served /health — started, not ready. (None, None) => GateOutcome::failed(GateFailure::BootHealthProbeFailed { last_error: last_probe_err, }), } } /// Substring search over raw bytes (the server's log output), for the /// `code_smoke` startup-log assertion. Avoids a lossy UTF-8 conversion of the /// whole buffer just to run `str::contains`. fn bytes_contain(haystack: &[u8], needle: &[u8]) -> bool { if needle.is_empty() || haystack.len() < needle.len() { return needle.is_empty(); } haystack.windows(needle.len()).any(|w| w == needle) } /// Apply the minimal env every `code_smoke` invocation shares: point the binary /// at the throwaway DB, force loopback (dev-mode config, no prod enforcement), /// hand it a dummy signing secret, and disable file scanning (no AV/YARA on the /// build host). The worktree is a clean git checkout, so no stray `.env` shadows /// these (and dotenvy never overrides already-set vars). /// /// This list has to carry EVERY var the server's `Config::from_env` treats as /// mandatory, because `code_smoke` is the only gate that reaches that function /// at all: `boot_smoke` and the docs check both short-circuit in `main` before /// config is loaded. So when the server makes a new var required, this is where /// it has to be answered, and nothing connects the two lists automatically. /// /// That has already bitten once. `CDN_BASE_URL` became mandatory on 2026-07-30 /// (server 3e3b1d15, closing a cover URL that could expire), 13 days after this /// function was written, and `code_smoke` failed with `MissingCdnBaseUrl` the /// first time it ran on the host — which was 2026-08-03, because the gate was /// configured in the repo and absent from `/etc/sando` in between. /// /// The values are deliberately throwaway. The gate asks whether this code can /// migrate, seed, boot and serve; whether a given deployment's env is complete /// is the `config_check_env_file` guard's job, on the node, against that node's /// real env file. fn code_smoke_env(cmd: &mut tokio::process::Command, ctx: &GateCtx, db_url: &str) { // `localhost`, not `127.0.0.1`, and the distinction is load-bearing. The // server derives its WebAuthn relying-party id from HOST_URL's host, and // `WebauthnBuilder::new` validates that id against `Url::domain()` — which // is `None` for an IP literal, so an origin of `http://127.0.0.1:` // fails with WebauthnError::Configuration before the server ever binds. It // is a real ceiling on the gate, not a preference: no IP-literal origin can // boot this binary. `localhost` is a domain, and matches the derived rp_id. // // HOST stays 127.0.0.1: that is the bind address, and the gate probes the // loopback address directly, so only the advertised origin changes. let origin = format!("http://localhost:{}", ctx.cfg.code_smoke_port); cmd.env("DATABASE_URL", db_url) .env("HOST", "127.0.0.1") .env("PORT", ctx.cfg.code_smoke_port.to_string()) .env("HOST_URL", &origin) // Required unconditionally by Config::from_env. Pointing it at the smoke // server's own origin keeps every rendered media URL resolvable within // the gate; no request is ever made to it. .env("CDN_BASE_URL", &origin) .env("SIGNING_SECRET", CODE_SMOKE_SIGNING_SECRET) .env("SCAN_ENABLED", "false") .env("INSECURE_COOKIES", "1"); } /// The throwaway smoke DB name for a version: `sando_code_smoke_` with /// every non-alphanumeric char folded to `_` and lowercased, capped at Postgres' /// 63-byte identifier limit. Sanitized to `[a-z0-9_]` so it's safe to quote into /// DDL. Deterministic per version, so a stale DB from a killed run is reclaimed /// by the next run's `DROP DATABASE IF EXISTS` rather than accumulating. fn code_smoke_db_name(version: &Version) -> String { let mut name = String::from("sando_code_smoke_"); for c in version.to_string().chars() { name.push(if c.is_ascii_alphanumeric() { c.to_ascii_lowercase() } else { '_' }); } name.truncate(63); name } /// Rewrite a `postgres://` URL to point at database `dbname`, preserving scheme, /// userinfo, host/port, and any query (e.g. the socket `?host=/var/run/postgresql` /// form) + fragment. Used to derive the maintenance connection (`postgres`) and /// the throwaway smoke DB URL from the configured `scratch_db_url`. fn pg_url_with_dbname(url: &str, dbname: &str) -> String { let Some(after_scheme) = url.find("://").map(|i| i + 3) else { return url.to_string(); }; let rest = &url[after_scheme..]; // Authority ends at the first '/', '?' or '#'; whatever follows is the // path (the old dbname) plus an optional query/fragment we must keep. let auth_end = rest.find(['/', '?', '#']).unwrap_or(rest.len()); let authority = &rest[..auth_end]; let tail = &rest[auth_end..]; let query_and_frag = match tail.find(['?', '#']) { Some(i) => &tail[i..], None => "", }; format!( "{}{}/{}{}", &url[..after_scheme], authority, dbname, query_and_frag ) } /// Create the throwaway smoke DB on the cluster `maintenance_url` points at, /// dropping any stale one first. `dbname` is sanitized to `[a-z0-9_]` by /// `code_smoke_db_name`, so quoting it is sufficient. `CREATE DATABASE` cannot /// run inside a transaction, so these go through the simple-query protocol (a /// raw `&str` execute), matching `reset_scratch`. async fn pg_create_db(maintenance_url: &str, dbname: &str) -> Result<()> { use sqlx::Executor; use sqlx::postgres::PgPoolOptions; let pool = PgPoolOptions::new() .max_connections(1) .connect(maintenance_url) .await?; pool.execute(sqlx::raw_sql(sqlx::AssertSqlSafe(format!( "DROP DATABASE IF EXISTS \"{dbname}\" WITH (FORCE)" )))) .await?; pool.execute(sqlx::raw_sql(sqlx::AssertSqlSafe(format!( "CREATE DATABASE \"{dbname}\"" )))) .await?; pool.close().await; Ok(()) } /// Drop the throwaway smoke DB, forcing off any lingering connection (the killed /// server's pool). Best-effort at the call site — a failure is logged, not fatal. async fn pg_drop_db(maintenance_url: &str, dbname: &str) -> Result<()> { use sqlx::Executor; use sqlx::postgres::PgPoolOptions; let pool = PgPoolOptions::new() .max_connections(1) .connect(maintenance_url) .await?; pool.execute(sqlx::raw_sql(sqlx::AssertSqlSafe(format!( "DROP DATABASE IF EXISTS \"{dbname}\" WITH (FORCE)" )))) .await?; pool.close().await; Ok(()) } async fn boot_smoke(ctx: &GateCtx, run_id: GateRunId) -> Result { let bin: Option<(String,)> = sqlx::query_as("SELECT artifact_path FROM versions WHERE app = ? AND version = ?") .bind(&ctx.cfg.id) .bind(&ctx.version) .fetch_optional(&ctx.pool) .await?; let Some((bin,)) = bin else { return Ok(GateOutcome::blocked(GateBlocker::ArtifactMissing { version: ctx.version.clone(), })); }; // Readiness smoke: start the binary and confirm it serves `GET /health` // within the window, not merely that the process stays up. Panics in main, // missing config, and port-bind failures still surface as an early exit; a // process that comes up but never serves /health is now its own failure. // // The server requires DATABASE_URL or it panics on config load before // we can observe anything. We point it at the scratch DB (already // migrated by the build step and refreshed by migration_dry_run if // that gate ran first). SCAN_ENABLED=false skips loading YARA rules // from /opt/makenotwork/yara-rules which doesn't exist on the build // host. SANDO_BOOT_SMOKE_PORT tells the smoke server which loopback port // to bind so we know where to probe. Other config has sane optional defaults. let mut cmd = tokio::process::Command::new(&bin); cmd.env("SANDO_BOOT_SMOKE", "1") .env("SANDO_BOOT_SMOKE_PORT", ctx.cfg.boot_smoke_port.to_string()) .env("SCAN_ENABLED", "false") .stdout(std::process::Stdio::piped()) .stderr(std::process::Stdio::piped()) .kill_on_drop(true); if let Some(scratch_url) = ctx.cfg.scratch_db_url.as_deref() { cmd.env("DATABASE_URL", scratch_url); } let log_path = gate_log_path(ctx, GateKind::BootSmoke); let log_ref = LogRef::new(&ctx.version, GateKind::BootSmoke); let mut child = match cmd.spawn() { Ok(c) => c, Err(e) => { // Spawn failures get a one-off log line via LiveLog so the // on-disk file still exists for `GET /logs/...`. let mut log = LiveLog::open(log_path, gate_chunk_cb(ctx.events.clone(), run_id)).await; log.write_chunk(format!("spawn: {e}\n").as_bytes()).await; log.close().await; return Ok(GateOutcome::failed(GateFailure::SpawnFailed { message: e.to_string(), }) .with_log_ref(log_ref)); } }; // The boot smoke window is 3s. Drain stdout/stderr concurrently through // a shared LiveLog sink so the operator sees panics/log lines stream in // real time before the kill, AND the on-disk log gets the full byte // stream for post-mortem reads. The drainers exit when their pipe // closes — which happens when the child exits naturally or after kill. let log = std::sync::Arc::new(tokio::sync::Mutex::new( LiveLog::open(log_path, gate_chunk_cb(ctx.events.clone(), run_id)).await, )); let stdout_task = tokio::spawn(stream_into_log(child.stdout.take(), log.clone())); let stderr_task = tokio::spawn(stream_into_log(child.stderr.take(), log.clone())); // Poll readiness across the 3s window instead of a flat sleep: GET /health // must return 2xx. A crash mid-window short-circuits to the exit-code // failure path (try_wait below); a process that stays up but never serves // /health is a distinct readiness failure. let probe_timeout = std::time::Duration::from_millis(500); let started = std::time::Instant::now(); let window = std::time::Duration::from_secs(3); let mut probe_ok_after: Option = None; let mut last_probe_err = "never responded".to_string(); let mut early_exit = None; while started.elapsed() < window { if let Some(status) = child.try_wait()? { early_exit = Some(status); break; } match tokio::time::timeout(probe_timeout, probe_health(ctx.cfg.boot_smoke_port)).await { Ok(Ok(())) => { probe_ok_after = Some(started.elapsed().as_millis() as u32); break; } Ok(Err(e)) => last_probe_err = e, Err(_) => last_probe_err = "probe timed out".to_string(), } tokio::time::sleep(std::time::Duration::from_millis(150)).await; } // Stop the child unless it already exited, then drain the log tasks. let exit = match early_exit { Some(status) => Some(status), None => { let e = child.try_wait()?; if e.is_none() { let _ = child.kill().await; } e } }; // The streamed bytes already landed in the live log and the on-disk file for // the post-mortem reader. Drain the join handles to avoid hangs. let _ = stdout_task.await; let _ = stderr_task.await; // Unique owner of the Arc at this point (both tasks dropped their clones). if let Ok(mutex) = std::sync::Arc::try_unwrap(log) { mutex.into_inner().close().await; } match (exit, probe_ok_after) { // Exited on its own within the window — a crash/panic/bind failure. (Some(status), _) => { let failure = classify::classify_boot_smoke(status.code()); Ok(GateOutcome::failed(failure).with_log_ref(log_ref)) } // Stayed up and served /health — readiness proven. (None, Some(after_ms)) => { Ok(GateOutcome::passed(PassNote::HealthyProbe { after_ms }).with_log_ref(log_ref)) } // Stayed up but never served /health — started, not ready. (None, None) => Ok(GateOutcome::failed(GateFailure::BootHealthProbeFailed { last_error: last_probe_err, }) .with_log_ref(log_ref)), } } /// One readiness probe of the boot-smoke server: connect to `127.0.0.1:port` /// and `GET /health`, returning `Ok(())` only on a `200`. A hand-rolled HTTP/1.0 /// request over a raw `TcpStream` keeps the outbound probe dependency-free /// (reqwest is dev-only); the smoke server serves the one route over axum, which /// speaks 1.0. `Err` carries a short reason for the operator's failure note. The /// caller wraps each call in a timeout. async fn probe_health(port: u16) -> std::result::Result<(), String> { use tokio::io::{AsyncReadExt, AsyncWriteExt}; let mut stream = tokio::net::TcpStream::connect((std::net::Ipv4Addr::LOCALHOST, port)) .await .map_err(|e| format!("connect: {e}"))?; stream .write_all(b"GET /health HTTP/1.0\r\nHost: localhost\r\nConnection: close\r\n\r\n") .await .map_err(|e| format!("write: {e}"))?; let mut buf = Vec::new(); stream .read_to_end(&mut buf) .await .map_err(|e| format!("read: {e}"))?; let text = String::from_utf8_lossy(&buf); let status_line = text.lines().next().unwrap_or(""); if status_line.contains(" 200 ") { Ok(()) } else { Err(format!("unexpected status line: {status_line:?}")) } } /// Sink that drops streamed output. `node_health` keeps the probe's stderr from /// its `RunOutput` for the failure note, so the live byte stream isn't needed. struct DiscardSink; #[async_trait::async_trait] impl ops_exec::LogSink for DiscardSink { async fn write_chunk(&mut self, _bytes: &[u8]) {} } /// `node_health` — the post-deploy gate that proves the *deployed nodes* are /// serving, recording one outcome per (tier, version) that the next promote /// checks. Distinct from `boot_smoke`, which boots the staged artifact on the /// build host: this probes each node over the same executor the deploy used, so /// a node that took a corrupt artifact, wrong-arch binary, or failed restart is /// caught here rather than waved through (Run-2 SERIOUS-3). Fails closed: any /// unhealthy node fails the gate, and an empty node set is `Blocked`. async fn node_health(ctx: &GateCtx) -> Result { if ctx.nodes.is_empty() { return Ok(GateOutcome::blocked(GateBlocker::NoNodesToProbe)); } for probe in &ctx.nodes { if let Err(detail) = probe_node(probe).await { return Ok(GateOutcome::failed(GateFailure::NodeUnhealthy { node: probe.node.to_string(), detail, })); } } Ok(GateOutcome::passed(PassNote::NodesHealthy { nodes: ctx.nodes.len() as u32, })) } /// Probe one node over its executor: confirm the unit is active post-restart /// and, when a `health_url` is configured, that it serves a 2xx. Retries across /// ~10s because the service may still be restarting / warming. Runs under the /// read-only `Observe(Health)` capability (every Sando node grants it), so the /// probe needs no deploy authority. `Ok(())` = healthy; `Err(detail)` carries a /// short reason for the gate's failure note. async fn probe_node(probe: &NodeProbe) -> std::result::Result<(), String> { use ops_exec::{Action, ObserveKind, Step, sh_quote}; let svc = sh_quote(&probe.service); let url = probe .health_url .as_deref() .map_or_else(|| "''".to_string(), sh_quote); // One executor round-trip with the retry loop on the node: is-active, then // (if a url is set) curl it for a 2xx. Exit 0 only when both hold. let script = format!( "svc={svc}; url={url}; \ for _ in $(seq 1 10); do \ if systemctl is-active --quiet \"$svc\"; then \ if [ -z \"$url\" ] || curl -fsS --max-time 5 \"$url\" >/dev/null 2>&1; then exit 0; fi; \ fi; \ sleep 1; \ done; \ echo 'service not active or health url not 2xx after retries' >&2; exit 1" ); let step = Step::shell(Action::Observe(ObserveKind::Health), script); let mut sink = DiscardSink; let out = probe .executor .run_streaming(&step, &mut sink) .await .map_err(|e| format!("probe spawn: {e}"))?; if out.status.success() { Ok(()) } else { let code = out .status .code() .map_or_else(|| "signal".to_string(), |c| c.to_string()); let stderr: String = String::from_utf8_lossy(&out.stderr) .chars() .take(200) .collect(); Err(format!("exit {code}: {stderr}")) } } /// Drain `stream` into the shared `LiveLog` (which forwards each chunk to /// the on-disk log file AND broadcasts a `GateLogChunk` event), and return /// the concatenated bytes so the classifier can still operate on the full /// output post-hoc. async fn stream_into_log( stream: Option, log: std::sync::Arc>, ) -> Vec where R: tokio::io::AsyncRead + Unpin + Send + 'static, { let mut total = Vec::new(); let Some(mut s) = stream else { return total }; let mut buf = [0u8; 4096]; loop { match s.read(&mut buf).await { Ok(0) => break, Err(_) => break, Ok(n) => { total.extend_from_slice(&buf[..n]); log.lock().await.write_chunk(&buf[..n]).await; } } } total } /// Spawn a child, drain its stdout/stderr through a `LiveLog`, return the /// combined buffers and exit status. Shared by `cargo_test` (no deadline) /// and ad-hoc callers — `boot_smoke` rolls its own variant because of its /// 3s kill window. async fn stream_child_to_live_log( child: &mut tokio::process::Child, events: EventTx, run_id: GateRunId, log_path: PathBuf, ) -> Result<(Vec, Vec, std::process::ExitStatus)> { let log = GateLog::new(LiveLog::open(log_path, gate_chunk_cb(events, run_id)).await); let out = log.stream_child(child).await; log.close().await; out } /// One gate's live log, held across every step of a multi-step gate. /// /// Single-child gates call [`stream_child_to_live_log`] and are done. The /// staged gates (`code_smoke`, `migration_dry_run`) instead run several /// children plus banner lines between them, and they hold one `GateLog` across /// the lot: a single sink means chunk sequence numbers stay monotonic for the /// whole gate, and the on-disk log reads in the order things actually happened /// rather than as stdout-then-stderr assembled at the end. /// /// Every write is best-effort in the same way `LiveLog` is: a log directory /// that cannot be written degrades to callback-only and never turns a passing /// gate red. struct GateLog { sink: Arc>, } impl GateLog { fn new(sink: LiveLog) -> Self { Self { sink: Arc::new(tokio::sync::Mutex::new(sink)), } } /// Open the sink for `gate`'s log file, streaming to the TUI under `run_id`. async fn open(ctx: &GateCtx, run_id: GateRunId, gate: GateKind) -> Self { Self::new( LiveLog::open( gate_log_path(ctx, gate), gate_chunk_cb(ctx.events.clone(), run_id), ) .await, ) } /// Emit a banner (or any line the gate itself produces) through the same /// sink the children stream to, so it lands in sequence with their output. async fn write(&self, bytes: &[u8]) { self.sink.lock().await.write_chunk(bytes).await; } /// Same, for the common `format!`-a-line case. async fn line(&self, s: &str) { self.write(s.as_bytes()).await; } /// Spawn `cmd` with both pipes captured, stream them into the sink as they /// arrive, and return the buffers plus the exit status. The buffers are /// what the classifiers still operate on post-hoc. async fn run( &self, cmd: &mut Command, ) -> std::io::Result<(Vec, Vec, std::process::ExitStatus)> { cmd.stdout(std::process::Stdio::piped()) .stderr(std::process::Stdio::piped()); let mut child = cmd.spawn()?; self.stream_child(&mut child) .await .map_err(std::io::Error::other) } /// Drain an already-spawned child's pipes into the sink and wait for it. async fn stream_child( &self, child: &mut tokio::process::Child, ) -> Result<(Vec, Vec, std::process::ExitStatus)> { let (stdout_task, stderr_task) = self.drain_pipes(child); let status = child.wait().await?; let stdout_buf = stdout_task.await.unwrap_or_default(); let stderr_buf = stderr_task.await.unwrap_or_default(); Ok((stdout_buf, stderr_buf, status)) } /// Start draining a child's pipes into the sink *without* waiting on the /// child. For `code_smoke`'s serve phase, which probes `/health` while the /// server is still up. The caller must await both handles for the buffers /// (and before [`Self::close`], so the flush isn't skipped). #[allow(clippy::type_complexity)] fn drain_pipes( &self, child: &mut tokio::process::Child, ) -> ( tokio::task::JoinHandle>, tokio::task::JoinHandle>, ) { ( tokio::spawn(stream_into_log(child.stdout.take(), self.sink.clone())), tokio::spawn(stream_into_log(child.stderr.take(), self.sink.clone())), ) } /// Flush the file. A still-outstanding streaming task (only possible if the /// gate returned without awaiting it) leaves the `Arc` shared, in which case /// the flush is skipped — `LiveLog` writes unbuffered to the OS either way, /// so nothing already written is lost. async fn close(self) { if let Ok(mutex) = Arc::try_unwrap(self.sink) { mutex.into_inner().close().await; } } } fn gate_log_path(ctx: &GateCtx, gate: GateKind) -> PathBuf { ctx.cfg .logs_root .join(ctx.version.to_string()) .join(format!("{}.log", gate.as_str())) } /// Live check: has `tier`'s burn-in window of `hours` elapsed since its clock /// (`tier_state.burn_in_started_at`, started by a promote onto the tier)? Used /// by the promote-time gate check (`unsatisfied_gates`) so a stale `blocked` /// row never masks an elapsed — or not-yet-elapsed — window. The `burn_in` gate /// runner below wraps the same state with a richer outcome for `/state`. pub async fn burn_in_satisfied( pool: &SqlitePool, app: &AppId, tier: &TierId, hours: u32, ) -> Result { let started: Option = sqlx::query_scalar("SELECT burn_in_started_at FROM tier_state WHERE app = ? AND tier = ?") .bind(app) .bind(tier) .fetch_optional(pool) .await? .flatten(); let Some(started) = started else { return Ok(false); }; let started = chrono::DateTime::parse_from_rfc3339(&started)?.with_timezone(&Utc); Ok(Utc::now() - started >= chrono::Duration::hours(hours as i64)) } async fn burn_in(ctx: &GateCtx, hours: u32) -> Result { // Check tier_state.burn_in_started_at on this tier; pass if enough time // has elapsed. The clock is started by /promote when a version lands on // the burn-in tier. let started: Option = sqlx::query_scalar("SELECT burn_in_started_at FROM tier_state WHERE app = ? AND tier = ?") .bind(&ctx.cfg.id) .bind(&ctx.tier) .fetch_optional(&ctx.pool) .await? .flatten(); let Some(started) = started else { return Ok(GateOutcome::blocked(GateBlocker::BurnInClockNotStarted)); }; let started = chrono::DateTime::parse_from_rfc3339(&started)?.with_timezone(&Utc); let elapsed = Utc::now() - started; let needed = chrono::Duration::hours(hours as i64); if elapsed >= needed { Ok(GateOutcome::passed(PassNote::BurnInElapsed { hours: elapsed.num_hours() as u32, })) } else { let remaining = (needed - elapsed).num_hours().max(0) as u32; Ok(GateOutcome::blocked(GateBlocker::BurnInRemaining { hours_remaining: remaining, hours_total: hours, })) } } async fn manual_confirm(ctx: &GateCtx) -> Result { // Pass iff a row in gate_runs exists with status='passed' for this // (tier, version, manual_confirm) that was inserted out-of-band by an // operator action. Since the harness inserts the in-flight row itself, // look for a prior confirmation row. let prior_at: Option = sqlx::query_scalar( "SELECT finished_at FROM gate_runs WHERE app = ? AND tier = ? AND version = ? AND gate_kind = 'manual_confirm' AND status = 'passed' ORDER BY id DESC LIMIT 1", ) .bind(&ctx.cfg.id) .bind(&ctx.tier) .bind(&ctx.version) .fetch_optional(&ctx.pool) .await?; match prior_at { Some(at_str) => { let at = chrono::DateTime::parse_from_rfc3339(&at_str) .map_or_else(|_| Utc::now(), |d| d.with_timezone(&Utc)); Ok(GateOutcome::passed(PassNote::OperatorConfirmed { at })) } None => Ok(GateOutcome::blocked( GateBlocker::AwaitingOperatorConfirmation, )), } } #[cfg(test)] mod tests { use super::*; use crate::events; use sqlx::sqlite::SqlitePoolOptions; fn target(dir: &str) -> crate::config::TestTarget { crate::config::TestTarget { dir: std::path::PathBuf::from(dir), aux_repo: None, features: Vec::new(), all_features: false, scratch_db: false, } } #[test] fn parse_check_docs_broken_count_reads_the_sentinel() { // The failing sentinel carries the count. assert_eq!( parse_check_docs_broken_count(b"some log\nMNW_CHECK_DOCS: 3 broken link(s)\n"), 3 ); // Count with a preceding log line still parses (first bare int wins). assert_eq!( parse_check_docs_broken_count( b" broken link: a -> b\nMNW_CHECK_DOCS: 1 broken link(s)\n" ), 1 ); // The parser only runs on failure; the "ok" sentinel is never fed to it, // and its "(2" token is not a bare int, so it yields 0 harmlessly. assert_eq!( parse_check_docs_broken_count(b"MNW_CHECK_DOCS: ok (2 collision(s) reported)\n"), 0 ); // Absent sentinel -> 0; the failure still stands, only the count is lost. assert_eq!(parse_check_docs_broken_count(b"unrelated output"), 0); } #[test] fn name_target_points_a_test_failure_at_its_crate() { // With one target the crate was implicit; with fifteen the operator // needs the summary to say which one broke. let f = name_target( GateFailure::CargoTest { failed_count: 3, first_failed: Some("workflows::sync::round_trip".into()), first_panic: None, }, std::path::Path::new("shared/synckit-client"), ); assert_eq!( f.summary(), "3 test(s) failed; first: shared/synckit-client: workflows::sync::round_trip", ); } #[test] fn name_target_points_a_compile_failure_at_its_crate() { let f = name_target( GateFailure::CompileError { error_count: 1, first_error: Some("error[E0063]".into()), }, std::path::Path::new("mnw-cli"), ); assert_eq!( f.summary(), "compile failed (1 error(s)); first: mnw-cli: error[E0063]" ); } #[test] fn name_target_names_the_crate_even_without_a_test_name() { let f = name_target( GateFailure::CargoTest { failed_count: 2, first_failed: None, first_panic: None, }, std::path::Path::new("pom"), ); assert_eq!(f.summary(), "2 test(s) failed; first: pom"); } #[test] fn name_target_leaves_unrelated_failures_alone() { let f = name_target( GateFailure::SpawnFailed { message: "no cargo".into(), }, std::path::Path::new("pom"), ); assert!(matches!(f, GateFailure::SpawnFailed { .. })); } /// `target()` above, but resolved against an aux repo's checkout. fn aux_target(dir: &str, repo: &str) -> crate::config::TestTarget { crate::config::TestTarget { aux_repo: Some(repo.to_string()), ..target(dir) } } /// True when the URL's host parses as a domain rather than an IP literal, /// which is the distinction `Url::domain()` draws and WebAuthn depends on. fn url_host_is_a_domain(url: &str) -> bool { let after = url.split("://").nth(1).unwrap_or(""); let host = after.split(['/', '?', '#']).next().unwrap_or(""); let host = host.rsplit('@').next().unwrap_or(host); let host = if let Some(rest) = host.strip_prefix('[') { rest.split(']').next().unwrap_or("") } else { host.split(':').next().unwrap_or("") }; !host.is_empty() && host.parse::().is_err() } fn resolving_ctx(worktree: &str, aux: &[(&str, &str)]) -> GateCtx { GateCtx { pool: SqlitePool::connect_lazy("sqlite::memory:").unwrap(), cfg: std::sync::Arc::new(crate::config::AppConfig::for_tests()), tier: TierId::new("host"), version: "0.1.0".parse().unwrap(), worktree: Some(PathBuf::from(worktree)), bundle: None, events: events::channel(), nodes: Vec::new(), build_id: None, aux_dirs: aux .iter() .map(|(n, d)| ((*n).to_string(), PathBuf::from(d))) .collect(), } } #[tokio::test] async fn a_plain_target_resolves_under_the_worktree() { let ctx = resolving_ctx("/w/abc123", &[]); assert_eq!( ctx.target_dir(&target("shared/tagtree")), Some(PathBuf::from("/w/abc123/shared/tagtree")), ); } #[tokio::test] async fn an_aux_target_resolves_beside_the_worktree_not_under_it() { // The whole point: `Libraries/docengine` is a sibling of the per-sha // worktree, so a worktree-relative path can never reach it. let ctx = resolving_ctx("/w/abc123", &[("docengine", "/w/Libraries/docengine")]); assert_eq!( ctx.target_dir(&aux_target("", "docengine")), Some(PathBuf::from("/w/Libraries/docengine")), ); // A subdirectory of an aux repo resolves under its checkout. assert_eq!( ctx.target_dir(&aux_target("crates/inner", "docengine")), Some(PathBuf::from("/w/Libraries/docengine/crates/inner")), ); } #[tokio::test] async fn an_aux_target_with_no_checkout_this_run_resolves_to_nothing() { // Promote-time gates carry no aux dirs. Resolving to a wrong path (say, // the worktree) would run the gate against whatever happened to sit // there; `None` makes the caller skip, and --check-config is what // catches a real typo. let ctx = resolving_ctx("/w/abc123", &[]); assert_eq!(ctx.target_dir(&aux_target("", "docengine")), None); } #[test] fn labels_name_the_repo_an_aux_target_lives_in() { assert_eq!(target("server").label(), "server"); assert_eq!(aux_target("", "docengine").label(), "docengine (aux)"); assert_eq!( aux_target("crates/inner", "docengine").label(), "docengine/crates/inner (aux)", ); } /// A `GateCtx` over `worktree` with the given frontend projects configured. /// No DB, no artifact — `code_smoke_frontends` touches neither. async fn frontend_ctx(worktree: &std::path::Path, dirs: &[&str]) -> GateCtx { let mut cfg = crate::config::AppConfig::for_tests(); cfg.frontend_builds = dirs .iter() .map(|d| crate::config::FrontendBuild { dir: PathBuf::from(d), script: "build".into(), }) .collect(); cfg.logs_root = worktree.join("logs"); GateCtx { pool: SqlitePoolOptions::new() .max_connections(1) .connect("sqlite::memory:") .await .unwrap(), cfg: std::sync::Arc::new(cfg), tier: TierId::new("host"), version: "0.1.0".parse().unwrap(), worktree: Some(worktree.to_path_buf()), bundle: None, events: events::channel(), nodes: Vec::new(), build_id: None, aux_dirs: HashMap::new(), } } /// A `GateCtx` for the `migration_dry_run` freshness checks: a migrated /// in-memory pool (so `backups` exists) and a scratch URL set, so the gate /// reaches the backup lookup instead of bailing on config. Nothing here /// touches postgres — every assertion below blocks before `reset_scratch`. async fn dry_run_ctx(worktree: &std::path::Path, max_age_hours: u32) -> GateCtx { let mut cfg = crate::config::AppConfig::for_tests(); cfg.scratch_db_url = Some("postgres:///sando_scratch".into()); cfg.backup_max_age_hours = max_age_hours; cfg.logs_root = worktree.join("logs"); let pool = SqlitePoolOptions::new() .max_connections(1) .connect("sqlite::memory:") .await .unwrap(); crate::db::migrate(&pool).await.unwrap(); GateCtx { pool, cfg: std::sync::Arc::new(cfg), tier: TierId::new("host"), version: "0.1.0".parse().unwrap(), worktree: Some(worktree.to_path_buf()), bundle: None, events: events::channel(), nodes: Vec::new(), build_id: None, aux_dirs: HashMap::new(), } } /// Record a `server` backup row fetched `hours_ago`, as `/backup/fetch` would. async fn seed_backup(ctx: &GateCtx, hours_ago: i64) { seed_named_backup(ctx, "server", hours_ago).await; } /// Record a backup row for one named dump. async fn seed_named_backup(ctx: &GateCtx, name: &str, hours_ago: i64) { let at = (Utc::now() - chrono::Duration::hours(hours_ago)).to_rfc3339(); sqlx::query( "INSERT INTO backups (name, fetched_at, source, local_path, byte_size) VALUES (?, ?, 'file:///x.sql.gz', '/tmp/sando-test-backup.sql.gz', 1000000)", ) .bind(name) .bind(at) .execute(&ctx.pool) .await .unwrap(); } /// The multithreaded check, as `sando-daemon.toml` configures it. fn mt_check() -> crate::config::MigrationCheck { crate::config::MigrationCheck { dir: std::path::PathBuf::from("multithreaded/migrations"), backup: "multithreaded".into(), scratch_db: Some("sando_scratch_mt".into()), owner_role: Some("multithreaded".into()), } } /// Re-point a `dry_run_ctx` at one check, keeping its pool and scratch URL. fn with_check(ctx: &mut GateCtx, check: crate::config::MigrationCheck) { let mut cfg = crate::config::AppConfig::for_tests(); cfg.scratch_db_url = ctx.cfg.scratch_db_url.clone(); cfg.backup_max_age_hours = ctx.cfg.backup_max_age_hours; cfg.logs_root = ctx.cfg.logs_root.clone(); cfg.migration_checks = vec![check]; ctx.cfg = std::sync::Arc::new(cfg); } #[tokio::test] async fn migration_dry_run_blocks_when_a_checks_own_dump_was_never_fetched() { // The hazard the check list exists for: multithreaded applies its own // migrations at boot against its own database, so the server's dump must // never stand in for it. A fetched `server` row with no `multithreaded` // row is exactly that substitution, and it has to block. let tmp = tempfile::tempdir().unwrap(); let mut ctx = dry_run_ctx(tmp.path(), 48).await; with_check(&mut ctx, mt_check()); seed_named_backup(&ctx, "server", 1).await; let log = GateLog::open(&ctx, GateRunId(0), GateKind::MigrationDryRun).await; let outcome = migration_dry_run_inner(&ctx, &log).await.unwrap(); log.close().await; let crate::outcome::GateStatus::Blocked { blocker } = outcome.status else { panic!("a missing multithreaded dump must block"); }; let GateBlocker::NoBackupAvailable { check, backup } = blocker else { panic!("expected NoBackupAvailable, got {blocker:?}"); }; assert_eq!(backup, "multithreaded", "names the dump that is missing"); assert!( check.contains("multithreaded/migrations"), "names the check that wanted it, got {check}" ); } #[tokio::test] async fn migration_dry_run_freshness_is_per_dump() { // A fresh server dump must not make a 45-day-old multithreaded dump look // current: the clock is per-database, or the second check inherits the // first's freshness and the gate is theatre. let tmp = tempfile::tempdir().unwrap(); let mut ctx = dry_run_ctx(tmp.path(), 48).await; with_check(&mut ctx, mt_check()); seed_named_backup(&ctx, "server", 1).await; seed_named_backup(&ctx, "multithreaded", 24 * 45).await; let log = GateLog::open(&ctx, GateRunId(0), GateKind::MigrationDryRun).await; let outcome = migration_dry_run_inner(&ctx, &log).await.unwrap(); log.close().await; let crate::outcome::GateStatus::Blocked { blocker } = outcome.status else { panic!("a 45-day-old multithreaded dump must block"); }; let GateBlocker::BackupStale { check, .. } = blocker else { panic!("expected BackupStale, got {blocker:?}"); }; assert!( check.contains("multithreaded/migrations"), "names the check whose dump is stale, got {check}" ); } #[tokio::test] async fn migration_dry_run_blocks_on_a_stale_backup() { // The failure this closes: the gate used to check only that a backups row // existed, so a fetch that silently stopped working left it green against // an ever-older schema. Sando ran 45 days that way. let tmp = tempfile::tempdir().unwrap(); let ctx = dry_run_ctx(tmp.path(), 48).await; seed_backup(&ctx, 24 * 45).await; let log = GateLog::open(&ctx, GateRunId(0), GateKind::MigrationDryRun).await; let outcome = migration_dry_run_inner(&ctx, &log).await.unwrap(); log.close().await; let crate::outcome::GateStatus::Blocked { blocker } = outcome.status else { panic!("a 45-day-old backup must block"); }; let GateBlocker::BackupStale { age_hours, max_age_hours, .. } = blocker else { panic!("expected BackupStale, got {blocker:?}"); }; assert_eq!(max_age_hours, 48); assert!( age_hours >= 24 * 45, "reports the real age, got {age_hours}" ); } #[tokio::test] async fn migration_dry_run_accepts_a_fresh_backup() { // The other side of the boundary: a backup inside the window must not be // blocked on freshness. It fails later (there is no such dump on disk), // which is exactly the proof the age check let it through. let tmp = tempfile::tempdir().unwrap(); let ctx = dry_run_ctx(tmp.path(), 48).await; seed_backup(&ctx, 6).await; let log = GateLog::open(&ctx, GateRunId(0), GateKind::MigrationDryRun).await; let outcome = migration_dry_run_inner(&ctx, &log).await.unwrap(); log.close().await; assert!( !matches!( outcome.status, crate::outcome::GateStatus::Blocked { blocker: GateBlocker::BackupStale { .. } } ), "a 6h-old backup is fresh, got {:?}", outcome.status, ); } #[tokio::test] async fn migration_dry_run_treats_an_unparsable_fetched_at_as_stale() { // Fail closed: `fetched_at` is daemon-written RFC 3339, so a value that // will not parse means the row is untrustworthy — and a freshness check // that shrugs at a timestamp it cannot read is not a freshness check. let tmp = tempfile::tempdir().unwrap(); let ctx = dry_run_ctx(tmp.path(), 48).await; sqlx::query( "INSERT INTO backups (fetched_at, source, local_path, byte_size) VALUES ('not-a-timestamp', 'file:///x.sql.gz', '/tmp/x.sql.gz', 1000000)", ) .execute(&ctx.pool) .await .unwrap(); let log = GateLog::open(&ctx, GateRunId(0), GateKind::MigrationDryRun).await; let outcome = migration_dry_run_inner(&ctx, &log).await.unwrap(); log.close().await; assert!( matches!( outcome.status, crate::outcome::GateStatus::Blocked { blocker: GateBlocker::BackupStale { .. } } ), "an unreadable fetched_at must block, got {:?}", outcome.status, ); } /// A `code_smoke` live log over `ctx.cfg.logs_root`, for the helpers that /// take one. `GateRunId(0)` never matches a real row; nothing reads the /// chunk events in these tests. async fn test_gate_log(ctx: &GateCtx) -> GateLog { GateLog::open(ctx, GateRunId(0), GateKind::CodeSmoke).await } /// Close `log` (flushing it) and read back what it wrote on disk. async fn read_gate_log(ctx: &GateCtx, log: GateLog) -> String { log.close().await; tokio::fs::read_to_string(gate_log_path(ctx, GateKind::CodeSmoke)) .await .expect("the gate log must exist on disk") } /// Write a minimal npm project at `worktree/` whose `build` script /// exits with `exit_code`. Pre-creates `node_modules` so the gate skips /// `npm ci` — these tests are about the build step, not the network. fn fake_npm_project(worktree: &std::path::Path, dir: &str, exit_code: u8) { let root = worktree.join(dir); std::fs::create_dir_all(root.join("node_modules")).unwrap(); std::fs::write( root.join("package.json"), format!( r#"{{"name":"fake","version":"0.0.0","private":true, "scripts":{{"build":"exit {exit_code}"}}}}"# ), ) .unwrap(); } #[tokio::test] async fn frontend_gate_fails_on_a_build_error_and_names_the_project() { // The whole point of the gate: the app build scripts downgrade this to a // cargo::warning, so if it passes here nothing stops a stale bundle. let tmp = tempfile::tempdir().unwrap(); fake_npm_project(tmp.path(), "server/frontend", 0); fake_npm_project(tmp.path(), "multithreaded/frontend", 2); let ctx = frontend_ctx(tmp.path(), &["server/frontend", "multithreaded/frontend"]).await; let log = test_gate_log(&ctx).await; let outcome = code_smoke_frontends(&ctx, &log) .await .expect("a failing tsc must fail the gate"); let crate::outcome::GateStatus::Failed { failure } = &outcome.status else { panic!("expected a failure, got {:?}", outcome.status) }; assert!( matches!( failure, GateFailure::CodeSmokeFrontend { dir, exit_code: Some(2) } if dir == "multithreaded/frontend" ), "got: {failure:?}" ); // The passing project ran first; its output belongs in the log too. let text = read_gate_log(&ctx, log).await; assert!(text.contains("server/frontend"), "log: {text}"); } #[tokio::test] async fn frontend_gate_passes_when_every_project_builds() { let tmp = tempfile::tempdir().unwrap(); fake_npm_project(tmp.path(), "server/frontend", 0); let ctx = frontend_ctx(tmp.path(), &["server/frontend"]).await; let log = test_gate_log(&ctx).await; assert!( code_smoke_frontends(&ctx, &log).await.is_none(), "a clean build must not fail the gate" ); } /// The point of routing `code_smoke` through `LiveLog`: its output reaches /// the operator *while* the gate runs, as `GateLogChunk` events, instead of /// appearing all at once when the gate finishes. #[tokio::test] async fn code_smoke_streams_chunks_as_it_runs() { let tmp = tempfile::tempdir().unwrap(); fake_npm_project(tmp.path(), "server/frontend", 0); let ctx = frontend_ctx(tmp.path(), &["server/frontend"]).await; let mut rx = ctx.events.subscribe_logs(); let log = GateLog::open(&ctx, GateRunId(7), GateKind::CodeSmoke).await; assert!(code_smoke_frontends(&ctx, &log).await.is_none()); log.close().await; let mut chunks = Vec::new(); while let Ok(envelope) = rx.try_recv() { if let Event::GateLogChunk { run_id, seq, text } = envelope.event { assert_eq!(run_id, GateRunId(7)); chunks.push((seq, text)); } } assert!(!chunks.is_empty(), "no chunk ever reached the bus"); // Sequence numbers are per-run and monotonic across every step of the // gate, which is why the whole gate shares one sink. let seqs: Vec = chunks.iter().map(|(seq, _)| *seq).collect(); assert!( seqs.windows(2).all(|w| w[0] < w[1]), "chunk seq must be monotonic, got {seqs:?}" ); let joined: String = chunks.into_iter().map(|(_, text)| text).collect(); assert!(joined.contains("server/frontend"), "chunks: {joined}"); } #[tokio::test] async fn frontend_gate_skips_a_project_absent_from_the_worktree() { // Rebuilding an older sha that predates the frontend must stay possible. let tmp = tempfile::tempdir().unwrap(); let ctx = frontend_ctx(tmp.path(), &["multithreaded/frontend"]).await; let log = test_gate_log(&ctx).await; assert!(code_smoke_frontends(&ctx, &log).await.is_none()); assert!( read_gate_log(&ctx, log).await.contains("skipping"), "the skip must be visible in the log" ); } #[tokio::test] async fn cargo_test_fails_closed_when_no_target_exists_in_the_worktree() { // A worktree missing every configured crate must not report "tests // passed" having run none. Uses an empty tempdir as the worktree, so // no cargo process is ever spawned. let tmp = tempfile::tempdir().unwrap(); let mut cfg = crate::config::AppConfig::for_tests(); cfg.test_targets = vec![target("server"), target("mnw-cli")]; cfg.logs_root = tmp.path().join("logs"); let pool = SqlitePoolOptions::new() .max_connections(1) .connect("sqlite::memory:") .await .unwrap(); let ctx = GateCtx { pool, cfg: std::sync::Arc::new(cfg), tier: TierId::new("host"), version: "0.1.0".parse().unwrap(), worktree: Some(tmp.path().to_path_buf()), bundle: None, events: events::channel(), nodes: Vec::new(), build_id: None, aux_dirs: HashMap::new(), }; let out = cargo_test(&ctx, GateRunId(1)).await.unwrap(); assert_eq!( out.status_str(), "failed", "green here would be a silent no-op gate" ); let crate::outcome::GateStatus::Failed { failure } = &out.status else { panic!("expected a failure") }; assert!( failure.summary().contains("ran no targets"), "got: {}", failure.summary() ); } #[tokio::test] async fn scratch_db_env_is_opt_in_per_target() { // Exporting DATABASE_URL knocks sqlx out of offline mode, so a crate // shipping .sqlx data must not see it. let mut cfg = crate::config::AppConfig::for_tests(); cfg.scratch_db_url = Some("postgres://sando@127.0.0.1/sando_scratch".into()); let ctx = GateCtx { pool: SqlitePoolOptions::new() .max_connections(1) .connect_lazy("sqlite::memory:") .unwrap(), cfg: std::sync::Arc::new(cfg), tier: TierId::new("host"), version: "0.1.0".parse().unwrap(), worktree: Some(std::path::PathBuf::from("/tmp/wt")), bundle: None, events: events::channel(), nodes: Vec::new(), build_id: None, aux_dirs: HashMap::new(), }; let dir = std::path::Path::new("/tmp/wt/x"); let off = cargo_test_command(&ctx, dir, &target("x"), &[], &[]); let has_db = |c: &Command| { c.as_std() .get_envs() .any(|(k, v)| k == "DATABASE_URL" && v.is_some()) }; assert!(!has_db(&off), "scratch_db defaults off"); let mut on_target = target("x"); on_target.scratch_db = true; assert!( has_db(&cargo_test_command(&ctx, dir, &on_target, &[], &[])), "opt-in exports it" ); } #[tokio::test] async fn all_features_replaces_the_feature_list() { let ctx = GateCtx { pool: SqlitePoolOptions::new() .max_connections(1) .connect_lazy("sqlite::memory:") .unwrap(), cfg: std::sync::Arc::new(crate::config::AppConfig::for_tests()), tier: TierId::new("host"), version: "0.1.0".parse().unwrap(), worktree: Some(std::path::PathBuf::from("/tmp/wt")), bundle: None, events: events::channel(), nodes: Vec::new(), build_id: None, aux_dirs: HashMap::new(), }; let mut t = target("shared/ops-exec"); t.all_features = true; let cmd = cargo_test_command(&ctx, std::path::Path::new("/tmp/wt"), &t, &[], &[]); let args: Vec<_> = cmd .as_std() .get_args() .map(|a| a.to_string_lossy().into_owned()) .collect(); assert!(args.iter().any(|a| a == "--all-features"), "got: {args:?}"); assert!(!args.iter().any(|a| a == "--features"), "got: {args:?}"); } #[test] fn every_gate_kind_round_trips_through_its_wire_string() { // gate_kind is a TEXT column and a WS event field; as_str and FromStr // disagreeing would make a gate's evidence unreadable by // unsatisfied_gates, which fails the promote closed with no explanation. for k in [ GateKind::CargoTest, GateKind::HardeningTest, GateKind::Clippy, GateKind::Fmt, GateKind::CargoAudit, GateKind::CargoDeny, GateKind::MigrationDryRun, GateKind::CodeSmoke, GateKind::BootSmoke, GateKind::NodeHealth, GateKind::BurnIn, GateKind::ManualConfirm, ] { assert_eq!( k.as_str().parse::().unwrap(), k, "round trip for {k:?}" ); } } #[test] fn first_meaningful_line_prefers_the_diagnostic() { let stderr = b" Updating crates.io index\nerror: 1 vulnerability found!\n"; assert_eq!( first_meaningful_line(b"", stderr), "error: 1 vulnerability found!" ); } #[test] fn first_meaningful_line_finds_a_deny_verdict() { let out = b"advisories FAILED, bans ok, licenses FAILED, sources ok\n"; assert!(first_meaningful_line(out, b"").contains("FAILED")); } #[test] fn first_meaningful_line_falls_back_rather_than_returning_empty() { assert!(first_meaningful_line(b"", b"").contains("see the gate log")); } #[tokio::test] async fn supply_chain_gates_skip_crates_without_their_config() { // Four crates in this repo fail `cargo audit` purely for want of a // triaged .cargo/audit.toml. Running it there would make the gate // permanently red, so a target only qualifies once it carries the file. // The worktree here has a Cargo.toml but no audit config, so nothing // qualifies and the gate fails closed rather than passing over zero work. let tmp = tempfile::tempdir().unwrap(); let crate_dir = tmp.path().join("server"); std::fs::create_dir_all(&crate_dir).unwrap(); std::fs::write(crate_dir.join("Cargo.toml"), "[package]\nname = \"x\"\n").unwrap(); let mut cfg = crate::config::AppConfig::for_tests(); cfg.test_targets = vec![target("server")]; cfg.logs_root = tmp.path().join("logs"); let ctx = GateCtx { pool: SqlitePoolOptions::new() .max_connections(1) .connect("sqlite::memory:") .await .unwrap(), cfg: std::sync::Arc::new(cfg), tier: TierId::new("host"), version: "0.1.0".parse().unwrap(), worktree: Some(tmp.path().to_path_buf()), bundle: None, events: events::channel(), nodes: Vec::new(), build_id: None, aux_dirs: HashMap::new(), }; let out = supply_chain(&ctx, GateRunId(1), GateKind::CargoAudit) .await .unwrap(); assert_eq!(out.status_str(), "failed"); let crate::outcome::GateStatus::Failed { failure } = &out.status else { panic!("expected a failure") }; assert!( failure.summary().contains("ran nothing"), "got: {}", failure.summary() ); // Drop the config in and the same target now qualifies. std::fs::create_dir_all(crate_dir.join(".cargo")).unwrap(); std::fs::write(crate_dir.join(".cargo/audit.toml"), "[advisories]\n").unwrap(); // The fixture crate is not a real cargo project, so the tool itself // still errors — but on its own terms, not with "ran nothing". That // distinction is the thing under test: the target was attempted. let out = supply_chain(&ctx, GateRunId(2), GateKind::CargoAudit) .await .unwrap(); if let crate::outcome::GateStatus::Failed { failure } = &out.status { assert!( !failure.summary().contains("ran nothing"), "a target carrying the config must be attempted, not skipped; got: {}", failure.summary(), ); } } #[test] fn tests_run_reads_the_libtest_summary() { let out = b"running 6 tests\ntest auth_rate_limit_triggers_on_burst ... ok\n\n\ test result: ok. 6 passed; 0 failed; 0 ignored; 0 measured; 118 filtered out\n"; assert_eq!(tests_run(out), 6); } #[test] fn tests_run_sums_across_test_binaries() { let out = b"test result: ok. 6 passed; 0 failed\ntest result: ok. 2 passed; 0 failed\n"; assert_eq!(tests_run(out), 8); } #[test] fn tests_run_is_zero_when_the_filter_matched_nothing() { // The case hardening_test fails closed on: a filter matching no tests // exits 0, so a rename would otherwise make the gate a green no-op. let out = b"running 0 tests\n\n\ test result: ok. 0 passed; 0 failed; 0 ignored; 0 measured; 124 filtered out\n"; assert_eq!(tests_run(out), 0); } #[test] fn tests_run_is_zero_without_a_summary_line() { assert_eq!(tests_run(b"error: could not compile `makenotwork`\n"), 0); } #[tokio::test] async fn hardening_test_command_carries_no_features() { // The entire point of the gate: production constants, which means no // `fast-tests`. A stray feature here silently restores the blind spot. let ctx = GateCtx { pool: SqlitePoolOptions::new() .max_connections(1) .connect_lazy("sqlite::memory:") .unwrap(), cfg: std::sync::Arc::new(crate::config::AppConfig::for_tests()), tier: TierId::new("host"), version: "0.1.0".parse().unwrap(), worktree: Some(std::path::PathBuf::from("/tmp/wt")), bundle: None, events: events::channel(), nodes: Vec::new(), build_id: None, aux_dirs: HashMap::new(), }; let plain = crate::config::TestTarget { dir: std::path::PathBuf::from("server"), aux_repo: None, features: Vec::new(), all_features: false, scratch_db: true, }; let cmd = cargo_test_command( &ctx, std::path::Path::new("/tmp/wt/server"), &plain, &[], &["--test", "integration"], ); let args: Vec<_> = cmd .as_std() .get_args() .map(|a| a.to_string_lossy().into_owned()) .collect(); assert!( !args.iter().any(|a| a == "--features"), "hardening_test must pass no features: {args:?}" ); assert!( !args.iter().any(|a| a.contains("fast-tests")), "got: {args:?}" ); let fast_target = crate::config::TestTarget { dir: std::path::PathBuf::from("server"), aux_repo: None, features: vec!["fast-tests".into()], all_features: false, scratch_db: true, }; let fast = cargo_test_command( &ctx, std::path::Path::new("/tmp/wt/server"), &fast_target, &["fast-tests"], &[], ); let fast_args: Vec<_> = fast .as_std() .get_args() .map(|a| a.to_string_lossy().into_owned()) .collect(); assert!( fast_args .windows(2) .any(|w| w == ["--features", "fast-tests"]), "got: {fast_args:?}" ); } /// Spawn a one-shot loopback server that answers the first connection with /// `status_line` + a tiny body, then closes. Returns the bound port. async fn oneshot_http(status_line: &'static str) -> u16 { use tokio::io::{AsyncReadExt, AsyncWriteExt}; let listener = tokio::net::TcpListener::bind((std::net::Ipv4Addr::LOCALHOST, 0)) .await .unwrap(); let port = listener.local_addr().unwrap().port(); tokio::spawn(async move { if let Ok((mut sock, _)) = listener.accept().await { let mut scratch = [0u8; 1024]; let _ = sock.read(&mut scratch).await; // drain the request line let resp = format!("{status_line}\r\nContent-Length: 2\r\nConnection: close\r\n\r\nok"); let _ = sock.write_all(resp.as_bytes()).await; } }); port } #[tokio::test] async fn probe_health_ok_on_200() { let port = oneshot_http("HTTP/1.1 200 OK").await; assert!(probe_health(port).await.is_ok()); } #[tokio::test] async fn probe_health_err_on_non_200() { let port = oneshot_http("HTTP/1.1 503 Service Unavailable").await; let err = probe_health(port).await.unwrap_err(); assert!(err.contains("status line"), "{err}"); } #[tokio::test] async fn probe_health_err_on_connection_refused() { // Bind then drop to get an almost-certainly-free port nothing listens on. let port = { let l = tokio::net::TcpListener::bind((std::net::Ipv4Addr::LOCALHOST, 0)) .await .unwrap(); l.local_addr().unwrap().port() }; let err = probe_health(port).await.unwrap_err(); assert!(err.contains("connect"), "{err}"); } /// burn_in returns a typed Blocked when the clock isn't started; the /// runner persists status='blocked' + outcome_json (the json carries /// blocker.kind = 'burn_in_clock_not_started'). #[tokio::test] async fn burn_in_blocked_persists_typed_outcome() { let pool = SqlitePoolOptions::new() .max_connections(1) .connect("sqlite::memory:") .await .unwrap(); crate::db::migrate(&pool).await.unwrap(); // Topology sync expects a tier row before gate_runs can reference it. sqlx::query("INSERT INTO tiers (name, ord, provisioned, canary) VALUES ('host', 0, 0, 'sequential')") .execute(&pool).await.unwrap(); sqlx::query("INSERT INTO tier_state (tier) VALUES ('host')") .execute(&pool) .await .unwrap(); // versions FK target. sqlx::query("INSERT INTO versions (version, git_sha, built_at, artifact_path) VALUES ('0.1.0', 'abc1234', '2026-01-01T00:00:00Z', '/tmp/x')") .execute(&pool).await.unwrap(); let cfg = std::sync::Arc::new(crate::config::AppConfig::for_tests()); let ctx = GateCtx { pool: pool.clone(), cfg, tier: TierId::new("host"), version: "0.1.0".parse().unwrap(), worktree: Some(std::path::PathBuf::from("/tmp/unused")), bundle: None, events: events::channel(), nodes: Vec::new(), build_id: None, aux_dirs: HashMap::new(), }; let out = run(&ctx, &Gate::BurnIn { hours: 24 }).await.unwrap(); assert_eq!(out.status_str(), "blocked"); assert!(!out.is_passed()); // Read the persisted row. let row: (Option, Option) = sqlx::query_as("SELECT status, outcome_json FROM gate_runs ORDER BY id DESC LIMIT 1") .fetch_one(&pool) .await .unwrap(); assert_eq!(row.0.as_deref(), Some("blocked"), "typed status"); let json: serde_json::Value = serde_json::from_str(row.1.as_deref().unwrap()).unwrap(); assert_eq!(json["status"]["kind"], "blocked"); assert_eq!( json["status"]["blocker"]["kind"], "burn_in_clock_not_started" ); } /// node_health fails closed when there are no nodes to probe: a serving tier /// should always carry nodes, so an empty set is a misconfiguration that must /// block promotion, not pass it. #[tokio::test] async fn node_health_blocks_with_no_nodes() { let pool = SqlitePoolOptions::new() .max_connections(1) .connect("sqlite::memory:") .await .unwrap(); crate::db::migrate(&pool).await.unwrap(); sqlx::query( "INSERT INTO tiers (name, ord, provisioned, canary) VALUES ('b', 2, 1, 'sequential')", ) .execute(&pool) .await .unwrap(); sqlx::query("INSERT INTO tier_state (tier) VALUES ('b')") .execute(&pool) .await .unwrap(); sqlx::query("INSERT INTO versions (version, git_sha, built_at, artifact_path) VALUES ('0.1.0', 'abc1234', '2026-01-01T00:00:00Z', '/tmp/x')") .execute(&pool).await.unwrap(); let cfg = std::sync::Arc::new(crate::config::AppConfig::for_tests()); let ctx = GateCtx { pool: pool.clone(), cfg, tier: TierId::new("b"), version: "0.1.0".parse().unwrap(), worktree: None, bundle: None, events: events::channel(), nodes: Vec::new(), // no nodes -> fail closed build_id: None, aux_dirs: HashMap::new(), }; let out = run(&ctx, &Gate::NodeHealth).await.unwrap(); assert_eq!(out.status_str(), "blocked"); assert!(!out.is_passed()); let row: (Option, Option) = sqlx::query_as("SELECT status, outcome_json FROM gate_runs ORDER BY id DESC LIMIT 1") .fetch_one(&pool) .await .unwrap(); let json: serde_json::Value = serde_json::from_str(row.1.as_deref().unwrap()).unwrap(); assert_eq!(json["status"]["blocker"]["kind"], "no_nodes_to_probe"); } /// reset_scratch must drop every non-system schema, not just `public` — /// otherwise migrations that create custom schemas (e.g. tower_sessions) /// collide on the next run. This regressed once (Phase 0) and the fix is /// load-bearing for migration_dry_run. /// /// Gated on `SANDO_TEST_PG_URL` so it only runs where postgres is /// available. Set `SANDO_TEST_PG_URL=postgres:///sando_scratch?host=/var/run/postgresql` /// (or similar) before `cargo test`. #[tokio::test] async fn reset_scratch_drops_all_non_system_schemas() { let Ok(url) = std::env::var("SANDO_TEST_PG_URL") else { eprintln!("skipping: SANDO_TEST_PG_URL not set"); return; }; use sqlx::Executor; use sqlx::postgres::PgPoolOptions; let pool = PgPoolOptions::new() .max_connections(1) .connect(&url) .await .unwrap(); // Plant two non-system schemas + a table in each. pool.execute( "DROP SCHEMA IF EXISTS foo CASCADE; CREATE SCHEMA foo; CREATE TABLE foo.t (i int);", ) .await .unwrap(); pool.execute("DROP SCHEMA IF EXISTS tower_sessions CASCADE; CREATE SCHEMA tower_sessions; CREATE TABLE tower_sessions.session (id text);") .await.unwrap(); pool.close().await; reset_scratch(&url, "makenotwork") .await .expect("reset_scratch"); let pool = PgPoolOptions::new() .max_connections(1) .connect(&url) .await .unwrap(); let rows: Vec<(String,)> = sqlx::query_as( "SELECT nspname FROM pg_namespace WHERE nspname NOT LIKE 'pg_%' AND nspname <> 'information_schema'", ) .fetch_all(&pool) .await .unwrap(); let names: Vec = rows.into_iter().map(|(s,)| s).collect(); // After reset, only `public` should remain among non-system schemas. assert_eq!(names, vec!["public".to_string()], "got: {names:?}"); pool.close().await; } /// reset_scratch must leave the dump's owner role existing and able to /// create in `public`, because a prod `pg_dump` carries `ALTER ... OWNER TO /// ` for every object. This was satisfied by a hand-created NOLOGIN /// role on fw13; nothing recorded it, so any other box failed /// migration_dry_run at the restore with "role does not exist". /// /// Uses a throwaway role name so it can prove the *creation* path rather /// than passing on fw13's pre-existing `makenotwork`. Same /// `SANDO_TEST_PG_URL` gate as above; needs a superuser connection. #[tokio::test] async fn reset_scratch_seeds_the_dump_owner_role_when_absent() { let Ok(url) = std::env::var("SANDO_TEST_PG_URL") else { eprintln!("skipping: SANDO_TEST_PG_URL not set"); return; }; use sqlx::Executor; use sqlx::postgres::PgPoolOptions; let role = "sando_test_owner_probe"; // `DROP ROLE` refuses while the role still holds the grants reset_scratch // gave it, so drop what it owns first. Idempotent, and a no-op when the // role is absent (the usual case on a first run). let drop_role = format!( "DO $$ BEGIN IF EXISTS (SELECT 1 FROM pg_roles WHERE rolname = '{role}') THEN EXECUTE 'DROP OWNED BY {role}'; EXECUTE 'DROP ROLE {role}'; END IF; END $$;" ); let pool = PgPoolOptions::new() .max_connections(1) .connect(&url) .await .unwrap(); pool.execute(sqlx::raw_sql(sqlx::AssertSqlSafe(drop_role.clone()))) .await .unwrap(); pool.close().await; reset_scratch(&url, role) .await .expect("reset_scratch creates the owner role"); let pool = PgPoolOptions::new() .max_connections(1) .connect(&url) .await .unwrap(); let (exists, can_login): (bool, bool) = sqlx::query_as("SELECT true, rolcanlogin FROM pg_roles WHERE rolname = $1") .bind(role) .fetch_one(&pool) .await .expect("owner role exists after reset"); assert!(exists); assert!( !can_login, "the owner role is an owner only, never a login identity" ); // The restore's owner-scoped DDL needs CREATE on public in the role's // own right (PG15+ dropped the implicit grant). let (has_create,): (bool,) = sqlx::query_as("SELECT pg_catalog.has_schema_privilege($1, 'public', 'CREATE')") .bind(role) .fetch_one(&pool) .await .unwrap(); assert!(has_create, "owner role must be able to create in public"); // Idempotent: a second reset must not error on the now-existing role. pool.close().await; reset_scratch(&url, role) .await .expect("reset_scratch is idempotent"); let pool = PgPoolOptions::new() .max_connections(1) .connect(&url) .await .unwrap(); pool.execute(sqlx::raw_sql(sqlx::AssertSqlSafe(drop_role))) .await .unwrap(); pool.close().await; } /// The preflight must pass against a privileged scratch connection. Guards /// the catalog query itself: a wrong column or a `current_user` that matches /// no `pg_roles` row would make `fetch_one` error (or, worse, a silently /// swapped pair would invert the check) and brick startup for everyone. /// `SANDO_TEST_PG_URL` is expected to be a superuser connection, as the /// gates require. #[tokio::test] async fn preflight_passes_on_a_privileged_scratch_connection() { let Ok(url) = std::env::var("SANDO_TEST_PG_URL") else { eprintln!("skipping: SANDO_TEST_PG_URL not set"); return; }; preflight_scratch_privileges(&url) .await .expect("a superuser scratch connection must satisfy the preflight"); } /// CF4: the restore pipeline must carry `ON_ERROR_STOP=1` (so psql fails on /// a bad statement instead of exiting 0 on a partial restore) and, for a /// gzip source, `set -o pipefail` (so a `gunzip` failure on a truncated /// archive isn't masked by psql's exit). Pure string check — no postgres. #[test] fn restore_shell_has_error_stop_and_pipefail() { let gz = restore_shell("postgres:///scratch", "/srv/sando/backups/latest.sql.gz"); assert!(gz.contains("ON_ERROR_STOP=1"), "gz: {gz}"); assert!(gz.contains("set -o pipefail"), "gz: {gz}"); assert!(gz.contains("gunzip -c"), "gz: {gz}"); let plain = restore_shell("postgres:///scratch", "/srv/sando/backups/dump.sql"); assert!(plain.contains("ON_ERROR_STOP=1"), "plain: {plain}"); // No pipeline for a plain .sql, so pipefail is unnecessary there. assert!(!plain.contains("gunzip"), "plain: {plain}"); // The db url is single-quote escaped in both forms. assert!(plain.contains("'postgres:///scratch'"), "plain: {plain}"); } #[test] fn split_pg_password_extracts_and_sanitizes() { // Password lifted out of the URL; the sanitized form keeps user/host/db. let (url, pw) = split_pg_password("postgres://sando:s3cret@db.host:5432/scratch"); assert_eq!(url, "postgres://sando@db.host:5432/scratch"); assert_eq!(pw.as_deref(), Some("s3cret")); // Percent-encoded password is decoded for PGPASSWORD. let (url, pw) = split_pg_password("postgresql://u:p%40ss%2Fword@h/d"); assert_eq!(url, "postgresql://u@h/d"); assert_eq!(pw.as_deref(), Some("p@ss/word")); } #[test] fn split_pg_password_noop_without_password() { // No userinfo password -> unchanged, None. (A ':' after the '@', e.g. a // port, must not be mistaken for the password delimiter.) assert_eq!( split_pg_password("postgres:///scratch"), ("postgres:///scratch".to_string(), None), ); assert_eq!( split_pg_password("postgres://sando@db.host:5432/scratch"), ("postgres://sando@db.host:5432/scratch".to_string(), None), ); } #[test] fn percent_decode_handles_escapes_and_malformed() { assert_eq!(percent_decode("plain"), "plain"); assert_eq!(percent_decode("a%2Fb"), "a/b"); // A malformed trailing escape is left literal, not dropped. assert_eq!(percent_decode("ab%2"), "ab%2"); assert_eq!(percent_decode("ab%zz"), "ab%zz"); } #[test] fn pg_url_with_dbname_rewrites_the_database() { // user:pass@host:port/db?query — swap db, keep everything else. assert_eq!( pg_url_with_dbname( "postgres://sando:pw@db.host:5432/sando_scratch?sslmode=require", "postgres" ), "postgres://sando:pw@db.host:5432/postgres?sslmode=require", ); // Socket form: the query carries `host=/var/run/postgresql` and must survive. assert_eq!( pg_url_with_dbname( "postgres:///sando_scratch?host=/var/run/postgresql", "sando_code_smoke_0_9_6" ), "postgres:///sando_code_smoke_0_9_6?host=/var/run/postgresql", ); // Plain host/db, no query. assert_eq!( pg_url_with_dbname("postgres://localhost/scratch", "postgres"), "postgres://localhost/scratch".replace("scratch", "postgres"), ); // No authority, no query (loopback socket, default db path). assert_eq!( pg_url_with_dbname("postgres:///scratch", "postgres"), "postgres:///postgres", ); } #[test] fn bytes_contain_matches_listening_in_log_output() { // JSON release log carries the message field verbatim. assert!(bytes_contain( br#"{"timestamp":"...","level":"INFO","fields":{"message":"listening","addr":"127.0.0.1:18182"}}"#, b"listening", )); // Human-format dev log. assert!(bytes_contain( b"2026-07-17 INFO makenotwork: listening addr=127.0.0.1:18182", b"listening" )); assert!(!bytes_contain( b"migrations complete; seeding catalog", b"listening" )); assert!(!bytes_contain(b"", b"listening")); } #[tokio::test] async fn code_smoke_env_supplies_every_mandatory_server_var() { // code_smoke is the only gate that reaches the server's Config::from_env // (boot_smoke and the docs check short-circuit before it), so anything // that function requires has to be answered here. CDN_BASE_URL became // mandatory 13 days after this env was written and went unnoticed until // the gate first ran on the host; this test is what makes the next one // fail here instead of in a promote. let ctx = resolving_ctx("/w/abc", &[]); let mut cmd = tokio::process::Command::new("true"); code_smoke_env(&mut cmd, &ctx, "postgres:///throwaway"); let set: std::collections::HashMap = cmd .as_std() .get_envs() .filter_map(|(k, v)| Some((k.to_str()?.to_string(), v?.to_str()?.to_string()))) .collect(); for key in [ "DATABASE_URL", "HOST", "PORT", "HOST_URL", "CDN_BASE_URL", "SIGNING_SECRET", ] { assert!(set.contains_key(key), "code_smoke_env must set {key}"); assert!(!set[key].is_empty(), "{key} must not be empty"); } // Loopback, so Config::from_env's is_production branch stays false and // the gate never trips MissingPublicBucket for want of an S3 bucket. assert!(set["HOST_URL"].starts_with("http://localhost")); // Not an IP literal: the server derives its WebAuthn rp_id from this // host, and WebauthnBuilder rejects an origin whose Url::domain() is // None, which is every IP address. An IP here cannot boot the server. assert!( url_host_is_a_domain(&set["HOST_URL"]), "HOST_URL host must be a domain, not an IP literal: {}", set["HOST_URL"], ); assert_eq!(set["HOST"], "127.0.0.1"); // The signing secret has to clear the server's 32-char floor, or the // gate fails with WeakSigningSecret instead of testing anything. assert!(set["SIGNING_SECRET"].len() >= 32); } #[test] fn code_smoke_db_name_sanitizes_and_caps() { assert_eq!( code_smoke_db_name(&"0.9.6".parse().unwrap()), "sando_code_smoke_0_9_6" ); // Pre-release/build metadata folds to underscores; result stays [a-z0-9_]. let n = code_smoke_db_name(&"1.0.0-rc.1+build".parse().unwrap()); assert_eq!(n, "sando_code_smoke_1_0_0_rc_1_build"); assert!( n.bytes() .all(|b| b.is_ascii_lowercase() || b.is_ascii_digit() || b == b'_') ); assert!(n.len() <= 63); } /// code_smoke is Blocked (not Failed) when the daemon has no scratch_db_url: /// there's no cluster to create the throwaway DB in, and that's an operator /// precondition, rendered yellow — the same shape as migration_dry_run. #[tokio::test] async fn code_smoke_blocks_without_scratch_db_url() { let pool = SqlitePoolOptions::new() .max_connections(1) .connect("sqlite::memory:") .await .unwrap(); crate::db::migrate(&pool).await.unwrap(); sqlx::query("INSERT INTO tiers (name, ord, provisioned, canary) VALUES ('host', 0, 1, 'sequential')") .execute(&pool).await.unwrap(); sqlx::query("INSERT INTO tier_state (tier) VALUES ('host')") .execute(&pool) .await .unwrap(); sqlx::query("INSERT INTO versions (version, git_sha, built_at, artifact_path) VALUES ('0.1.0', 'abc1234', '2026-01-01T00:00:00Z', '/tmp/x')") .execute(&pool).await.unwrap(); let cfg = std::sync::Arc::new(crate::config::AppConfig::for_tests()); // scratch_db_url: None let ctx = GateCtx { pool: pool.clone(), cfg, tier: TierId::new("host"), version: "0.1.0".parse().unwrap(), worktree: Some(std::path::PathBuf::from("/tmp/unused")), bundle: None, events: events::channel(), nodes: Vec::new(), build_id: None, aux_dirs: HashMap::new(), }; let out = run(&ctx, &Gate::CodeSmoke).await.unwrap(); assert_eq!(out.status_str(), "blocked"); assert!(!out.is_passed()); let row: (Option, Option) = sqlx::query_as("SELECT status, outcome_json FROM gate_runs ORDER BY id DESC LIMIT 1") .fetch_one(&pool) .await .unwrap(); assert_eq!(row.0.as_deref(), Some("blocked")); let json: serde_json::Value = serde_json::from_str(row.1.as_deref().unwrap()).unwrap(); assert_eq!(json["status"]["blocker"]["kind"], "scratch_db_url_unset"); } /// Sanity: applying MNW migrations from a *non-existent* dir errors, /// rather than silently no-op'ing. Cheap pure check, no postgres needed /// (the sqlx::Migrator::new constructor itself reads the dir). #[tokio::test] async fn run_migrator_errors_on_missing_dir() { // The first thing run_migrator does is `Migrator::new(dir)`, which // needs a real dir to read migration files from. let res = run_migrator( "postgres:///does-not-matter", std::path::Path::new("/nonexistent/sando-test-migrations"), ) .await; assert!(res.is_err()); } }