//! Tests for [`super`]. use super::promotion::{ Evidence, PromotedBuild, RollbackReport, rollback_deployed_nodes, unsatisfied_gates, }; use super::*; use crate::config::AppConfig; use crate::topology::{BackupConfig, CanaryPolicy, Gate, Node, RepoConfig, Tier, Topology}; use async_trait::async_trait; use axum::body::Body; use axum::http::{Request, StatusCode}; use http_body_util::BodyExt; use ops_exec::{CapabilitySet, Executor, LogSink, RunOutput, Step, SyncOpts}; use sqlx::SqlitePool; use sqlx::sqlite::SqlitePoolOptions; use std::collections::BTreeMap; use std::os::unix::process::ExitStatusExt; use std::path::PathBuf; use std::sync::{Arc, Mutex as StdMutex}; use tower::ServiceExt; async fn fresh_pool() -> SqlitePool { let pool = SqlitePoolOptions::new() .max_connections(1) .connect("sqlite::memory:") .await .unwrap(); sqlx::migrate!("./migrations").run(&pool).await.unwrap(); pool } /// Two-tier topology used by the route tests: mm (provisioned, no nodes) /// → a (provisioned, one local node). Mirrors the production shape /// without involving real ssh / postgres. fn test_topo() -> Topology { Topology { repo: Some(RepoConfig { bare_path: "/tmp/test.git".into(), branch: "main".into(), upstream: None, }), backup: vec![BackupConfig { name: "server".into(), source: "file:///tmp/test-backup.sql".into(), local_path: "/tmp/local-backup.sql".into(), }], tiers: vec![ Tier { public_url: None, name: "host".into(), provisioned: true, gates: vec![], canary: CanaryPolicy::Sequential, nodes: vec![], }, Tier { public_url: None, name: "a".into(), provisioned: true, gates: vec![Gate::BootSmoke], canary: CanaryPolicy::Sequential, nodes: vec![Node { platform: None, base_image: None, libc: None, name: "a-local".into(), ssh_target: "local".into(), release_root: "/tmp/a-node".into(), service_name: "makenotwork.service".into(), health_url: None, config_check_env_file: None, actuate: crate::topology::default_actuate(), observe: crate::topology::default_observe(), companions: Vec::new(), }], }, ], aux_repos: Vec::new(), } } fn test_cfg() -> AppConfig { AppConfig { page_smoke_cmd: None, platform: None, code_smoke_env: BTreeMap::default(), id: crate::domain::AppId::default(), topology_path: PathBuf::from("/tmp/test-sando.toml"), build_host: Some("test-host".into()), workdir: PathBuf::from("/tmp/sando-work"), release_root: PathBuf::from("/tmp/sando-releases"), scratch_db_url: None, scratch_owner_role: "makenotwork".into(), boot_smoke_port: 18181, code_smoke_port: 18182, bin_names: vec!["makenotwork".into()], logs_root: PathBuf::from("/tmp/sando-logs"), release_contents: vec![], cargo_target_dir: None, gate_timeout_secs: 2400, companions: Vec::new(), test_targets: vec![crate::config::TestTarget { dir: PathBuf::from("server"), aux_repo: None, features: vec!["fast-tests".into()], all_features: false, scratch_db: true, }], migration_checks: vec![], frontend_builds: vec![], backup_max_age_hours: 48, } } /// Two products mounted on one daemon address different state. /// /// The mount-per-product shape is what makes this true: each router carries /// its own product's config, so `/apps/pom/state` cannot answer from MNW's /// tiers even if a handler forgets the product exists. The root mount keeps /// meaning the default product, which is what the runbook and the TUI call. #[tokio::test] async fn each_app_is_addressable_and_the_root_stays_the_default() { let pool = fresh_pool().await; // MNW ships host + a; pom ships one tier of its own, named differently // so the response says which product answered. for (app, tiers) in [("mnw", vec!["host", "a"]), ("pom", vec!["pom-host"])] { for (i, name) in tiers.iter().enumerate() { sqlx::query("INSERT INTO tiers (app, name, ord, provisioned) VALUES (?, ?, ?, 1)") .bind(app) .bind(name) .bind(i as i64) .execute(&pool) .await .unwrap(); sqlx::query("INSERT INTO tier_state (app, tier) VALUES (?, ?)") .bind(app) .bind(name) .execute(&pool) .await .unwrap(); } } let mnw_topo = Arc::new(test_topo()); let mut pom_topo = test_topo(); pom_topo.tiers = vec![crate::topology::Tier { public_url: None, name: "pom-host".into(), provisioned: true, gates: vec![], canary: crate::topology::CanaryPolicy::Sequential, nodes: vec![], }]; let pom_topo = Arc::new(pom_topo); let mnw_cfg = Arc::new(test_cfg()); let mut pom = test_cfg(); pom.id = crate::domain::AppId::new("pom"); let pom_cfg = Arc::new(pom); let mut apps = crate::state::AppMap::new(); for (id, cfg, topo) in [ (mnw_cfg.id.clone(), mnw_cfg.clone(), mnw_topo.clone()), (pom_cfg.id.clone(), pom_cfg.clone(), pom_topo.clone()), ] { let executors = Arc::new(crate::state::build_executors(&topo)); apps.insert( id, Arc::new(crate::state::App { cfg, topo, executors, }), ); } let state = AppState { pool, apps: Arc::new(apps), default_app: mnw_cfg.id.clone(), topo: mnw_topo, cfg: mnw_cfg, active_build: Arc::new(tokio::sync::Mutex::new(None)), deploy_lock: Arc::new(tokio::sync::Mutex::new(())), events: crate::events::channel(), executors: Arc::new(std::collections::HashMap::new()), api_token: None, }; let get = async |uri: &str| -> String { let resp = router_for_apps(state.clone()) .oneshot(Request::builder().uri(uri).body(Body::empty()).unwrap()) .await .unwrap(); assert_eq!(resp.status(), StatusCode::OK, "GET {uri}"); body_string(resp).await }; // The root is the default product. let root = get("/state").await; assert!(root.contains("\"host\""), "root /state: {root}"); assert!(!root.contains("pom-host"), "root must not show pom: {root}"); // Each product answers under its own mount. let mnw = get("/apps/mnw/state").await; assert_eq!(mnw, root, "the default mount and the root are one product"); let pom = get("/apps/pom/state").await; assert!(pom.contains("pom-host"), "/apps/pom/state: {pom}"); assert!( !pom.contains("\"host\""), "pom must not see mnw's tiers: {pom}" ); // And the index says what is mounted. let index = get("/apps").await; assert!( index.contains("\"mnw\"") && index.contains("\"pom\""), "{index}" ); assert!(index.contains("\"default_app\":\"mnw\""), "{index}"); // An unconfigured product is not a route. let resp = router_for_apps(state.clone()) .oneshot( Request::builder() .uri("/apps/nope/state") .body(Body::empty()) .unwrap(), ) .await .unwrap(); assert_eq!(resp.status(), StatusCode::NOT_FOUND); } async fn test_state() -> AppState { let pool = fresh_pool().await; // Seed tier rows so FKs on tier_state / gate_runs are satisfied. for (i, name) in ["host", "a"].iter().enumerate() { sqlx::query( "INSERT INTO tiers (name, ord, provisioned, canary) VALUES (?, ?, 1, 'sequential')", ) .bind(name) .bind(i as i64) .execute(&pool) .await .unwrap(); sqlx::query("INSERT INTO tier_state (tier) VALUES (?)") .bind(name) .execute(&pool) .await .unwrap(); } // Don't call install_recorder in tests — it touches a process-global // and conflicts when tests run in parallel. let topo = test_topo(); let executors = Arc::new(crate::state::build_executors(&topo)); let topo = Arc::new(topo); let cfg = Arc::new(test_cfg()); let (apps, default_app) = crate::state::one_app(cfg.clone(), topo.clone(), executors.clone()); AppState { pool, apps, default_app, topo, cfg, active_build: Arc::new(tokio::sync::Mutex::new(None)), deploy_lock: Arc::new(tokio::sync::Mutex::new(())), events: crate::events::channel(), executors, api_token: None, } } async fn body_string(resp: axum::response::Response) -> String { let bytes = resp.into_body().collect().await.unwrap().to_bytes(); String::from_utf8(bytes.to_vec()).unwrap() } /// Insert the FK prerequisites for inserting gate_runs/tier_state rows. async fn seed(pool: &SqlitePool, tier: &str, version: &str) { sqlx::query("INSERT INTO tiers (name, ord, provisioned, canary) VALUES (?, 0, 1, 'sequential') ON CONFLICT DO NOTHING") .bind(tier).execute(pool).await.unwrap(); sqlx::query("INSERT INTO versions (version, git_sha, built_at, artifact_path) VALUES (?, 'sha', datetime('now'), '/tmp/x') ON CONFLICT DO NOTHING") .bind(version).execute(pool).await.unwrap(); sqlx::query( "INSERT INTO tier_state (tier, current_version) VALUES (?, NULL) ON CONFLICT DO NOTHING", ) .bind(tier) .execute(pool) .await .unwrap(); } async fn insert_gate(pool: &SqlitePool, tier: &str, version: &str, kind: &str, passed: i64) { let status = if passed == 1 { "passed" } else { "failed" }; sqlx::query( "INSERT INTO gate_runs (version, tier, gate_kind, started_at, finished_at, status) \ VALUES (?, ?, ?, datetime('now'), datetime('now'), ?)", ) .bind(version) .bind(tier) .bind(kind) .bind(status) .execute(pool) .await .unwrap(); } /// `/state` says when a tier's artifact is gone, and stays quiet when it is /// not. /// /// The gap this closes: `versions`/`build_runs` keep naming a path long /// after gc removed the bytes, and nothing reconciled the two. Both times it /// happened, the tier read green until an rsync failed mid-promote. #[tokio::test] async fn state_reports_a_referenced_artifact_whose_bytes_are_gone() { let tmp = tempfile::tempdir().unwrap(); let present = tmp.path().join("releases").join("1111111111111111"); tokio::fs::create_dir_all(&present).await.unwrap(); let state = test_state().await; let bin = state.cfg.primary_bin().to_string(); tokio::fs::write(present.join(&bin), b"bin").await.unwrap(); seed(&state.pool, "a", "0.11.20").await; let build = seed_build( &state.pool, "2a53c900", "0.11.20", present.to_string_lossy().as_ref(), ) .await; sqlx::query("UPDATE tier_state SET current_version = ?, current_build_id = ? WHERE tier = 'a'") .bind("0.11.20") .bind(build) .execute(&state.pool) .await .unwrap(); let tier_a = |v: &StateView| { v.tiers .iter() .find(|t| t.name == "a") .expect("tier a") .missing_artifact .clone() }; assert_eq!( tier_a(&state_view(&state).await.unwrap()), None, "the artifact is on disk; nothing to report" ); // gc takes it, as it did twice in production. Nothing in the database // changes, which is the whole defect. tokio::fs::remove_dir_all(&present).await.unwrap(); let said = tier_a(&state_view(&state).await.unwrap()).expect("a missing-artifact report"); assert!(said.contains("0.11.20"), "{said}"); assert!(said.contains("current"), "{said}"); } /// A rebuild at an unchanged version must not move what a tier reports. /// /// Gate rows keyed on (tier, version) let two runs of one version interleave, /// with each gate showing whichever had written it last: two reads of the /// same tier a minute apart disagree about whether `hardening_test` passed or /// never ran. #[tokio::test] async fn state_reports_the_build_the_tier_is_running() { let state = test_state().await; seed(&state.pool, "a", "0.11.20").await; let first = seed_build(&state.pool, "2a53c900", "0.11.20", "/rel/1111111111111111").await; let second = seed_build(&state.pool, "adf56cd9", "0.11.20", "/rel/2222222222222222").await; // The tier is running the first build, and it went green there. insert_gate_build(&state.pool, "a", "0.11.20", "hardening_test", 1, first).await; sqlx::query("UPDATE tier_state SET current_version = ?, current_build_id = ? WHERE tier = 'a'") .bind("0.11.20") .bind(first) .execute(&state.pool) .await .unwrap(); let before = state_view(&state).await.unwrap(); let gates_of = |v: &StateView| { v.tiers .iter() .find(|t| t.name == "a") .unwrap() .gates .iter() .map(|g| (g.kind.clone(), g.status.clone())) .collect::>() }; assert_eq!( gates_of(&before), vec![("hardening_test".to_string(), Some("passed".to_string()))], ); // A retry of the same version fails the same gate. The tier still runs // the first build, so what it reports is unchanged. insert_gate_build(&state.pool, "a", "0.11.20", "hardening_test", 0, second).await; assert_eq!( gates_of(&state_view(&state).await.unwrap()), gates_of(&before), "a sibling rebuild rewrote a tier it was never deployed to", ); } // ---- unsatisfied_gates ---- fn tid(s: &str) -> crate::domain::TierId { crate::domain::TierId::new(s) } #[tokio::test] async fn unsatisfied_gates_empty_when_no_configured_gates() { // A tier that configures no gates has nothing to satisfy. let pool = fresh_pool().await; seed(&pool, "host", "0.8.12").await; let pending = unsatisfied_gates( &pool, &crate::domain::AppId::default(), &tid("host"), &[], &Evidence { version: "0.8.12", builds: &[], tier_build: None, }, false, ) .await .unwrap(); assert_eq!(pending, Vec::::new()); } #[tokio::test] async fn unsatisfied_gates_flags_configured_gate_that_never_ran() { // THE CF1 FIX: a configured gate with no gate_runs row is unsatisfied // (fail closed), NOT silently treated as green. Before this, an A tier // whose boot_smoke never executed exposed zero rows and waved promotion // straight through to prod. let pool = fresh_pool().await; seed(&pool, "a", "0.8.12").await; let pending = unsatisfied_gates( &pool, &crate::domain::AppId::default(), &tid("a"), &[Gate::BootSmoke], &Evidence { version: "0.8.12", builds: &[], tier_build: None, }, false, ) .await .unwrap(); assert_eq!(pending, vec!["boot_smoke".to_string()]); } #[tokio::test] async fn unsatisfied_gates_flags_failed_kind() { let pool = fresh_pool().await; seed(&pool, "host", "0.8.12").await; insert_gate(&pool, "host", "0.8.12", "cargo_test", 0).await; insert_gate(&pool, "host", "0.8.12", "boot_smoke", 1).await; let pending = unsatisfied_gates( &pool, &crate::domain::AppId::default(), &tid("host"), &[Gate::CargoTest, Gate::BootSmoke], &Evidence { version: "0.8.12", builds: &[], tier_build: None, }, false, ) .await .unwrap(); assert_eq!(pending, vec!["cargo_test".to_string()]); } #[tokio::test] async fn unsatisfied_gates_latest_row_wins() { // Two runs of the same gate; only the latest counts. A flap from // red to green should clear the pending entry. let pool = fresh_pool().await; seed(&pool, "host", "0.8.12").await; insert_gate(&pool, "host", "0.8.12", "cargo_test", 0).await; insert_gate(&pool, "host", "0.8.12", "cargo_test", 1).await; let pending = unsatisfied_gates( &pool, &crate::domain::AppId::default(), &tid("host"), &[Gate::CargoTest], &Evidence { version: "0.8.12", builds: &[], tier_build: None, }, false, ) .await .unwrap(); assert!(pending.is_empty()); } async fn insert_confirm( pool: &SqlitePool, tier: &str, version: &str, at: chrono::DateTime, ) { sqlx::query( "INSERT INTO gate_runs (version, tier, gate_kind, started_at, finished_at, status) \ VALUES (?, ?, 'manual_confirm', ?, ?, 'passed')", ) .bind(version) .bind(tier) .bind(at.to_rfc3339()) .bind(at.to_rfc3339()) .execute(pool) .await .unwrap(); } #[tokio::test] async fn unsatisfied_gates_manual_confirm_requires_fresh_confirmation() { // A confirmation only satisfies the gate if it post-dates the version's // current landing on the tier (burn_in_started_at). A stale confirm left // over from before a rollback + rollback-forward must NOT wave it through. let pool = fresh_pool().await; seed(&pool, "a", "0.8.12").await; let landed = chrono::Utc::now(); sqlx::query("UPDATE tier_state SET burn_in_started_at = ? WHERE tier = 'a'") .bind(landed.to_rfc3339()) .execute(&pool) .await .unwrap(); // Stale confirmation (recorded before this landing) -> unsatisfied. insert_confirm(&pool, "a", "0.8.12", landed - chrono::Duration::hours(1)).await; let pending = unsatisfied_gates( &pool, &crate::domain::AppId::default(), &tid("a"), &[Gate::ManualConfirm], &Evidence { version: "0.8.12", builds: &[], tier_build: None, }, false, ) .await .unwrap(); assert_eq!( pending, vec!["manual_confirm".to_string()], "stale confirm must not satisfy" ); // Fresh confirmation (after this landing) -> satisfied. insert_confirm(&pool, "a", "0.8.12", landed + chrono::Duration::minutes(5)).await; let pending = unsatisfied_gates( &pool, &crate::domain::AppId::default(), &tid("a"), &[Gate::ManualConfirm], &Evidence { version: "0.8.12", builds: &[], tier_build: None, }, false, ) .await .unwrap(); assert!(pending.is_empty(), "fresh confirm satisfies"); } #[tokio::test] async fn unsatisfied_gates_manual_confirm_fails_closed_without_baseline() { // No landing clock (burn_in_started_at NULL) -> a passed confirm row is // not provably fresh, so fail closed and require a new confirmation. let pool = fresh_pool().await; seed(&pool, "a", "0.8.12").await; // leaves burn_in_started_at NULL insert_confirm(&pool, "a", "0.8.12", chrono::Utc::now()).await; let pending = unsatisfied_gates( &pool, &crate::domain::AppId::default(), &tid("a"), &[Gate::ManualConfirm], &Evidence { version: "0.8.12", builds: &[], tier_build: None, }, false, ) .await .unwrap(); assert_eq!( pending, vec!["manual_confirm".to_string()], "no baseline -> fail closed" ); } #[tokio::test] async fn unsatisfied_gates_hotfix_skips_only_burn_in() { // burn_in is evaluated live (no clock started -> not elapsed); cargo_test // has a failing row. Normal: both unsatisfied, in configured order. // hotfix: burn_in suppressed, cargo_test still flagged. Lock the semantic // so a future change doesn't widen the hotfix bypass. let pool = fresh_pool().await; seed(&pool, "a", "0.8.12").await; insert_gate(&pool, "a", "0.8.12", "cargo_test", 0).await; let gates = [Gate::BurnIn { hours: 48 }, Gate::CargoTest]; let normal = unsatisfied_gates( &pool, &crate::domain::AppId::default(), &tid("a"), &gates, &Evidence { version: "0.8.12", builds: &[], tier_build: None, }, false, ) .await .unwrap(); assert_eq!( normal, vec!["burn_in".to_string(), "cargo_test".to_string()] ); let with_hotfix = unsatisfied_gates( &pool, &crate::domain::AppId::default(), &tid("a"), &gates, &Evidence { version: "0.8.12", builds: &[], tier_build: None, }, true, ) .await .unwrap(); assert_eq!(with_hotfix, vec!["cargo_test".to_string()]); } #[tokio::test] async fn unsatisfied_gates_burn_in_passes_when_window_elapsed() { // A burn-in clock started far enough in the past satisfies the gate // live — no gate_runs row needed. let pool = fresh_pool().await; seed(&pool, "a", "0.8.12").await; sqlx::query("UPDATE tier_state SET burn_in_started_at = ? WHERE tier = 'a'") .bind((chrono::Utc::now() - chrono::Duration::hours(50)).to_rfc3339()) .execute(&pool) .await .unwrap(); let pending = unsatisfied_gates( &pool, &crate::domain::AppId::default(), &tid("a"), &[Gate::BurnIn { hours: 48 }], &Evidence { version: "0.8.12", builds: &[], tier_build: None, }, false, ) .await .unwrap(); assert!(pending.is_empty(), "50h elapsed satisfies a 48h burn-in"); } #[tokio::test] async fn unsatisfied_gates_ignores_other_tiers_and_versions() { let pool = fresh_pool().await; seed(&pool, "host", "0.8.12").await; seed(&pool, "host", "0.8.11").await; seed(&pool, "a", "0.8.12").await; // Mark host/0.8.12 cargo_test failing, but unrelated tiers/versions // shouldn't pollute the query. insert_gate(&pool, "host", "0.8.12", "cargo_test", 0).await; insert_gate(&pool, "a", "0.8.12", "cargo_test", 0).await; insert_gate(&pool, "host", "0.8.11", "cargo_test", 0).await; let pending = unsatisfied_gates( &pool, &crate::domain::AppId::default(), &tid("host"), &[Gate::CargoTest], &Evidence { version: "0.8.12", builds: &[], tier_build: None, }, false, ) .await .unwrap(); assert_eq!(pending, vec!["cargo_test".to_string()]); } #[tokio::test] async fn unsatisfied_gates_null_status_is_treated_as_failing() { // An in-flight gate (started_at set, finished_at + status NULL) // should NOT be treated as green. Otherwise a race could promote // before the gate concludes. let pool = fresh_pool().await; seed(&pool, "host", "0.8.12").await; sqlx::query( "INSERT INTO gate_runs (version, tier, gate_kind, started_at) \ VALUES ('0.8.12', 'host', 'cargo_test', datetime('now'))", ) .execute(&pool) .await .unwrap(); let pending = unsatisfied_gates( &pool, &crate::domain::AppId::default(), &tid("host"), &[Gate::CargoTest], &Evidence { version: "0.8.12", builds: &[], tier_build: None, }, false, ) .await .unwrap(); assert_eq!(pending, vec!["cargo_test".to_string()]); } // ---- /confirm/{tier} ---- #[tokio::test] async fn confirm_rejects_when_tier_has_no_current_version() { // tier_state.a.current_version is NULL by default. /confirm has // nothing to confirm against → GateBlocked (400). let state = test_state().await; let app = router(state.clone()); let resp = app .oneshot( Request::builder() .method("POST") .uri("/confirm/a") .body(Body::empty()) .unwrap(), ) .await .unwrap(); assert_eq!(resp.status(), StatusCode::CONFLICT); let body = body_string(resp).await; assert!(body.contains("no current_version"), "got: {body}"); } #[tokio::test] async fn confirm_accepts_when_current_version_set_and_inserts_row() { let state = test_state().await; // Seed a version + advance tier a's state to it. sqlx::query("INSERT INTO versions (version, git_sha, built_at, artifact_path) VALUES ('0.8.12','sha',datetime('now'),'/tmp/x')") .execute(&state.pool).await.unwrap(); sqlx::query("UPDATE tier_state SET current_version = '0.8.12' WHERE tier = 'a'") .execute(&state.pool) .await .unwrap(); let app = router(state.clone()); let resp = app .oneshot( Request::builder() .method("POST") .uri("/confirm/a") .body(Body::empty()) .unwrap(), ) .await .unwrap(); assert_eq!(resp.status(), StatusCode::OK); let body = body_string(resp).await; assert!(body.contains("\"tier\":\"a\"")); assert!(body.contains("\"version\":\"0.8.12\"")); // A passing gate_runs row was inserted. let count: (i64,) = sqlx::query_as( "SELECT COUNT(*) FROM gate_runs WHERE tier='a' AND gate_kind='manual_confirm' AND status='passed'", ) .fetch_one(&state.pool) .await .unwrap(); assert_eq!(count.0, 1); } #[tokio::test] async fn confirm_404s_for_unknown_tier() { let state = test_state().await; let app = router(state); let resp = app .oneshot( Request::builder() .method("POST") .uri("/confirm/zzzz") .body(Body::empty()) .unwrap(), ) .await .unwrap(); assert_eq!(resp.status(), StatusCode::NOT_FOUND); } #[tokio::test] async fn get_run_404s_for_unknown_id() { let state = test_state().await; let app = router(state); let resp = app .oneshot( Request::builder() .uri("/runs/999") .body(Body::empty()) .unwrap(), ) .await .unwrap(); assert_eq!(resp.status(), StatusCode::NOT_FOUND); } #[tokio::test] async fn get_run_returns_view_with_gates() { let state = test_state().await; // A run that reached version 0.10.2 and ran two host gates (one red). let run_id = crate::runs::create(&state.pool, &state.cfg.id, "abc1234def") .await .unwrap(); let ver: crate::domain::Version = "0.10.2".parse().unwrap(); seed(&state.pool, "host", "0.10.2").await; crate::runs::set_version(&state.pool, run_id, &ver) .await .unwrap(); // Keyed on the run, not on the version: these are the rows that carry // this run's build id, and a sibling rebuild of 0.10.2 writing its own // rows must not change what this run reports. insert_gate_build(&state.pool, "host", "0.10.2", "cargo_test", 0, run_id.0).await; insert_gate_build(&state.pool, "host", "0.10.2", "boot_smoke", 1, run_id.0).await; let app = router(state); let resp = app .oneshot( Request::builder() .uri(format!("/runs/{}", run_id.0)) .body(Body::empty()) .unwrap(), ) .await .unwrap(); assert_eq!(resp.status(), StatusCode::OK); let v: serde_json::Value = serde_json::from_str(&body_string(resp).await).unwrap(); assert_eq!(v["run_id"], run_id.0); assert_eq!(v["sha"], "abc1234def"); assert_eq!(v["version"], "0.10.2"); assert_eq!(v["result"], "building"); // Both host gates surface, latest-per-kind, alphabetized by kind. assert_eq!(v["gates"].as_array().unwrap().len(), 2); assert_eq!(v["gates"][0]["kind"], "boot_smoke"); assert_eq!(v["gates"][0]["status"], "passed"); assert_eq!(v["gates"][1]["kind"], "cargo_test"); assert_eq!(v["gates"][1]["status"], "failed"); } #[tokio::test] async fn get_run_ignores_a_sibling_rebuild_of_the_same_version() { // A rebuild at an unchanged version is the normal way to retry a red // build. Keyed on (tier, version), the two runs' rows interleaved and // each gate reported whichever run wrote it last — a reader of /runs/{id} // could watch a passed gate become "not run" a minute later with nothing // touched. Runs 60-62 of mnw-server 0.11.20, 2026-08-19. let state = test_state().await; seed(&state.pool, "host", "0.11.20").await; let ver: crate::domain::Version = "0.11.20".parse().unwrap(); let first = crate::runs::create(&state.pool, &state.cfg.id, "2a53c900") .await .unwrap(); crate::runs::set_version(&state.pool, first, &ver) .await .unwrap(); insert_gate_build(&state.pool, "host", "0.11.20", "cargo_deny", 0, first.0).await; let second = crate::runs::create(&state.pool, &state.cfg.id, "adf56cd9") .await .unwrap(); crate::runs::set_version(&state.pool, second, &ver) .await .unwrap(); insert_gate_build(&state.pool, "host", "0.11.20", "cargo_deny", 1, second.0).await; // The later run passing does not turn the earlier run green, and the // earlier run failing does not follow the later one. for (run, want) in [(first, "failed"), (second, "passed")] { let app = router(state.clone()); let resp = app .oneshot( Request::builder() .uri(format!("/runs/{}", run.0)) .body(Body::empty()) .unwrap(), ) .await .unwrap(); assert_eq!(resp.status(), StatusCode::OK); let v: serde_json::Value = serde_json::from_str(&body_string(resp).await).unwrap(); assert_eq!(v["gates"].as_array().unwrap().len(), 1, "run {}", run.0); assert_eq!(v["gates"][0]["kind"], "cargo_deny"); assert_eq!(v["gates"][0]["status"], want, "run {}", run.0); } } #[tokio::test] async fn get_run_wait_returns_immediately_when_settled() { let state = test_state().await; let run_id = crate::runs::create(&state.pool, &state.cfg.id, "abc1234def") .await .unwrap(); crate::runs::mark_passed(&state.pool, run_id).await.unwrap(); let app = router(state); // Generous timeout, but an already-settled run must not wait for it. let resp = app .oneshot( Request::builder() .uri(format!("/runs/{}/wait?timeout_ms=60000", run_id.0)) .body(Body::empty()) .unwrap(), ) .await .unwrap(); assert_eq!(resp.status(), StatusCode::OK); let v: serde_json::Value = serde_json::from_str(&body_string(resp).await).unwrap(); assert_eq!(v["result"], "passed"); } #[tokio::test] async fn get_run_wait_returns_building_at_timeout() { let state = test_state().await; let run_id = crate::runs::create(&state.pool, &state.cfg.id, "abc1234def") .await .unwrap(); let app = router(state); // timeout_ms=0 → deadline is now → the first poll returns the // still-building run rather than blocking. let resp = app .oneshot( Request::builder() .uri(format!("/runs/{}/wait?timeout_ms=0", run_id.0)) .body(Body::empty()) .unwrap(), ) .await .unwrap(); assert_eq!(resp.status(), StatusCode::OK); let v: serde_json::Value = serde_json::from_str(&body_string(resp).await).unwrap(); assert_eq!(v["result"], "building"); } #[tokio::test] async fn get_run_wait_404s_for_unknown_id() { let state = test_state().await; let app = router(state); let resp = app .oneshot( Request::builder() .uri("/runs/999/wait?timeout_ms=0") .body(Body::empty()) .unwrap(), ) .await .unwrap(); assert_eq!(resp.status(), StatusCode::NOT_FOUND); } #[test] fn self_update_unit_maps_sha_to_instance() { let sha = crate::domain::GitSha::parse("abc1234def5678").unwrap(); assert_eq!( self_update_unit(&sha), "sando-update@abc1234def5678.service" ); } #[tokio::test] async fn self_update_rejects_bad_sha_with_400() { // A malformed sha is a client error and must be rejected *before* any // privileged unit is triggered (so this test never shells out). let state = test_state().await; let app = router(state); let resp = app .oneshot( Request::builder() .method("POST") .uri("/self-update") .header("Content-Type", "application/json") .body(Body::from(r#"{"sha":"not-a-sha!"}"#)) .unwrap(), ) .await .unwrap(); assert_eq!(resp.status(), StatusCode::BAD_REQUEST); } // ---- /promote/{tier} default-version resolution ---- #[tokio::test] async fn promote_to_first_tier_is_rejected() { // tier 0 is host — you /rebuild, not /promote. let state = test_state().await; let app = router(state); let resp = app .oneshot( Request::builder() .method("POST") .uri("/promote/host") .body(Body::empty()) .unwrap(), ) .await .unwrap(); assert_eq!(resp.status(), StatusCode::CONFLICT); let body = body_string(resp).await; assert!( body.contains("cannot /promote to the first tier"), "got: {body}" ); } #[tokio::test] async fn promote_without_body_and_no_predecessor_version_errors() { // tier a has no body version supplied AND its predecessor mm has // current_version=NULL. Should fail before any deploy. let state = test_state().await; let app = router(state); let resp = app .oneshot( Request::builder() .method("POST") .uri("/promote/a") .body(Body::empty()) .unwrap(), ) .await .unwrap(); assert_eq!(resp.status(), StatusCode::CONFLICT); let body = body_string(resp).await; assert!( body.contains("no version specified") || body.contains("no current_version"), "got: {body}" ); } #[tokio::test] async fn promote_blocked_when_predecessor_gate_never_ran() { // End-to-end CF1: the host tier configures boot_smoke but it never ran // (no gate_runs row). Promoting host -> a must be GateBlocked, citing the // unsatisfied gate, instead of waving through on zero evidence. A real // `versions` row is present so the ONLY thing that can block is the gate. let pool = fresh_pool().await; for (i, name) in ["host", "a"].iter().enumerate() { sqlx::query( "INSERT INTO tiers (name, ord, provisioned, canary) VALUES (?, ?, 1, 'sequential')", ) .bind(name) .bind(i as i64) .execute(&pool) .await .unwrap(); sqlx::query("INSERT INTO tier_state (tier) VALUES (?)") .bind(name) .execute(&pool) .await .unwrap(); } let mut topo = test_topo(); topo.tiers[0].gates = vec![Gate::BootSmoke]; // host configures a gate... let executors = Arc::new(crate::state::build_executors(&topo)); let topo = Arc::new(topo); let cfg = Arc::new(test_cfg()); let (apps, default_app) = crate::state::one_app(cfg.clone(), topo.clone(), executors.clone()); let state = AppState { pool, apps, default_app, topo, cfg, active_build: Arc::new(tokio::sync::Mutex::new(None)), deploy_lock: Arc::new(tokio::sync::Mutex::new(())), events: crate::events::channel(), executors, api_token: None, }; sqlx::query("INSERT INTO versions (version, git_sha, built_at, artifact_path) VALUES ('0.8.12','sha',datetime('now'),'/tmp/x')") .execute(&state.pool).await.unwrap(); sqlx::query("UPDATE tier_state SET current_version = '0.8.12' WHERE tier = 'host'") .execute(&state.pool) .await .unwrap(); let app = router(state); let resp = app .oneshot( Request::builder() .method("POST") .uri("/promote/a") .body(Body::empty()) .unwrap(), ) .await .unwrap(); assert_eq!(resp.status(), StatusCode::CONFLICT); let body = body_string(resp).await; assert!( body.contains("boot_smoke"), "expected boot_smoke to block; got: {body}" ); } #[tokio::test] async fn migration_bearing_promote_requires_fresh_confirm() { // The predecessor (host) configures NO gates, so nothing would normally // block host -> a. A `bears_migration` promote must still be blocked on a // fresh `manual_confirm` it does not have: rollback restores the binary // only, so the one-way advance needs a conscious operator sign-off. let pool = fresh_pool().await; for (i, name) in ["host", "a"].iter().enumerate() { sqlx::query( "INSERT INTO tiers (name, ord, provisioned, canary) VALUES (?, ?, 1, 'sequential')", ) .bind(name) .bind(i as i64) .execute(&pool) .await .unwrap(); sqlx::query("INSERT INTO tier_state (tier) VALUES (?)") .bind(name) .execute(&pool) .await .unwrap(); } let mut topo = test_topo(); topo.tiers[0].gates = vec![]; // host configures no gates at all let executors = Arc::new(crate::state::build_executors(&topo)); let topo = Arc::new(topo); let cfg = Arc::new(test_cfg()); let (apps, default_app) = crate::state::one_app(cfg.clone(), topo.clone(), executors.clone()); let state = AppState { pool, apps, default_app, topo, cfg, active_build: Arc::new(tokio::sync::Mutex::new(None)), deploy_lock: Arc::new(tokio::sync::Mutex::new(())), events: crate::events::channel(), executors, api_token: None, }; sqlx::query("INSERT INTO versions (version, git_sha, built_at, artifact_path) VALUES ('0.8.12','sha',datetime('now'),'/tmp/x')") .execute(&state.pool).await.unwrap(); sqlx::query("UPDATE tier_state SET current_version = '0.8.12' WHERE tier = 'host'") .execute(&state.pool) .await .unwrap(); let app = router(state); let resp = app .oneshot( Request::builder() .method("POST") .uri("/promote/a") .header("content-type", "application/json") .body(Body::from(r#"{"bears_migration": true}"#)) .unwrap(), ) .await .unwrap(); assert_eq!(resp.status(), StatusCode::CONFLICT); let body = body_string(resp).await; assert!( body.contains("manual_confirm"), "expected the migration promote to block on manual_confirm; got: {body}" ); } #[tokio::test] async fn tier_state_advance_is_atomic_previous_from_old_current() { // CF3: the promote advance is a single UPDATE where previous_version is // set from the row's *old* current_version (SQLite evaluates RHS against // the original row). No read-modify-write to lose under concurrency. let pool = fresh_pool().await; seed(&pool, "a", "1.0.0").await; // current_version FKs into versions, so the target must exist too. sqlx::query("INSERT INTO versions (version, git_sha, built_at, artifact_path) VALUES ('2.0.0','sha',datetime('now'),'/tmp/x')") .execute(&pool).await.unwrap(); sqlx::query("UPDATE tier_state SET current_version = '1.0.0' WHERE tier = 'a'") .execute(&pool) .await .unwrap(); // Exercise the sealed forward-advance primitive itself — the same op // /promote and the host build path both call (S1), not a copy of its SQL. let v = crate::domain::Version::parse("2.0.0").unwrap(); crate::runs::advance_tier(&pool, &crate::domain::AppId::default(), "a", &v, None) .await .unwrap(); let (cur, prev): (Option, Option) = sqlx::query_as("SELECT current_version, previous_version FROM tier_state WHERE tier = 'a'") .fetch_one(&pool) .await .unwrap(); assert_eq!(cur.as_deref(), Some("2.0.0")); assert_eq!( prev.as_deref(), Some("1.0.0"), "previous = the pre-update current, atomically" ); } #[tokio::test] async fn canary_rollback_restores_deployed_nodes_to_previous_version() { use crate::topology::{Node, default_actuate, default_observe}; let tmp = tempfile::tempdir().unwrap(); // Two local nodes, each pre-seeded as if a promote had flipped them to // 2.0.0 (current -> releases/2.0.0), with the prior 1.0.0 still on disk. let mut nodes = Vec::new(); for name in ["n1", "n2"] { let rr = tmp.path().join(name); for v in ["1.0.0", "2.0.0"] { tokio::fs::create_dir_all(rr.join("releases").join(v)) .await .unwrap(); } tokio::fs::symlink("releases/2.0.0", rr.join("current")) .await .unwrap(); nodes.push(Node { platform: None, base_image: None, libc: None, name: name.into(), ssh_target: "local".into(), release_root: rr.to_string_lossy().into_owned(), service_name: "x.service".into(), health_url: None, config_check_env_file: None, actuate: default_actuate(), observe: default_observe(), companions: Vec::new(), }); } let mut state = test_state().await; // The rollback target needs a versions row. The release dir name comes // from the artifact_path's parent (legacy layout: releases/). sqlx::query("INSERT INTO versions (version, git_sha, built_at, artifact_path) VALUES ('1.0.0','sha',datetime('now'),'/tmp/staged/releases/1.0.0/makenotwork')") .execute(&state.pool).await.unwrap(); let execs: crate::state::ExecutorMap = nodes .iter() .map(|n| (n.name.clone(), crate::state::build_executor(n))) .collect(); state.executors = std::sync::Arc::new(execs); let refs: Vec<&Node> = nodes.iter().collect(); let report = rollback_deployed_nodes(&state, &tid("a"), &refs, "1.0.0").await; assert_eq!(report.restored, 2, "both deployed nodes should be restored"); assert!( report.is_consistent(), "nothing should be indeterminate: {report:?}" ); for n in &nodes { let cur = tokio::fs::read_link(std::path::Path::new(&n.release_root).join("current")) .await .unwrap(); assert_eq!( cur.to_string_lossy(), "releases/1.0.0", "node {} rolled back", n.name ); } } #[tokio::test] async fn canary_rollback_is_noop_without_a_previous_artifact() { use crate::topology::{Node, default_actuate, default_observe}; let tmp = tempfile::tempdir().unwrap(); let rr = tmp.path().join("n1"); tokio::fs::create_dir_all(&rr).await.unwrap(); let node = Node { platform: None, base_image: None, libc: None, name: "n1".into(), ssh_target: "local".into(), release_root: rr.to_string_lossy().into_owned(), service_name: "x.service".into(), health_url: None, config_check_env_file: None, actuate: default_actuate(), observe: default_observe(), companions: Vec::new(), }; let state = test_state().await; // no versions row for "9.9.9" let report = rollback_deployed_nodes(&state, &tid("a"), &[&node], "9.9.9").await; assert_eq!( report.restored, 0, "no artifact to roll back to -> nothing restored, no panic" ); assert_eq!( report.touched(), 1, "the node must be accounted for somewhere" ); // No rollback could even be attempted, so the node's version is not // knowable here. That must read as indeterminate, not as safe. assert_eq!(report.indeterminate, 1); assert!(!report.is_consistent()); } /// A rollback that fails before the symlink swap leaves the node on the /// version it was already running, the one being rolled back to. Counting /// that as "not restored" would report `restored=0 of=1` and send an /// operator to inspect a production box that is entirely fine. #[test] fn a_rollback_that_failed_before_the_swap_is_not_an_incident() { let report = RollbackReport { restored: 0, already_on_previous: 1, indeterminate: 0, }; assert_eq!(report.touched(), 1); assert!( report.is_consistent(), "a node that never left the previous version is not split-brain" ); // Contrast: the same zero restored, but the swap had run. This one does // warrant a human, and the two must not report the same way. let real = RollbackReport { restored: 0, already_on_previous: 0, indeterminate: 1, }; assert_eq!(real.touched(), 1); assert!(!real.is_consistent()); } /// A genuine split-brain still reports as one: some nodes back on the old /// version, one stranded. #[test] fn a_mixed_outcome_is_inconsistent_if_any_node_is_unknown() { let report = RollbackReport { restored: 2, already_on_previous: 1, indeterminate: 1, }; assert_eq!(report.touched(), 4); assert!(!report.is_consistent()); } // ---- FleetFake: a multi-node promote across recorded fake executors ---- // // The route-level promote tests until now ran a single local node against a // real LocalExec, so the sequential-canary fan-out, the cross-node deploy // ordering, and the mid-canary rollback of already-flipped nodes had no // coverage. FleetFake records every deploy op (tagged by node) into one // shared log so ordering is visible across the fleet, and can fail any op // whose shell script or rsync target contains a marker — used to fail a // node's forward deploy of the new version while its rollback to the prior // version (a different `releases/` path) still succeeds. struct FleetFake { tag: String, caps: CapabilitySet, log: Arc>>, fail_if_contains: Option, } impl FleetFake { fn fails(&self, text: &str) -> bool { self.fail_if_contains .as_deref() .is_some_and(|m| text.contains(m)) } } #[async_trait] impl Executor for FleetFake { async fn run_streaming( &self, step: &Step, _sink: &mut dyn LogSink, ) -> anyhow::Result { let script = step.argv.last().cloned().unwrap_or_default(); self.log .lock() .unwrap() .push(format!("{}:run:{script}", self.tag)); Ok(RunOutput { status: std::process::ExitStatus::from_raw(if self.fails(&script) { 1 << 8 } else { 0 }), stdout: Vec::new(), stderr: Vec::new(), }) } async fn pull_file( &self, _r: &std::path::Path, _l: &std::path::Path, _o: &SyncOpts, ) -> anyhow::Result<()> { Ok(()) } async fn pull_dir( &self, _r: &std::path::Path, _l: &std::path::Path, _o: &SyncOpts, ) -> anyhow::Result<()> { Ok(()) } async fn pull_glob(&self, _g: &str, _l: &std::path::Path, _o: &SyncOpts) -> anyhow::Result<()> { Ok(()) } async fn push_dir( &self, _local: &std::path::Path, remote: &std::path::Path, _o: &SyncOpts, ) -> anyhow::Result<()> { let dst = remote.display().to_string(); self.log .lock() .unwrap() .push(format!("{}:push:{dst}", self.tag)); if self.fails(&dst) { anyhow::bail!("fake rsync failure on {}", self.tag); } Ok(()) } fn capabilities(&self) -> &CapabilitySet { &self.caps } } /// Rebuild tier "a" with `names` as remote fake nodes sharing one op log, /// seed the version/tier_state prerequisites for a promote of 3.0.0 up from /// `host` (tier "a" starts on 2.0.0 with 1.0.0 behind it), and optionally /// make `fail_node` fail any op containing `fail_marker`. Returns the state /// and the shared log. async fn fleet_fixture( names: &[&str], fail_node: Option<&str>, fail_marker: &str, ) -> (AppState, Arc>>) { use crate::topology::{default_actuate, default_observe}; let mut state = test_state().await; let log = Arc::new(StdMutex::new(Vec::::new())); let nodes: Vec = names .iter() .map(|name| Node { platform: None, base_image: None, libc: None, name: (*name).into(), ssh_target: format!("deploy@{name}"), release_root: format!("/tmp/fleet/{name}"), service_name: "makenotwork.service".into(), health_url: None, config_check_env_file: None, actuate: default_actuate(), observe: default_observe(), companions: Vec::new(), }) .collect(); let mut topo = (*state.topo).clone(); topo.tiers[1].nodes = nodes.clone(); // Drop the tier's post-deploy gate: the deploy fan-out is the subject // here, and node_health would need its own probe wiring. topo.tiers[1].gates = vec![]; state.topo = Arc::new(topo); let execs: crate::state::ExecutorMap = nodes .iter() .map(|n| { let nm = n.name.to_string(); let fail = fail_node == Some(nm.as_str()); let exec: Arc = Arc::new(FleetFake { tag: nm, caps: CapabilitySet::from_tokens(["deploy", "restart"], ["health"]), log: log.clone(), fail_if_contains: fail.then(|| fail_marker.to_string()), }); (n.name.clone(), exec) }) .collect(); state.executors = Arc::new(execs); for n in &nodes { // deploys.node FKs into `nodes`. sqlx::query( "INSERT INTO nodes (name, tier, ssh_target, release_root) VALUES (?, 'a', ?, ?)", ) .bind(&n.name) .bind(&n.ssh_target) .bind(&n.release_root) .execute(&state.pool) .await .unwrap(); } for v in ["1.0.0", "2.0.0", "3.0.0"] { // Legacy (pre-identity) artifact_path is `releases//`, // so the release dir the node mirrors is named for the version. The // fixture reflects that real layout (parent basename == version). sqlx::query("INSERT INTO versions (version, git_sha, built_at, artifact_path) VALUES (?, 'sha', datetime('now'), ?)") .bind(v).bind(format!("/tmp/staged/releases/{v}/makenotwork")).execute(&state.pool).await.unwrap(); } sqlx::query("UPDATE tier_state SET current_version = '3.0.0' WHERE tier = 'host'") .execute(&state.pool) .await .unwrap(); sqlx::query( "UPDATE tier_state SET current_version = '2.0.0', previous_version = '1.0.0' WHERE tier = 'a'", ) .execute(&state.pool) .await .unwrap(); (state, log) } /// Index of the first logged op belonging to `node` (panics if the node was /// never touched — the message names it). fn first_touch(log: &[String], node: &str) -> usize { let prefix = format!("{node}:"); log.iter() .position(|e| e.starts_with(&prefix)) .unwrap_or_else(|| panic!("node {node:?} was never deployed to; log: {log:#?}")) } #[tokio::test] async fn promote_deploys_every_node_in_tier_order_and_advances() { let (state, log) = fleet_fixture(&["a1", "a2", "a3"], None, "").await; let pool = state.pool.clone(); let Json(body) = promote_inner(state, "a".into(), PromoteBody::default()) .await .expect("all nodes deploy, so the promote succeeds"); assert_eq!( body["nodes_deployed"], serde_json::json!(["a1", "a2", "a3"]), "the response names every node the promote reached", ); let (cur, prev) = tier_versions(&pool, "a").await; assert_eq!(cur.as_deref(), Some("3.0.0")); assert_eq!(prev.as_deref(), Some("2.0.0")); // Sequential canary: a1 is fully touched before a2, a2 before a3. let log = log.lock().unwrap().clone(); assert!( first_touch(&log, "a1") < first_touch(&log, "a2") && first_touch(&log, "a2") < first_touch(&log, "a3"), "nodes must deploy in tier order: {log:#?}", ); // Every node has a green deploy row for the promoted version. let ok: i64 = sqlx::query_scalar( "SELECT COUNT(*) FROM deploys WHERE version = '3.0.0' AND outcome = 'ok'", ) .fetch_one(&pool) .await .unwrap(); assert_eq!(ok, 3, "one ok deploy row per node"); } #[tokio::test] async fn a_mid_canary_deploy_failure_rolls_touched_nodes_back_and_does_not_advance() { // a2 fails its forward deploy of 3.0.0; a1 was already flipped, a3 is // never reached. The touched nodes (a1, a2) roll back to 2.0.0 — their // rollback ops target `releases/2.0.0`, which the marker does not match — // and tier_state must NOT advance. let (state, log) = fleet_fixture(&["a1", "a2", "a3"], Some("a2"), "releases/3.0.0").await; let pool = state.pool.clone(); let err = promote_inner(state, "a".into(), PromoteBody::default()) .await .expect_err("a mid-canary deploy failure must fail the promote"); assert!( matches!(err, crate::error::Error::Other(_)), "a deploy failure propagates as Other, got: {err:?}", ); // tier_state untouched: the failure returns before advance_tier. let (cur, prev) = tier_versions(&pool, "a").await; assert_eq!( cur.as_deref(), Some("2.0.0"), "a failed rollout must not advance" ); assert_eq!(prev.as_deref(), Some("1.0.0")); let log = log.lock().unwrap().clone(); // a3 sits after the failed a2 in the sequence and is never touched. assert!( !log.iter().any(|e| e.starts_with("a3:")), "nodes after the failure must not be deployed to: {log:#?}", ); // Both touched nodes were rolled back to the prior version. for n in ["a1", "a2"] { assert!( log.iter() .any(|e| e.starts_with(&format!("{n}:")) && e.contains("releases/2.0.0")), "touched node {n} must be restored to 2.0.0: {log:#?}", ); } // The forward attempt is on the record: a1 ok, a2 failed. let a1: String = sqlx::query_scalar("SELECT outcome FROM deploys WHERE version = '3.0.0' AND node = 'a1'") .fetch_one(&pool) .await .unwrap(); assert_eq!(a1, "ok"); let a2: String = sqlx::query_scalar("SELECT outcome FROM deploys WHERE version = '3.0.0' AND node = 'a2'") .fetch_one(&pool) .await .unwrap(); assert_eq!(a2, "failed"); // Both touched nodes restored => the tier is consistent on 2.0.0, so the // partial flag is cleared, not set. let reason: Option = sqlx::query_scalar("SELECT partial_reason FROM tier_state WHERE tier = 'a'") .fetch_one(&pool) .await .unwrap(); assert_eq!( reason, None, "a fully-restored canary leaves the tier consistent, not partial", ); } /// Build a one-node tier "a" on a tempdir release root, pre-seeded as if a /// promote had flipped it to `current` with `prev` still staged on disk. /// Returns the state (topology rewired to the tempdir node) and the tempdir, /// which the caller must keep alive. async fn rollback_fixture(prev: &str, current: &str) -> (AppState, tempfile::TempDir) { use crate::topology::{Node, default_actuate, default_observe}; let tmp = tempfile::tempdir().unwrap(); let rr = tmp.path().join("a-local"); for v in [prev, current] { tokio::fs::create_dir_all(rr.join("releases").join(v)) .await .unwrap(); } tokio::fs::symlink(format!("releases/{current}"), rr.join("current")) .await .unwrap(); let node = Node { platform: None, base_image: None, libc: None, name: "a-local".into(), ssh_target: "local".into(), release_root: rr.to_string_lossy().into_owned(), service_name: "x.service".into(), health_url: None, config_check_env_file: None, actuate: default_actuate(), observe: default_observe(), companions: Vec::new(), }; let mut state = test_state().await; let mut topo = (*state.topo).clone(); topo.tiers[1].nodes = vec![node]; state.executors = Arc::new(crate::state::build_executors(&topo)); state.topo = Arc::new(topo); // deploys.node FKs into `nodes`, so the promote path needs the row. sqlx::query("INSERT INTO nodes (name, tier, ssh_target, release_root) VALUES ('a-local', 'a', 'local', ?)") .bind(rr.to_string_lossy().into_owned()) .execute(&state.pool).await.unwrap(); for v in [prev, current] { sqlx::query("INSERT INTO versions (version, git_sha, built_at, artifact_path) VALUES (?, 'sha', datetime('now'), ?)") .bind(v).bind(format!("/tmp/staged/releases/{v}/makenotwork")).execute(&state.pool).await.unwrap(); } sqlx::query("UPDATE tier_state SET current_version = ?, previous_version = ? WHERE tier = 'a'") .bind(current) .bind(prev) .execute(&state.pool) .await .unwrap(); (state, tmp) } async fn tier_versions(pool: &SqlitePool, tier: &str) -> (Option, Option) { sqlx::query_as("SELECT current_version, previous_version FROM tier_state WHERE tier = ?") .bind(tier) .fetch_one(pool) .await .unwrap() } #[tokio::test] async fn rollback_clears_previous_version_rather_than_swapping_it() { // The swap bug: writing the version we just rolled OFF into // previous_version made a second /rollback roll FORWARD onto the broken // build the operator was escaping. previous_version must go NULL. let (state, _tmp) = rollback_fixture("1.0.0", "2.0.0").await; let pool = state.pool.clone(); let _ = rollback(State(state), Path("a".to_string())).await.unwrap(); let (cur, prev) = tier_versions(&pool, "a").await; assert_eq!( cur.as_deref(), Some("1.0.0"), "rolled back to the previous version" ); assert_eq!( prev, None, "the version we rolled off must NOT become the rollback target" ); } #[tokio::test] async fn second_rollback_refuses_instead_of_rolling_forward() { let (state, _tmp) = rollback_fixture("1.0.0", "2.0.0").await; let pool = state.pool.clone(); let release_root = state.topo.tiers[1].nodes[0].release_root.clone(); let _ = rollback(State(state.clone()), Path("a".to_string())) .await .unwrap(); let err = rollback(State(state), Path("a".to_string())) .await .expect_err("only one step of history is tracked; a second rollback must refuse"); assert!( matches!(&err, crate::error::Error::GateBlocked(m) if m.contains("no previous_version")), "expected a loud refusal, got: {err:?}", ); // The refusal is total: neither the DB nor the node moved back to 2.0.0. let (cur, _) = tier_versions(&pool, "a").await; assert_eq!(cur.as_deref(), Some("1.0.0")); let link = tokio::fs::read_link(std::path::Path::new(&release_root).join("current")) .await .unwrap(); assert_eq!( link.to_string_lossy(), "releases/1.0.0", "node stays on the rolled-back version" ); } #[tokio::test] async fn promote_refuses_an_unprovisioned_tier() { // Every step of a promote to a node-less tier is a silent no-op that // still reports success: the deploy loop iterates nothing and // advance_tier records a current_version the tier never received. let (mut state, _tmp) = rollback_fixture("1.0.0", "2.0.0").await; let mut topo = (*state.topo).clone(); topo.tiers[1].provisioned = false; topo.tiers[1].nodes.clear(); state.topo = Arc::new(topo); let pool = state.pool.clone(); sqlx::query("UPDATE tier_state SET current_version = '2.0.0' WHERE tier = 'host'") .execute(&pool) .await .unwrap(); let err = promote_inner(state, "a".into(), PromoteBody::default()) .await .expect_err("promoting to an unprovisioned tier must be refused"); assert!( matches!(&err, crate::error::Error::GateBlocked(m) if m.contains("not provisioned")), "got: {err:?}", ); // And nothing was recorded: the tier keeps whatever it had before. let (cur, _) = tier_versions(&pool, "a").await; assert_eq!( cur.as_deref(), Some("2.0.0"), "a refused promote must not advance tier_state", ); } #[tokio::test] async fn promote_advances_the_tier_and_flips_the_symlink_when_gates_are_green() { // The happy path. Every other promote test asserts a refusal or a red // outcome, so nothing pinned what a *successful* promote actually does: // deploy reaches the node, the `current` symlink flips, tier_state // advances with previous_version = the version we came off, any stale // partial flag clears, and the handler reports the nodes it touched. use crate::topology::Gate; let (mut state, _tmp) = rollback_fixture("1.0.0", "2.0.0").await; // Source tier `host` gates the promote on cargo_test. Satisfying it is // the point of the test: the sibling case asserts an unsatisfied gate // blocks, this one asserts a satisfied gate lets the promote through. let mut topo = (*state.topo).clone(); topo.tiers[0].gates = vec![Gate::CargoTest]; state.topo = Arc::new(topo); let pool = state.pool.clone(); let release_root = state.topo.tiers[1].nodes[0].release_root.clone(); // 3.0.0 is staged on the build host and green on `host`. tokio::fs::create_dir_all( std::path::Path::new(&release_root) .join("releases") .join("3.0.0"), ) .await .unwrap(); sqlx::query("INSERT INTO versions (version, git_sha, built_at, artifact_path) VALUES ('3.0.0','sha',datetime('now'),'/tmp/staged/releases/3.0.0/makenotwork')") .execute(&pool).await.unwrap(); sqlx::query("UPDATE tier_state SET current_version = '3.0.0' WHERE tier = 'host'") .execute(&pool) .await .unwrap(); insert_gate(&pool, "host", "3.0.0", "cargo_test", 1).await; // A stale flag from an earlier incident, which a clean rollout clears. set_partial(&state, &tid("a"), "left over from a previous canary").await; let Json(body) = promote_inner(state, "a".into(), PromoteBody::default()) .await .expect("gates are green and the node deploys, so the promote succeeds"); assert_eq!(body["tier"], "a"); assert_eq!(body["version"], "3.0.0"); assert_eq!( body["nodes_deployed"], serde_json::json!(["a-local"]), "the response names every node the promote reached", ); let (cur, prev) = tier_versions(&pool, "a").await; assert_eq!(cur.as_deref(), Some("3.0.0")); assert_eq!( prev.as_deref(), Some("2.0.0"), "previous_version is the version we came off, so a rollback aims at it", ); // The node genuinely moved: the promote is not just bookkeeping. let link = tokio::fs::read_link(std::path::Path::new(&release_root).join("current")) .await .unwrap(); assert_eq!(link.to_string_lossy(), "releases/3.0.0"); let reason: Option = sqlx::query_scalar("SELECT partial_reason FROM tier_state WHERE tier = 'a'") .fetch_one(&pool) .await .unwrap(); assert_eq!( reason, None, "a clean full rollout clears a stale partial flag", ); // The deploy is on the record as having succeeded, which is what the // next promote's gate check and /state both read. let (node, outcome): (String, String) = sqlx::query_as("SELECT node, outcome FROM deploys WHERE version = '3.0.0'") .fetch_one(&pool) .await .unwrap(); assert_eq!(node, "a-local"); assert_eq!(outcome, "ok"); } #[tokio::test] async fn promote_fails_and_flags_the_tier_when_post_deploy_gates_are_red() { // The deploy reached every node, so tier_state advances (a stale // current_version would aim a later rollback at the wrong artifact), but // the promote must NOT report success: the tier is flagged partial and // the handler returns the gate failure. use crate::topology::Gate; let (mut state, _tmp) = rollback_fixture("1.0.0", "2.0.0").await; // node_health is the tier's only gate, and it fails closed with no // probes — an empty executor map gives it nothing to probe while // deploy_node still falls back to a built executor and succeeds. let mut topo = (*state.topo).clone(); topo.tiers[1].gates = vec![Gate::NodeHealth]; state.topo = Arc::new(topo); state.executors = Arc::new(crate::state::ExecutorMap::new()); let pool = state.pool.clone(); // Promote 3.0.0 up from host, which configures no gates of its own. sqlx::query("INSERT INTO versions (version, git_sha, built_at, artifact_path) VALUES ('3.0.0','sha',datetime('now'),'/tmp/staged/releases/3.0.0/makenotwork')") .execute(&pool).await.unwrap(); sqlx::query("UPDATE tier_state SET current_version = '3.0.0' WHERE tier = 'host'") .execute(&pool) .await .unwrap(); let err = promote_inner(state, "a".into(), PromoteBody::default()) .await .expect_err("red post-deploy gates must fail the promote"); assert!( matches!(&err, crate::error::Error::GateBlocked(m) if m.contains("node_health")), "the failure must name the red gate, got: {err:?}", ); let (cur, _) = tier_versions(&pool, "a").await; assert_eq!( cur.as_deref(), Some("3.0.0"), "tier_state tracks what the nodes actually run" ); let reason: Option = sqlx::query_scalar("SELECT partial_reason FROM tier_state WHERE tier = 'a'") .fetch_one(&pool) .await .unwrap(); assert!( reason.as_deref().is_some_and(|r| r.contains("node_health")), "the tier must be flagged for /state and the TUI, got: {reason:?}", ); } #[tokio::test] async fn set_partial_then_clear_roundtrips() { let state = test_state().await; let read = || async { sqlx::query_scalar::<_, Option>( "SELECT partial_reason FROM tier_state WHERE tier = 'a'", ) .fetch_one(&state.pool) .await .unwrap() }; assert_eq!(read().await, None, "consistent tier starts clean"); set_partial(&state, &tid("a"), "canary rollback incomplete: 1/2").await; assert_eq!( read().await.as_deref(), Some("canary rollback incomplete: 1/2") ); clear_partial(&state, &tid("a")).await; assert_eq!(read().await, None, "clear nulls it back out"); } #[tokio::test] async fn state_surfaces_partial_reason() { use axum::extract::State; let state = test_state().await; set_partial( &state, &tid("a"), "first-deploy canary failed: 1 node(s) on 2.0.0", ) .await; let Json(view) = get_state(State(state)).await.unwrap(); let a = view.tiers.iter().find(|t| t.name == "a").unwrap(); assert_eq!( a.partial_reason.as_deref(), Some("first-deploy canary failed: 1 node(s) on 2.0.0"), ); let host = view.tiers.iter().find(|t| t.name == "host").unwrap(); assert_eq!( host.partial_reason, None, "untouched tier stays clean in /state" ); } #[tokio::test] async fn state_build_is_null_until_first_rebuild_then_surfaces_latest() { use axum::extract::State; let state = test_state().await; // No build runs yet → build is null, so /state doesn't pretend a build // is happening. let Json(view) = get_state(State(state.clone())).await.unwrap(); assert!(view.build.is_none()); // A failed run must surface its cause in /state, not just in /runs. let run_id = crate::runs::create(&state.pool, &state.cfg.id, "deadbeef") .await .unwrap(); crate::runs::mark_failed(&state.pool, run_id, "cargo_test: 3 test(s) failed") .await .unwrap(); let Json(view) = get_state(State(state)).await.unwrap(); let b = view.build.expect("build surfaced"); assert_eq!(b.run_id, run_id.0); assert_eq!(b.result, "failed"); assert_eq!( b.failure_summary.as_deref(), Some("cargo_test: 3 test(s) failed") ); } #[tokio::test] async fn status_json_serves_the_shared_payload_over_the_real_router() { // The mapping itself is tested in `crate::status`. This asserts the // route is wired, serves valid JSON, and stays internally consistent // (no dangling child or action references) against a real topology // rather than a hand-built fixture. let state = test_state().await; set_partial(&state, &tid("a"), "canary rollback incomplete: 1/2").await; let resp = router(state) .oneshot( Request::builder() .uri("/status.json") .body(Body::empty()) .unwrap(), ) .await .unwrap(); assert_eq!(resp.status(), StatusCode::OK); let body = http_body_util::BodyExt::collect(resp.into_body()) .await .unwrap() .to_bytes(); let payload: ops_status::Payload = serde_json::from_slice(&body).unwrap(); assert_eq!(payload.source, crate::status::SOURCE); assert_eq!(payload.schema, ops_status::SCHEMA_VERSION); assert_eq!(payload.validate(), Ok(())); assert_eq!( payload.node("tier:a").unwrap().status, ops_status::Status::Failed, "a partial tier must surface as failed over the wire" ); assert_eq!(payload.worst_status(), ops_status::Status::Failed); } #[tokio::test] async fn promote_with_explicit_version_but_missing_artifact_404s() { // Explicit version supplied, gates trivially pass (mm has none in // test_topo), but `versions` table has no row → 404. let state = test_state().await; let app = router(state); let resp = app .oneshot( Request::builder() .method("POST") .uri("/promote/a") .header("content-type", "application/json") .body(Body::from(r#"{"version":"9.9.9"}"#)) .unwrap(), ) .await .unwrap(); assert_eq!(resp.status(), StatusCode::NOT_FOUND); } // ---- GET /logs/{version}/{gate} ---- async fn state_with_logs_root(logs_root: PathBuf) -> AppState { let mut s = test_state().await; let mut cfg = (*s.cfg).clone(); cfg.logs_root = logs_root; s.cfg = Arc::new(cfg); s } #[tokio::test] async fn get_gate_log_returns_file_contents() { let tmp = tempfile::tempdir().unwrap(); let dir = tmp.path().join("0.9.5"); tokio::fs::create_dir_all(&dir).await.unwrap(); tokio::fs::write(dir.join("cargo_test.log"), b"hello sandod\n") .await .unwrap(); let state = state_with_logs_root(tmp.path().to_path_buf()).await; let app = router(state); let resp = app .oneshot( Request::builder() .uri("/logs/0.9.5/cargo_test") .body(Body::empty()) .unwrap(), ) .await .unwrap(); assert_eq!(resp.status(), StatusCode::OK); assert_eq!(body_string(resp).await, "hello sandod\n"); } #[tokio::test] async fn get_gate_log_accepts_a_log_ref_verbatim() { // `log_ref` on a gate row is `/.log`; appending it to // `/logs/` must work as-is, so following the ref is the path of least // effort. Guessing from the version is what returns another run's output. let tmp = tempfile::tempdir().unwrap(); tokio::fs::create_dir_all(tmp.path().join("62")) .await .unwrap(); tokio::fs::write(tmp.path().join("62/cargo_deny.log"), b"run 62 only") .await .unwrap(); let state = state_with_logs_root(tmp.path().to_path_buf()).await; let app = router(state); let resp = app .oneshot( Request::builder() .uri("/logs/62/cargo_deny.log") .body(Body::empty()) .unwrap(), ) .await .unwrap(); assert_eq!(resp.status(), StatusCode::OK); assert_eq!(body_string(resp).await, "run 62 only"); } #[tokio::test] async fn get_gate_log_404s_when_missing() { let tmp = tempfile::tempdir().unwrap(); let state = state_with_logs_root(tmp.path().to_path_buf()).await; let app = router(state); let resp = app .oneshot( Request::builder() .uri("/logs/0.9.5/cargo_test") .body(Body::empty()) .unwrap(), ) .await .unwrap(); assert_eq!(resp.status(), StatusCode::NOT_FOUND); } // ---- CF2: bearer-token auth on deploy mutators ---- #[tokio::test] async fn mutating_route_requires_bearer_when_token_set() { let mut state = test_state().await; state.api_token = Some(std::sync::Arc::from("s3cr3t")); let app = router(state); // No Authorization header -> 401, before any deploy logic runs. let resp = app .clone() .oneshot( Request::builder() .method("POST") .uri("/promote/a") .body(Body::empty()) .unwrap(), ) .await .unwrap(); assert_eq!(resp.status(), StatusCode::UNAUTHORIZED); // Wrong token -> 401. let resp = app .clone() .oneshot( Request::builder() .method("POST") .uri("/promote/a") .header("authorization", "Bearer nope") .body(Body::empty()) .unwrap(), ) .await .unwrap(); assert_eq!(resp.status(), StatusCode::UNAUTHORIZED); // Correct token -> passes auth (then blocked downstream by gates / // missing predecessor version, but specifically NOT 401). let resp = app .clone() .oneshot( Request::builder() .method("POST") .uri("/promote/a") .header("authorization", "Bearer s3cr3t") .body(Body::empty()) .unwrap(), ) .await .unwrap(); assert_ne!(resp.status(), StatusCode::UNAUTHORIZED); } #[tokio::test] async fn read_routes_require_token_when_set() { // Reads expose prod state (versions, SHAs, gate logs, the event stream), // so they are bearer-gated too — not just the mutators. A tailnet peer // without the token gets 401; the TUI presents the token and gets 200. let mut state = test_state().await; state.api_token = Some(std::sync::Arc::from("s3cr3t")); let app = router(state); // No token -> 401 on a read. let resp = app .clone() .oneshot( Request::builder() .uri("/state") .body(Body::empty()) .unwrap(), ) .await .unwrap(); assert_eq!(resp.status(), StatusCode::UNAUTHORIZED); // Correct token -> 200. let resp = app .oneshot( Request::builder() .uri("/state") .header("authorization", "Bearer s3cr3t") .body(Body::empty()) .unwrap(), ) .await .unwrap(); assert_eq!(resp.status(), StatusCode::OK); } #[tokio::test] async fn read_routes_open_without_a_token() { // The loopback/dev posture: no token configured, reads pass through so a // local TUI needs no credential. let state = test_state().await; let app = router(state); let resp = app .oneshot( Request::builder() .uri("/state") .body(Body::empty()) .unwrap(), ) .await .unwrap(); assert_eq!(resp.status(), StatusCode::OK); } #[tokio::test] async fn self_update_malformed_body_is_400_not_422() { // TypedBody funnels a JSON deserialize failure through the Error envelope // (400), not axum's raw 422 — keeping every mutator on one error contract. let state = test_state().await; // no token -> auth passes, body is the gate let app = router(state); let resp = app .oneshot( Request::builder() .method("POST") .uri("/self-update") .header("content-type", "application/json") .body(Body::from("{ not valid json")) .unwrap(), ) .await .unwrap(); assert_eq!(resp.status(), StatusCode::BAD_REQUEST); } #[tokio::test] async fn gate_log_rejects_unknown_gate_kind() { // The gate segment is an allowlisted GateKind, not a free-form filename: // an unknown kind is a 404, so `*.log` basenames can't be probed. let state = test_state().await; let app = router(state); let resp = app .oneshot( Request::builder() .uri("/logs/0.9.6/passwd") .body(Body::empty()) .unwrap(), ) .await .unwrap(); assert_eq!(resp.status(), StatusCode::NOT_FOUND); } /// Path-traversal guard: a `..` segment must not escape logs_root. /// axum's `{name}` param already rejects a literal `/` in the value, but /// `..` as a whole segment is structurally valid and must be blocked at /// the handler. #[tokio::test] async fn get_gate_log_rejects_dotdot_segments() { let tmp = tempfile::tempdir().unwrap(); let state = state_with_logs_root(tmp.path().to_path_buf()).await; let app = router(state); let resp = app .oneshot( Request::builder() .uri("/logs/../etc/passwd") .body(Body::empty()) .unwrap(), ) .await .unwrap(); assert_eq!(resp.status(), StatusCode::NOT_FOUND); } // ---- build-identity path (wiki release-artifact-identity) ---- /// Insert a settled (passed) build_runs row with content identity and return /// its id. `staged_path` is the content-addressed release dir the bundle was /// published to (`releases/`). async fn seed_build(pool: &SqlitePool, sha: &str, version: &str, staged_path: &str) -> i64 { sqlx::query_scalar( "INSERT INTO build_runs (sha, version, phase, result, started_at, bundle_digest, staged_path) VALUES (?, ?, 'done', 'passed', datetime('now'), ?, ?) RETURNING id", ) .bind(sha) .bind(version) .bind(format!("{sha}-digest")) .bind(staged_path) .fetch_one(pool) .await .unwrap() } async fn insert_gate_build( pool: &SqlitePool, tier: &str, version: &str, kind: &str, passed: i64, build_id: i64, ) { let status = if passed == 1 { "passed" } else { "failed" }; sqlx::query( "INSERT INTO gate_runs (version, tier, gate_kind, started_at, finished_at, status, build_id) \ VALUES (?, ?, ?, datetime('now'), datetime('now'), ?, ?)", ) .bind(version) .bind(tier) .bind(kind) .bind(status) .bind(build_id) .execute(pool) .await .unwrap(); } #[tokio::test] async fn unsatisfied_gates_keys_build_evidence_on_build_id() { let pool = fresh_pool().await; seed(&pool, "a", "3.0.0").await; // Two builds of the SAME version string. Only b1's gate ran. let b1 = seed_build(&pool, "sha1", "3.0.0", "/rel/1111111111111111").await; let b2 = seed_build(&pool, "sha2", "3.0.0", "/rel/2222222222222222").await; insert_gate_build(&pool, "a", "3.0.0", "cargo_test", 1, b1).await; // b1's own evidence satisfies. let ok = unsatisfied_gates( &pool, &crate::domain::AppId::default(), &tid("a"), &[Gate::CargoTest], &Evidence { version: "3.0.0", builds: &[PromotedBuild { platform: None, build_id: Some(b1), }], tier_build: None, }, false, ) .await .unwrap(); assert!(ok.is_empty(), "the build that passed its gate is satisfied"); // b2 shares the version but has no gate row of its own: fail closed. This // is the hole — a rebuild reusing the version must not ride b1's evidence. let bad = unsatisfied_gates( &pool, &crate::domain::AppId::default(), &tid("a"), &[Gate::CargoTest], &Evidence { version: "3.0.0", builds: &[PromotedBuild { platform: None, build_id: Some(b2), }], tier_build: None, }, false, ) .await .unwrap(); assert_eq!( bad, vec!["cargo_test".to_string()], "a different build of the same version does not inherit the evidence" ); // Legacy (pre-identity) callers still resolve by version string. let legacy = unsatisfied_gates( &pool, &crate::domain::AppId::default(), &tid("a"), &[Gate::CargoTest], &Evidence { version: "3.0.0", builds: &[], tier_build: None, }, false, ) .await .unwrap(); assert!(legacy.is_empty(), "version-keyed legacy path unchanged"); } fn plat(s: &str) -> crate::domain::Platform { crate::domain::Platform::parse(s).unwrap() } /// The structural block that made a cross-architecture promote impossible. /// /// A pom astra -> hetzner promote ships the x86_64 bundle. astra is aarch64 /// and runs only the aarch64 one, so a `node_health` looked up against the /// x86_64 build finds nothing and fail-closed refuses, while `/state` /// correctly reports astra's node_health as passed off astra's own row. No /// amount of re-running or re-confirming clears that: astra will never run /// that bundle. /// /// node_health is evidence about the tier, so it is keyed on the tier's own /// build. cargo_test is evidence about the bytes, so it stays per shipped /// build. The assertion below holds both halves at once. #[tokio::test] async fn tier_gates_key_on_the_tier_s_build_not_on_what_it_ships_onward() { let pool = fresh_pool().await; seed(&pool, "astra", "0.4.3").await; let arm = seed_build(&pool, "sha-arm", "0.4.3", "/rel/aaaaaaaaaaaaaaaa").await; let x86 = seed_build(&pool, "sha-x86", "0.4.3", "/rel/bbbbbbbbbbbbbbbb").await; // What astra actually has: its own node_health, and each bundle's own // artifact evidence from its own intake. insert_gate_build(&pool, "astra", "0.4.3", "node_health", 1, arm).await; insert_gate_build(&pool, "astra", "0.4.3", "cargo_test", 1, arm).await; insert_gate_build(&pool, "astra", "0.4.3", "cargo_test", 1, x86).await; let ships_x86 = [PromotedBuild { platform: Some(plat("linux/x86_64")), build_id: Some(x86), }]; let pending = unsatisfied_gates( &pool, &crate::domain::AppId::default(), &tid("astra"), &[Gate::NodeHealth, Gate::CargoTest], &Evidence { version: "0.4.3", builds: &ships_x86, tier_build: Some(arm), }, // astra's own build false, ) .await .unwrap(); assert!( pending.is_empty(), "astra's node_health vouches for astra, not for the bundle leaving it: {pending:?}" ); // And it is still a real gate: a tier whose own node_health never passed // is refused, however green the bundle it is shipping. seed(&pool, "hetzner", "0.4.3").await; insert_gate_build(&pool, "hetzner", "0.4.3", "cargo_test", 1, x86).await; let pending = unsatisfied_gates( &pool, &crate::domain::AppId::default(), &tid("hetzner"), &[Gate::NodeHealth, Gate::CargoTest], &Evidence { version: "0.4.3", builds: &ships_x86, tier_build: Some(x86), }, false, ) .await .unwrap(); assert_eq!(pending, vec!["node_health".to_string()]); } /// The looseness this closes: one pom version is two bundles with two /// digests, each accepted through its own intake. Checking only the build the /// source tier points at let the sibling ship on gate rows nobody read. /// Every build the promote will ship must show its own passed row. #[tokio::test] async fn every_shipped_build_must_show_its_own_gate_evidence() { let pool = fresh_pool().await; seed(&pool, "a", "4.0.0").await; let arm = seed_build(&pool, "sha-arm", "4.0.0", "/rel/aaaaaaaaaaaaaaaa").await; let x86 = seed_build(&pool, "sha-x86", "4.0.0", "/rel/bbbbbbbbbbbbbbbb").await; // Only the aarch64 half was gated. This is exactly the state a // two-architecture release passes through while the second build runs. insert_gate_build(&pool, "a", "4.0.0", "cargo_test", 1, arm).await; let both = [ PromotedBuild { platform: Some(plat("linux/aarch64")), build_id: Some(arm), }, PromotedBuild { platform: Some(plat("linux/x86_64")), build_id: Some(x86), }, ]; let pending = unsatisfied_gates( &pool, &crate::domain::AppId::default(), &tid("a"), &[Gate::CargoTest], &Evidence { version: "4.0.0", builds: &both, tier_build: None, }, false, ) .await .unwrap(); assert_eq!( pending, vec!["cargo_test (linux/x86_64)".to_string()], "the ungated half blocks the promote, and the message says which half" ); // Gate the sibling and the promote clears. Each architecture stands on // its own evidence; neither inherits the other's. insert_gate_build(&pool, "a", "4.0.0", "cargo_test", 1, x86).await; let pending = unsatisfied_gates( &pool, &crate::domain::AppId::default(), &tid("a"), &[Gate::CargoTest], &Evidence { version: "4.0.0", builds: &both, tier_build: None, }, false, ) .await .unwrap(); assert!(pending.is_empty(), "both halves gated: {pending:?}"); } /// A single-platform product's error message is exactly what it always was. /// The platform qualifier is for telling two halves apart, so adding it to a /// product that has one half would be noise in the one message an operator /// reads under pressure. #[tokio::test] async fn one_shipped_build_reports_an_unqualified_gate_name() { let pool = fresh_pool().await; seed(&pool, "a", "5.0.0").await; let only = seed_build(&pool, "sha-one", "5.0.0", "/rel/cccccccccccccccc").await; let pending = unsatisfied_gates( &pool, &crate::domain::AppId::default(), &tid("a"), &[Gate::CargoTest], &Evidence { version: "5.0.0", builds: &[PromotedBuild { platform: Some(plat("linux/x86_64")), build_id: Some(only), }], tier_build: None, }, false, ) .await .unwrap(); assert_eq!(pending, vec!["cargo_test".to_string()]); } /// `burn_in` is keyed on the tier's clock, not on a build, so a /// two-architecture promote must ask about it once rather than name it twice /// in the failure. #[tokio::test] async fn a_tier_scoped_gate_is_reported_once_across_several_builds() { let pool = fresh_pool().await; seed(&pool, "a", "6.0.0").await; let arm = seed_build(&pool, "sha-arm6", "6.0.0", "/rel/dddddddddddddddd").await; let x86 = seed_build(&pool, "sha-x866", "6.0.0", "/rel/eeeeeeeeeeeeeeee").await; let pending = unsatisfied_gates( &pool, &crate::domain::AppId::default(), &tid("a"), &[Gate::BurnIn { hours: 48 }], &Evidence { version: "6.0.0", builds: &[ PromotedBuild { platform: Some(plat("linux/aarch64")), build_id: Some(arm), }, PromotedBuild { platform: Some(plat("linux/x86_64")), build_id: Some(x86), }, ], tier_build: None, }, false, ) .await .unwrap(); assert_eq!( pending, vec!["burn_in".to_string()], "a tier-scoped gate belongs to the tier, not to each build" ); } /// A tier is usually several nodes on one architecture. Deduplicating means /// three x86_64 nodes ask about one build once, rather than repeating the /// same gate name three times in the error. #[test] fn distinct_builds_collapses_nodes_that_share_a_build() { use super::promotion::distinct_builds; let node = Node { platform: None, base_image: None, libc: None, name: "n1".into(), ssh_target: "local".into(), release_root: "/tmp/n1".into(), service_name: "makenotwork.service".into(), health_url: None, config_check_env_file: None, actuate: crate::topology::default_actuate(), observe: crate::topology::default_observe(), companions: Vec::new(), }; let bundles = vec![ ( &node, std::path::PathBuf::from("/rel/a"), Some(plat("linux/x86_64")), Some(7), ), ( &node, std::path::PathBuf::from("/rel/a"), Some(plat("linux/x86_64")), Some(7), ), ( &node, std::path::PathBuf::from("/rel/b"), Some(plat("linux/aarch64")), Some(8), ), ]; let builds = distinct_builds(&bundles); assert_eq!(builds.len(), 2); assert_eq!(builds[0].build_id, Some(7)); assert_eq!(builds[1].build_id, Some(8)); } #[tokio::test] async fn promote_rejects_an_explicit_version_that_is_not_the_source_build() { // The burn-in hole: `promote --version Y` used to check the SOURCE tier's // clock/evidence (which belong to whatever is current there), letting Y // inherit another build's 48h. Now promote resolves the source's current // build and refuses an explicit version that isn't it. let state = test_state().await; seed_version(&state.pool, "3.0.0").await; let b = seed_build(&state.pool, "shaB", "3.0.0", "/rel/deadbeefdeadbeef").await; sqlx::query( "UPDATE tier_state SET current_version='3.0.0', current_build_id=? WHERE tier='host'", ) .bind(b) .execute(&state.pool) .await .unwrap(); let err = promote_inner( state, "a".into(), PromoteBody { version: Some("2.0.0".into()), ..Default::default() }, ) .await .expect_err("promoting a version other than the source build must be refused"); assert!( matches!(&err, crate::error::Error::GateBlocked(m) if m.contains("vouched for")), "the refusal must name the mismatch, got: {err:?}", ); } #[tokio::test] async fn promote_identity_path_deploys_the_source_build_and_advances_build_id() { // Full promote through the identity path: source tier points at a build, // promote resolves the artifact through it, deploys build_runs.staged_path // (content-addressed), and advances the target's current_build_id. let mut state = test_state().await; let node_root = tempfile::tempdir().unwrap(); // Point the a-local node at a real tempdir so the symlink swap is checkable. let mut topo = (*state.topo).clone(); topo.tiers[1].nodes[0].release_root = node_root.path().to_string_lossy().into_owned(); topo.tiers[1].gates = vec![]; // isolate the deploy fan-out state.topo = Arc::new(topo); state.executors = Arc::new(crate::state::build_executors(&state.topo)); // deploys.node FKs into `nodes`, so the promote path needs the row. sqlx::query("INSERT INTO nodes (name, tier, ssh_target, release_root) VALUES ('a-local', 'a', 'local', ?)") .bind(node_root.path().to_string_lossy().into_owned()) .execute(&state.pool) .await .unwrap(); seed_version(&state.pool, "3.0.0").await; let staged = format!( "{}/releases/abc123abc123abc1", node_root.path().to_string_lossy() ); let b = seed_build(&state.pool, "shaB", "3.0.0", &staged).await; sqlx::query( "UPDATE tier_state SET current_version='3.0.0', current_build_id=? WHERE tier='host'", ) .bind(b) .execute(&state.pool) .await .unwrap(); let pool = state.pool.clone(); let Json(body) = promote_inner(state, "a".into(), PromoteBody::default()) .await .expect("identity-path promote succeeds"); assert_eq!(body["version"], "3.0.0"); // Target tier advanced with the build identity, not just the version. let (cur_v, cur_b): (Option, Option) = sqlx::query_as("SELECT current_version, current_build_id FROM tier_state WHERE tier = 'a'") .fetch_one(&pool) .await .unwrap(); assert_eq!(cur_v.as_deref(), Some("3.0.0")); assert_eq!(cur_b, Some(b), "target records the promoted build id"); // The deploy row is attributed to the build. let deploy_b: Option = sqlx::query_scalar( "SELECT build_id FROM deploys WHERE tier = 'a' ORDER BY id DESC LIMIT 1", ) .fetch_one(&pool) .await .unwrap(); assert_eq!(deploy_b, Some(b)); // The node's `current` points at the content-addressed release dir, whose // name is the staged_path's basename — not the version. let link = tokio::fs::read_link(node_root.path().join("current")) .await .unwrap(); assert_eq!(link.to_string_lossy(), "releases/abc123abc123abc1"); } /// Insert a bare `versions` row (FK target for tier_state.current_version). async fn seed_version(pool: &SqlitePool, version: &str) { sqlx::query("INSERT INTO versions (version, git_sha, built_at, artifact_path) VALUES (?, 'sha', datetime('now'), '/tmp/x') ON CONFLICT DO NOTHING") .bind(version).execute(pool).await.unwrap(); } // ---- per-node bundle resolution (wiki release-artifact-identity) ---- /// A settled build of `version` recorded as `platform`'s half of it. async fn seed_platform_build( pool: &SqlitePool, sha: &str, version: &str, platform: &str, staged_path: &str, ) -> i64 { sqlx::query_scalar( "INSERT INTO build_runs (sha, version, phase, result, started_at, bundle_digest, staged_path, platform) VALUES (?, ?, 'done', 'passed', datetime('now'), ?, ?, ?) RETURNING id", ) .bind(sha) .bind(version) .bind(format!("{sha}-digest")) .bind(staged_path) .bind(platform) .fetch_one(pool) .await .unwrap() } fn node_on(name: &str, platform: Option<&str>) -> Node { Node { platform: platform.map(plat), base_image: None, libc: None, name: name.into(), ssh_target: "local".into(), release_root: format!("/tmp/{name}"), service_name: "makenotwork.service".into(), health_url: None, config_check_env_file: None, actuate: crate::topology::default_actuate(), observe: crate::topology::default_observe(), companions: Vec::new(), } } /// One pom version is two bundles with two digests. The node states which /// architecture it can run, and the sibling bundle is resolved out of /// `build_runs` rather than the caller's own half being shipped everywhere. #[tokio::test] async fn a_node_on_the_other_architecture_gets_its_own_bundle() { let state = test_state().await; seed_version(&state.pool, "5.0.0").await; let arm = seed_platform_build( &state.pool, "sha-arm", "5.0.0", "linux/aarch64", "/rel/aaaaaaaaaaaaaaaa", ) .await; let x86 = seed_platform_build( &state.pool, "sha-x86", "5.0.0", "linux/x86_64", "/rel/bbbbbbbbbbbbbbbb", ) .await; let n_arm = node_on("astra", Some("linux/aarch64")); let n_x86 = node_on("hetzner", Some("linux/x86_64")); let bundles = super::promotion::bundles_for_nodes( &state, "5.0.0", &[&n_arm, &n_x86], std::path::Path::new("/rel/aaaaaaaaaaaaaaaa"), Some(&plat("linux/aarch64")), Some(arm), ) .await .expect("both architectures have a green bundle at this version"); assert_eq!(bundles.len(), 2); assert_eq!( bundles[0].1, std::path::PathBuf::from("/rel/aaaaaaaaaaaaaaaa") ); assert_eq!(bundles[0].2, Some(plat("linux/aarch64"))); assert_eq!(bundles[0].3, Some(arm)); // The one the caller never held: resolved by platform, and it carries // the sibling's own build id so its own gate evidence is what gets // checked. assert_eq!( bundles[1].1, std::path::PathBuf::from("/rel/bbbbbbbbbbbbbbbb") ); assert_eq!(bundles[1].2, Some(plat("linux/x86_64"))); assert_eq!(bundles[1].3, Some(x86)); } /// A version with no green bundle for the node's architecture fails the /// whole resolution, before any node is touched. #[tokio::test] async fn a_missing_architecture_half_refuses_the_promote_rather_than_defaulting() { let state = test_state().await; seed_version(&state.pool, "5.0.0").await; seed_platform_build( &state.pool, "sha-arm", "5.0.0", "linux/aarch64", "/rel/aaaaaaaaaaaaaaaa", ) .await; let n_x86 = node_on("hetzner", Some("linux/x86_64")); let err = super::promotion::bundles_for_nodes( &state, "5.0.0", &[&n_x86], std::path::Path::new("/rel/aaaaaaaaaaaaaaaa"), Some(&plat("linux/aarch64")), None, ) .await .expect_err("the x86_64 half was never built, so there is nothing to ship"); assert!( format!("{err:?}").contains("no green linux/x86_64 bundle"), "{err:?}" ); } /// When the node and the caller name the same platform, the caller's bundle /// is the answer and no lookup happens. A single-platform product whose /// nodes have started stating a platform still has no `build_runs.platform` /// row to find, so a lookup here would refuse a promote that is fine. #[tokio::test] async fn a_node_that_agrees_with_the_caller_takes_the_callers_bundle() { let state = test_state().await; seed_version(&state.pool, "5.0.0").await; let node = node_on("hetzner", Some("linux/x86_64")); let bundles = super::promotion::bundles_for_nodes( &state, "5.0.0", &[&node], std::path::Path::new("/rel/legacy"), Some(&plat("linux/x86_64")), None, ) .await .expect("the caller already holds the bundle this node wants"); assert_eq!(bundles[0].1, std::path::PathBuf::from("/rel/legacy")); assert_eq!(bundles[0].2, Some(plat("linux/x86_64"))); assert_eq!(bundles[0].3, None); } /// The previous version is two bundles too. If the one this node runs /// cannot be resolved, nothing is attempted anywhere and every touched node /// is indeterminate — which is the truth, not a default. #[tokio::test] async fn a_rollback_that_cannot_resolve_a_bundle_reports_every_node_indeterminate() { let state = test_state().await; sqlx::query("INSERT INTO versions (version, git_sha, built_at, artifact_path) VALUES ('1.0.0','sha',datetime('now'),'/tmp/staged/releases/1.0.0/makenotwork')") .execute(&state.pool).await.unwrap(); // No build_runs row for either node's architecture at 1.0.0. let n1 = node_on("n1", Some("linux/aarch64")); let n2 = node_on("n2", Some("linux/aarch64")); let report = rollback_deployed_nodes(&state, &tid("a"), &[&n1, &n2], "1.0.0").await; assert_eq!(report.restored, 0); assert_eq!( report.indeterminate, 2, "one per node the rollback never reached: {report:?}" ); assert!(!report.is_consistent()); } /// A rollback that fails at the symlink swap on one node of several leaves /// that node unknown and the rest restored. The count has to be per node: /// a tier reported wholesale indeterminate sends an operator to inspect /// boxes that are fine, and a tier reported wholesale restored hides the /// one that is not. #[tokio::test] async fn a_rollback_failing_on_one_node_counts_only_that_node_indeterminate() { // The marker matches the swap-and-restart script, which is the only op // annotated AtOrAfterSwap; a1's rollback therefore lands in the // indeterminate arm rather than the already-on-previous one. let (state, _log) = fleet_fixture(&["a1", "a2"], Some("a1"), "reload-or-restart").await; let nodes: Vec = state.topo.tiers[1].nodes.clone(); let refs: Vec<&Node> = nodes.iter().collect(); let report = rollback_deployed_nodes(&state, &tid("a"), &refs, "1.0.0").await; assert_eq!(report.restored, 1, "a2 came back: {report:?}"); assert_eq!( report.indeterminate, 1, "only the node whose swap failed is unknown: {report:?}" ); assert_eq!(report.already_on_previous, 0); assert_eq!(report.touched(), 2); assert!(!report.is_consistent()); }