max / makenotwork
3 files changed,
+973 insertions,
-500 deletions
| @@ -1,2130 +0,0 @@ | |||
| 1 | - | use crate::error::Result; | |
| 2 | - | use crate::state::AppState; | |
| 3 | - | use axum::extract::{Path, Query, State, WebSocketUpgrade}; | |
| 4 | - | use axum::response::IntoResponse; | |
| 5 | - | use axum::routing::{get, post}; | |
| 6 | - | use axum::{Json, Router}; | |
| 7 | - | use serde::{Deserialize, Serialize}; | |
| 8 | - | use sqlx::Row; | |
| 9 | - | ||
| 10 | - | pub fn router(state: AppState) -> Router { | |
| 11 | - | let prom = state.prom.clone(); | |
| 12 | - | let token = state.api_token.clone(); | |
| 13 | - | ||
| 14 | - | // Every route requires the bearer token (when one is configured): the | |
| 15 | - | // mutators carry prod-deploy authority (CF2), and the reads expose prod | |
| 16 | - | // state — deployed versions, build SHAs, full gate logs, the live event | |
| 17 | - | // stream — that a tailnet peer should not get unauthenticated. `/metrics` | |
| 18 | - | // is the one deliberate opt-out (Prometheus convention, no secrets), added | |
| 19 | - | // outside the layered group below. The bearer-success path logs the caller | |
| 20 | - | // identity for mutating (POST) requests so a prod ship is attributable. | |
| 21 | - | let protected = Router::new() | |
| 22 | - | // mutators — prod-deploy authority | |
| 23 | - | .route("/promote/{tier}", post(promote)) | |
| 24 | - | .route("/rollback/{tier}", post(rollback)) | |
| 25 | - | .route("/rebuild", post(rebuild)) | |
| 26 | - | .route("/self-update", post(self_update)) | |
| 27 | - | .route("/confirm/{tier}", post(confirm)) | |
| 28 | - | .route("/backup/fetch", post(backup_fetch)) | |
| 29 | - | // reads — prod state, now also gated | |
| 30 | - | .route("/state", get(get_state)) | |
| 31 | - | .route("/runs/{id}", get(get_run)) | |
| 32 | - | .route("/runs/{id}/wait", get(get_run_wait)) | |
| 33 | - | .route("/logs/{version}/{gate}", get(get_gate_log)) | |
| 34 | - | .route("/events", get(events_ws)) | |
| 35 | - | .route_layer(axum::middleware::from_fn(move |req, next| { | |
| 36 | - | require_bearer(token.clone(), req, next) | |
| 37 | - | })); | |
| 38 | - | ||
| 39 | - | Router::new() | |
| 40 | - | .merge(protected) | |
| 41 | - | .with_state(state) | |
| 42 | - | .route("/metrics", get(crate::metrics::render).with_state(prom)) | |
| 43 | - | } | |
| 44 | - | ||
| 45 | - | /// Bearer-token gate for the deploy mutators. When no token is configured the | |
| 46 | - | /// request passes (main() only allows that on a loopback bind). Comparison is | |
| 47 | - | /// constant-time to avoid leaking the token via timing. | |
| 48 | - | async fn require_bearer( | |
| 49 | - | token: Option<std::sync::Arc<str>>, | |
| 50 | - | req: axum::extract::Request, | |
| 51 | - | next: axum::middleware::Next, | |
| 52 | - | ) -> axum::response::Response { | |
| 53 | - | let Some(expected) = token.as_deref() else { | |
| 54 | - | return next.run(req).await; | |
| 55 | - | }; | |
| 56 | - | let path = req.uri().path().to_string(); | |
| 57 | - | let method = req.method().clone(); | |
| 58 | - | let ok = req | |
| 59 | - | .headers() | |
| 60 | - | .get(axum::http::header::AUTHORIZATION) | |
| 61 | - | .and_then(|h| h.to_str().ok()) | |
| 62 | - | .and_then(|h| h.strip_prefix("Bearer ")) | |
| 63 | - | .is_some_and(|t| ct_eq(t, expected)); | |
| 64 | - | if ok { | |
| 65 | - | // Attribute mutations (POST) to a caller. Reads are GET and poll every | |
| 66 | - | // few seconds, so logging them would drown the journal; the audit value | |
| 67 | - | // is in *who shipped/rolled back prod*, which is always a POST. The peer | |
| 68 | - | // address is the closest identity we have without per-operator tokens | |
| 69 | - | // (`tailscale whois` is a future refinement); a shared token at least | |
| 70 | - | // gets a source host into the trail. | |
| 71 | - | if method == axum::http::Method::POST { | |
| 72 | - | let peer = req | |
| 73 | - | .extensions() | |
| 74 | - | .get::<axum::extract::ConnectInfo<std::net::SocketAddr>>() | |
| 75 | - | .map(|ci| ci.0.to_string()) | |
| 76 | - | .unwrap_or_else(|| "unknown".into()); | |
| 77 | - | tracing::info!(path = %path, peer = %peer, "authenticated deploy mutation"); | |
| 78 | - | } | |
| 79 | - | next.run(req).await | |
| 80 | - | } else { | |
| 81 | - | tracing::warn!(path = %path, "rejected unauthenticated request"); | |
| 82 | - | (axum::http::StatusCode::UNAUTHORIZED, "missing or invalid bearer token\n").into_response() | |
| 83 | - | } | |
| 84 | - | } | |
| 85 | - | ||
| 86 | - | /// Constant-time string compare. Length is allowed to leak (a token's length is | |
| 87 | - | /// not the secret); the byte comparison does not short-circuit. | |
| 88 | - | fn ct_eq(a: &str, b: &str) -> bool { | |
| 89 | - | let (a, b) = (a.as_bytes(), b.as_bytes()); | |
| 90 | - | if a.len() != b.len() { | |
| 91 | - | return false; | |
| 92 | - | } | |
| 93 | - | let mut diff = 0u8; | |
| 94 | - | for (x, y) in a.iter().zip(b) { | |
| 95 | - | diff |= x ^ y; | |
| 96 | - | } | |
| 97 | - | diff == 0 | |
| 98 | - | } | |
| 99 | - | ||
| 100 | - | #[derive(Serialize)] | |
| 101 | - | struct StateView { | |
| 102 | - | /// The running sandod's own package version. Lets a self-update caller | |
| 103 | - | /// confirm the new binary is live after the restart (the tier versions are | |
| 104 | - | /// the *deployed product*, not the controller). | |
| 105 | - | sandod_version: &'static str, | |
| 106 | - | tiers: Vec<TierView>, | |
| 107 | - | /// The most recent build run (the resource `GET /runs/{id}` exposes in | |
| 108 | - | /// full). Surfaced here so a `/state` poller sees an in-flight or failed | |
| 109 | - | /// build — the tier versions only ever reflect the last *success*, so | |
| 110 | - | /// without this `/state` looks frozen for the whole build. `null` until | |
| 111 | - | /// the first `/rebuild`. | |
| 112 | - | build: Option<crate::runs::BuildSummary>, | |
| 113 | - | } | |
| 114 | - | ||
| 115 | - | #[derive(Serialize)] | |
| 116 | - | struct TierView { | |
| 117 | - | name: String, | |
| 118 | - | ord: i64, | |
| 119 | - | provisioned: bool, | |
| 120 | - | canary: String, | |
| 121 | - | current_version: Option<String>, | |
| 122 | - | previous_version: Option<String>, | |
| 123 | - | burn_in_started_at: Option<String>, | |
| 124 | - | /// Non-null when the tier was left in a partial / mixed-version state by a | |
| 125 | - | /// failed promote or rollback whose compensation could not fully restore | |
| 126 | - | /// consistency. NULL when the tier is consistent. The TUI flags it red. | |
| 127 | - | partial_reason: Option<String>, | |
| 128 | - | nodes: Vec<String>, | |
| 129 | - | gates: Vec<GateView>, | |
| 130 | - | } | |
| 131 | - | ||
| 132 | - | #[derive(Serialize)] | |
| 133 | - | struct GateView { | |
| 134 | - | kind: String, | |
| 135 | - | finished_at: Option<String>, | |
| 136 | - | /// `'passed' | 'failed' | 'blocked'` or NULL while in-flight. The TUI | |
| 137 | - | /// uses this to choose green/red/yellow rendering. | |
| 138 | - | status: Option<String>, | |
| 139 | - | /// Full typed `GateOutcome` as a JSON object, when present. | |
| 140 | - | /// Deserialized lazily by the consumer; sandod doesn't re-parse it. | |
| 141 | - | outcome: Option<serde_json::Value>, | |
| 142 | - | /// Relative path under `cfg.logs_root` to the persisted stdout/stderr. | |
| 143 | - | log_ref: Option<String>, | |
| 144 | - | } | |
| 145 | - | ||
| 146 | - | async fn get_state(State(s): State<AppState>) -> Result<Json<StateView>> { | |
| 147 | - | let rows = sqlx::query( | |
| 148 | - | "SELECT t.name, t.ord, t.provisioned, t.canary, | |
| 149 | - | ts.current_version, ts.previous_version, ts.burn_in_started_at, | |
| 150 | - | ts.partial_reason | |
| 151 | - | FROM tiers t | |
| 152 | - | LEFT JOIN tier_state ts ON ts.tier = t.name | |
| 153 | - | ORDER BY t.ord", | |
| 154 | - | ) | |
| 155 | - | .fetch_all(&s.pool) | |
| 156 | - | .await?; | |
| 157 | - | ||
| 158 | - | let mut tiers = Vec::with_capacity(rows.len()); | |
| 159 | - | for r in rows { | |
| 160 | - | let name: String = r.get("name"); | |
| 161 | - | let current_version: Option<String> = r.get("current_version"); | |
| 162 | - | ||
| 163 | - | let nodes: Vec<String> = sqlx::query_scalar("SELECT name FROM nodes WHERE tier = ? ORDER BY name") | |
| 164 | - | .bind(&name) | |
| 165 | - | .fetch_all(&s.pool) | |
| 166 | - | .await?; | |
| 167 | - | ||
| 168 | - | // Surface gates for current_version when set, otherwise for the most | |
| 169 | - | // recently attempted version on this tier. Without the fallback, a | |
| 170 | - | // tier that has never gone green (MM after a build failure, B before | |
| 171 | - | // first deploy) exposes no gate detail via /state — debugging required | |
| 172 | - | // SSH and direct SQLite access. See sando todo: gate observability. | |
| 173 | - | let gate_version: Option<String> = if current_version.is_some() { | |
| 174 | - | current_version.clone() | |
| 175 | - | } else { | |
| 176 | - | sqlx::query_scalar( | |
| 177 | - | "SELECT version FROM gate_runs WHERE tier = ? | |
| 178 | - | ORDER BY id DESC LIMIT 1", | |
| 179 | - | ) | |
| 180 | - | .bind(&name) | |
| 181 | - | .fetch_optional(&s.pool) | |
| 182 | - | .await? | |
| 183 | - | }; | |
| 184 | - | ||
| 185 | - | let gates: Vec<GateView> = if let Some(ver) = gate_version.as_ref() { | |
| 186 | - | // Most recent gate_runs row per gate_kind for (tier, ver). | |
| 187 | - | sqlx::query( | |
| 188 | - | "SELECT gate_kind, finished_at, status, outcome_json, log_ref | |
| 189 | - | FROM gate_runs g | |
| 190 | - | WHERE tier = ?1 AND version = ?2 | |
| 191 | - | AND id = (SELECT MAX(id) FROM gate_runs | |
| 192 | - | WHERE tier = ?1 AND version = ?2 AND gate_kind = g.gate_kind) | |
| 193 | - | ORDER BY gate_kind", | |
| 194 | - | ) | |
| 195 | - | .bind(&name) | |
| 196 | - | .bind(ver) | |
| 197 | - | .fetch_all(&s.pool) | |
| 198 | - | .await? | |
| 199 | - | .into_iter() | |
| 200 | - | .map(|gr| GateView { | |
| 201 | - | kind: gr.get("gate_kind"), | |
| 202 | - | finished_at: gr.get("finished_at"), | |
| 203 | - | status: gr.get("status"), | |
| 204 | - | outcome: gr.get::<Option<String>, _>("outcome_json") | |
| 205 | - | .and_then(|s| serde_json::from_str(&s).ok()), | |
| 206 | - | log_ref: gr.get("log_ref"), | |
| 207 | - | }) | |
| 208 | - | .collect() | |
| 209 | - | } else { | |
| 210 | - | Vec::new() | |
| 211 | - | }; | |
| 212 | - | ||
| 213 | - | tiers.push(TierView { | |
| 214 | - | name, | |
| 215 | - | ord: r.get("ord"), | |
| 216 | - | provisioned: r.get::<i64, _>("provisioned") != 0, | |
| 217 | - | canary: r.get("canary"), | |
| 218 | - | current_version, | |
| 219 | - | previous_version: r.get("previous_version"), | |
| 220 | - | burn_in_started_at: r.get("burn_in_started_at"), | |
| 221 | - | partial_reason: r.get("partial_reason"), | |
| 222 | - | nodes, | |
| 223 | - | gates, | |
| 224 | - | }); | |
| 225 | - | } | |
| 226 | - | ||
| 227 | - | let build = crate::runs::latest_summary(&s.pool).await?; | |
| 228 | - | Ok(Json(StateView { sandod_version: env!("CARGO_PKG_VERSION"), tiers, build })) | |
| 229 | - | } | |
| 230 | - | ||
| 231 | - | #[derive(Deserialize, Default)] | |
| 232 | - | struct PromoteBody { | |
| 233 | - | /// Optional. If absent, defaults to the predecessor tier's `current_version` | |
| 234 | - | /// (i.e. promote whatever just finished baking on the previous tier). | |
| 235 | - | #[serde(default)] | |
| 236 | - | version: Option<String>, | |
| 237 | - | #[serde(default)] | |
| 238 | - | hotfix: bool, | |
| 239 | - | #[serde(default)] | |
| 240 | - | reset_burn_in: bool, | |
| 241 | - | } | |
| 242 | - | ||
| 243 | - | async fn promote( | |
| 244 | - | State(s): State<AppState>, | |
| 245 | - | Path(tier): Path<String>, | |
| 246 | - | body: Option<Json<PromoteBody>>, | |
| 247 | - | ) -> Result<Json<serde_json::Value>> { | |
| 248 | - | let body = body.map(|Json(b)| b).unwrap_or_default(); | |
| 249 | - | // Run the deploy in a detached task, not inline in the request future. | |
| 250 | - | // A promote's rsync + restart can outlast a client's socket timeout; if the | |
| 251 | - | // client disconnects, axum drops the handler future — and if the deploy ran | |
| 252 | - | // inline that would abandon the rollout mid-flight (symlink half-swapped, | |
| 253 | - | // tier_state not advanced, and — before the kill_on_drop fix — an orphaned | |
| 254 | - | // rsync). Spawning detaches the work from the connection: the task owns the | |
| 255 | - | // deploy lock and runs to completion regardless, while a still-connected | |
| 256 | - | // client still receives the result. Mirrors how /rebuild already runs its | |
| 257 | - | // build off-request. | |
| 258 | - | tokio::spawn(promote_inner(s, tier, body)) | |
| 259 | - | .await | |
| 260 | - | .map_err(|e| crate::error::Error::Other(anyhow::anyhow!("promote task failed to join: {e}")))? | |
| 261 | - | } | |
| 262 | - | ||
| 263 | - | async fn promote_inner( | |
| 264 | - | s: AppState, | |
| 265 | - | tier: String, | |
| 266 | - | body: PromoteBody, | |
| 267 | - | ) -> Result<Json<serde_json::Value>> { | |
| 268 | - | // Serialize the whole check -> deploy -> advance against any concurrent | |
| 269 | - | // promote/rollback (CF3). Held for the entire task. | |
| 270 | - | let _deploy_guard = s.deploy_lock.lock().await; | |
| 271 | - | let tier = crate::domain::TierId::new(tier); | |
| 272 | - | let idx = s.topo.tiers.iter().position(|t| t.name == tier) | |
| 273 | - | .ok_or(crate::error::Error::NotFound)?; | |
| 274 | - | if idx == 0 { | |
| 275 | - | return Err(crate::error::Error::GateBlocked( | |
| 276 | - | "cannot /promote to the first tier; use /rebuild".into(), | |
| 277 | - | )); | |
| 278 | - | } | |
| 279 | - | let target = &s.topo.tiers[idx]; | |
| 280 | - | let source = &s.topo.tiers[idx - 1]; | |
| 281 | - | ||
| 282 | - | // Resolve version: explicit if given, else the source tier's current. | |
| 283 | - | let version_str = match body.version { | |
| 284 | - | Some(v) => v, | |
| 285 | - | None => sqlx::query_scalar::<_, Option<String>>( | |
| 286 | - | "SELECT current_version FROM tier_state WHERE tier = ?", | |
| 287 | - | ) | |
| 288 | - | .bind(&source.name) | |
| 289 | - | .fetch_optional(&s.pool).await | |
| 290 | - | .map_err(crate::error::Error::Db)? | |
| 291 | - | .flatten() | |
| 292 | - | .ok_or_else(|| crate::error::Error::GateBlocked( | |
| 293 | - | format!("no version specified and tier {} has no current_version", source.name), | |
| 294 | - | ))?, | |
| 295 | - | }; | |
| 296 | - | let version = crate::domain::Version::parse(&version_str) | |
| 297 | - | .map_err(|e| crate::error::Error::Other(anyhow::anyhow!(e)))?; | |
| 298 | - | ||
| 299 | - | // 1. Predecessor must have all of its configured gates satisfied for this | |
| 300 | - | // version (with optional hotfix override that skips burn_in). Evaluated | |
| 301 | - | // against the topology gate list, so a gate that never ran blocks the | |
| 302 | - | // promote instead of being treated as green. | |
| 303 | - | let pending = unsatisfied_gates(&s.pool, &source.name, &source.gates, &version_str, body.hotfix).await?; | |
| 304 | - | if !pending.is_empty() { | |
| 305 | - | return Err(crate::error::Error::GateBlocked(format!( | |
| 306 | - | "{} gate(s) not satisfied on tier {}: {}", | |
| 307 | - | pending.len(), | |
| 308 | - | source.name, | |
| 309 | - | pending.join(", "), | |
| 310 | - | ))); | |
| 311 | - | } | |
| 312 | - | ||
| 313 | - | // 2. Look up the artifact for this version. | |
| 314 | - | let bin: Option<(String,)> = sqlx::query_as( | |
| 315 | - | "SELECT artifact_path FROM versions WHERE version = ?", | |
| 316 | - | ) | |
| 317 | - | .bind(&version) | |
| 318 | - | .fetch_optional(&s.pool) | |
| 319 | - | .await | |
| 320 | - | .map_err(crate::error::Error::Db)?; | |
| 321 | - | let Some((bin,)) = bin else { | |
| 322 | - | return Err(crate::error::Error::NotFound); | |
| 323 | - | }; | |
| 324 | - | let bin_path = std::path::PathBuf::from(bin); | |
| 325 | - | // `artifact_path` is the primary binary; the staged release dir is its parent. | |
| 326 | - | let staged_dir = bin_path.parent() | |
| 327 | - | .ok_or_else(|| crate::error::Error::Other(anyhow::anyhow!("artifact_path has no parent")))? | |
| 328 | - | .to_path_buf(); | |
| 329 | - | ||
| 330 | - | // The version this tier was running before this promote — the rollback | |
| 331 | - | // target if a canary node fails partway through a multi-node rollout. | |
| 332 | - | let prev_version: Option<String> = sqlx::query_scalar( | |
| 333 | - | "SELECT current_version FROM tier_state WHERE tier = ?", | |
| 334 | - | ) | |
| 335 | - | .bind(&target.name) | |
| 336 | - | .fetch_optional(&s.pool).await.map_err(crate::error::Error::Db)?.flatten(); | |
| 337 | - | ||
| 338 | - | // 3. Deploy to each node. Sequential canary is the only policy | |
| 339 | - | // implemented in v0; parallel is a one-line change once we trust the | |
| 340 | - | // sequential path. Track the nodes already flipped to the new version so | |
| 341 | - | // a mid-rollout failure can roll them back (canary rollback). | |
| 342 | - | let mut deployed: Vec<&crate::topology::Node> = Vec::new(); | |
| 343 | - | for node in &target.nodes { | |
| 344 | - | let started = chrono::Utc::now().to_rfc3339(); | |
| 345 | - | crate::events::emit(&s.events, crate::events::Event::DeployStart { | |
| 346 | - | tier: target.name.clone(), node: node.name.clone(), version: version.clone(), | |
| 347 | - | }); | |
| 348 | - | let executor = s.executors.get(&node.name).cloned() | |
| 349 | - | .unwrap_or_else(|| crate::state::build_executor(node)); | |
| 350 | - | let result = crate::deploy::deploy_node(executor.as_ref(), node, &version_str, &staged_dir, s.cfg.primary_bin()).await; | |
| 351 | - | let finished = chrono::Utc::now().to_rfc3339(); | |
| 352 | - | let (outcome_obj, err_for_propagation) = match result { | |
| 353 | - | Ok(_) => (crate::outcome::DeployOutcome::ok(), None), | |
| 354 | - | Err(e) => { | |
| 355 | - | let msg = format!("{e:#}"); | |
| 356 | - | let kind = crate::classify::classify_deploy_error(&msg); | |
| 357 | - | (crate::outcome::DeployOutcome::failed(kind), Some(e)) | |
| 358 | - | } | |
| 359 | - | }; | |
| 360 | - | let outcome_json = serde_json::to_string(&outcome_obj) | |
| 361 | - | .unwrap_or_else(|e| format!("{{\"_serialize_error\":{e:?}}}")); | |
| 362 | - | sqlx::query( | |
| 363 | - | "INSERT INTO deploys (version, tier, node, started_at, finished_at, outcome, outcome_json, hotfix, reset_burn_in) | |
| 364 | - | VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)", | |
| 365 | - | ) | |
| 366 | - | .bind(&version).bind(&target.name).bind(&node.name) | |
| 367 | - | .bind(&started).bind(&finished).bind(outcome_obj.status_str()) | |
| 368 | - | .bind(&outcome_json) | |
| 369 | - | .bind(body.hotfix as i64).bind(body.reset_burn_in as i64) | |
| 370 | - | .execute(&s.pool).await.map_err(crate::error::Error::Db)?; | |
| 371 | - | if let Some(e) = err_for_propagation { | |
| 372 | - | let crate::outcome::DeployStatus::Failed { failure } = outcome_obj.status else { | |
| 373 | - | unreachable!("err_for_propagation is Some iff status is Failed"); | |
| 374 | - | }; | |
| 375 | - | tracing::error!( | |
| 376 | - | tier = %target.name, node = %node.name, version = %version, | |
| 377 | - | failure = failure.summary(), | |
| 378 | - | "deploy failed; current symlink left intact, tier_state not advanced" | |
| 379 | - | ); | |
| 380 | - | crate::events::emit(&s.events, crate::events::Event::DeployFailed { | |
| 381 | - | tier: target.name.clone(), node: node.name.clone(), | |
| 382 | - | version: version.clone(), failure, | |
| 383 | - | }); | |
| 384 | - | ||
| 385 | - | // Canary rollback: restore every node this promote touched — the | |
| 386 | - | // ones already flipped to the new version AND this failed node | |
| 387 | - | // (whose state is indeterminate: the symlink swap may have landed | |
| 388 | - | // before the restart failed) — back to the tier's prior version, so | |
| 389 | - | // the fleet is left consistent on `prev` rather than split-brain. | |
| 390 | - | // Nodes after this one were never touched and stay on `prev`. | |
| 391 | - | deployed.push(node); | |
| 392 | - | let touched = deployed.len(); | |
| 393 | - | match prev_version.as_deref() { | |
| 394 | - | Some(prev) => { | |
| 395 | - | let restored = rollback_deployed_nodes(&s, &target.name, &deployed, prev).await; | |
| 396 | - | tracing::warn!( | |
| 397 | - | tier = %target.name, restored, of = touched, | |
| 398 | - | from = %version, to = prev, | |
| 399 | - | "canary failed mid-rollout; rolled touched nodes back to the previous version", | |
| 400 | - | ); | |
| 401 | - | if restored > 0 | |
| 402 | - | && let Ok(prev_v) = crate::domain::Version::parse(prev) | |
| 403 | - | { | |
| 404 | - | crate::events::emit(&s.events, crate::events::Event::Rollback { | |
| 405 | - | tier: target.name.clone(), from: version.clone(), to: prev_v, | |
| 406 | - | }); | |
| 407 | - | } | |
| 408 | - | // If every touched node was restored, the tier is consistent on | |
| 409 | - | // `prev` — clear any stale flag. Otherwise it is genuinely | |
| 410 | - | // split-brain: record exactly how, for /state. | |
| 411 | - | if restored == touched { | |
| 412 | - | clear_partial(&s, &target.name).await; | |
| 413 | - | } else { | |
| 414 | - | set_partial(&s, &target.name, &format!( | |
| 415 | - | "canary rollback incomplete: {restored}/{touched} nodes restored to {prev}; \ | |
| 416 | - | {} may still be on {version} — manual check needed", | |
| 417 | - | touched - restored, | |
| 418 | - | )).await; | |
| 419 | - | } | |
| 420 | - | } | |
| 421 | - | None => { | |
| 422 | - | tracing::error!( | |
| 423 | - | tier = %target.name, count = touched, version = %version, | |
| 424 | - | "canary failed on a first deploy (no previous version to restore to); \ | |
| 425 | - | touched nodes remain on the new version — manual cleanup needed", | |
| 426 | - | ); | |
| 427 | - | set_partial(&s, &target.name, &format!( | |
| 428 | - | "first-deploy canary failed: {touched} node(s) left on {version}, \ | |
| 429 | - | no prior version to restore — manual cleanup needed", | |
| 430 | - | )).await; | |
| 431 | - | } | |
| 432 | - | } | |
| 433 | - | return Err(crate::error::Error::Other(e)); | |
| 434 | - | } | |
| 435 | - | deployed.push(node); | |
| 436 | - | crate::events::emit(&s.events, crate::events::Event::DeployOk { | |
| 437 | - | tier: target.name.clone(), node: node.name.clone(), version: version.clone(), | |
| 438 | - | }); | |
| 439 | - | } | |
| 440 | - | ||
| 441 | - | // 3b. Run this tier's post-deploy gates (node_health) against the freshly | |
| 442 | - | // deployed nodes and record their outcomes. These rows are the evidence | |
| 443 | - | // the NEXT promote (this tier -> the following one) checks via | |
| 444 | - | // `unsatisfied_gates`. Before CF1, only the host tier ran gates, so | |
| 445 | - | // A/B/C had no evidence and promotion waved through; node_health now | |
| 446 | - | // proves the deployed nodes are serving (Run-2 SERIOUS-3: boot_smoke | |
| 447 | - | // used to re-run the staged binary locally and proved nothing about the | |
| 448 | - | // node). burn_in / manual_confirm are not run here — they are evaluated | |
| 449 | - | // live / by the operator at the next promote. A failed gate does not | |
| 450 | - | // unwind this deploy (the artifact is already live on the tier); it | |
| 451 | - | // blocks the next promote, which is the fail-closed behavior we want. | |
| 452 | - | let post_deploy: Vec<crate::topology::Gate> = | |
| 453 | - | target.gates.iter().filter(|g| g.runs_post_deploy()).cloned().collect(); | |
| 454 | - | if !post_deploy.is_empty() { | |
| 455 | - | // node_health probes each node the deploy just shipped to, over the same | |
| 456 | - | // executor the deploy used. Build the probe set from the tier's nodes and | |
| 457 | - | // the startup executor map; a node missing an executor (shouldn't happen | |
| 458 | - | // — both come from the same topology) is skipped, and an empty set makes | |
| 459 | - | // node_health Blocked (fail closed). | |
| 460 | - | let nodes: Vec<crate::gates::NodeProbe> = target | |
| 461 | - | .nodes | |
| 462 | - | .iter() | |
| 463 | - | .filter_map(|n| { | |
| 464 | - | s.executors.get(&n.name).map(|exec| crate::gates::NodeProbe { | |
| 465 | - | node: n.name.clone(), | |
| 466 | - | service: n.service_name.clone(), | |
| 467 | - | health_url: n.health_url.clone(), | |
| 468 | - | executor: exec.clone(), | |
| 469 | - | }) | |
| 470 | - | }) | |
| 471 | - | .collect(); | |
| 472 | - | let ctx = crate::gates::GateCtx { | |
| 473 | - | pool: s.pool.clone(), | |
| 474 | - | cfg: s.cfg.clone(), | |
| 475 | - | tier: target.name.clone(), | |
| 476 | - | version: version.clone(), | |
| 477 | - | // No worktree at promote time; node_health works over executors, not | |
| 478 | - | // a checkout. | |
| 479 | - | worktree: std::path::PathBuf::new(), | |
| 480 | - | events: s.events.clone(), | |
| 481 | - | nodes, | |
| 482 | - | }; | |
| 483 | - | match crate::gates::run_all(&ctx, &post_deploy).await { | |
| 484 | - | Ok(true) => {} | |
| 485 | - | Ok(false) => tracing::warn!( | |
| 486 | - | tier = %target.name, version = %version, | |
| 487 | - | "post-deploy gate(s) failed; tier advanced but promotion to the next tier will be blocked", | |
| 488 | - | ), | |
| 489 | - | Err(e) => tracing::error!( | |
| 490 | - | tier = %target.name, version = %version, error = %e, | |
| 491 | - | "post-deploy gate execution errored; promotion to the next tier will be blocked", | |
| 492 | - | ), | |
| 493 | - | } | |
| 494 | - | } | |
| 495 | - | ||
| 496 | - | // 4. Advance tier_state through the single sealed forward-advance op (atomic | |
| 497 | - | // self-referential UPDATE; no read-modify-write to lose under concurrency, | |
| 498 | - | // CF3). We hold deploy_lock for this whole handler, so the advance is | |
| 499 | - | // serialized against rollback and the host build path's advance. | |
| 500 | - | // reset_burn_in on the *source* tier nulls its clock only when the operator |
Lines truncated
| @@ -0,0 +1,1667 @@ | |||
| 1 | + | use crate::error::Result; | |
| 2 | + | use crate::state::AppState; | |
| 3 | + | use axum::extract::{Path, Query, State, WebSocketUpgrade}; | |
| 4 | + | use axum::response::IntoResponse; | |
| 5 | + | use axum::routing::{get, post}; | |
| 6 | + | use axum::{Json, Router}; | |
| 7 | + | use serde::{Deserialize, Serialize}; | |
| 8 | + | use sqlx::Row; | |
| 9 | + | ||
| 10 | + | mod promotion; | |
| 11 | + | use promotion::{promote_inner, set_partial, clear_partial}; | |
| 12 | + | ||
| 13 | + | pub fn router(state: AppState) -> Router { | |
| 14 | + | let prom = state.prom.clone(); | |
| 15 | + | let token = state.api_token.clone(); | |
| 16 | + | ||
| 17 | + | // Every route requires the bearer token (when one is configured): the | |
| 18 | + | // mutators carry prod-deploy authority (CF2), and the reads expose prod | |
| 19 | + | // state — deployed versions, build SHAs, full gate logs, the live event | |
| 20 | + | // stream — that a tailnet peer should not get unauthenticated. `/metrics` | |
| 21 | + | // is the one deliberate opt-out (Prometheus convention, no secrets), added | |
| 22 | + | // outside the layered group below. The bearer-success path logs the caller | |
| 23 | + | // identity for mutating (POST) requests so a prod ship is attributable. | |
| 24 | + | let protected = Router::new() | |
| 25 | + | // mutators — prod-deploy authority | |
| 26 | + | .route("/promote/{tier}", post(promote)) | |
| 27 | + | .route("/rollback/{tier}", post(rollback)) | |
| 28 | + | .route("/rebuild", post(rebuild)) | |
| 29 | + | .route("/self-update", post(self_update)) | |
| 30 | + | .route("/confirm/{tier}", post(confirm)) | |
| 31 | + | .route("/backup/fetch", post(backup_fetch)) | |
| 32 | + | // reads — prod state, now also gated | |
| 33 | + | .route("/state", get(get_state)) | |
| 34 | + | .route("/runs/{id}", get(get_run)) | |
| 35 | + | .route("/runs/{id}/wait", get(get_run_wait)) | |
| 36 | + | .route("/logs/{version}/{gate}", get(get_gate_log)) | |
| 37 | + | .route("/events", get(events_ws)) | |
| 38 | + | .route_layer(axum::middleware::from_fn(move |req, next| { | |
| 39 | + | require_bearer(token.clone(), req, next) | |
| 40 | + | })); | |
| 41 | + | ||
| 42 | + | Router::new() | |
| 43 | + | .merge(protected) | |
| 44 | + | .with_state(state) | |
| 45 | + | .route("/metrics", get(crate::metrics::render).with_state(prom)) | |
| 46 | + | } | |
| 47 | + | ||
| 48 | + | /// Bearer-token gate for the deploy mutators. When no token is configured the | |
| 49 | + | /// request passes (main() only allows that on a loopback bind). Comparison is | |
| 50 | + | /// constant-time to avoid leaking the token via timing. | |
| 51 | + | async fn require_bearer( | |
| 52 | + | token: Option<std::sync::Arc<str>>, | |
| 53 | + | req: axum::extract::Request, | |
| 54 | + | next: axum::middleware::Next, | |
| 55 | + | ) -> axum::response::Response { | |
| 56 | + | let Some(expected) = token.as_deref() else { | |
| 57 | + | return next.run(req).await; | |
| 58 | + | }; | |
| 59 | + | let path = req.uri().path().to_string(); | |
| 60 | + | let method = req.method().clone(); | |
| 61 | + | let ok = req | |
| 62 | + | .headers() | |
| 63 | + | .get(axum::http::header::AUTHORIZATION) | |
| 64 | + | .and_then(|h| h.to_str().ok()) | |
| 65 | + | .and_then(|h| h.strip_prefix("Bearer ")) | |
| 66 | + | .is_some_and(|t| ct_eq(t, expected)); | |
| 67 | + | if ok { | |
| 68 | + | // Attribute mutations (POST) to a caller. Reads are GET and poll every | |
| 69 | + | // few seconds, so logging them would drown the journal; the audit value | |
| 70 | + | // is in *who shipped/rolled back prod*, which is always a POST. The peer | |
| 71 | + | // address is the closest identity we have without per-operator tokens | |
| 72 | + | // (`tailscale whois` is a future refinement); a shared token at least | |
| 73 | + | // gets a source host into the trail. | |
| 74 | + | if method == axum::http::Method::POST { | |
| 75 | + | let peer = req | |
| 76 | + | .extensions() | |
| 77 | + | .get::<axum::extract::ConnectInfo<std::net::SocketAddr>>() | |
| 78 | + | .map(|ci| ci.0.to_string()) | |
| 79 | + | .unwrap_or_else(|| "unknown".into()); | |
| 80 | + | tracing::info!(path = %path, peer = %peer, "authenticated deploy mutation"); | |
| 81 | + | } | |
| 82 | + | next.run(req).await | |
| 83 | + | } else { | |
| 84 | + | tracing::warn!(path = %path, "rejected unauthenticated request"); | |
| 85 | + | (axum::http::StatusCode::UNAUTHORIZED, "missing or invalid bearer token\n").into_response() | |
| 86 | + | } | |
| 87 | + | } | |
| 88 | + | ||
| 89 | + | /// Constant-time string compare. Length is allowed to leak (a token's length is | |
| 90 | + | /// not the secret); the byte comparison does not short-circuit. | |
| 91 | + | fn ct_eq(a: &str, b: &str) -> bool { | |
| 92 | + | let (a, b) = (a.as_bytes(), b.as_bytes()); | |
| 93 | + | if a.len() != b.len() { | |
| 94 | + | return false; | |
| 95 | + | } | |
| 96 | + | let mut diff = 0u8; | |
| 97 | + | for (x, y) in a.iter().zip(b) { | |
| 98 | + | diff |= x ^ y; | |
| 99 | + | } | |
| 100 | + | diff == 0 | |
| 101 | + | } | |
| 102 | + | ||
| 103 | + | #[derive(Serialize)] | |
| 104 | + | struct StateView { | |
| 105 | + | /// The running sandod's own package version. Lets a self-update caller | |
| 106 | + | /// confirm the new binary is live after the restart (the tier versions are | |
| 107 | + | /// the *deployed product*, not the controller). | |
| 108 | + | sandod_version: &'static str, | |
| 109 | + | tiers: Vec<TierView>, | |
| 110 | + | /// The most recent build run (the resource `GET /runs/{id}` exposes in | |
| 111 | + | /// full). Surfaced here so a `/state` poller sees an in-flight or failed | |
| 112 | + | /// build — the tier versions only ever reflect the last *success*, so | |
| 113 | + | /// without this `/state` looks frozen for the whole build. `null` until | |
| 114 | + | /// the first `/rebuild`. | |
| 115 | + | build: Option<crate::runs::BuildSummary>, | |
| 116 | + | } | |
| 117 | + | ||
| 118 | + | #[derive(Serialize)] | |
| 119 | + | struct TierView { | |
| 120 | + | name: String, | |
| 121 | + | ord: i64, | |
| 122 | + | provisioned: bool, | |
| 123 | + | canary: String, | |
| 124 | + | current_version: Option<String>, | |
| 125 | + | previous_version: Option<String>, | |
| 126 | + | burn_in_started_at: Option<String>, | |
| 127 | + | /// Non-null when the tier was left in a partial / mixed-version state by a | |
| 128 | + | /// failed promote or rollback whose compensation could not fully restore | |
| 129 | + | /// consistency. NULL when the tier is consistent. The TUI flags it red. | |
| 130 | + | partial_reason: Option<String>, | |
| 131 | + | nodes: Vec<String>, | |
| 132 | + | gates: Vec<GateView>, | |
| 133 | + | } | |
| 134 | + | ||
| 135 | + | #[derive(Serialize)] | |
| 136 | + | struct GateView { | |
| 137 | + | kind: String, | |
| 138 | + | finished_at: Option<String>, | |
| 139 | + | /// `'passed' | 'failed' | 'blocked'` or NULL while in-flight. The TUI | |
| 140 | + | /// uses this to choose green/red/yellow rendering. | |
| 141 | + | status: Option<String>, | |
| 142 | + | /// Full typed `GateOutcome` as a JSON object, when present. | |
| 143 | + | /// Deserialized lazily by the consumer; sandod doesn't re-parse it. | |
| 144 | + | outcome: Option<serde_json::Value>, | |
| 145 | + | /// Relative path under `cfg.logs_root` to the persisted stdout/stderr. | |
| 146 | + | log_ref: Option<String>, | |
| 147 | + | } | |
| 148 | + | ||
| 149 | + | async fn get_state(State(s): State<AppState>) -> Result<Json<StateView>> { | |
| 150 | + | let rows = sqlx::query( | |
| 151 | + | "SELECT t.name, t.ord, t.provisioned, t.canary, | |
| 152 | + | ts.current_version, ts.previous_version, ts.burn_in_started_at, | |
| 153 | + | ts.partial_reason | |
| 154 | + | FROM tiers t | |
| 155 | + | LEFT JOIN tier_state ts ON ts.tier = t.name | |
| 156 | + | ORDER BY t.ord", | |
| 157 | + | ) | |
| 158 | + | .fetch_all(&s.pool) | |
| 159 | + | .await?; | |
| 160 | + | ||
| 161 | + | let mut tiers = Vec::with_capacity(rows.len()); | |
| 162 | + | for r in rows { | |
| 163 | + | let name: String = r.get("name"); | |
| 164 | + | let current_version: Option<String> = r.get("current_version"); | |
| 165 | + | ||
| 166 | + | let nodes: Vec<String> = sqlx::query_scalar("SELECT name FROM nodes WHERE tier = ? ORDER BY name") | |
| 167 | + | .bind(&name) | |
| 168 | + | .fetch_all(&s.pool) | |
| 169 | + | .await?; | |
| 170 | + | ||
| 171 | + | // Surface gates for current_version when set, otherwise for the most | |
| 172 | + | // recently attempted version on this tier. Without the fallback, a | |
| 173 | + | // tier that has never gone green (MM after a build failure, B before | |
| 174 | + | // first deploy) exposes no gate detail via /state — debugging required | |
| 175 | + | // SSH and direct SQLite access. See sando todo: gate observability. | |
| 176 | + | let gate_version: Option<String> = if current_version.is_some() { | |
| 177 | + | current_version.clone() | |
| 178 | + | } else { | |
| 179 | + | sqlx::query_scalar( | |
| 180 | + | "SELECT version FROM gate_runs WHERE tier = ? | |
| 181 | + | ORDER BY id DESC LIMIT 1", | |
| 182 | + | ) | |
| 183 | + | .bind(&name) | |
| 184 | + | .fetch_optional(&s.pool) | |
| 185 | + | .await? | |
| 186 | + | }; | |
| 187 | + | ||
| 188 | + | let gates: Vec<GateView> = if let Some(ver) = gate_version.as_ref() { | |
| 189 | + | // Most recent gate_runs row per gate_kind for (tier, ver). | |
| 190 | + | sqlx::query( | |
| 191 | + | "SELECT gate_kind, finished_at, status, outcome_json, log_ref | |
| 192 | + | FROM gate_runs g | |
| 193 | + | WHERE tier = ?1 AND version = ?2 | |
| 194 | + | AND id = (SELECT MAX(id) FROM gate_runs | |
| 195 | + | WHERE tier = ?1 AND version = ?2 AND gate_kind = g.gate_kind) | |
| 196 | + | ORDER BY gate_kind", | |
| 197 | + | ) | |
| 198 | + | .bind(&name) | |
| 199 | + | .bind(ver) | |
| 200 | + | .fetch_all(&s.pool) | |
| 201 | + | .await? | |
| 202 | + | .into_iter() | |
| 203 | + | .map(|gr| GateView { | |
| 204 | + | kind: gr.get("gate_kind"), | |
| 205 | + | finished_at: gr.get("finished_at"), | |
| 206 | + | status: gr.get("status"), | |
| 207 | + | outcome: gr.get::<Option<String>, _>("outcome_json") | |
| 208 | + | .and_then(|s| serde_json::from_str(&s).ok()), | |
| 209 | + | log_ref: gr.get("log_ref"), | |
| 210 | + | }) | |
| 211 | + | .collect() | |
| 212 | + | } else { | |
| 213 | + | Vec::new() | |
| 214 | + | }; | |
| 215 | + | ||
| 216 | + | tiers.push(TierView { | |
| 217 | + | name, | |
| 218 | + | ord: r.get("ord"), | |
| 219 | + | provisioned: r.get::<i64, _>("provisioned") != 0, | |
| 220 | + | canary: r.get("canary"), | |
| 221 | + | current_version, | |
| 222 | + | previous_version: r.get("previous_version"), | |
| 223 | + | burn_in_started_at: r.get("burn_in_started_at"), | |
| 224 | + | partial_reason: r.get("partial_reason"), | |
| 225 | + | nodes, | |
| 226 | + | gates, | |
| 227 | + | }); | |
| 228 | + | } | |
| 229 | + | ||
| 230 | + | let build = crate::runs::latest_summary(&s.pool).await?; | |
| 231 | + | Ok(Json(StateView { sandod_version: env!("CARGO_PKG_VERSION"), tiers, build })) | |
| 232 | + | } | |
| 233 | + | ||
| 234 | + | #[derive(Deserialize, Default)] | |
| 235 | + | struct PromoteBody { | |
| 236 | + | /// Optional. If absent, defaults to the predecessor tier's `current_version` | |
| 237 | + | /// (i.e. promote whatever just finished baking on the previous tier). | |
| 238 | + | #[serde(default)] | |
| 239 | + | version: Option<String>, | |
| 240 | + | #[serde(default)] | |
| 241 | + | hotfix: bool, | |
| 242 | + | #[serde(default)] | |
| 243 | + | reset_burn_in: bool, | |
| 244 | + | } | |
| 245 | + | ||
| 246 | + | async fn promote( | |
| 247 | + | State(s): State<AppState>, | |
| 248 | + | Path(tier): Path<String>, | |
| 249 | + | body: Option<Json<PromoteBody>>, | |
| 250 | + | ) -> Result<Json<serde_json::Value>> { | |
| 251 | + | let body = body.map(|Json(b)| b).unwrap_or_default(); | |
| 252 | + | // Run the deploy in a detached task, not inline in the request future. | |
| 253 | + | // A promote's rsync + restart can outlast a client's socket timeout; if the | |
| 254 | + | // client disconnects, axum drops the handler future — and if the deploy ran | |
| 255 | + | // inline that would abandon the rollout mid-flight (symlink half-swapped, | |
| 256 | + | // tier_state not advanced, and — before the kill_on_drop fix — an orphaned | |
| 257 | + | // rsync). Spawning detaches the work from the connection: the task owns the | |
| 258 | + | // deploy lock and runs to completion regardless, while a still-connected | |
| 259 | + | // client still receives the result. Mirrors how /rebuild already runs its | |
| 260 | + | // build off-request. | |
| 261 | + | tokio::spawn(promote_inner(s, tier, body)) | |
| 262 | + | .await | |
| 263 | + | .map_err(|e| crate::error::Error::Other(anyhow::anyhow!("promote task failed to join: {e}")))? | |
| 264 | + | } | |
| 265 | + | ||
| 266 | + | async fn rollback( | |
| 267 | + | State(s): State<AppState>, | |
| 268 | + | Path(tier): Path<String>, | |
| 269 | + | ) -> Result<Json<serde_json::Value>> { | |
| 270 | + | // Serialize against any concurrent promote/rollback (CF3). | |
| 271 | + | let _deploy_guard = s.deploy_lock.lock().await; | |
| 272 | + | let tier = crate::domain::TierId::new(tier); | |
| 273 | + | let target = s.topo.tiers.iter().find(|t| t.name == tier) | |
| 274 | + | .ok_or(crate::error::Error::NotFound)?; | |
| 275 | + | ||
| 276 | + | let row: Option<(Option<String>, Option<String>)> = sqlx::query_as( | |
| 277 | + | "SELECT current_version, previous_version FROM tier_state WHERE tier = ?", | |
| 278 | + | ) | |
| 279 | + | .bind(&tier) | |
| 280 | + | .fetch_optional(&s.pool).await.map_err(crate::error::Error::Db)?; | |
| 281 | + | let (Some(current_str), Some(previous_str)) = row.unwrap_or((None, None)) else { | |
| 282 | + | return Err(crate::error::Error::GateBlocked( | |
| 283 | + | "no previous_version to roll back to".into(), | |
| 284 | + | )); | |
| 285 | + | }; | |
| 286 | + | let current = crate::domain::Version::parse(¤t_str) | |
| 287 | + | .map_err(|e| crate::error::Error::Other(anyhow::anyhow!(e)))?; | |
| 288 | + | let previous = crate::domain::Version::parse(&previous_str) | |
| 289 | + | .map_err(|e| crate::error::Error::Other(anyhow::anyhow!(e)))?; | |
| 290 | + | ||
| 291 | + | let bin: Option<(String,)> = sqlx::query_as( | |
| 292 | + | "SELECT artifact_path FROM versions WHERE version = ?", | |
| 293 | + | ) | |
| 294 | + | .bind(&previous) | |
| 295 | + | .fetch_optional(&s.pool).await.map_err(crate::error::Error::Db)?; | |
| 296 | + | let Some((bin,)) = bin else { | |
| 297 | + | return Err(crate::error::Error::GateBlocked( | |
| 298 | + | format!("previous version {previous} has no artifact_path; rollback impossible"), | |
| 299 | + | )); | |
| 300 | + | }; | |
| 301 | + | let bin_path = std::path::PathBuf::from(bin); | |
| 302 | + | let staged_dir = bin_path.parent() | |
| 303 | + | .ok_or_else(|| crate::error::Error::Other(anyhow::anyhow!("artifact_path has no parent")))? | |
| 304 | + | .to_path_buf(); | |
| 305 | + | ||
| 306 | + | // Deploy `previous` to every node. A failure partway through leaves the fleet | |
| 307 | + | // split-brain (some on `previous`, some still on `current`); record exactly how | |
| 308 | + | // far we got so /state surfaces it rather than reporting a clean rollback. | |
| 309 | + | let total = target.nodes.len(); | |
| 310 | + | for (i, node) in target.nodes.iter().enumerate() { | |
| 311 | + | let executor = s.executors.get(&node.name).cloned() | |
| 312 | + | .unwrap_or_else(|| crate::state::build_executor(node)); | |
| 313 | + | if let Err(e) = crate::deploy::deploy_node(executor.as_ref(), node, &previous_str, &staged_dir, s.cfg.primary_bin()).await { | |
| 314 | + | set_partial(&s, &tier, &format!( | |
| 315 | + | "rollback to {previous} incomplete: {i}/{total} nodes rolled back, \ | |
| 316 | + | node {} failed; remaining nodes still on {current} — manual check needed", | |
| 317 | + | node.name, | |
| 318 | + | )).await; | |
| 319 | + | tracing::error!(tier = %tier, node = %node.name, done = i, of = total, | |
| 320 | + | to = %previous, "rollback failed mid-fleet; tier left partial"); | |
| 321 | + | return Err(crate::error::Error::Other(e)); | |
| 322 | + | } | |
| 323 | + | } | |
| 324 | + | ||
| 325 | + | sqlx::query( | |
| 326 | + | "UPDATE tier_state SET current_version = ?, previous_version = ?, burn_in_started_at = NULL | |
| 327 | + | WHERE tier = ?", | |
| 328 | + | ) | |
| 329 | + | .bind(&previous) | |
| 330 | + | .bind(¤t) | |
| 331 | + | .bind(&tier) | |
| 332 | + | .execute(&s.pool).await.map_err(crate::error::Error::Db)?; | |
| 333 | + | ||
| 334 | + | // Full rollback succeeded on every node: the tier is consistent again. | |
| 335 | + | clear_partial(&s, &tier).await; | |
| 336 | + | ||
| 337 | + | tracing::warn!(tier = %tier, from = %current, to = %previous, "rollback complete"); | |
| 338 | + | crate::events::emit(&s.events, crate::events::Event::Rollback { | |
| 339 | + | tier: tier.clone(), from: current.clone(), to: previous.clone(), | |
| 340 | + | }); | |
| 341 | + | metrics::counter!("sando_rollbacks_total", "tier" => tier.to_string()).increment(1); | |
| 342 | + | ||
| 343 | + | Ok(Json(serde_json::json!({ | |
| 344 | + | "tier": tier, | |
| 345 | + | "rolled_back_from": current, | |
| 346 | + | "now_running": previous, | |
| 347 | + | }))) | |
| 348 | + | } | |
| 349 | + | ||
| 350 | + | #[derive(Deserialize, Default)] | |
| 351 | + | struct RebuildBody { | |
| 352 | + | /// Specific sha to build. If absent, resolve `topo.repo.branch` from the bare repo. | |
| 353 | + | #[serde(default)] | |
| 354 | + | sha: Option<String>, | |
| 355 | + | } | |
| 356 | + | ||
| 357 | + | async fn rebuild( | |
| 358 | + | State(s): State<AppState>, | |
| 359 | + | body: Option<Json<RebuildBody>>, | |
| 360 | + | ) -> Result<Json<serde_json::Value>> { | |
| 361 | + | let body = body.map(|Json(b)| b).unwrap_or_default(); | |
| 362 | + | let sha = match body.sha { | |
| 363 | + | Some(sha) => sha, | |
| 364 | + | None => { | |
| 365 | + | // Omitted sha = "build the deploy branch's tip". Fetch upstream | |
| 366 | + | // first so we resolve the *upstream* HEAD, not a possibly-stale | |
| 367 | + | // local branch ref — the build task fetches too, but only after the | |
| 368 | + | // sha is already chosen, so without this `/rebuild {}` could build | |
| 369 | + | // an old commit. A fetch failure is non-fatal: fall back to the | |
| 370 | + | // current bare-repo tip (same policy as the build task). | |
| 371 | + | let bare = std::path::Path::new(&s.topo.repo.bare_path); | |
| 372 | + | if let Some(upstream) = s.topo.repo.upstream.as_deref() | |
| 373 | + | && let Err(e) = crate::git::fetch_upstream(bare, upstream, &s.topo.repo.branch).await | |
| 374 | + | { | |
| 375 | + | tracing::warn!(error = %e, "pre-resolve upstream fetch failed; resolving current bare-repo branch tip"); | |
| 376 | + | } | |
| 377 | + | crate::git::resolve_ref(bare, &s.topo.repo.branch) | |
| 378 | + | .await | |
| 379 | + | .map_err(crate::error::Error::Other)? | |
| 380 | + | } | |
| 381 | + | }; | |
| 382 | + | ||
| 383 | + | // Boundary parse: a sha entering Sando must be hex of plausible length. | |
| 384 | + | // The build pipeline downstream only ever sees `GitSha`. | |
| 385 | + | let sha = crate::domain::GitSha::parse(&sha) | |
| 386 | + | .map_err(|e| crate::error::Error::Other(anyhow::anyhow!(e)))?; | |
| 387 | + | ||
| 388 | + | tracing::info!(sha = %sha, "rebuild requested"); | |
| 389 | + | crate::events::emit(&s.events, crate::events::Event::RebuildRequested { sha: sha.clone() }); | |
| 390 | + | ||
| 391 | + | // One pollable resource per triggered build. Created before the spawn so | |
| 392 | + | // the run id is in the response even if the task is aborted milliseconds | |
| 393 | + | // later by a still-newer /rebuild. | |
| 394 | + | let run_id = crate::runs::create(&s.pool, sha.as_str()) | |
| 395 | + | .await | |
| 396 | + | .map_err(crate::error::Error::Other)?; | |
| 397 | + | ||
| 398 | + | // Latest /rebuild wins: abort any in-flight build before spawning a new | |
| 399 | + | // one. Aborting drops the spawned task's future, which drops any | |
| 400 | + | // tokio::process::Child it owns; with `kill_on_drop(true)` set on the | |
| 401 | + | // cargo Command, SIGKILL propagates to cargo + its rustc children. | |
| 402 | + | let mut slot = s.active_build.lock().await; | |
| 403 | + | if let Some(prev) = slot.take() | |
| 404 | + | && !prev.handle.is_finished() | |
| 405 | + | { | |
| 406 | + | tracing::warn!("aborting in-flight build for newer /rebuild request"); | |
| 407 | + | crate::events::emit(&s.events, crate::events::Event::BuildAborted { sha_aborted: sha.clone() }); | |
| 408 | + | prev.handle.abort(); | |
| 409 | + | // Aborting drops the task before it can settle its own row, so | |
| 410 | + | // record the supersession here. | |
| 411 | + | crate::runs::mark_aborted(&s.pool, prev.run_id).await.ok(); | |
| 412 | + | } | |
| 413 | + | ||
| 414 | + | let pool = s.pool.clone(); | |
| 415 | + | let cfg = s.cfg.clone(); | |
| 416 | + | let topo = s.topo.clone(); | |
| 417 | + | let events_for_task = s.events.clone(); | |
| 418 | + | let sha_for_task = sha.clone(); | |
| 419 | + | let sha_response = sha.to_string(); | |
| 420 | + | let pool_for_task = s.pool.clone(); | |
| 421 | + | let deploy_lock = s.deploy_lock.clone(); | |
| 422 | + | let handle = tokio::spawn(async move { | |
| 423 | + | if let Err(e) = crate::build::build_and_run_host(pool, cfg, topo, sha_for_task.clone(), events_for_task, run_id, deploy_lock).await { | |
| 424 | + | tracing::error!(sha = %sha_for_task, error = %e, "rebuild pipeline failed"); | |
| 425 | + | // Pre-gate bails (fetch/checkout/version/scratch) don't settle the | |
| 426 | + | // run themselves; the build-step compile error already did. First | |
| 427 | + | // terminal write wins, so this is a safety net for the rest. | |
| 428 | + | crate::runs::mark_failed(&pool_for_task, run_id, &format!("{e:#}")).await.ok(); | |
| 429 | + | } | |
| 430 | + | }); | |
| 431 | + | *slot = Some(crate::state::ActiveBuild { handle: handle.abort_handle(), run_id }); | |
| 432 | + | ||
| 433 | + | Ok(Json(serde_json::json!({ "accepted": true, "sha": sha_response, "run_id": run_id.0 }))) | |
| 434 | + | } | |
| 435 | + | ||
| 436 | + | /// `GET /runs/{id}` — the build-status resource a non-TUI driver polls after | |
| 437 | + | /// `/rebuild`. Open (read-only) like `/state` and `/logs`. | |
| 438 | + | async fn get_run( | |
| 439 | + | State(s): State<AppState>, | |
| 440 | + | Path(id): Path<i64>, | |
| 441 | + | ) -> Result<Json<crate::runs::RunView>> { | |
| 442 | + | crate::runs::get(&s.pool, crate::domain::RunId(id)) | |
| 443 | + | .await | |
| 444 | + | .map_err(crate::error::Error::Other)? | |
| 445 | + | .map(Json) | |
| 446 | + | .ok_or(crate::error::Error::NotFound) | |
| 447 | + | } | |
| 448 | + | ||
| 449 | + | #[derive(Deserialize)] | |
| 450 | + | struct WaitParams { | |
| 451 | + | /// How long to hold the request open before returning a still-building | |
| 452 | + | /// run. Default 30s, capped at 120s. | |
| 453 | + | #[serde(default)] | |
| 454 | + | timeout_ms: Option<u64>, | |
| 455 | + | } | |
| 456 | + | ||
| 457 | + | /// `GET /runs/{id}/wait` — long-poll: hold the request open until the run | |
| 458 | + | /// settles (`result != building`) or the timeout elapses, then return the | |
| 459 | + | /// current `RunView`. Removes polling-cadence guessing for a headless driver | |
| 460 | + | /// (fire `/rebuild`, block on `/wait`). On timeout the run is returned | |
| 461 | + | /// still-building (200) and the caller re-issues `/wait`. 404 if unknown. | |
| 462 | + | async fn get_run_wait( | |
| 463 | + | State(s): State<AppState>, | |
| 464 | + | Path(id): Path<i64>, | |
| 465 | + | Query(p): Query<WaitParams>, | |
| 466 | + | ) -> Result<Json<crate::runs::RunView>> { | |
| 467 | + | let run_id = crate::domain::RunId(id); | |
| 468 | + | let timeout = std::time::Duration::from_millis(p.timeout_ms.unwrap_or(30_000).min(120_000)); | |
| 469 | + | let deadline = tokio::time::Instant::now() + timeout; | |
| 470 | + | // Poll the row rather than wiring a per-run notifier: a build settles on | |
| 471 | + | // the minute scale, so a sub-second tick is plenty responsive and the | |
| 472 | + | // query is a single indexed read. The request releases its pool handle | |
| 473 | + | // between ticks. | |
| 474 | + | let tick = std::time::Duration::from_millis(750); | |
| 475 | + | loop { | |
| 476 | + | let view = crate::runs::get(&s.pool, run_id) | |
| 477 | + | .await | |
| 478 | + | .map_err(crate::error::Error::Other)? | |
| 479 | + | .ok_or(crate::error::Error::NotFound)?; | |
| 480 | + | let now = tokio::time::Instant::now(); | |
| 481 | + | if view.result != "building" || now >= deadline { | |
| 482 | + | return Ok(Json(view)); | |
| 483 | + | } | |
| 484 | + | tokio::time::sleep((deadline - now).min(tick)).await; | |
| 485 | + | } | |
| 486 | + | } | |
| 487 | + | ||
| 488 | + | #[derive(Deserialize)] | |
| 489 | + | struct SelfUpdateBody { | |
| 490 | + | /// The commit to rebuild sandod from. Must already be on the canonical | |
| 491 | + | /// remote (the updater `git fetch`es it). | |
| 492 | + | sha: String, | |
| 493 | + | } | |
| 494 | + | ||
| 495 | + | /// The privileged updater unit instance for `sha`. A git sha is hex-only, so it | |
| 496 | + | /// is a safe systemd instance name with no escaping needed. | |
| 497 | + | fn self_update_unit(sha: &crate::domain::GitSha) -> String { | |
| 498 | + | format!("sando-update@{sha}.service") | |
| 499 | + | } | |
| 500 | + |
Lines truncated
| @@ -0,0 +1,473 @@ | |||
| 1 | + | //! Promotion service: the deploy state machine lifted out of the HTTP | |
| 2 | + | //! layer — promote orchestration, canary rollback of deployed nodes, | |
| 3 | + | //! partial-state flags, and the promote-time gate-satisfaction check. | |
| 4 | + | //! routes/mod.rs stays HTTP glue that calls into these. | |
| 5 | + | ||
| 6 | + | use super::*; | |
| 7 | + | ||
| 8 | + | pub(super) async fn promote_inner( | |
| 9 | + | s: AppState, | |
| 10 | + | tier: String, | |
| 11 | + | body: PromoteBody, | |
| 12 | + | ) -> Result<Json<serde_json::Value>> { | |
| 13 | + | // Serialize the whole check -> deploy -> advance against any concurrent | |
| 14 | + | // promote/rollback (CF3). Held for the entire task. | |
| 15 | + | let _deploy_guard = s.deploy_lock.lock().await; | |
| 16 | + | let tier = crate::domain::TierId::new(tier); | |
| 17 | + | let idx = s.topo.tiers.iter().position(|t| t.name == tier) | |
| 18 | + | .ok_or(crate::error::Error::NotFound)?; | |
| 19 | + | if idx == 0 { | |
| 20 | + | return Err(crate::error::Error::GateBlocked( | |
| 21 | + | "cannot /promote to the first tier; use /rebuild".into(), | |
| 22 | + | )); | |
| 23 | + | } | |
| 24 | + | let target = &s.topo.tiers[idx]; | |
| 25 | + | let source = &s.topo.tiers[idx - 1]; | |
| 26 | + | ||
| 27 | + | // Resolve version: explicit if given, else the source tier's current. | |
| 28 | + | let version_str = match body.version { | |
| 29 | + | Some(v) => v, | |
| 30 | + | None => sqlx::query_scalar::<_, Option<String>>( | |
| 31 | + | "SELECT current_version FROM tier_state WHERE tier = ?", | |
| 32 | + | ) | |
| 33 | + | .bind(&source.name) | |
| 34 | + | .fetch_optional(&s.pool).await | |
| 35 | + | .map_err(crate::error::Error::Db)? | |
| 36 | + | .flatten() | |
| 37 | + | .ok_or_else(|| crate::error::Error::GateBlocked( | |
| 38 | + | format!("no version specified and tier {} has no current_version", source.name), | |
| 39 | + | ))?, | |
| 40 | + | }; | |
| 41 | + | let version = crate::domain::Version::parse(&version_str) | |
| 42 | + | .map_err(|e| crate::error::Error::Other(anyhow::anyhow!(e)))?; | |
| 43 | + | ||
| 44 | + | // 1. Predecessor must have all of its configured gates satisfied for this | |
| 45 | + | // version (with optional hotfix override that skips burn_in). Evaluated | |
| 46 | + | // against the topology gate list, so a gate that never ran blocks the | |
| 47 | + | // promote instead of being treated as green. | |
| 48 | + | let pending = unsatisfied_gates(&s.pool, &source.name, &source.gates, &version_str, body.hotfix).await?; | |
| 49 | + | if !pending.is_empty() { | |
| 50 | + | return Err(crate::error::Error::GateBlocked(format!( | |
| 51 | + | "{} gate(s) not satisfied on tier {}: {}", | |
| 52 | + | pending.len(), | |
| 53 | + | source.name, | |
| 54 | + | pending.join(", "), | |
| 55 | + | ))); | |
| 56 | + | } | |
| 57 | + | ||
| 58 | + | // 2. Look up the artifact for this version. | |
| 59 | + | let bin: Option<(String,)> = sqlx::query_as( | |
| 60 | + | "SELECT artifact_path FROM versions WHERE version = ?", | |
| 61 | + | ) | |
| 62 | + | .bind(&version) | |
| 63 | + | .fetch_optional(&s.pool) | |
| 64 | + | .await | |
| 65 | + | .map_err(crate::error::Error::Db)?; | |
| 66 | + | let Some((bin,)) = bin else { | |
| 67 | + | return Err(crate::error::Error::NotFound); | |
| 68 | + | }; | |
| 69 | + | let bin_path = std::path::PathBuf::from(bin); | |
| 70 | + | // `artifact_path` is the primary binary; the staged release dir is its parent. | |
| 71 | + | let staged_dir = bin_path.parent() | |
| 72 | + | .ok_or_else(|| crate::error::Error::Other(anyhow::anyhow!("artifact_path has no parent")))? | |
| 73 | + | .to_path_buf(); | |
| 74 | + | ||
| 75 | + | // The version this tier was running before this promote — the rollback | |
| 76 | + | // target if a canary node fails partway through a multi-node rollout. | |
| 77 | + | let prev_version: Option<String> = sqlx::query_scalar( | |
| 78 | + | "SELECT current_version FROM tier_state WHERE tier = ?", | |
| 79 | + | ) | |
| 80 | + | .bind(&target.name) | |
| 81 | + | .fetch_optional(&s.pool).await.map_err(crate::error::Error::Db)?.flatten(); | |
| 82 | + | ||
| 83 | + | // 3. Deploy to each node. Sequential canary is the only policy | |
| 84 | + | // implemented in v0; parallel is a one-line change once we trust the | |
| 85 | + | // sequential path. Track the nodes already flipped to the new version so | |
| 86 | + | // a mid-rollout failure can roll them back (canary rollback). | |
| 87 | + | let mut deployed: Vec<&crate::topology::Node> = Vec::new(); | |
| 88 | + | for node in &target.nodes { | |
| 89 | + | let started = chrono::Utc::now().to_rfc3339(); | |
| 90 | + | crate::events::emit(&s.events, crate::events::Event::DeployStart { | |
| 91 | + | tier: target.name.clone(), node: node.name.clone(), version: version.clone(), | |
| 92 | + | }); | |
| 93 | + | let executor = s.executors.get(&node.name).cloned() | |
| 94 | + | .unwrap_or_else(|| crate::state::build_executor(node)); | |
| 95 | + | let result = crate::deploy::deploy_node(executor.as_ref(), node, &version_str, &staged_dir, s.cfg.primary_bin()).await; | |
| 96 | + | let finished = chrono::Utc::now().to_rfc3339(); | |
| 97 | + | let (outcome_obj, err_for_propagation) = match result { | |
| 98 | + | Ok(_) => (crate::outcome::DeployOutcome::ok(), None), | |
| 99 | + | Err(e) => { | |
| 100 | + | let msg = format!("{e:#}"); | |
| 101 | + | let kind = crate::classify::classify_deploy_error(&msg); | |
| 102 | + | (crate::outcome::DeployOutcome::failed(kind), Some(e)) | |
| 103 | + | } | |
| 104 | + | }; | |
| 105 | + | let outcome_json = serde_json::to_string(&outcome_obj) | |
| 106 | + | .unwrap_or_else(|e| format!("{{\"_serialize_error\":{e:?}}}")); | |
| 107 | + | sqlx::query( | |
| 108 | + | "INSERT INTO deploys (version, tier, node, started_at, finished_at, outcome, outcome_json, hotfix, reset_burn_in) | |
| 109 | + | VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)", | |
| 110 | + | ) | |
| 111 | + | .bind(&version).bind(&target.name).bind(&node.name) | |
| 112 | + | .bind(&started).bind(&finished).bind(outcome_obj.status_str()) | |
| 113 | + | .bind(&outcome_json) | |
| 114 | + | .bind(body.hotfix as i64).bind(body.reset_burn_in as i64) | |
| 115 | + | .execute(&s.pool).await.map_err(crate::error::Error::Db)?; | |
| 116 | + | if let Some(e) = err_for_propagation { | |
| 117 | + | let crate::outcome::DeployStatus::Failed { failure } = outcome_obj.status else { | |
| 118 | + | unreachable!("err_for_propagation is Some iff status is Failed"); | |
| 119 | + | }; | |
| 120 | + | tracing::error!( | |
| 121 | + | tier = %target.name, node = %node.name, version = %version, | |
| 122 | + | failure = failure.summary(), | |
| 123 | + | "deploy failed; current symlink left intact, tier_state not advanced" | |
| 124 | + | ); | |
| 125 | + | crate::events::emit(&s.events, crate::events::Event::DeployFailed { | |
| 126 | + | tier: target.name.clone(), node: node.name.clone(), | |
| 127 | + | version: version.clone(), failure, | |
| 128 | + | }); | |
| 129 | + | ||
| 130 | + | // Canary rollback: restore every node this promote touched — the | |
| 131 | + | // ones already flipped to the new version AND this failed node | |
| 132 | + | // (whose state is indeterminate: the symlink swap may have landed | |
| 133 | + | // before the restart failed) — back to the tier's prior version, so | |
| 134 | + | // the fleet is left consistent on `prev` rather than split-brain. | |
| 135 | + | // Nodes after this one were never touched and stay on `prev`. | |
| 136 | + | deployed.push(node); | |
| 137 | + | let touched = deployed.len(); | |
| 138 | + | match prev_version.as_deref() { | |
| 139 | + | Some(prev) => { | |
| 140 | + | let restored = rollback_deployed_nodes(&s, &target.name, &deployed, prev).await; | |
| 141 | + | tracing::warn!( | |
| 142 | + | tier = %target.name, restored, of = touched, | |
| 143 | + | from = %version, to = prev, | |
| 144 | + | "canary failed mid-rollout; rolled touched nodes back to the previous version", | |
| 145 | + | ); | |
| 146 | + | if restored > 0 | |
| 147 | + | && let Ok(prev_v) = crate::domain::Version::parse(prev) | |
| 148 | + | { | |
| 149 | + | crate::events::emit(&s.events, crate::events::Event::Rollback { | |
| 150 | + | tier: target.name.clone(), from: version.clone(), to: prev_v, | |
| 151 | + | }); | |
| 152 | + | } | |
| 153 | + | // If every touched node was restored, the tier is consistent on | |
| 154 | + | // `prev` — clear any stale flag. Otherwise it is genuinely | |
| 155 | + | // split-brain: record exactly how, for /state. | |
| 156 | + | if restored == touched { | |
| 157 | + | clear_partial(&s, &target.name).await; | |
| 158 | + | } else { | |
| 159 | + | set_partial(&s, &target.name, &format!( | |
| 160 | + | "canary rollback incomplete: {restored}/{touched} nodes restored to {prev}; \ | |
| 161 | + | {} may still be on {version} — manual check needed", | |
| 162 | + | touched - restored, | |
| 163 | + | )).await; | |
| 164 | + | } | |
| 165 | + | } | |
| 166 | + | None => { | |
| 167 | + | tracing::error!( | |
| 168 | + | tier = %target.name, count = touched, version = %version, | |
| 169 | + | "canary failed on a first deploy (no previous version to restore to); \ | |
| 170 | + | touched nodes remain on the new version — manual cleanup needed", | |
| 171 | + | ); | |
| 172 | + | set_partial(&s, &target.name, &format!( | |
| 173 | + | "first-deploy canary failed: {touched} node(s) left on {version}, \ | |
| 174 | + | no prior version to restore — manual cleanup needed", | |
| 175 | + | )).await; | |
| 176 | + | } | |
| 177 | + | } | |
| 178 | + | return Err(crate::error::Error::Other(e)); | |
| 179 | + | } | |
| 180 | + | deployed.push(node); | |
| 181 | + | crate::events::emit(&s.events, crate::events::Event::DeployOk { | |
| 182 | + | tier: target.name.clone(), node: node.name.clone(), version: version.clone(), | |
| 183 | + | }); | |
| 184 | + | } | |
| 185 | + | ||
| 186 | + | // 3b. Run this tier's post-deploy gates (node_health) against the freshly | |
| 187 | + | // deployed nodes and record their outcomes. These rows are the evidence | |
| 188 | + | // the NEXT promote (this tier -> the following one) checks via | |
| 189 | + | // `unsatisfied_gates`. Before CF1, only the host tier ran gates, so | |
| 190 | + | // A/B/C had no evidence and promotion waved through; node_health now | |
| 191 | + | // proves the deployed nodes are serving (Run-2 SERIOUS-3: boot_smoke | |
| 192 | + | // used to re-run the staged binary locally and proved nothing about the | |
| 193 | + | // node). burn_in / manual_confirm are not run here — they are evaluated | |
| 194 | + | // live / by the operator at the next promote. A failed gate does not | |
| 195 | + | // unwind this deploy (the artifact is already live on the tier); it | |
| 196 | + | // blocks the next promote, which is the fail-closed behavior we want. | |
| 197 | + | let post_deploy: Vec<crate::topology::Gate> = | |
| 198 | + | target.gates.iter().filter(|g| g.runs_post_deploy()).cloned().collect(); | |
| 199 | + | if !post_deploy.is_empty() { | |
| 200 | + | // node_health probes each node the deploy just shipped to, over the same | |
| 201 | + | // executor the deploy used. Build the probe set from the tier's nodes and | |
| 202 | + | // the startup executor map; a node missing an executor (shouldn't happen | |
| 203 | + | // — both come from the same topology) is skipped, and an empty set makes | |
| 204 | + | // node_health Blocked (fail closed). | |
| 205 | + | let nodes: Vec<crate::gates::NodeProbe> = target | |
| 206 | + | .nodes | |
| 207 | + | .iter() | |
| 208 | + | .filter_map(|n| { | |
| 209 | + | s.executors.get(&n.name).map(|exec| crate::gates::NodeProbe { | |
| 210 | + | node: n.name.clone(), | |
| 211 | + | service: n.service_name.clone(), | |
| 212 | + | health_url: n.health_url.clone(), | |
| 213 | + | executor: exec.clone(), | |
| 214 | + | }) | |
| 215 | + | }) | |
| 216 | + | .collect(); | |
| 217 | + | let ctx = crate::gates::GateCtx { | |
| 218 | + | pool: s.pool.clone(), | |
| 219 | + | cfg: s.cfg.clone(), | |
| 220 | + | tier: target.name.clone(), | |
| 221 | + | version: version.clone(), | |
| 222 | + | // No worktree at promote time; node_health works over executors, not | |
| 223 | + | // a checkout. | |
| 224 | + | worktree: std::path::PathBuf::new(), | |
| 225 | + | events: s.events.clone(), | |
| 226 | + | nodes, | |
| 227 | + | }; | |
| 228 | + | match crate::gates::run_all(&ctx, &post_deploy).await { | |
| 229 | + | Ok(true) => {} | |
| 230 | + | Ok(false) => tracing::warn!( | |
| 231 | + | tier = %target.name, version = %version, | |
| 232 | + | "post-deploy gate(s) failed; tier advanced but promotion to the next tier will be blocked", | |
| 233 | + | ), | |
| 234 | + | Err(e) => tracing::error!( | |
| 235 | + | tier = %target.name, version = %version, error = %e, | |
| 236 | + | "post-deploy gate execution errored; promotion to the next tier will be blocked", | |
| 237 | + | ), | |
| 238 | + | } | |
| 239 | + | } | |
| 240 | + | ||
| 241 | + | // 4. Advance tier_state through the single sealed forward-advance op (atomic | |
| 242 | + | // self-referential UPDATE; no read-modify-write to lose under concurrency, | |
| 243 | + | // CF3). We hold deploy_lock for this whole handler, so the advance is | |
| 244 | + | // serialized against rollback and the host build path's advance. | |
| 245 | + | // reset_burn_in on the *source* tier nulls its clock only when the operator | |
| 246 | + | // explicitly asked. | |
| 247 | + | crate::runs::advance_tier(&s.pool, target.name.as_str(), &version) | |
| 248 | + | .await | |
| 249 | + | .map_err(crate::error::Error::Db)?; | |
| 250 | + | ||
| 251 | + | if body.reset_burn_in { | |
| 252 | + | sqlx::query("UPDATE tier_state SET burn_in_started_at = NULL WHERE tier = ?") | |
| 253 | + | .bind(&source.name) | |
| 254 | + | .execute(&s.pool).await.map_err(crate::error::Error::Db)?; | |
| 255 | + | } | |
| 256 | + | ||
| 257 | + | // A clean full rollout to every node clears any prior partial flag on this tier. | |
| 258 | + | clear_partial(&s, &target.name).await; | |
| 259 | + | ||
| 260 | + | crate::events::emit(&s.events, crate::events::Event::PromoteComplete { | |
| 261 | + | tier: target.name.clone(), version: version.clone(), | |
| 262 | + | }); | |
| 263 | + | metrics::counter!("sando_promotes_total", "tier" => target.name.to_string()).increment(1); | |
| 264 | + | tracing::info!( | |
| 265 | + | version = %version, tier = %target.name, | |
| 266 | + | hotfix = body.hotfix, reset_burn_in = body.reset_burn_in, | |
| 267 | + | "promote complete", | |
| 268 | + | ); | |
| 269 | + | ||
| 270 | + | Ok(Json(serde_json::json!({ | |
| 271 | + | "tier": target.name, | |
| 272 | + | "version": version, | |
| 273 | + | "nodes_deployed": target.nodes.iter().map(|n| n.name.clone()).collect::<Vec<_>>(), | |
| 274 | + | }))) | |
| 275 | + | } | |
| 276 | + | ||
| 277 | + | /// After a canary node fails mid-promote, restore the nodes already flipped to | |
| 278 | + | /// the new version back to `prev_version`, leaving the tier consistent (all on | |
| 279 | + | /// the old version) rather than split-brain. Best-effort: every node is | |
| 280 | + | /// attempted; a per-node failure is logged but never propagated (the promote is | |
| 281 | + | /// already failing). Returns how many nodes were successfully restored. Returns | |
| 282 | + | /// 0 (with an error log) when the previous version has no recorded artifact to | |
| 283 | + | /// roll back to. | |
| 284 | + | pub(super) async fn rollback_deployed_nodes( | |
| 285 | + | s: &AppState, | |
| 286 | + | tier: &crate::domain::TierId, | |
| 287 | + | nodes: &[&crate::topology::Node], | |
| 288 | + | prev_version: &str, | |
| 289 | + | ) -> usize { | |
| 290 | + | let bin: Option<(String,)> = match sqlx::query_as( | |
| 291 | + | "SELECT artifact_path FROM versions WHERE version = ?", | |
| 292 | + | ) | |
| 293 | + | .bind(prev_version) | |
| 294 | + | .fetch_optional(&s.pool) | |
| 295 | + | .await | |
| 296 | + | { | |
| 297 | + | Ok(b) => b, | |
| 298 | + | Err(e) => { | |
| 299 | + | tracing::error!(tier = %tier, prev = prev_version, error = %e, | |
| 300 | + | "canary rollback: looking up the previous artifact failed; nodes left on the new version"); | |
| 301 | + | return 0; | |
| 302 | + | } | |
| 303 | + | }; | |
| 304 | + | let Some((bin,)) = bin else { | |
| 305 | + | tracing::error!(tier = %tier, prev = prev_version, nodes = nodes.len(), | |
| 306 | + | "canary rollback: previous version has no artifact_path; nodes left on the new version"); | |
| 307 | + | return 0; | |
| 308 | + | }; | |
| 309 | + | let Some(staged_dir) = std::path::PathBuf::from(&bin).parent().map(|p| p.to_path_buf()) else { | |
| 310 | + | tracing::error!(tier = %tier, prev = prev_version, | |
| 311 | + | "canary rollback: previous artifact_path has no parent dir; nodes left on the new version"); | |
| 312 | + | return 0; | |
| 313 | + | }; | |
| 314 | + | ||
| 315 | + | let mut restored = 0usize; | |
| 316 | + | for node in nodes { | |
| 317 | + | let executor = s.executors.get(&node.name).cloned() | |
| 318 | + | .unwrap_or_else(|| crate::state::build_executor(node)); | |
| 319 | + | match crate::deploy::deploy_node(executor.as_ref(), node, prev_version, &staged_dir, s.cfg.primary_bin()).await { | |
| 320 | + | Ok(_) => { | |
| 321 | + | restored += 1; | |
| 322 | + | tracing::warn!(tier = %tier, node = %node.name, version = prev_version, | |
| 323 | + | "canary rollback: node restored to the previous version"); | |
| 324 | + | } | |
| 325 | + | Err(e) => tracing::error!( | |
| 326 | + | tier = %tier, node = %node.name, version = prev_version, error = %format!("{e:#}"), | |
| 327 | + | "canary rollback FAILED for node; it remains on the new version — manual intervention needed", | |
| 328 | + | ), | |
| 329 | + | } | |
| 330 | + | } | |
| 331 | + | restored | |
| 332 | + | } | |
| 333 | + | ||
| 334 | + | /// Flag a tier as left in a partial / mixed-version state, with a human-readable | |
| 335 | + | /// reason surfaced through `/state` and the TUI. Best-effort: a failure to record | |
| 336 | + | /// the flag is logged, never propagated — the caller is already on an error path | |
| 337 | + | /// and the worse outcome is to mask the original failure with a bookkeeping one. | |
| 338 | + | pub(super) async fn set_partial(s: &AppState, tier: &crate::domain::TierId, reason: &str) { | |
| 339 | + | if let Err(e) = sqlx::query("UPDATE tier_state SET partial_reason = ? WHERE tier = ?") | |
| 340 | + | .bind(reason) | |
| 341 | + | .bind(tier) | |
| 342 | + | .execute(&s.pool) | |
| 343 | + | .await | |
| 344 | + | { | |
| 345 | + | tracing::error!(tier = %tier, reason, error = %e, | |
| 346 | + | "failed to record tier partial state; the fleet may be inconsistent without a /state flag"); | |
| 347 | + | } | |
| 348 | + | metrics::gauge!("sando_tier_partial", "tier" => tier.to_string()).set(1.0); | |
| 349 | + | } | |
| 350 | + | ||
| 351 | + | /// Clear a tier's partial flag after a clean full promote or rollback. Errors are | |
| 352 | + | /// logged but not propagated: the deploy itself succeeded, and a stale flag is a | |
| 353 | + | /// visible nuisance, not a safety regression (the operator sees a partial marker | |
| 354 | + | /// on a tier that is actually fine, and re-checks). | |
| 355 | + | pub(super) async fn clear_partial(s: &AppState, tier: &crate::domain::TierId) { | |
| 356 | + | if let Err(e) = sqlx::query("UPDATE tier_state SET partial_reason = NULL WHERE tier = ?") | |
| 357 | + | .bind(tier) | |
| 358 | + | .execute(&s.pool) | |
| 359 | + | .await | |
| 360 | + | { | |
| 361 | + | tracing::warn!(tier = %tier, error = %e, "failed to clear tier partial flag"); | |
| 362 | + | } | |
| 363 | + | metrics::gauge!("sando_tier_partial", "tier" => tier.to_string()).set(0.0); | |
| 364 | + | } | |
| 365 | + | ||
| 366 | + | /// Returns the kinds of `tier`'s *configured* gates that are not satisfied for | |
| 367 | + | /// `version`. `hotfix` suppresses the `burn_in` requirement only. | |
| 368 | + | /// | |
| 369 | + | /// Fail-closed against the topology gate list (the CF1 fix). The previous | |
| 370 | + | /// version inspected only existing `gate_runs` rows, so a configured gate that | |
| 371 | + | /// had *never run* produced no row and was invisibly treated as green — letting | |
| 372 | + | /// a promote wave through with zero evidence (it shipped 0.9.5 to prod with | |
| 373 | + | /// tier A's `boot_smoke` never recorded). Now every configured gate must show | |
| 374 | + | /// positive evidence: | |
| 375 | + | /// - `burn_in` is evaluated live against the tier's clock (a stored `blocked` | |
| 376 | + | /// row would otherwise never flip to passed as time elapses); | |
| 377 | + | /// - every other kind requires a `passed` row for (tier, version) — a missing | |
| 378 | + | /// or non-passed latest row counts as unsatisfied. | |
| 379 | + | pub(super) async fn unsatisfied_gates( | |
| 380 | + | pool: &sqlx::SqlitePool, | |
| 381 | + | tier: &crate::domain::TierId, | |
| 382 | + | gates: &[crate::topology::Gate], | |
| 383 | + | version: &str, | |
| 384 | + | hotfix: bool, | |
| 385 | + | ) -> std::result::Result<Vec<String>, crate::error::Error> { | |
| 386 | + | use crate::topology::Gate; | |
| 387 | + | let mut bad = Vec::new(); | |
| 388 | + | for gate in gates { | |
| 389 | + | let kind = gate.kind(); | |
| 390 | + | match gate { | |
| 391 | + | Gate::BurnIn { hours } => { | |
| 392 | + | if hotfix { | |
| 393 | + | continue; | |
| 394 | + | } | |
| 395 | + | let ok = crate::gates::burn_in_satisfied(pool, tier, *hours) | |
| 396 | + | .await | |
| 397 | + | .map_err(crate::error::Error::Other)?; | |
| 398 | + | if !ok { | |
| 399 | + | bad.push(kind.as_str().to_string()); | |
| 400 | + | } | |
| 401 | + | } | |
| 402 | + | Gate::ManualConfirm => { | |
| 403 | + | // A confirmation must be *fresh*: recorded at or after the | |
| 404 | + | // version's current landing on this tier (tier_state | |
| 405 | + | // .burn_in_started_at, the per-deploy clock). Without this a | |
| 406 | + | // confirmation row survives a rollback + rollback-forward and | |
| 407 | + | // waves a re-deploy of the same version through with no fresh | |
| 408 | + | // operator sign-off — weaker than burn_in, which is clock-based. | |
| 409 | + | // No baseline (NULL) => fail closed: require a fresh confirm. | |
| 410 | + | let confirmed_at: Option<String> = sqlx::query_scalar( | |
| 411 | + | "SELECT finished_at FROM gate_runs | |
| 412 | + | WHERE tier = ?1 AND version = ?2 AND gate_kind = 'manual_confirm' AND status = 'passed' | |
| 413 | + | ORDER BY id DESC LIMIT 1", | |
| 414 | + | ) | |
| 415 | + | .bind(tier.as_str()) | |
| 416 | + | .bind(version) | |
| 417 | + | .fetch_optional(pool) | |
| 418 | + | .await | |
| 419 | + | .map_err(crate::error::Error::Db)? | |
| 420 | + | .flatten(); | |
| 421 | + | let landed_at: Option<String> = sqlx::query_scalar( | |
| 422 | + | "SELECT burn_in_started_at FROM tier_state WHERE tier = ?", | |
| 423 | + | ) | |
| 424 | + | .bind(tier.as_str()) | |
| 425 | + | .fetch_optional(pool) | |
| 426 | + | .await | |
| 427 | + | .map_err(crate::error::Error::Db)? | |
| 428 | + | .flatten(); | |
| 429 | + | let fresh = match (confirmed_at, landed_at) { | |
| 430 | + | (Some(c), Some(l)) => { | |
| 431 | + | match ( | |
| 432 | + | chrono::DateTime::parse_from_rfc3339(&c), | |
| 433 | + | chrono::DateTime::parse_from_rfc3339(&l), | |
| 434 | + | ) { | |
| 435 | + | (Ok(cd), Ok(ld)) => cd >= ld, | |
| 436 | + | _ => false, // unparseable timestamp -> fail closed | |
| 437 | + | } | |
| 438 | + | } | |
| 439 | + | _ => false, | |
| 440 | + | }; | |
| 441 | + | if !fresh { | |
| 442 | + | bad.push(kind.as_str().to_string()); | |
| 443 | + | } | |
| 444 | + | } | |
| 445 | + | // Build/post-deploy gates that leave a `gate_runs` row: the latest | |
| 446 | + | // row for this (tier, version, kind) must be `passed`. Listed | |
| 447 | + | // explicitly (no `_` catch-all) so adding a new `Gate` variant is a | |
| 448 | + | // compile error here until its promotion semantics are decided — | |
| 449 | + | // a transient-`blocked` kind silently falling into "needs a passed | |
| 450 | + | // row" would be permanently unsatisfiable. | |
| 451 | + | Gate::CargoTest | Gate::MigrationDryRun | Gate::BootSmoke | Gate::NodeHealth => { | |
| 452 | + | // Latest row for this configured gate kind; NULL/missing/any | |
| 453 | + | // non-'passed' status all count as unsatisfied (fail closed). | |
| 454 | + | let status: Option<String> = sqlx::query_scalar( | |
| 455 | + | "SELECT status FROM gate_runs | |
| 456 | + | WHERE tier = ?1 AND version = ?2 AND gate_kind = ?3 | |
| 457 | + | ORDER BY id DESC LIMIT 1", | |
| 458 | + | ) | |
| 459 | + | .bind(tier.as_str()) | |
| 460 | + | .bind(version) | |
| 461 | + | .bind(kind.as_str()) | |
| 462 | + | .fetch_optional(pool) | |
| 463 | + | .await | |
| 464 | + | .map_err(crate::error::Error::Db)? | |
| 465 | + | .flatten(); | |
| 466 | + | if status.as_deref() != Some("passed") { | |
| 467 | + | bad.push(kind.as_str().to_string()); | |
| 468 | + | } | |
| 469 | + | } | |
| 470 | + | } | |
| 471 | + | } | |
| 472 | + | Ok(bad) | |
| 473 | + | } |