Skip to main content

max / makenotwork

164.5 KB · 4252 lines History Blame Raw
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::{clear_partial, promote_inner, set_partial};
12
13 pub fn router(state: AppState) -> Router {
14 let token = state.api_token.clone();
15
16 // Every route requires the bearer token (when one is configured): the
17 // mutators carry prod-deploy authority (CF2), and the reads expose prod
18 // state — deployed versions, build SHAs, full gate logs, the live event
19 // stream — that a tailnet peer should not get unauthenticated. There are no
20 // unauthenticated routes: /metrics used to be the one opt-out, but the
21 // Prometheus exporter was removed rather than leave an auth posture on it
22 // that no scraper actually used. 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("/intake", post(intake))
30 .route("/self-update", post(self_update))
31 .route("/confirm/{tier}", post(confirm))
32 .route("/backup/fetch", post(backup_fetch))
33 // reads — prod state, now also gated
34 .route("/state", get(get_state))
35 .route("/status.json", get(get_status_json))
36 .route("/runs/{id}", get(get_run))
37 .route("/runs/{id}/wait", get(get_run_wait))
38 .route("/logs/{run}/{gate}", get(get_gate_log))
39 .route("/events", get(events_ws))
40 .route_layer(axum::middleware::from_fn(move |req, next| {
41 require_bearer(token.clone(), req, next)
42 }));
43
44 Router::new().merge(protected).with_state(state)
45 }
46
47 /// The daemon's whole HTTP surface: one product's routes at the root, and every
48 /// product's under `/apps/<id>`.
49 ///
50 /// A product is addressed by mounting a whole router per product rather than by
51 /// a `{app}` path parameter every handler has to read. [`AppState`] already
52 /// carries one product's view — its config, topology and node executors — so
53 /// giving each mount its own state means no handler can address the wrong
54 /// product by forgetting to look at a parameter. The mistake is unavailable
55 /// rather than guarded against.
56 ///
57 /// The unprefixed paths stay, meaning the default product. `/promote/b` is what
58 /// the deploy runbook says, what the TUI calls, and what an operator types under
59 /// pressure; making all of that conditional on a rename would be a cost paid by
60 /// the wrong people.
61 pub fn router_for_apps(state: AppState) -> Router {
62 // The root mount. `state` already holds the default product's view.
63 let mut r = router(state.clone());
64 for (id, app) in state.apps.iter() {
65 let scoped = AppState {
66 cfg: app.cfg.clone(),
67 topo: app.topo.clone(),
68 executors: app.executors.clone(),
69 ..state.clone()
70 };
71 r = r.nest(&format!("/apps/{id}"), router(scoped));
72 }
73 r.merge(apps_index(state))
74 }
75
76 /// `GET /apps` — which products this daemon ships, and which one the unprefixed
77 /// routes address. Behind the same bearer gate as everything else: it describes
78 /// the deploy surface.
79 fn apps_index(state: AppState) -> Router {
80 let token = state.api_token.clone();
81 let protected =
82 Router::new()
83 .route("/apps", get(list_apps))
84 .route_layer(axum::middleware::from_fn(move |req, next| {
85 require_bearer(token.clone(), req, next)
86 }));
87 Router::new().merge(protected).with_state(state)
88 }
89
90 #[derive(serde::Serialize)]
91 pub struct AppsView {
92 /// Every product, in configured order.
93 pub apps: Vec<String>,
94 /// The one the unprefixed routes act on.
95 pub default_app: String,
96 }
97
98 async fn list_apps(State(s): State<AppState>) -> Json<AppsView> {
99 Json(AppsView {
100 apps: s.app_ids().iter().map(ToString::to_string).collect(),
101 default_app: s.default_app.to_string(),
102 })
103 }
104
105 /// Bearer-token gate for the deploy mutators. When no token is configured the
106 /// request passes (main() only allows that on a loopback bind). Comparison is
107 /// constant-time to avoid leaking the token via timing.
108 async fn require_bearer(
109 token: Option<std::sync::Arc<str>>,
110 req: axum::extract::Request,
111 next: axum::middleware::Next,
112 ) -> axum::response::Response {
113 let Some(expected) = token.as_deref() else {
114 return next.run(req).await;
115 };
116 let path = req.uri().path().to_string();
117 let method = req.method().clone();
118 let ok = req
119 .headers()
120 .get(axum::http::header::AUTHORIZATION)
121 .and_then(|h| h.to_str().ok())
122 .and_then(|h| h.strip_prefix("Bearer "))
123 .is_some_and(|t| ops_core::daemon::ct_eq(t, expected));
124 if ok {
125 // Attribute mutations (POST) to a caller. Reads are GET and poll every
126 // few seconds, so logging them would drown the journal; the audit value
127 // is in *who shipped/rolled back prod*, which is always a POST. The peer
128 // address is the closest identity we have without per-operator tokens
129 // (`tailscale whois` is a future refinement); a shared token at least
130 // gets a source host into the trail.
131 if method == axum::http::Method::POST {
132 let peer = req
133 .extensions()
134 .get::<axum::extract::ConnectInfo<std::net::SocketAddr>>()
135 .map_or_else(|| "unknown".into(), |ci| ci.0.to_string());
136 tracing::info!(path = %path, peer = %peer, "authenticated deploy mutation");
137 }
138 next.run(req).await
139 } else {
140 tracing::warn!(path = %path, "rejected unauthenticated request");
141 (
142 axum::http::StatusCode::UNAUTHORIZED,
143 "missing or invalid bearer token\n",
144 )
145 .into_response()
146 }
147 }
148
149 #[derive(Serialize)]
150 pub(crate) struct StateView {
151 /// The running sandod's own package version. Lets a self-update caller
152 /// confirm the new binary is live after the restart (the tier versions are
153 /// the *deployed product*, not the controller).
154 pub(crate) sandod_version: &'static str,
155 pub(crate) tiers: Vec<TierView>,
156 /// The most recent build run (the resource `GET /runs/{id}` exposes in
157 /// full). Surfaced here so a `/state` poller sees an in-flight or failed
158 /// build — the tier versions only ever reflect the last *success*, so
159 /// without this `/state` looks frozen for the whole build. `null` until
160 /// the first `/rebuild`.
161 pub(crate) build: Option<crate::runs::BuildSummary>,
162 }
163
164 #[derive(Serialize)]
165 pub(crate) struct TierView {
166 pub(crate) name: String,
167 pub(crate) ord: i64,
168 pub(crate) provisioned: bool,
169 pub(crate) canary: String,
170 pub(crate) current_version: Option<String>,
171 pub(crate) previous_version: Option<String>,
172 pub(crate) burn_in_started_at: Option<String>,
173 /// Non-null when the tier was left in a partial / mixed-version state by a
174 /// failed promote or rollback whose compensation could not fully restore
175 /// consistency. NULL when the tier is consistent. The TUI flags it red.
176 pub(crate) partial_reason: Option<String>,
177 /// Non-null when an artifact this tier still names is no longer on disk:
178 /// the row points at a release dir whose primary binary is gone, so a
179 /// promote or rollback resolving through it would fail at the rsync.
180 ///
181 /// The tier keeps serving — the running process holds its own inode — which
182 /// is why this was invisible until an operator reached for a rollback that
183 /// was not there. Read live on every `/state`, so a store repaired by hand
184 /// clears it on the next poll with nothing to reset.
185 pub(crate) missing_artifact: Option<String>,
186 pub(crate) nodes: Vec<String>,
187 pub(crate) gates: Vec<GateView>,
188 }
189
190 #[derive(Serialize)]
191 pub(crate) struct GateView {
192 pub(crate) kind: String,
193 pub(crate) finished_at: Option<String>,
194 /// `'passed' | 'failed' | 'blocked'` or NULL while in-flight. The TUI
195 /// uses this to choose green/red/yellow rendering.
196 pub(crate) status: Option<String>,
197 /// Full typed `GateOutcome` as a JSON object, when present.
198 /// Deserialized lazily by the consumer; sandod doesn't re-parse it.
199 pub(crate) outcome: Option<serde_json::Value>,
200 /// Relative path under `cfg.logs_root` to the persisted stdout/stderr.
201 pub(crate) log_ref: Option<String>,
202 }
203
204 async fn get_state(State(s): State<AppState>) -> Result<Json<StateView>> {
205 Ok(Json(state_view(&s).await?))
206 }
207
208 /// `GET /status.json` — the same state, in the shared cross-service payload
209 /// every operator surface renders. See `crate::status`.
210 async fn get_status_json(State(s): State<AppState>) -> Result<Json<ops_status::Payload>> {
211 let view = state_view(&s).await?;
212 Ok(Json(crate::status::payload(&view, chrono::Utc::now())))
213 }
214
215 /// The shared read behind `/state` and `/status.json`.
216 pub(crate) async fn state_view(s: &AppState) -> Result<StateView> {
217 let rows = sqlx::query(
218 "SELECT t.name, t.ord, t.provisioned, t.canary,
219 ts.current_version, ts.previous_version, ts.burn_in_started_at,
220 ts.partial_reason, ts.current_build_id
221 FROM tiers t
222 LEFT JOIN tier_state ts ON ts.app = t.app AND ts.tier = t.name
223 WHERE t.app = ?
224 ORDER BY t.ord",
225 )
226 .bind(&s.cfg.id)
227 .fetch_all(&s.pool)
228 .await?;
229
230 // What the deployed state points at, and which of it is actually there.
231 // One pass for every tier: the read is a handful of stats, and doing it here
232 // rather than on a timer means `/state` cannot report a stale verdict.
233 let refs = crate::retention::tier_refs(&s.pool, &s.cfg.id, s.cfg.primary_bin()).await?;
234 let gone = crate::retention::missing(&refs).await;
235
236 let mut tiers: Vec<TierView> = Vec::with_capacity(rows.len());
237 for r in rows {
238 let name: String = r.get("name");
239 let current_version: Option<String> = r.get("current_version");
240 let current_build_id: Option<i64> = r.get("current_build_id");
241
242 let nodes: Vec<String> =
243 sqlx::query_scalar("SELECT name FROM nodes WHERE app = ? AND tier = ? ORDER BY name")
244 .bind(&s.cfg.id)
245 .bind(&name)
246 .fetch_all(&s.pool)
247 .await?;
248
249 // Which run's gates this tier reports.
250 //
251 // The build the tier is actually running, when it has an identity: gate
252 // rows are evidence about a build, and a version string is not one. This
253 // was keyed on the version until 2026-08-20, and a rebuild at an
254 // unchanged version — the normal way to retry a red build — made the
255 // rows of two runs interleave, each gate showing whichever run wrote it
256 // last. Statuses appeared to move backwards on a tier nobody had touched
257 // (`hardening_test passed` then `hardening_test None`, a minute apart),
258 // because a reader cannot tell a not-yet-run gate from one a sibling run
259 // has just replaced.
260 //
261 // `GateScope::Version` is the fallback, and it covers two live cases: a
262 // pre-identity tier (NULL `current_build_id`, the same legacy path
263 // `unsatisfied_gates` takes), and a tier that has never gone green at all
264 // — MM after a build failure, B before its first deploy — where the most
265 // recently attempted version is the only handle there is. Without that
266 // second one, debugging a red tier meant SSH and direct SQLite access.
267 enum GateScope {
268 Build(i64),
269 Version(String),
270 }
271 let scope: Option<GateScope> = match (current_build_id, current_version.clone()) {
272 (Some(bid), _) => Some(GateScope::Build(bid)),
273 (None, Some(ver)) => Some(GateScope::Version(ver)),
274 (None, None) => sqlx::query_scalar::<_, String>(
275 "SELECT version FROM gate_runs WHERE app = ? AND tier = ?
276 ORDER BY id DESC LIMIT 1",
277 )
278 .bind(&s.cfg.id)
279 .bind(&name)
280 .fetch_optional(&s.pool)
281 .await?
282 .map(GateScope::Version),
283 };
284
285 // Most recent row per gate_kind within the scope. Both arms share the
286 // shape; only the key differs.
287 let gate_rows = match &scope {
288 Some(GateScope::Build(bid)) => {
289 sqlx::query(
290 "SELECT gate_kind, finished_at, status, outcome_json, log_ref
291 FROM gate_runs g
292 WHERE app = ?1 AND tier = ?2 AND build_id = ?3
293 AND id = (SELECT MAX(id) FROM gate_runs
294 WHERE app = ?1 AND tier = ?2 AND build_id = ?3
295 AND gate_kind = g.gate_kind)
296 ORDER BY gate_kind",
297 )
298 .bind(&s.cfg.id)
299 .bind(&name)
300 .bind(bid)
301 .fetch_all(&s.pool)
302 .await?
303 }
304 Some(GateScope::Version(ver)) => {
305 sqlx::query(
306 "SELECT gate_kind, finished_at, status, outcome_json, log_ref
307 FROM gate_runs g
308 WHERE app = ?1 AND tier = ?2 AND version = ?3
309 AND id = (SELECT MAX(id) FROM gate_runs
310 WHERE app = ?1 AND tier = ?2 AND version = ?3
311 AND gate_kind = g.gate_kind)
312 ORDER BY gate_kind",
313 )
314 .bind(&s.cfg.id)
315 .bind(&name)
316 .bind(ver)
317 .fetch_all(&s.pool)
318 .await?
319 }
320 None => Vec::new(),
321 };
322 let gates: Vec<GateView> = gate_rows
323 .into_iter()
324 .map(|gr| GateView {
325 kind: gr.get("gate_kind"),
326 finished_at: gr.get("finished_at"),
327 status: gr.get("status"),
328 outcome: gr
329 .get::<Option<String>, _>("outcome_json")
330 .and_then(|s| serde_json::from_str(&s).ok()),
331 log_ref: gr.get("log_ref"),
332 })
333 .collect();
334
335 let missing_artifact = {
336 let mine: Vec<_> = gone.iter().filter(|g| g.tier == name).cloned().collect();
337 (!mine.is_empty()).then(|| crate::retention::describe(&mine))
338 };
339
340 tiers.push(TierView {
341 name,
342 ord: r.get("ord"),
343 provisioned: r.get::<i64, _>("provisioned") != 0,
344 canary: r.get("canary"),
345 current_version,
346 previous_version: r.get("previous_version"),
347 burn_in_started_at: r.get("burn_in_started_at"),
348 partial_reason: r.get("partial_reason"),
349 missing_artifact,
350 nodes,
351 gates,
352 });
353 }
354
355 let build = crate::runs::latest_summary(&s.pool, &s.cfg.id).await?;
356 Ok(StateView {
357 sandod_version: env!("CARGO_PKG_VERSION"),
358 tiers,
359 build,
360 })
361 }
362
363 #[derive(Deserialize, Default)]
364 struct PromoteBody {
365 /// Optional. If absent, defaults to the predecessor tier's `current_version`
366 /// (i.e. promote whatever just finished baking on the previous tier).
367 #[serde(default)]
368 version: Option<String>,
369 #[serde(default)]
370 hotfix: bool,
371 #[serde(default)]
372 reset_burn_in: bool,
373 /// Set when the release carries a database migration. Rollback restores the
374 /// binary and release_contents only, never the schema (MNW migrates forward
375 /// on boot and has no down path), so a migration-bearing promote is one-way.
376 /// This forces a fresh `manual_confirm` on the predecessor tier even if the
377 /// tier does not configure one, so the operator consciously acknowledges the
378 /// rollback contract. `hotfix` does not suppress it. See
379 /// `deploy/README.md` "Rollback contract".
380 #[serde(default)]
381 bears_migration: bool,
382 }
383
384 async fn promote(
385 State(s): State<AppState>,
386 Path(tier): Path<String>,
387 body: Option<Json<PromoteBody>>,
388 ) -> Result<Json<serde_json::Value>> {
389 let body = body.map(|Json(b)| b).unwrap_or_default();
390 // Run the deploy in a detached task, not inline in the request future.
391 // A promote's rsync + restart can outlast a client's socket timeout; if the
392 // client disconnects, axum drops the handler future — and if the deploy ran
393 // inline that would abandon the rollout mid-flight (symlink half-swapped,
394 // tier_state not advanced, and — before the kill_on_drop fix — an orphaned
395 // rsync). Spawning detaches the work from the connection: the task owns the
396 // deploy lock and runs to completion regardless, while a still-connected
397 // client still receives the result. Mirrors how /rebuild already runs its
398 // build off-request.
399 tokio::spawn(promote_inner(s, tier, body))
400 .await
401 .map_err(|e| {
402 crate::error::Error::Other(anyhow::anyhow!("promote task failed to join: {e}"))
403 })?
404 }
405
406 async fn rollback(
407 State(s): State<AppState>,
408 Path(tier): Path<String>,
409 ) -> Result<Json<serde_json::Value>> {
410 // Serialize against any concurrent promote/rollback (CF3).
411 let _deploy_guard = s.deploy_lock.lock().await;
412 let tier = crate::domain::TierId::new(tier);
413 let target = s
414 .topo
415 .tiers
416 .iter()
417 .find(|t| t.name == tier)
418 .ok_or(crate::error::Error::NotFound)?;
419
420 let row: Option<(Option<String>, Option<String>)> = sqlx::query_as(
421 "SELECT current_version, previous_version FROM tier_state WHERE app = ? AND tier = ?",
422 )
423 .bind(&s.cfg.id)
424 .bind(&tier)
425 .fetch_optional(&s.pool)
426 .await
427 .map_err(crate::error::Error::Db)?;
428 let (Some(current_str), Some(previous_str)) = row.unwrap_or((None, None)) else {
429 return Err(crate::error::Error::GateBlocked(
430 "no previous_version to roll back to".into(),
431 ));
432 };
433 let current = crate::domain::Version::parse(&current_str)
434 .map_err(|e| crate::error::Error::Other(anyhow::anyhow!(e)))?;
435 let previous = crate::domain::Version::parse(&previous_str)
436 .map_err(|e| crate::error::Error::Other(anyhow::anyhow!(e)))?;
437
438 let bin: Option<(String,)> =
439 sqlx::query_as("SELECT artifact_path FROM versions WHERE app = ? AND version = ?")
440 .bind(&s.cfg.id)
441 .bind(&previous)
442 .fetch_optional(&s.pool)
443 .await
444 .map_err(crate::error::Error::Db)?;
445 let Some((bin,)) = bin else {
446 return Err(crate::error::Error::GateBlocked(format!(
447 "previous version {previous} has no artifact_path; rollback impossible"
448 )));
449 };
450 let bin_path = std::path::PathBuf::from(bin);
451 let staged_dir = bin_path
452 .parent()
453 .ok_or_else(|| crate::error::Error::Other(anyhow::anyhow!("artifact_path has no parent")))?
454 .to_path_buf();
455
456 // Deploy `previous` to every node. A failure partway through leaves the fleet
457 // split-brain (some on `previous`, some still on `current`); record exactly how
458 // far we got so /state surfaces it rather than reporting a clean rollback.
459 let total = target.nodes.len();
460 // Same per-node resolution as a promote: on a two-architecture product the
461 // version being rolled back to is also two bundles, and each node takes its
462 // own. Resolved before any node is touched, so a version whose other half
463 // was never accepted fails the rollback whole rather than partway.
464 let target_nodes: Vec<&crate::topology::Node> = target.nodes.iter().collect();
465 // No fallback build id: a rollback re-runs no gates, so there is no evidence
466 // to key on. These are bytes the tier already ran.
467 let bundles =
468 promotion::bundles_for_nodes(&s, &previous_str, &target_nodes, &staged_dir, None, None)
469 .await?;
470 // The node gc's pinned set, read before the first node is touched. An
471 // operator-driven rollback has not started shipping bytes yet, so failing
472 // here is free and matches the host store's rule: never gc without knowing
473 // what is referenced. The canary rollback inside a promote takes the softer
474 // line (skip the gc, roll back anyway) because by then a tier is already
475 // failing.
476 let pinned = crate::retention::pinned_dirs(&s.pool, &s.cfg.id)
477 .await
478 .map_err(crate::error::Error::Other)?;
479
480 for (i, (node, node_bundle, node_bundle_platform, _)) in bundles.iter().enumerate() {
481 let executor = s
482 .executors
483 .get(&node.name)
484 .cloned()
485 .unwrap_or_else(|| crate::state::build_executor(node));
486 let placement =
487 crate::deploy::Placement::check(node, node_bundle, node_bundle_platform.as_ref())
488 .map_err(|e| crate::error::Error::GateBlocked(e.to_string()))?;
489 if let Err(e) = crate::deploy::deploy_node(
490 executor.as_ref(),
491 placement,
492 &previous_str,
493 s.cfg.primary_bin(),
494 Some(&pinned),
495 )
496 .await
497 {
498 set_partial(
499 &s,
500 &tier,
501 &format!(
502 "rollback to {previous} incomplete: {i}/{total} nodes rolled back, \
503 node {} failed; remaining nodes still on {current} — manual check needed",
504 node.name,
505 ),
506 )
507 .await;
508 tracing::error!(tier = %tier, node = %node.name, done = i, of = total,
509 to = %previous, "rollback failed mid-fleet; tier left partial");
510 return Err(crate::error::Error::Other(e));
511 }
512 }
513
514 // `previous_version` becomes NULL, not `current`. Sando tracks exactly one
515 // step of history, so writing the version we just rolled *off* into the
516 // rollback slot makes a second /rollback roll FORWARD onto the build the
517 // operator was escaping. NULL makes the second call fail loudly with "no
518 // previous_version to roll back to" — honest about the depth we keep.
519 // Stepping back further is the deploys-table walk we deliberately don't do.
520 //
521 // `advanced_at = now` even though burn_in_started_at is nulled: a rollback IS
522 // a change to the deployed identity, so the startup reconcile must treat the
523 // roll-off `deploys` row as predating this state — otherwise a cleanly
524 // rolled-back tier would look like an unrecorded deploy (migration 009 /
525 // [`crate::reconcile`]).
526 let now = chrono::Utc::now().to_rfc3339();
527 sqlx::query(
528 "UPDATE tier_state
529 SET current_version = ?, previous_version = NULL,
530 burn_in_started_at = NULL, advanced_at = ?
531 WHERE app = ? AND tier = ?",
532 )
533 .bind(&previous)
534 .bind(&now)
535 .bind(&s.cfg.id)
536 .bind(&tier)
537 .execute(&s.pool)
538 .await
539 .map_err(crate::error::Error::Db)?;
540
541 // Full rollback succeeded on every node: the tier is consistent again.
542 clear_partial(&s, &tier).await;
543
544 tracing::warn!(tier = %tier, from = %current, to = %previous, "rollback complete");
545 crate::events::emit(
546 &s.events,
547 crate::events::Event::Rollback {
548 tier: tier.clone(),
549 from: current.clone(),
550 to: previous.clone(),
551 },
552 );
553
554 Ok(Json(serde_json::json!({
555 "tier": tier,
556 "rolled_back_from": current,
557 "now_running": previous,
558 })))
559 }
560
561 #[derive(Deserialize, Default)]
562 struct RebuildBody {
563 /// Specific sha to build. If absent, resolve `topo.repo.branch` from the bare repo.
564 #[serde(default)]
565 sha: Option<String>,
566 }
567
568 #[derive(Deserialize)]
569 struct IntakeBody {
570 /// Directory holding the assembled bundle, which must already be under this
571 /// app's `release_root/staging/`. Publishing is an atomic same-filesystem
572 /// rename, so a path anywhere else would silently become a copy. Getting the
573 /// bytes there is the builder's job; this endpoint decides whether they may
574 /// stay.
575 staged: String,
576 /// The builder's `ArtifactRecord`, verbatim. Bento writes one per target at
577 /// `dist_root/<app>/<version>/<target-slug>/record.json` and sends it here
578 /// rather than inside the bundle: the record names the digest of the bundle,
579 /// and the digest covers every file in it, so a record shipped among the
580 /// artifacts would change the digest it names.
581 record: String,
582 }
583
584 /// `POST /intake` — accept an artifact this daemon did not build, and take it
585 /// through the same host-tier gating a locally-built one gets.
586 ///
587 /// This is the Sando half of the Bento boundary (wiki [[sando-bento-boundary]]):
588 /// Bento builds and packages, Sando decides whether a thing advances a stage.
589 /// The bundle arrives already assembled, with no worktree and no compiler run,
590 /// and joins the pipeline at the point where a Sando build has just published
591 /// its own bundle.
592 ///
593 /// The bytes are proved against the record before anything else happens, so a
594 /// bundle that drifted in transit is refused with the offending file named
595 /// rather than gated and shipped.
596 async fn intake(
597 State(s): State<AppState>,
598 Json(body): Json<IntakeBody>,
599 ) -> Result<Json<serde_json::Value>> {
600 let staged = std::path::PathBuf::from(&body.staged);
601 // The staging dir has to be under this app's release root, and this is the
602 // one place that can be enforced: `intake::accept` renames out of it, and a
603 // path elsewhere would either cross a filesystem or reach somewhere the
604 // caller should not be able to name.
605 let staging_root = s.cfg.release_root.join("staging");
606 let staged_abs = staged.canonicalize().map_err(|e| {
607 crate::error::Error::Other(anyhow::anyhow!("staged path {}: {e}", body.staged))
608 })?;
609 let staging_root_abs = staging_root.canonicalize().map_err(|e| {
610 crate::error::Error::Other(anyhow::anyhow!(
611 "staging root {}: {e}",
612 staging_root.display()
613 ))
614 })?;
615 if !staged_abs.starts_with(&staging_root_abs) {
616 return Err(crate::error::Error::GateBlocked(format!(
617 "staged bundle {} is not under {}; publishing is an atomic rename \
618 within the release root",
619 staged_abs.display(),
620 staging_root_abs.display(),
621 )));
622 }
623
624 // The run row is created before the record is parsed so a refused intake is
625 // still visible as an attempt. Its sha comes from the record's provenance,
626 // which is the artifact's own claim about what it was built from; `accept`
627 // is what decides whether to believe the document at all.
628 let sha = serde_json::from_str::<serde_json::Value>(&body.record)
629 .ok()
630 .and_then(|v| {
631 v.get("provenance")
632 .and_then(|p| p.get("git_sha"))
633 .and_then(|g| g.as_str())
634 .map(str::to_owned)
635 })
636 .unwrap_or_default();
637 let run_id = crate::runs::create(&s.pool, &s.cfg.id, &sha)
638 .await
639 .map_err(crate::error::Error::Other)?;
640
641 tracing::info!(staged = %staged_abs.display(), run_id = %run_id, "artifact intake requested");
642
643 // Prove the bytes BEFORE answering. This has to be synchronous, and the
644 // reason is the whole contract: the producer treats a non-success here as a
645 // failed handoff and fails its own build on it. Proving in a spawned task
646 // and answering `accepted: true` first would tell Bento the artifact landed
647 // and only then discover it had not — a corrupt bundle would go green
648 // upstream and be visible solely in this daemon's run row, which is exactly
649 // the "evidence vouches for one thing, the deploy ships another" class the
650 // boundary exists to close.
651 //
652 // It is affordable: hashing the bundle is the work the transfer just did.
653 let accepted =
654 match crate::build::accept_intake(&s.pool, &s.cfg, &staged_abs, &body.record, run_id).await
655 {
656 Ok(p) => p,
657 Err(e) => {
658 let msg = format!("{e:#}");
659 tracing::error!(error = %msg, "refused an intake");
660 crate::runs::mark_failed(&s.pool, run_id, &msg).await.ok();
661 return Err(crate::error::Error::GateBlocked(msg));
662 }
663 };
664
665 // Gating is not. It runs the host tier's gates, which can take an hour, and
666 // whether the artifact advances is this daemon's business rather than the
667 // builder's. The producer's job ended when the bytes were believed.
668 let pool = s.pool.clone();
669 let pool_for_task = s.pool.clone();
670 let cfg = s.cfg.clone();
671 let topo = s.topo.clone();
672 let events = s.events.clone();
673 let deploy_lock = s.deploy_lock.clone();
674 tokio::spawn(async move {
675 if let Err(e) =
676 crate::build::gate_intake(pool, cfg, topo, accepted, events, run_id, deploy_lock).await
677 {
678 tracing::error!(error = %e, "gating an accepted artifact failed");
679 crate::runs::mark_failed(&pool_for_task, run_id, &format!("{e:#}"))
680 .await
681 .ok();
682 }
683 });
684
685 Ok(Json(
686 serde_json::json!({ "accepted": true, "run_id": run_id.0 }),
687 ))
688 }
689
690 async fn rebuild(
691 State(s): State<AppState>,
692 body: Option<Json<RebuildBody>>,
693 ) -> Result<Json<serde_json::Value>> {
694 let body = body.map(|Json(b)| b).unwrap_or_default();
695 let sha = match body.sha {
696 Some(sha) => sha,
697 None => {
698 // Omitted sha = "build the deploy branch's tip". Fetch upstream
699 // first so we resolve the *upstream* HEAD, not a possibly-stale
700 // local branch ref — the build task fetches too, but only after the
701 // sha is already chosen, so without this `/rebuild {}` could build
702 // an old commit. A fetch failure is non-fatal: fall back to the
703 // current bare-repo tip (same policy as the build task).
704 let repo = s.topo.repo.as_ref().ok_or_else(|| {
705 crate::error::Error::GateBlocked(format!(
706 "{} is intake-only: it declares no [repo], so there is nothing to \
707 build from. Ship it with POST /intake.",
708 s.cfg.id
709 ))
710 })?;
711 let bare = std::path::Path::new(&repo.bare_path);
712 if let Some(upstream) = repo.upstream.as_deref()
713 && let Err(e) = crate::git::fetch_upstream(bare, upstream, &repo.branch).await
714 {
715 tracing::warn!(error = %e, "pre-resolve upstream fetch failed; resolving current bare-repo branch tip");
716 }
717 crate::git::resolve_ref(bare, &repo.branch)
718 .await
719 .map_err(crate::error::Error::Other)?
720 }
721 };
722
723 // Boundary parse: a sha entering Sando must be hex of plausible length.
724 // The build pipeline downstream only ever sees `GitSha`.
725 let sha = crate::domain::GitSha::parse(&sha)
726 .map_err(|e| crate::error::Error::Other(anyhow::anyhow!(e)))?;
727
728 tracing::info!(sha = %sha, "rebuild requested");
729 crate::events::emit(
730 &s.events,
731 crate::events::Event::RebuildRequested { sha: sha.clone() },
732 );
733
734 // One pollable resource per triggered build. Created before the spawn so
735 // the run id is in the response even if the task is aborted milliseconds
736 // later by a still-newer /rebuild.
737 let run_id = crate::runs::create(&s.pool, &s.cfg.id, sha.as_str())
738 .await
739 .map_err(crate::error::Error::Other)?;
740
741 // Latest /rebuild wins: abort any in-flight build before spawning a new
742 // one. Aborting drops the spawned task's future, which drops any
743 // tokio::process::Child it owns; with `kill_on_drop(true)` set on the
744 // cargo Command, SIGKILL propagates to cargo + its rustc children.
745 let mut slot = s.active_build.lock().await;
746 if let Some(prev) = slot.take()
747 && !prev.handle.is_finished()
748 {
749 tracing::warn!("aborting in-flight build for newer /rebuild request");
750 crate::events::emit(
751 &s.events,
752 crate::events::Event::BuildAborted {
753 sha_aborted: sha.clone(),
754 },
755 );
756 prev.handle.abort();
757 // Aborting drops the task before it can settle its own row, so
758 // record the supersession here.
759 crate::runs::mark_aborted(&s.pool, prev.run_id).await.ok();
760 }
761
762 let pool = s.pool.clone();
763 let cfg = s.cfg.clone();
764 let topo = s.topo.clone();
765 let events_for_task = s.events.clone();
766 let sha_for_task = sha.clone();
767 let sha_response = sha.to_string();
768 let pool_for_task = s.pool.clone();
769 let deploy_lock = s.deploy_lock.clone();
770 let handle = tokio::spawn(async move {
771 if let Err(e) = crate::build::build_and_run_host(
772 pool,
773 cfg,
774 topo,
775 sha_for_task.clone(),
776 events_for_task,
777 run_id,
778 deploy_lock,
779 )
780 .await
781 {
782 tracing::error!(sha = %sha_for_task, error = %e, "rebuild pipeline failed");
783 // Pre-gate bails (fetch/checkout/version/scratch) don't settle the
784 // run themselves; the build-step compile error already did. First
785 // terminal write wins, so this is a safety net for the rest.
786 crate::runs::mark_failed(&pool_for_task, run_id, &format!("{e:#}"))
787 .await
788 .ok();
789 }
790 });
791 *slot = Some(crate::state::ActiveBuild {
792 handle: handle.abort_handle(),
793 run_id,
794 });
795
796 Ok(Json(
797 serde_json::json!({ "accepted": true, "sha": sha_response, "run_id": run_id.0 }),
798 ))
799 }
800
801 /// `GET /runs/{id}` — the build-status resource a non-TUI driver polls after
802 /// `/rebuild`. Open (read-only) like `/state` and `/logs`.
803 async fn get_run(
804 State(s): State<AppState>,
805 Path(id): Path<i64>,
806 ) -> Result<Json<crate::runs::RunView>> {
807 crate::runs::get(&s.pool, crate::domain::RunId(id))
808 .await
809 .map_err(crate::error::Error::Other)?
810 .map(Json)
811 .ok_or(crate::error::Error::NotFound)
812 }
813
814 #[derive(Deserialize)]
815 struct WaitParams {
816 /// How long to hold the request open before returning a still-building
817 /// run. Default 30s, capped at 120s.
818 #[serde(default)]
819 timeout_ms: Option<u64>,
820 }
821
822 /// `GET /runs/{id}/wait` — long-poll: hold the request open until the run
823 /// settles (`result != building`) or the timeout elapses, then return the
824 /// current `RunView`. Removes polling-cadence guessing for a headless driver
825 /// (fire `/rebuild`, block on `/wait`). On timeout the run is returned
826 /// still-building (200) and the caller re-issues `/wait`. 404 if unknown.
827 async fn get_run_wait(
828 State(s): State<AppState>,
829 Path(id): Path<i64>,
830 Query(p): Query<WaitParams>,
831 ) -> Result<Json<crate::runs::RunView>> {
832 let run_id = crate::domain::RunId(id);
833 let timeout = std::time::Duration::from_millis(p.timeout_ms.unwrap_or(30_000).min(120_000));
834 let deadline = tokio::time::Instant::now() + timeout;
835 // Poll the row rather than wiring a per-run notifier: a build settles on
836 // the minute scale, so a sub-second tick is plenty responsive and the
837 // query is a single indexed read. The request releases its pool handle
838 // between ticks.
839 let tick = std::time::Duration::from_millis(750);
840 loop {
841 let view = crate::runs::get(&s.pool, run_id)
842 .await
843 .map_err(crate::error::Error::Other)?
844 .ok_or(crate::error::Error::NotFound)?;
845 let now = tokio::time::Instant::now();
846 if view.result != "building" || now >= deadline {
847 return Ok(Json(view));
848 }
849 tokio::time::sleep((deadline - now).min(tick)).await;
850 }
851 }
852
853 #[derive(Deserialize)]
854 struct SelfUpdateBody {
855 /// The commit to rebuild sandod from. Must already be on the canonical
856 /// remote (the updater `git fetch`es it).
857 sha: String,
858 }
859
860 /// The privileged updater unit instance for `sha`. A git sha is hex-only, so it
861 /// is a safe systemd instance name with no escaping needed.
862 fn self_update_unit(sha: &crate::domain::GitSha) -> String {
863 format!("sando-update@{sha}.service")
864 }
865
866 /// The bare repo sandod's own source lives in, and the branch that seals it.
867 ///
868 /// sandod is built from `sando/daemon` inside the **default product's** repo, so
869 /// that product's `[repo]` is the one the self-update path has to keep current.
870 /// Resolved here rather than from `s.topo` because `/self-update` is mounted
871 /// under every product (`/apps/pom/self-update` reaches the same daemon), and
872 /// pom's topology is intake-only — it declares no `[repo]` at all. Reading the
873 /// per-request product would make the fetch below silently a no-op on every
874 /// mount but one.
875 fn self_update_source(s: &AppState) -> Option<(std::path::PathBuf, String, Option<String>)> {
876 let repo = s.apps.get(&s.default_app)?.topo.repo.as_ref()?;
877 Some((
878 std::path::PathBuf::from(&repo.bare_path),
879 repo.branch.clone(),
880 repo.upstream.clone(),
881 ))
882 }
883
884 /// Trigger a rebuild + restart of sandod *itself* to `sha`. sandod runs
885 /// unprivileged (User=sando, NoNewPrivileges, ProtectSystem=strict) and cannot
886 /// write `/usr/local/bin/sandod` or restart its own service — so it only
887 /// *triggers* the root oneshot `sando-update@<sha>.service` (which the sando
888 /// user is authorized to start by a scoped polkit rule). That unit builds
889 /// `sando/daemon` as the sando user in a dedicated checkout, installs the new
890 /// binary, and restarts sandod. Bearer-gated like the other mutators; the new
891 /// version shows up in `/state`'s `sandod_version` once the restart lands.
892 ///
893 /// **The upstream fetch below is what makes a fresh sha reachable at all.** The
894 /// updater builds from the local bare repo and refuses any sha that is not an
895 /// ancestor of its `main` (see `deploy/sando-self-update.sh`, "Provenance"). That
896 /// bare repo used to advance only when `/rebuild` fetched upstream during a
897 /// *server* build, which chained the controller's currency to the server's
898 /// release cadence and meant an urgent controller fix could not ship while the
899 /// server was red. Fetching here decouples the two: the rule stays exactly as
900 /// strict, it is simply checked against the canonical remote rather than against
901 /// a local cache of it that only a server build refreshed.
902 async fn self_update(
903 State(s): State<AppState>,
904 crate::error::TypedBody(body): crate::error::TypedBody<SelfUpdateBody>,
905 ) -> Result<Json<serde_json::Value>> {
906 let sha = crate::domain::GitSha::parse(&body.sha)
907 .map_err(|e| crate::error::Error::BadRequest(format!("invalid sha: {e}")))?;
908
909 let unit = self_update_unit(&sha);
910
911 // Refresh the bare repo from the canonical remote, then check the sha against
912 // it here — before the trigger, so a bad sha is a synchronous 4xx rather than
913 // an `accepted: true` and an exit 4 buried in the updater's journal.
914 //
915 // Only sandod writes this repo, and only by fetching the deploy branch from
916 // the authenticated upstream, which is the whole basis of the updater's
917 // provenance seal. Doing the fetch here keeps that true: no new writer.
918 if let Some((bare, branch, upstream)) = self_update_source(&s) {
919 // A fetch failure is non-fatal, matching the build path: the sha may
920 // already be present from an earlier fetch, and the ancestry check below
921 // is the real gate. What must not happen is proceeding *silently*.
922 if let Some(upstream) = upstream.as_deref()
923 && let Err(e) = crate::git::fetch_upstream(&bare, upstream, &branch).await
924 {
925 tracing::warn!(error = %e, upstream, "self-update upstream fetch failed; checking the sha against the current bare-repo state");
926 }
927 match crate::git::is_ancestor(&bare, sha.as_str(), &branch).await {
928 Ok(true) => {}
929 Ok(false) => {
930 return Err(crate::error::Error::BadRequest(format!(
931 "sha {sha} is not an ancestor of {branch} in {} — push it to the upstream remote first (the updater refuses any sha off the deploy branch)",
932 bare.display(),
933 )));
934 }
935 // Never block the update on our own pre-check being unable to answer:
936 // the updater re-runs the identical check as root before it builds, so
937 // the seal holds regardless. Losing the good error message is the only
938 // cost of passing through here.
939 Err(e) => {
940 tracing::warn!(error = %e, "self-update ancestry pre-check could not run; deferring to the updater's own check");
941 }
942 }
943 }
944
945 // Don't restart the controller out from under an in-flight server build — the
946 // restart would SIGKILL it mid-deploy. Hold the active_build slot across the
947 // trigger, not just the check: releasing it before triggering let a /rebuild
948 // claim the slot in the gap and start a build the updater then killed (the
949 // Run 2 TOCTOU). With the guard held, no build can start until the updater is
950 // enqueued.
951 {
952 let slot = s.active_build.lock().await;
953 if slot.as_ref().is_some_and(|b| !b.handle.is_finished()) {
954 return Err(crate::error::Error::GateBlocked(
955 "a server build is in flight; retry /self-update once it settles".into(),
956 ));
957 }
958 // Also refuse during an in-flight promote/rollback: the updater's restart
959 // would SIGKILL the daemon mid-deploy. try_lock (not lock) so we reject
960 // rather than queue behind a long deploy; the guard is dropped immediately
961 // — it only needs to observe that no deploy holds it right now.
962 if s.deploy_lock.try_lock().is_err() {
963 return Err(crate::error::Error::GateBlocked(
964 "a promote/rollback is in flight; retry /self-update once it settles".into(),
965 ));
966 }
967
968 tracing::warn!(sha = %sha, unit = %unit, "self-update requested; triggering privileged updater");
969 // `--no-block`: return as soon as the job is enqueued. The build+restart
970 // outcome lands in `journalctl -u <unit>`; sandod is restarted out from
971 // under this request, so there is nothing more to await here.
972 let status = tokio::process::Command::new("systemctl")
973 .args(["start", "--no-block", &unit])
974 .status()
975 .await
976 .map_err(|e| crate::error::Error::Other(anyhow::anyhow!("spawning systemctl: {e}")))?;
977 if !status.success() {
978 return Err(crate::error::Error::Other(anyhow::anyhow!(
979 "systemctl start {unit} exited {status}; is sando-update@.service installed and the sando-user polkit rule in place?"
980 )));
981 }
982 }
983 Ok(Json(
984 serde_json::json!({ "accepted": true, "sha": sha.to_string(), "unit": unit }),
985 ))
986 }
987
988 async fn confirm(
989 State(s): State<AppState>,
990 Path(tier): Path<String>,
991 ) -> Result<Json<serde_json::Value>> {
992 // Operator-driven satisfaction of a `manual_confirm` gate. Looks up the
993 // pending version (current MM version, or the tier's own if non-mm) and
994 // inserts a passing gate_runs row so /promote can advance.
995 //
996 // Hold deploy_lock so the version we read and the row we insert reference the
997 // same tier state — a concurrent rollback can't change current_version between
998 // the read and the insert (ultra-fuzz Run 2, F3).
999 let _deploy_guard = s.deploy_lock.lock().await;
1000 let tier = crate::domain::TierId::new(tier);
1001 let target = s
1002 .topo
1003 .tiers
1004 .iter()
1005 .find(|t| t.name == tier)
1006 .ok_or(crate::error::Error::NotFound)?;
1007
1008 // The build id comes back with the version so the row this writes carries the
1009 // identity every other gate row on this tier carries. Without it a
1010 // `manual_confirm` was the one gate `/state` could not scope to a build, and
1011 // it read as "not run" on a tier reporting per build.
1012 let confirmed: Option<(Option<String>, Option<i64>)> = sqlx::query_as(
1013 "SELECT current_version, current_build_id FROM tier_state WHERE app = ? AND tier = ?",
1014 )
1015 .bind(&s.cfg.id)
1016 .bind(&target.name)
1017 .fetch_optional(&s.pool)
1018 .await
1019 .map_err(crate::error::Error::Db)?;
1020 let (version_str, build_id) = confirmed.unwrap_or((None, None));
1021 let version_str = version_str.ok_or_else(|| {
1022 crate::error::Error::GateBlocked(format!(
1023 "tier {tier} has no current_version; nothing to confirm"
1024 ))
1025 })?;
1026 let version = crate::domain::Version::parse(&version_str)
1027 .map_err(|e| crate::error::Error::Other(anyhow::anyhow!(e)))?;
1028
1029 let now = chrono::Utc::now().to_rfc3339();
1030 let outcome =
1031 crate::outcome::GateOutcome::passed(crate::outcome::PassNote::OperatorConfirmed {
1032 at: chrono::Utc::now(),
1033 });
1034 let outcome_json = serde_json::to_string(&outcome)
1035 .map_err(|e| crate::error::Error::Other(anyhow::anyhow!(e)))?;
1036 sqlx::query(
1037 "INSERT INTO gate_runs (app, version, tier, gate_kind, started_at, finished_at, status, outcome_json, build_id)
1038 VALUES (?, ?, ?, 'manual_confirm', ?, ?, 'passed', ?, ?)",
1039 )
1040 .bind(&s.cfg.id)
1041 .bind(&version).bind(&target.name).bind(&now).bind(&now).bind(&outcome_json)
1042 .bind(build_id)
1043 .execute(&s.pool).await.map_err(crate::error::Error::Db)?;
1044
1045 tracing::info!(tier = %tier, version = %version, "manual_confirm recorded");
1046 crate::events::emit(
1047 &s.events,
1048 crate::events::Event::ManualConfirm {
1049 tier: tier.clone(),
1050 version: version.clone(),
1051 },
1052 );
1053
1054 Ok(Json(
1055 serde_json::json!({ "tier": tier, "version": version }),
1056 ))
1057 }
1058
1059 #[derive(Deserialize, Default)]
1060 struct BackupFetchBody {
1061 /// Accept a dump that falls below the plausibility floor, re-baselining it to
1062 /// this fetch's size. For the case where the source legitimately shrank and
1063 /// the floor has wedged itself; the `gzip -t` integrity check still applies.
1064 /// Operator-only — never set this from a timer.
1065 #[serde(default)]
1066 force: bool,
1067 /// Fetch only this configured dump (`[[backup]]` name). Omitted, every
1068 /// configured dump is fetched — which is what the daily timer wants, since
1069 /// each one gates a different database.
1070 #[serde(default)]
1071 name: Option<String>,
1072 }
1073
1074 async fn backup_fetch(
1075 State(s): State<AppState>,
1076 body: Option<Json<BackupFetchBody>>,
1077 ) -> Result<Json<serde_json::Value>> {
1078 let body = body.map(|Json(b)| b).unwrap_or_default();
1079 let fetched = crate::backup::fetch(&s.pool, &s.cfg, &s.topo, body.force, body.name.as_deref())
1080 .await
1081 .map_err(crate::error::Error::Other)?;
1082 for fb in &fetched {
1083 crate::events::emit(
1084 &s.events,
1085 crate::events::Event::BackupFetched {
1086 source: fb.source.clone(),
1087 byte_size: fb.byte_size.unwrap_or(0),
1088 },
1089 );
1090 }
1091 // `backups` stays an array even for a one-entry topology: a caller that
1092 // parses the response should not start failing the day a second database
1093 // gets a gate. The top-level keys mirror the first entry for the operator
1094 // reading `curl` output, and for the pre-list callers that read them.
1095 let first = fetched.first();
1096 Ok(Json(serde_json::json!({
1097 "source": first.map(|f| f.source.clone()),
1098 "local_path": first.map(|f| f.local_path.clone()),
1099 "byte_size": first.and_then(|f| f.byte_size),
1100 "backups": fetched.iter().map(|f| serde_json::json!({
1101 "name": f.name,
1102 "source": f.source,
1103 "local_path": f.local_path,
1104 "byte_size": f.byte_size,
1105 })).collect::<Vec<_>>(),
1106 })))
1107 }
1108
1109 /// The tail of one gate's log.
1110 ///
1111 /// `run` is the `build_runs.id` the gate ran for — the directory
1112 /// [`GateCtx::log_scope`](crate::gates::GateCtx::log_scope) wrote under. It was
1113 /// the version string until 2026-08-20, when every run of a version shared one
1114 /// file and a reader of a rebuild got the previous attempt's output. Callers
1115 /// follow the `log_ref` on the gate row rather than building this path, so both
1116 /// scopes resolve and logs written under the old scheme stay readable at the
1117 /// path their row records.
1118 async fn get_gate_log(
1119 State(s): State<AppState>,
1120 Path((run, gate)): Path<(String, String)>,
1121 ) -> Result<axum::response::Response> {
1122 // Guard against `..` / absolute paths — the scope segment must be a single
1123 // safe component. Without this, `GET /logs/..%2Fetc/passwd` would escape
1124 // logs_root.
1125 fn safe(seg: &str) -> bool {
1126 !seg.is_empty() && !seg.contains('/') && !seg.contains('\\') && seg != "." && seg != ".."
1127 }
1128 if !safe(&run) {
1129 return Err(crate::error::Error::NotFound);
1130 }
1131 // The gate segment is an allowlisted kind, not a free-form filename: parse it
1132 // to `GateKind` so only the known `<kind>.log` files are reachable. This
1133 // closes `*.log` filename probing within a scope dir (traversal was already
1134 // blocked; this bounds the basename to the known set).
1135 //
1136 // A trailing `.log` is accepted so a caller can append a gate row's `log_ref`
1137 // to `/logs/` verbatim rather than taking it apart and putting it back
1138 // together. Following the ref is the correct way to reach a log — building
1139 // the path from the version reaches a different run's output on any version
1140 // that was rebuilt — so the shape that gets it right should be the easy one.
1141 let kind: crate::domain::GateKind = gate
1142 .strip_suffix(".log")
1143 .unwrap_or(&gate)
1144 .parse()
1145 .map_err(|_| crate::error::Error::NotFound)?;
1146 let path = s
1147 .cfg
1148 .logs_root
1149 .join(&run)
1150 .join(format!("{}.log", kind.as_str()));
1151 // Bound how much a single request can pull into the daemon's memory: a
1152 // runaway gate log shouldn't be read whole. Past the cap we return the tail
1153 // (the recent, relevant output) behind a truncation marker.
1154 const MAX_LOG_BYTES: u64 = 8 * 1024 * 1024;
1155 use tokio::io::{AsyncReadExt, AsyncSeekExt};
1156 let mut file = match tokio::fs::File::open(&path).await {
1157 Ok(f) => f,
1158 Err(e) if e.kind() == std::io::ErrorKind::NotFound => {
1159 return Err(crate::error::Error::NotFound);
1160 }
1161 Err(e) => return Err(crate::error::Error::Other(e.into())),
1162 };
1163 let len = file
1164 .metadata()
1165 .await
1166 .map_err(|e| crate::error::Error::Other(e.into()))?
1167 .len();
1168 let body: Vec<u8> = if len <= MAX_LOG_BYTES {
1169 let mut buf = Vec::with_capacity(len as usize);
1170 file.read_to_end(&mut buf)
1171 .await
1172 .map_err(|e| crate::error::Error::Other(e.into()))?;
1173 buf
1174 } else {
1175 file.seek(std::io::SeekFrom::End(-(MAX_LOG_BYTES as i64)))
1176 .await
1177 .map_err(|e| crate::error::Error::Other(e.into()))?;
1178 let mut buf = format!(
1179 "[log truncated: showing the last {} MiB of {len} bytes]\n",
1180 MAX_LOG_BYTES / 1024 / 1024,
1181 )
1182 .into_bytes();
1183 file.read_to_end(&mut buf)
1184 .await
1185 .map_err(|e| crate::error::Error::Other(e.into()))?;
1186 buf
1187 };
1188 Ok((
1189 [(
1190 axum::http::header::CONTENT_TYPE,
1191 "text/plain; charset=utf-8",
1192 )],
1193 body,
1194 )
1195 .into_response())
1196 }
1197
1198 async fn events_ws(ws: WebSocketUpgrade, State(s): State<AppState>) -> impl IntoResponse {
1199 use axum::extract::ws::Message;
1200 use tokio::sync::broadcast::error::RecvError;
1201
1202 ws.on_upgrade(move |mut socket| async move {
1203 // Subscribe to both channels and merge them: a lag on the high-rate log
1204 // stream emits its own `lagged` frame without dropping anything on the
1205 // status stream (and vice versa), so a busy gate's chunk firehose can't
1206 // evict a PromoteComplete/GateDone the operator needs to see.
1207 let mut status_rx = s.events.subscribe_status();
1208 let mut logs_rx = s.events.subscribe_logs();
1209 loop {
1210 let recv = tokio::select! {
1211 r = status_rx.recv() => r,
1212 r = logs_rx.recv() => r,
1213 };
1214 match recv {
1215 Ok(env) => {
1216 let json = match serde_json::to_string(&env) {
1217 Ok(s) => s,
1218 Err(e) => {
1219 tracing::warn!(error = %e, "events ws: serialize failed");
1220 continue;
1221 }
1222 };
1223 if socket.send(Message::Text(json.into())).await.is_err() {
1224 break;
1225 }
1226 }
1227 Err(RecvError::Lagged(n)) => {
1228 let _ = socket
1229 .send(Message::Text(
1230 format!(r#"{{"kind":"lagged","skipped":{n}}}"#).into(),
1231 ))
1232 .await;
1233 }
1234 Err(RecvError::Closed) => break,
1235 }
1236 }
1237 })
1238 }
1239
1240 #[cfg(test)]
1241 mod tests {
1242 use super::promotion::{
1243 Evidence, PromotedBuild, RollbackReport, rollback_deployed_nodes, unsatisfied_gates,
1244 };
1245 use super::*;
1246 use crate::config::AppConfig;
1247 use crate::topology::{BackupConfig, CanaryPolicy, Gate, Node, RepoConfig, Tier, Topology};
1248 use async_trait::async_trait;
1249 use axum::body::Body;
1250 use axum::http::{Request, StatusCode};
1251 use http_body_util::BodyExt;
1252 use ops_exec::{CapabilitySet, Executor, LogSink, RunOutput, Step, SyncOpts};
1253 use sqlx::SqlitePool;
1254 use sqlx::sqlite::SqlitePoolOptions;
1255 use std::collections::BTreeMap;
1256 use std::os::unix::process::ExitStatusExt;
1257 use std::path::PathBuf;
1258 use std::sync::{Arc, Mutex as StdMutex};
1259 use tower::ServiceExt;
1260
1261 async fn fresh_pool() -> SqlitePool {
1262 let pool = SqlitePoolOptions::new()
1263 .max_connections(1)
1264 .connect("sqlite::memory:")
1265 .await
1266 .unwrap();
1267 sqlx::migrate!("./migrations").run(&pool).await.unwrap();
1268 pool
1269 }
1270
1271 /// Two-tier topology used by the route tests: mm (provisioned, no nodes)
1272 /// → a (provisioned, one local node). Mirrors the production shape
1273 /// without involving real ssh / postgres.
1274 fn test_topo() -> Topology {
1275 Topology {
1276 repo: Some(RepoConfig {
1277 bare_path: "/tmp/test.git".into(),
1278 branch: "main".into(),
1279 upstream: None,
1280 }),
1281 backup: vec![BackupConfig {
1282 name: "server".into(),
1283 source: "file:///tmp/test-backup.sql".into(),
1284 local_path: "/tmp/local-backup.sql".into(),
1285 }],
1286 tiers: vec![
1287 Tier {
1288 public_url: None,
1289 name: "host".into(),
1290 provisioned: true,
1291 gates: vec![],
1292 canary: CanaryPolicy::Sequential,
1293 nodes: vec![],
1294 },
1295 Tier {
1296 public_url: None,
1297 name: "a".into(),
1298 provisioned: true,
1299 gates: vec![Gate::BootSmoke],
1300 canary: CanaryPolicy::Sequential,
1301 nodes: vec![Node {
1302 platform: None,
1303 base_image: None,
1304 libc: None,
1305 name: "a-local".into(),
1306 ssh_target: "local".into(),
1307 release_root: "/tmp/a-node".into(),
1308 service_name: "makenotwork.service".into(),
1309 health_url: None,
1310 config_check_env_file: None,
1311 actuate: crate::topology::default_actuate(),
1312 observe: crate::topology::default_observe(),
1313 companions: Vec::new(),
1314 }],
1315 },
1316 ],
1317 aux_repos: Vec::new(),
1318 }
1319 }
1320
1321 fn test_cfg() -> AppConfig {
1322 AppConfig {
1323 page_smoke_cmd: None,
1324 platform: None,
1325 code_smoke_env: BTreeMap::default(),
1326 id: crate::domain::AppId::default(),
1327 topology_path: PathBuf::from("/tmp/test-sando.toml"),
1328 build_host: Some("test-host".into()),
1329 workdir: PathBuf::from("/tmp/sando-work"),
1330 release_root: PathBuf::from("/tmp/sando-releases"),
1331 scratch_db_url: None,
1332 scratch_owner_role: "makenotwork".into(),
1333 boot_smoke_port: 18181,
1334 code_smoke_port: 18182,
1335 bin_names: vec!["makenotwork".into()],
1336 logs_root: PathBuf::from("/tmp/sando-logs"),
1337 release_contents: vec![],
1338 cargo_target_dir: None,
1339 gate_timeout_secs: 2400,
1340 companions: Vec::new(),
1341 test_targets: vec![crate::config::TestTarget {
1342 dir: PathBuf::from("server"),
1343 aux_repo: None,
1344 features: vec!["fast-tests".into()],
1345 all_features: false,
1346 scratch_db: true,
1347 }],
1348 migration_checks: vec![],
1349 frontend_builds: vec![],
1350 backup_max_age_hours: 48,
1351 }
1352 }
1353
1354 /// Two products mounted on one daemon address different state.
1355 ///
1356 /// The mount-per-product shape is what makes this true: each router carries
1357 /// its own product's config, so `/apps/pom/state` cannot answer from MNW's
1358 /// tiers even if a handler forgets the product exists. The root mount keeps
1359 /// meaning the default product, which is what the runbook and the TUI call.
1360 #[tokio::test]
1361 async fn each_app_is_addressable_and_the_root_stays_the_default() {
1362 let pool = fresh_pool().await;
1363 // MNW ships host + a; pom ships one tier of its own, named differently
1364 // so the response says which product answered.
1365 for (app, tiers) in [("mnw", vec!["host", "a"]), ("pom", vec!["pom-host"])] {
1366 for (i, name) in tiers.iter().enumerate() {
1367 sqlx::query("INSERT INTO tiers (app, name, ord, provisioned) VALUES (?, ?, ?, 1)")
1368 .bind(app)
1369 .bind(name)
1370 .bind(i as i64)
1371 .execute(&pool)
1372 .await
1373 .unwrap();
1374 sqlx::query("INSERT INTO tier_state (app, tier) VALUES (?, ?)")
1375 .bind(app)
1376 .bind(name)
1377 .execute(&pool)
1378 .await
1379 .unwrap();
1380 }
1381 }
1382
1383 let mnw_topo = Arc::new(test_topo());
1384 let mut pom_topo = test_topo();
1385 pom_topo.tiers = vec![crate::topology::Tier {
1386 public_url: None,
1387 name: "pom-host".into(),
1388 provisioned: true,
1389 gates: vec![],
1390 canary: crate::topology::CanaryPolicy::Sequential,
1391 nodes: vec![],
1392 }];
1393 let pom_topo = Arc::new(pom_topo);
1394
1395 let mnw_cfg = Arc::new(test_cfg());
1396 let mut pom = test_cfg();
1397 pom.id = crate::domain::AppId::new("pom");
1398 let pom_cfg = Arc::new(pom);
1399
1400 let mut apps = crate::state::AppMap::new();
1401 for (id, cfg, topo) in [
1402 (mnw_cfg.id.clone(), mnw_cfg.clone(), mnw_topo.clone()),
1403 (pom_cfg.id.clone(), pom_cfg.clone(), pom_topo.clone()),
1404 ] {
1405 let executors = Arc::new(crate::state::build_executors(&topo));
1406 apps.insert(
1407 id,
1408 Arc::new(crate::state::App {
1409 cfg,
1410 topo,
1411 executors,
1412 }),
1413 );
1414 }
1415 let state = AppState {
1416 pool,
1417 apps: Arc::new(apps),
1418 default_app: mnw_cfg.id.clone(),
1419 topo: mnw_topo,
1420 cfg: mnw_cfg,
1421 active_build: Arc::new(tokio::sync::Mutex::new(None)),
1422 deploy_lock: Arc::new(tokio::sync::Mutex::new(())),
1423 events: crate::events::channel(),
1424 executors: Arc::new(std::collections::HashMap::new()),
1425 api_token: None,
1426 };
1427
1428 let get = async |uri: &str| -> String {
1429 let resp = router_for_apps(state.clone())
1430 .oneshot(Request::builder().uri(uri).body(Body::empty()).unwrap())
1431 .await
1432 .unwrap();
1433 assert_eq!(resp.status(), StatusCode::OK, "GET {uri}");
1434 body_string(resp).await
1435 };
1436
1437 // The root is the default product.
1438 let root = get("/state").await;
1439 assert!(root.contains("\"host\""), "root /state: {root}");
1440 assert!(!root.contains("pom-host"), "root must not show pom: {root}");
1441
1442 // Each product answers under its own mount.
1443 let mnw = get("/apps/mnw/state").await;
1444 assert_eq!(mnw, root, "the default mount and the root are one product");
1445 let pom = get("/apps/pom/state").await;
1446 assert!(pom.contains("pom-host"), "/apps/pom/state: {pom}");
1447 assert!(
1448 !pom.contains("\"host\""),
1449 "pom must not see mnw's tiers: {pom}"
1450 );
1451
1452 // And the index says what is mounted.
1453 let index = get("/apps").await;
1454 assert!(
1455 index.contains("\"mnw\"") && index.contains("\"pom\""),
1456 "{index}"
1457 );
1458 assert!(index.contains("\"default_app\":\"mnw\""), "{index}");
1459
1460 // An unconfigured product is not a route.
1461 let resp = router_for_apps(state.clone())
1462 .oneshot(
1463 Request::builder()
1464 .uri("/apps/nope/state")
1465 .body(Body::empty())
1466 .unwrap(),
1467 )
1468 .await
1469 .unwrap();
1470 assert_eq!(resp.status(), StatusCode::NOT_FOUND);
1471 }
1472
1473 async fn test_state() -> AppState {
1474 let pool = fresh_pool().await;
1475 // Seed tier rows so FKs on tier_state / gate_runs are satisfied.
1476 for (i, name) in ["host", "a"].iter().enumerate() {
1477 sqlx::query(
1478 "INSERT INTO tiers (name, ord, provisioned, canary) VALUES (?, ?, 1, 'sequential')",
1479 )
1480 .bind(name)
1481 .bind(i as i64)
1482 .execute(&pool)
1483 .await
1484 .unwrap();
1485 sqlx::query("INSERT INTO tier_state (tier) VALUES (?)")
1486 .bind(name)
1487 .execute(&pool)
1488 .await
1489 .unwrap();
1490 }
1491 // Don't call install_recorder in tests — it touches a process-global
1492 // and conflicts when tests run in parallel.
1493 let topo = test_topo();
1494 let executors = Arc::new(crate::state::build_executors(&topo));
1495 let topo = Arc::new(topo);
1496 let cfg = Arc::new(test_cfg());
1497 let (apps, default_app) =
1498 crate::state::one_app(cfg.clone(), topo.clone(), executors.clone());
1499 AppState {
1500 pool,
1501 apps,
1502 default_app,
1503 topo,
1504 cfg,
1505 active_build: Arc::new(tokio::sync::Mutex::new(None)),
1506 deploy_lock: Arc::new(tokio::sync::Mutex::new(())),
1507 events: crate::events::channel(),
1508 executors,
1509 api_token: None,
1510 }
1511 }
1512
1513 async fn body_string(resp: axum::response::Response) -> String {
1514 let bytes = resp.into_body().collect().await.unwrap().to_bytes();
1515 String::from_utf8(bytes.to_vec()).unwrap()
1516 }
1517
1518 /// Insert the FK prerequisites for inserting gate_runs/tier_state rows.
1519 async fn seed(pool: &SqlitePool, tier: &str, version: &str) {
1520 sqlx::query("INSERT INTO tiers (name, ord, provisioned, canary) VALUES (?, 0, 1, 'sequential') ON CONFLICT DO NOTHING")
1521 .bind(tier).execute(pool).await.unwrap();
1522 sqlx::query("INSERT INTO versions (version, git_sha, built_at, artifact_path) VALUES (?, 'sha', datetime('now'), '/tmp/x') ON CONFLICT DO NOTHING")
1523 .bind(version).execute(pool).await.unwrap();
1524 sqlx::query("INSERT INTO tier_state (tier, current_version) VALUES (?, NULL) ON CONFLICT DO NOTHING")
1525 .bind(tier).execute(pool).await.unwrap();
1526 }
1527
1528 async fn insert_gate(pool: &SqlitePool, tier: &str, version: &str, kind: &str, passed: i64) {
1529 let status = if passed == 1 { "passed" } else { "failed" };
1530 sqlx::query(
1531 "INSERT INTO gate_runs (version, tier, gate_kind, started_at, finished_at, status) \
1532 VALUES (?, ?, ?, datetime('now'), datetime('now'), ?)",
1533 )
1534 .bind(version)
1535 .bind(tier)
1536 .bind(kind)
1537 .bind(status)
1538 .execute(pool)
1539 .await
1540 .unwrap();
1541 }
1542
1543 /// `/state` says when a tier's artifact is gone, and stays quiet when it is
1544 /// not.
1545 ///
1546 /// The gap this closes: `versions`/`build_runs` keep naming a path long
1547 /// after gc removed the bytes, and nothing reconciled the two. Both times it
1548 /// happened, the tier read green until an rsync failed mid-promote.
1549 #[tokio::test]
1550 async fn state_reports_a_referenced_artifact_whose_bytes_are_gone() {
1551 let tmp = tempfile::tempdir().unwrap();
1552 let present = tmp.path().join("releases").join("1111111111111111");
1553 tokio::fs::create_dir_all(&present).await.unwrap();
1554
1555 let state = test_state().await;
1556 let bin = state.cfg.primary_bin().to_string();
1557 tokio::fs::write(present.join(&bin), b"bin").await.unwrap();
1558
1559 seed(&state.pool, "a", "0.11.20").await;
1560 let build = seed_build(
1561 &state.pool,
1562 "2a53c900",
1563 "0.11.20",
1564 present.to_string_lossy().as_ref(),
1565 )
1566 .await;
1567 sqlx::query(
1568 "UPDATE tier_state SET current_version = ?, current_build_id = ? WHERE tier = 'a'",
1569 )
1570 .bind("0.11.20")
1571 .bind(build)
1572 .execute(&state.pool)
1573 .await
1574 .unwrap();
1575
1576 let tier_a = |v: &StateView| {
1577 v.tiers
1578 .iter()
1579 .find(|t| t.name == "a")
1580 .expect("tier a")
1581 .missing_artifact
1582 .clone()
1583 };
1584
1585 assert_eq!(
1586 tier_a(&state_view(&state).await.unwrap()),
1587 None,
1588 "the artifact is on disk; nothing to report"
1589 );
1590
1591 // gc takes it, as it did twice in production. Nothing in the database
1592 // changes, which is the whole defect.
1593 tokio::fs::remove_dir_all(&present).await.unwrap();
1594
1595 let said = tier_a(&state_view(&state).await.unwrap()).expect("a missing-artifact report");
1596 assert!(said.contains("0.11.20"), "{said}");
1597 assert!(said.contains("current"), "{said}");
1598 }
1599
1600 /// A rebuild at an unchanged version must not move what a tier reports.
1601 ///
1602 /// `/state` keyed its gate rows on (tier, version), so two runs of one
1603 /// version interleaved and each gate showed whichever had written it last.
1604 /// The visible symptom was statuses that moved while nothing was touched:
1605 /// two reads of the same tier a minute apart disagreed about whether
1606 /// `hardening_test` had passed or never run. Runs 60-62 of mnw-server
1607 /// 0.11.20, 2026-08-19.
1608 #[tokio::test]
1609 async fn state_reports_the_build_the_tier_is_running() {
1610 let state = test_state().await;
1611 seed(&state.pool, "a", "0.11.20").await;
1612 let first = seed_build(&state.pool, "2a53c900", "0.11.20", "/rel/1111111111111111").await;
1613 let second = seed_build(&state.pool, "adf56cd9", "0.11.20", "/rel/2222222222222222").await;
1614 // The tier is running the first build, and it went green there.
1615 insert_gate_build(&state.pool, "a", "0.11.20", "hardening_test", 1, first).await;
1616 sqlx::query(
1617 "UPDATE tier_state SET current_version = ?, current_build_id = ? WHERE tier = 'a'",
1618 )
1619 .bind("0.11.20")
1620 .bind(first)
1621 .execute(&state.pool)
1622 .await
1623 .unwrap();
1624
1625 let before = state_view(&state).await.unwrap();
1626 let gates_of = |v: &StateView| {
1627 v.tiers
1628 .iter()
1629 .find(|t| t.name == "a")
1630 .unwrap()
1631 .gates
1632 .iter()
1633 .map(|g| (g.kind.clone(), g.status.clone()))
1634 .collect::<Vec<_>>()
1635 };
1636 assert_eq!(
1637 gates_of(&before),
1638 vec![("hardening_test".to_string(), Some("passed".to_string()))],
1639 );
1640
1641 // A retry of the same version fails the same gate. The tier still runs
1642 // the first build, so what it reports is unchanged.
1643 insert_gate_build(&state.pool, "a", "0.11.20", "hardening_test", 0, second).await;
1644 assert_eq!(
1645 gates_of(&state_view(&state).await.unwrap()),
1646 gates_of(&before),
1647 "a sibling rebuild rewrote a tier it was never deployed to",
1648 );
1649 }
1650
1651 // ---- unsatisfied_gates ----
1652
1653 fn tid(s: &str) -> crate::domain::TierId {
1654 crate::domain::TierId::new(s)
1655 }
1656
1657 #[tokio::test]
1658 async fn unsatisfied_gates_empty_when_no_configured_gates() {
1659 // A tier that configures no gates has nothing to satisfy.
1660 let pool = fresh_pool().await;
1661 seed(&pool, "host", "0.8.12").await;
1662 let pending = unsatisfied_gates(
1663 &pool,
1664 &crate::domain::AppId::default(),
1665 &tid("host"),
1666 &[],
1667 &Evidence {
1668 version: "0.8.12",
1669 builds: &[],
1670 tier_build: None,
1671 },
1672 false,
1673 )
1674 .await
1675 .unwrap();
1676 assert_eq!(pending, Vec::<String>::new());
1677 }
1678
1679 #[tokio::test]
1680 async fn unsatisfied_gates_flags_configured_gate_that_never_ran() {
1681 // THE CF1 FIX: a configured gate with no gate_runs row is unsatisfied
1682 // (fail closed), NOT silently treated as green. Before this, an A tier
1683 // whose boot_smoke never executed exposed zero rows and waved promotion
1684 // straight through to prod.
1685 let pool = fresh_pool().await;
1686 seed(&pool, "a", "0.8.12").await;
1687 let pending = unsatisfied_gates(
1688 &pool,
1689 &crate::domain::AppId::default(),
1690 &tid("a"),
1691 &[Gate::BootSmoke],
1692 &Evidence {
1693 version: "0.8.12",
1694 builds: &[],
1695 tier_build: None,
1696 },
1697 false,
1698 )
1699 .await
1700 .unwrap();
1701 assert_eq!(pending, vec!["boot_smoke".to_string()]);
1702 }
1703
1704 #[tokio::test]
1705 async fn unsatisfied_gates_flags_failed_kind() {
1706 let pool = fresh_pool().await;
1707 seed(&pool, "host", "0.8.12").await;
1708 insert_gate(&pool, "host", "0.8.12", "cargo_test", 0).await;
1709 insert_gate(&pool, "host", "0.8.12", "boot_smoke", 1).await;
1710 let pending = unsatisfied_gates(
1711 &pool,
1712 &crate::domain::AppId::default(),
1713 &tid("host"),
1714 &[Gate::CargoTest, Gate::BootSmoke],
1715 &Evidence {
1716 version: "0.8.12",
1717 builds: &[],
1718 tier_build: None,
1719 },
1720 false,
1721 )
1722 .await
1723 .unwrap();
1724 assert_eq!(pending, vec!["cargo_test".to_string()]);
1725 }
1726
1727 #[tokio::test]
1728 async fn unsatisfied_gates_latest_row_wins() {
1729 // Two runs of the same gate; only the latest counts. A flap from
1730 // red to green should clear the pending entry.
1731 let pool = fresh_pool().await;
1732 seed(&pool, "host", "0.8.12").await;
1733 insert_gate(&pool, "host", "0.8.12", "cargo_test", 0).await;
1734 insert_gate(&pool, "host", "0.8.12", "cargo_test", 1).await;
1735 let pending = unsatisfied_gates(
1736 &pool,
1737 &crate::domain::AppId::default(),
1738 &tid("host"),
1739 &[Gate::CargoTest],
1740 &Evidence {
1741 version: "0.8.12",
1742 builds: &[],
1743 tier_build: None,
1744 },
1745 false,
1746 )
1747 .await
1748 .unwrap();
1749 assert!(pending.is_empty());
1750 }
1751
1752 async fn insert_confirm(
1753 pool: &SqlitePool,
1754 tier: &str,
1755 version: &str,
1756 at: chrono::DateTime<chrono::Utc>,
1757 ) {
1758 sqlx::query(
1759 "INSERT INTO gate_runs (version, tier, gate_kind, started_at, finished_at, status) \
1760 VALUES (?, ?, 'manual_confirm', ?, ?, 'passed')",
1761 )
1762 .bind(version)
1763 .bind(tier)
1764 .bind(at.to_rfc3339())
1765 .bind(at.to_rfc3339())
1766 .execute(pool)
1767 .await
1768 .unwrap();
1769 }
1770
1771 #[tokio::test]
1772 async fn unsatisfied_gates_manual_confirm_requires_fresh_confirmation() {
1773 // A confirmation only satisfies the gate if it post-dates the version's
1774 // current landing on the tier (burn_in_started_at). A stale confirm left
1775 // over from before a rollback + rollback-forward must NOT wave it through.
1776 let pool = fresh_pool().await;
1777 seed(&pool, "a", "0.8.12").await;
1778 let landed = chrono::Utc::now();
1779 sqlx::query("UPDATE tier_state SET burn_in_started_at = ? WHERE tier = 'a'")
1780 .bind(landed.to_rfc3339())
1781 .execute(&pool)
1782 .await
1783 .unwrap();
1784
1785 // Stale confirmation (recorded before this landing) -> unsatisfied.
1786 insert_confirm(&pool, "a", "0.8.12", landed - chrono::Duration::hours(1)).await;
1787 let pending = unsatisfied_gates(
1788 &pool,
1789 &crate::domain::AppId::default(),
1790 &tid("a"),
1791 &[Gate::ManualConfirm],
1792 &Evidence {
1793 version: "0.8.12",
1794 builds: &[],
1795 tier_build: None,
1796 },
1797 false,
1798 )
1799 .await
1800 .unwrap();
1801 assert_eq!(
1802 pending,
1803 vec!["manual_confirm".to_string()],
1804 "stale confirm must not satisfy"
1805 );
1806
1807 // Fresh confirmation (after this landing) -> satisfied.
1808 insert_confirm(&pool, "a", "0.8.12", landed + chrono::Duration::minutes(5)).await;
1809 let pending = unsatisfied_gates(
1810 &pool,
1811 &crate::domain::AppId::default(),
1812 &tid("a"),
1813 &[Gate::ManualConfirm],
1814 &Evidence {
1815 version: "0.8.12",
1816 builds: &[],
1817 tier_build: None,
1818 },
1819 false,
1820 )
1821 .await
1822 .unwrap();
1823 assert!(pending.is_empty(), "fresh confirm satisfies");
1824 }
1825
1826 #[tokio::test]
1827 async fn unsatisfied_gates_manual_confirm_fails_closed_without_baseline() {
1828 // No landing clock (burn_in_started_at NULL) -> a passed confirm row is
1829 // not provably fresh, so fail closed and require a new confirmation.
1830 let pool = fresh_pool().await;
1831 seed(&pool, "a", "0.8.12").await; // leaves burn_in_started_at NULL
1832 insert_confirm(&pool, "a", "0.8.12", chrono::Utc::now()).await;
1833 let pending = unsatisfied_gates(
1834 &pool,
1835 &crate::domain::AppId::default(),
1836 &tid("a"),
1837 &[Gate::ManualConfirm],
1838 &Evidence {
1839 version: "0.8.12",
1840 builds: &[],
1841 tier_build: None,
1842 },
1843 false,
1844 )
1845 .await
1846 .unwrap();
1847 assert_eq!(
1848 pending,
1849 vec!["manual_confirm".to_string()],
1850 "no baseline -> fail closed"
1851 );
1852 }
1853
1854 #[tokio::test]
1855 async fn unsatisfied_gates_hotfix_skips_only_burn_in() {
1856 // burn_in is evaluated live (no clock started -> not elapsed); cargo_test
1857 // has a failing row. Normal: both unsatisfied, in configured order.
1858 // hotfix: burn_in suppressed, cargo_test still flagged. Lock the semantic
1859 // so a future change doesn't widen the hotfix bypass.
1860 let pool = fresh_pool().await;
1861 seed(&pool, "a", "0.8.12").await;
1862 insert_gate(&pool, "a", "0.8.12", "cargo_test", 0).await;
1863 let gates = [Gate::BurnIn { hours: 48 }, Gate::CargoTest];
1864
1865 let normal = unsatisfied_gates(
1866 &pool,
1867 &crate::domain::AppId::default(),
1868 &tid("a"),
1869 &gates,
1870 &Evidence {
1871 version: "0.8.12",
1872 builds: &[],
1873 tier_build: None,
1874 },
1875 false,
1876 )
1877 .await
1878 .unwrap();
1879 assert_eq!(
1880 normal,
1881 vec!["burn_in".to_string(), "cargo_test".to_string()]
1882 );
1883
1884 let with_hotfix = unsatisfied_gates(
1885 &pool,
1886 &crate::domain::AppId::default(),
1887 &tid("a"),
1888 &gates,
1889 &Evidence {
1890 version: "0.8.12",
1891 builds: &[],
1892 tier_build: None,
1893 },
1894 true,
1895 )
1896 .await
1897 .unwrap();
1898 assert_eq!(with_hotfix, vec!["cargo_test".to_string()]);
1899 }
1900
1901 #[tokio::test]
1902 async fn unsatisfied_gates_burn_in_passes_when_window_elapsed() {
1903 // A burn-in clock started far enough in the past satisfies the gate
1904 // live — no gate_runs row needed.
1905 let pool = fresh_pool().await;
1906 seed(&pool, "a", "0.8.12").await;
1907 sqlx::query("UPDATE tier_state SET burn_in_started_at = ? WHERE tier = 'a'")
1908 .bind((chrono::Utc::now() - chrono::Duration::hours(50)).to_rfc3339())
1909 .execute(&pool)
1910 .await
1911 .unwrap();
1912 let pending = unsatisfied_gates(
1913 &pool,
1914 &crate::domain::AppId::default(),
1915 &tid("a"),
1916 &[Gate::BurnIn { hours: 48 }],
1917 &Evidence {
1918 version: "0.8.12",
1919 builds: &[],
1920 tier_build: None,
1921 },
1922 false,
1923 )
1924 .await
1925 .unwrap();
1926 assert!(pending.is_empty(), "50h elapsed satisfies a 48h burn-in");
1927 }
1928
1929 #[tokio::test]
1930 async fn unsatisfied_gates_ignores_other_tiers_and_versions() {
1931 let pool = fresh_pool().await;
1932 seed(&pool, "host", "0.8.12").await;
1933 seed(&pool, "host", "0.8.11").await;
1934 seed(&pool, "a", "0.8.12").await;
1935 // Mark host/0.8.12 cargo_test failing, but unrelated tiers/versions
1936 // shouldn't pollute the query.
1937 insert_gate(&pool, "host", "0.8.12", "cargo_test", 0).await;
1938 insert_gate(&pool, "a", "0.8.12", "cargo_test", 0).await;
1939 insert_gate(&pool, "host", "0.8.11", "cargo_test", 0).await;
1940
1941 let pending = unsatisfied_gates(
1942 &pool,
1943 &crate::domain::AppId::default(),
1944 &tid("host"),
1945 &[Gate::CargoTest],
1946 &Evidence {
1947 version: "0.8.12",
1948 builds: &[],
1949 tier_build: None,
1950 },
1951 false,
1952 )
1953 .await
1954 .unwrap();
1955 assert_eq!(pending, vec!["cargo_test".to_string()]);
1956 }
1957
1958 #[tokio::test]
1959 async fn unsatisfied_gates_null_status_is_treated_as_failing() {
1960 // An in-flight gate (started_at set, finished_at + status NULL)
1961 // should NOT be treated as green. Otherwise a race could promote
1962 // before the gate concludes.
1963 let pool = fresh_pool().await;
1964 seed(&pool, "host", "0.8.12").await;
1965 sqlx::query(
1966 "INSERT INTO gate_runs (version, tier, gate_kind, started_at) \
1967 VALUES ('0.8.12', 'host', 'cargo_test', datetime('now'))",
1968 )
1969 .execute(&pool)
1970 .await
1971 .unwrap();
1972
1973 let pending = unsatisfied_gates(
1974 &pool,
1975 &crate::domain::AppId::default(),
1976 &tid("host"),
1977 &[Gate::CargoTest],
1978 &Evidence {
1979 version: "0.8.12",
1980 builds: &[],
1981 tier_build: None,
1982 },
1983 false,
1984 )
1985 .await
1986 .unwrap();
1987 assert_eq!(pending, vec!["cargo_test".to_string()]);
1988 }
1989
1990 // ---- /confirm/{tier} ----
1991
1992 #[tokio::test]
1993 async fn confirm_rejects_when_tier_has_no_current_version() {
1994 // tier_state.a.current_version is NULL by default. /confirm has
1995 // nothing to confirm against → GateBlocked (400).
1996 let state = test_state().await;
1997 let app = router(state.clone());
1998 let resp = app
1999 .oneshot(
2000 Request::builder()
2001 .method("POST")
2002 .uri("/confirm/a")
2003 .body(Body::empty())
2004 .unwrap(),
2005 )
2006 .await
2007 .unwrap();
2008 assert_eq!(resp.status(), StatusCode::CONFLICT);
2009 let body = body_string(resp).await;
2010 assert!(body.contains("no current_version"), "got: {body}");
2011 }
2012
2013 #[tokio::test]
2014 async fn confirm_accepts_when_current_version_set_and_inserts_row() {
2015 let state = test_state().await;
2016 // Seed a version + advance tier a's state to it.
2017 sqlx::query("INSERT INTO versions (version, git_sha, built_at, artifact_path) VALUES ('0.8.12','sha',datetime('now'),'/tmp/x')")
2018 .execute(&state.pool).await.unwrap();
2019 sqlx::query("UPDATE tier_state SET current_version = '0.8.12' WHERE tier = 'a'")
2020 .execute(&state.pool)
2021 .await
2022 .unwrap();
2023
2024 let app = router(state.clone());
2025 let resp = app
2026 .oneshot(
2027 Request::builder()
2028 .method("POST")
2029 .uri("/confirm/a")
2030 .body(Body::empty())
2031 .unwrap(),
2032 )
2033 .await
2034 .unwrap();
2035 assert_eq!(resp.status(), StatusCode::OK);
2036 let body = body_string(resp).await;
2037 assert!(body.contains("\"tier\":\"a\""));
2038 assert!(body.contains("\"version\":\"0.8.12\""));
2039
2040 // A passing gate_runs row was inserted.
2041 let count: (i64,) = sqlx::query_as(
2042 "SELECT COUNT(*) FROM gate_runs WHERE tier='a' AND gate_kind='manual_confirm' AND status='passed'",
2043 )
2044 .fetch_one(&state.pool)
2045 .await
2046 .unwrap();
2047 assert_eq!(count.0, 1);
2048 }
2049
2050 #[tokio::test]
2051 async fn confirm_404s_for_unknown_tier() {
2052 let state = test_state().await;
2053 let app = router(state);
2054 let resp = app
2055 .oneshot(
2056 Request::builder()
2057 .method("POST")
2058 .uri("/confirm/zzzz")
2059 .body(Body::empty())
2060 .unwrap(),
2061 )
2062 .await
2063 .unwrap();
2064 assert_eq!(resp.status(), StatusCode::NOT_FOUND);
2065 }
2066
2067 #[tokio::test]
2068 async fn get_run_404s_for_unknown_id() {
2069 let state = test_state().await;
2070 let app = router(state);
2071 let resp = app
2072 .oneshot(
2073 Request::builder()
2074 .uri("/runs/999")
2075 .body(Body::empty())
2076 .unwrap(),
2077 )
2078 .await
2079 .unwrap();
2080 assert_eq!(resp.status(), StatusCode::NOT_FOUND);
2081 }
2082
2083 #[tokio::test]
2084 async fn get_run_returns_view_with_gates() {
2085 let state = test_state().await;
2086 // A run that reached version 0.10.2 and ran two host gates (one red).
2087 let run_id = crate::runs::create(&state.pool, &state.cfg.id, "abc1234def")
2088 .await
2089 .unwrap();
2090 let ver: crate::domain::Version = "0.10.2".parse().unwrap();
2091 seed(&state.pool, "host", "0.10.2").await;
2092 crate::runs::set_version(&state.pool, run_id, &ver)
2093 .await
2094 .unwrap();
2095 // Keyed on the run, not on the version: these are the rows that carry
2096 // this run's build id, and a sibling rebuild of 0.10.2 writing its own
2097 // rows must not change what this run reports.
2098 insert_gate_build(&state.pool, "host", "0.10.2", "cargo_test", 0, run_id.0).await;
2099 insert_gate_build(&state.pool, "host", "0.10.2", "boot_smoke", 1, run_id.0).await;
2100
2101 let app = router(state);
2102 let resp = app
2103 .oneshot(
2104 Request::builder()
2105 .uri(format!("/runs/{}", run_id.0))
2106 .body(Body::empty())
2107 .unwrap(),
2108 )
2109 .await
2110 .unwrap();
2111 assert_eq!(resp.status(), StatusCode::OK);
2112 let v: serde_json::Value = serde_json::from_str(&body_string(resp).await).unwrap();
2113 assert_eq!(v["run_id"], run_id.0);
2114 assert_eq!(v["sha"], "abc1234def");
2115 assert_eq!(v["version"], "0.10.2");
2116 assert_eq!(v["result"], "building");
2117 // Both host gates surface, latest-per-kind, alphabetized by kind.
2118 assert_eq!(v["gates"].as_array().unwrap().len(), 2);
2119 assert_eq!(v["gates"][0]["kind"], "boot_smoke");
2120 assert_eq!(v["gates"][0]["status"], "passed");
2121 assert_eq!(v["gates"][1]["kind"], "cargo_test");
2122 assert_eq!(v["gates"][1]["status"], "failed");
2123 }
2124
2125 #[tokio::test]
2126 async fn get_run_ignores_a_sibling_rebuild_of_the_same_version() {
2127 // A rebuild at an unchanged version is the normal way to retry a red
2128 // build. Keyed on (tier, version), the two runs' rows interleaved and
2129 // each gate reported whichever run wrote it last — a reader of /runs/{id}
2130 // could watch a passed gate become "not run" a minute later with nothing
2131 // touched. Runs 60-62 of mnw-server 0.11.20, 2026-08-19.
2132 let state = test_state().await;
2133 seed(&state.pool, "host", "0.11.20").await;
2134 let ver: crate::domain::Version = "0.11.20".parse().unwrap();
2135
2136 let first = crate::runs::create(&state.pool, &state.cfg.id, "2a53c900")
2137 .await
2138 .unwrap();
2139 crate::runs::set_version(&state.pool, first, &ver)
2140 .await
2141 .unwrap();
2142 insert_gate_build(&state.pool, "host", "0.11.20", "cargo_deny", 0, first.0).await;
2143
2144 let second = crate::runs::create(&state.pool, &state.cfg.id, "adf56cd9")
2145 .await
2146 .unwrap();
2147 crate::runs::set_version(&state.pool, second, &ver)
2148 .await
2149 .unwrap();
2150 insert_gate_build(&state.pool, "host", "0.11.20", "cargo_deny", 1, second.0).await;
2151
2152 // The later run passing does not turn the earlier run green, and the
2153 // earlier run failing does not follow the later one.
2154 for (run, want) in [(first, "failed"), (second, "passed")] {
2155 let app = router(state.clone());
2156 let resp = app
2157 .oneshot(
2158 Request::builder()
2159 .uri(format!("/runs/{}", run.0))
2160 .body(Body::empty())
2161 .unwrap(),
2162 )
2163 .await
2164 .unwrap();
2165 assert_eq!(resp.status(), StatusCode::OK);
2166 let v: serde_json::Value = serde_json::from_str(&body_string(resp).await).unwrap();
2167 assert_eq!(v["gates"].as_array().unwrap().len(), 1, "run {}", run.0);
2168 assert_eq!(v["gates"][0]["kind"], "cargo_deny");
2169 assert_eq!(v["gates"][0]["status"], want, "run {}", run.0);
2170 }
2171 }
2172
2173 #[tokio::test]
2174 async fn get_run_wait_returns_immediately_when_settled() {
2175 let state = test_state().await;
2176 let run_id = crate::runs::create(&state.pool, &state.cfg.id, "abc1234def")
2177 .await
2178 .unwrap();
2179 crate::runs::mark_passed(&state.pool, run_id).await.unwrap();
2180
2181 let app = router(state);
2182 // Generous timeout, but an already-settled run must not wait for it.
2183 let resp = app
2184 .oneshot(
2185 Request::builder()
2186 .uri(format!("/runs/{}/wait?timeout_ms=60000", run_id.0))
2187 .body(Body::empty())
2188 .unwrap(),
2189 )
2190 .await
2191 .unwrap();
2192 assert_eq!(resp.status(), StatusCode::OK);
2193 let v: serde_json::Value = serde_json::from_str(&body_string(resp).await).unwrap();
2194 assert_eq!(v["result"], "passed");
2195 }
2196
2197 #[tokio::test]
2198 async fn get_run_wait_returns_building_at_timeout() {
2199 let state = test_state().await;
2200 let run_id = crate::runs::create(&state.pool, &state.cfg.id, "abc1234def")
2201 .await
2202 .unwrap();
2203
2204 let app = router(state);
2205 // timeout_ms=0 → deadline is now → the first poll returns the
2206 // still-building run rather than blocking.
2207 let resp = app
2208 .oneshot(
2209 Request::builder()
2210 .uri(format!("/runs/{}/wait?timeout_ms=0", run_id.0))
2211 .body(Body::empty())
2212 .unwrap(),
2213 )
2214 .await
2215 .unwrap();
2216 assert_eq!(resp.status(), StatusCode::OK);
2217 let v: serde_json::Value = serde_json::from_str(&body_string(resp).await).unwrap();
2218 assert_eq!(v["result"], "building");
2219 }
2220
2221 #[tokio::test]
2222 async fn get_run_wait_404s_for_unknown_id() {
2223 let state = test_state().await;
2224 let app = router(state);
2225 let resp = app
2226 .oneshot(
2227 Request::builder()
2228 .uri("/runs/999/wait?timeout_ms=0")
2229 .body(Body::empty())
2230 .unwrap(),
2231 )
2232 .await
2233 .unwrap();
2234 assert_eq!(resp.status(), StatusCode::NOT_FOUND);
2235 }
2236
2237 #[test]
2238 fn self_update_unit_maps_sha_to_instance() {
2239 let sha = crate::domain::GitSha::parse("abc1234def5678").unwrap();
2240 assert_eq!(
2241 self_update_unit(&sha),
2242 "sando-update@abc1234def5678.service"
2243 );
2244 }
2245
2246 #[tokio::test]
2247 async fn self_update_rejects_bad_sha_with_400() {
2248 // A malformed sha is a client error and must be rejected *before* any
2249 // privileged unit is triggered (so this test never shells out).
2250 let state = test_state().await;
2251 let app = router(state);
2252 let resp = app
2253 .oneshot(
2254 Request::builder()
2255 .method("POST")
2256 .uri("/self-update")
2257 .header("Content-Type", "application/json")
2258 .body(Body::from(r#"{"sha":"not-a-sha!"}"#))
2259 .unwrap(),
2260 )
2261 .await
2262 .unwrap();
2263 assert_eq!(resp.status(), StatusCode::BAD_REQUEST);
2264 }
2265
2266 // ---- /promote/{tier} default-version resolution ----
2267
2268 #[tokio::test]
2269 async fn promote_to_first_tier_is_rejected() {
2270 // tier 0 is host — you /rebuild, not /promote.
2271 let state = test_state().await;
2272 let app = router(state);
2273 let resp = app
2274 .oneshot(
2275 Request::builder()
2276 .method("POST")
2277 .uri("/promote/host")
2278 .body(Body::empty())
2279 .unwrap(),
2280 )
2281 .await
2282 .unwrap();
2283 assert_eq!(resp.status(), StatusCode::CONFLICT);
2284 let body = body_string(resp).await;
2285 assert!(
2286 body.contains("cannot /promote to the first tier"),
2287 "got: {body}"
2288 );
2289 }
2290
2291 #[tokio::test]
2292 async fn promote_without_body_and_no_predecessor_version_errors() {
2293 // tier a has no body version supplied AND its predecessor mm has
2294 // current_version=NULL. Should fail before any deploy.
2295 let state = test_state().await;
2296 let app = router(state);
2297 let resp = app
2298 .oneshot(
2299 Request::builder()
2300 .method("POST")
2301 .uri("/promote/a")
2302 .body(Body::empty())
2303 .unwrap(),
2304 )
2305 .await
2306 .unwrap();
2307 assert_eq!(resp.status(), StatusCode::CONFLICT);
2308 let body = body_string(resp).await;
2309 assert!(
2310 body.contains("no version specified") || body.contains("no current_version"),
2311 "got: {body}"
2312 );
2313 }
2314
2315 #[tokio::test]
2316 async fn promote_blocked_when_predecessor_gate_never_ran() {
2317 // End-to-end CF1: the host tier configures boot_smoke but it never ran
2318 // (no gate_runs row). Promoting host -> a must be GateBlocked, citing the
2319 // unsatisfied gate, instead of waving through on zero evidence. A real
2320 // `versions` row is present so the ONLY thing that can block is the gate.
2321 let pool = fresh_pool().await;
2322 for (i, name) in ["host", "a"].iter().enumerate() {
2323 sqlx::query(
2324 "INSERT INTO tiers (name, ord, provisioned, canary) VALUES (?, ?, 1, 'sequential')",
2325 )
2326 .bind(name)
2327 .bind(i as i64)
2328 .execute(&pool)
2329 .await
2330 .unwrap();
2331 sqlx::query("INSERT INTO tier_state (tier) VALUES (?)")
2332 .bind(name)
2333 .execute(&pool)
2334 .await
2335 .unwrap();
2336 }
2337 let mut topo = test_topo();
2338 topo.tiers[0].gates = vec![Gate::BootSmoke]; // host configures a gate...
2339 let executors = Arc::new(crate::state::build_executors(&topo));
2340 let topo = Arc::new(topo);
2341 let cfg = Arc::new(test_cfg());
2342 let (apps, default_app) =
2343 crate::state::one_app(cfg.clone(), topo.clone(), executors.clone());
2344 let state = AppState {
2345 pool,
2346 apps,
2347 default_app,
2348 topo,
2349 cfg,
2350 active_build: Arc::new(tokio::sync::Mutex::new(None)),
2351 deploy_lock: Arc::new(tokio::sync::Mutex::new(())),
2352 events: crate::events::channel(),
2353 executors,
2354 api_token: None,
2355 };
2356 sqlx::query("INSERT INTO versions (version, git_sha, built_at, artifact_path) VALUES ('0.8.12','sha',datetime('now'),'/tmp/x')")
2357 .execute(&state.pool).await.unwrap();
2358 sqlx::query("UPDATE tier_state SET current_version = '0.8.12' WHERE tier = 'host'")
2359 .execute(&state.pool)
2360 .await
2361 .unwrap();
2362
2363 let app = router(state);
2364 let resp = app
2365 .oneshot(
2366 Request::builder()
2367 .method("POST")
2368 .uri("/promote/a")
2369 .body(Body::empty())
2370 .unwrap(),
2371 )
2372 .await
2373 .unwrap();
2374 assert_eq!(resp.status(), StatusCode::CONFLICT);
2375 let body = body_string(resp).await;
2376 assert!(
2377 body.contains("boot_smoke"),
2378 "expected boot_smoke to block; got: {body}"
2379 );
2380 }
2381
2382 #[tokio::test]
2383 async fn migration_bearing_promote_requires_fresh_confirm() {
2384 // The predecessor (host) configures NO gates, so nothing would normally
2385 // block host -> a. A `bears_migration` promote must still be blocked on a
2386 // fresh `manual_confirm` it does not have: rollback restores the binary
2387 // only, so the one-way advance needs a conscious operator sign-off.
2388 let pool = fresh_pool().await;
2389 for (i, name) in ["host", "a"].iter().enumerate() {
2390 sqlx::query(
2391 "INSERT INTO tiers (name, ord, provisioned, canary) VALUES (?, ?, 1, 'sequential')",
2392 )
2393 .bind(name)
2394 .bind(i as i64)
2395 .execute(&pool)
2396 .await
2397 .unwrap();
2398 sqlx::query("INSERT INTO tier_state (tier) VALUES (?)")
2399 .bind(name)
2400 .execute(&pool)
2401 .await
2402 .unwrap();
2403 }
2404 let mut topo = test_topo();
2405 topo.tiers[0].gates = vec![]; // host configures no gates at all
2406 let executors = Arc::new(crate::state::build_executors(&topo));
2407 let topo = Arc::new(topo);
2408 let cfg = Arc::new(test_cfg());
2409 let (apps, default_app) =
2410 crate::state::one_app(cfg.clone(), topo.clone(), executors.clone());
2411 let state = AppState {
2412 pool,
2413 apps,
2414 default_app,
2415 topo,
2416 cfg,
2417 active_build: Arc::new(tokio::sync::Mutex::new(None)),
2418 deploy_lock: Arc::new(tokio::sync::Mutex::new(())),
2419 events: crate::events::channel(),
2420 executors,
2421 api_token: None,
2422 };
2423 sqlx::query("INSERT INTO versions (version, git_sha, built_at, artifact_path) VALUES ('0.8.12','sha',datetime('now'),'/tmp/x')")
2424 .execute(&state.pool).await.unwrap();
2425 sqlx::query("UPDATE tier_state SET current_version = '0.8.12' WHERE tier = 'host'")
2426 .execute(&state.pool)
2427 .await
2428 .unwrap();
2429
2430 let app = router(state);
2431 let resp = app
2432 .oneshot(
2433 Request::builder()
2434 .method("POST")
2435 .uri("/promote/a")
2436 .header("content-type", "application/json")
2437 .body(Body::from(r#"{"bears_migration": true}"#))
2438 .unwrap(),
2439 )
2440 .await
2441 .unwrap();
2442 assert_eq!(resp.status(), StatusCode::CONFLICT);
2443 let body = body_string(resp).await;
2444 assert!(
2445 body.contains("manual_confirm"),
2446 "expected the migration promote to block on manual_confirm; got: {body}"
2447 );
2448 }
2449
2450 #[tokio::test]
2451 async fn tier_state_advance_is_atomic_previous_from_old_current() {
2452 // CF3: the promote advance is a single UPDATE where previous_version is
2453 // set from the row's *old* current_version (SQLite evaluates RHS against
2454 // the original row). No read-modify-write to lose under concurrency.
2455 let pool = fresh_pool().await;
2456 seed(&pool, "a", "1.0.0").await;
2457 // current_version FKs into versions, so the target must exist too.
2458 sqlx::query("INSERT INTO versions (version, git_sha, built_at, artifact_path) VALUES ('2.0.0','sha',datetime('now'),'/tmp/x')")
2459 .execute(&pool).await.unwrap();
2460 sqlx::query("UPDATE tier_state SET current_version = '1.0.0' WHERE tier = 'a'")
2461 .execute(&pool)
2462 .await
2463 .unwrap();
2464
2465 // Exercise the sealed forward-advance primitive itself — the same op
2466 // /promote and the host build path both call (S1), not a copy of its SQL.
2467 let v = crate::domain::Version::parse("2.0.0").unwrap();
2468 crate::runs::advance_tier(&pool, &crate::domain::AppId::default(), "a", &v, None)
2469 .await
2470 .unwrap();
2471
2472 let (cur, prev): (Option<String>, Option<String>) = sqlx::query_as(
2473 "SELECT current_version, previous_version FROM tier_state WHERE tier = 'a'",
2474 )
2475 .fetch_one(&pool)
2476 .await
2477 .unwrap();
2478 assert_eq!(cur.as_deref(), Some("2.0.0"));
2479 assert_eq!(
2480 prev.as_deref(),
2481 Some("1.0.0"),
2482 "previous = the pre-update current, atomically"
2483 );
2484 }
2485
2486 #[tokio::test]
2487 async fn canary_rollback_restores_deployed_nodes_to_previous_version() {
2488 use crate::topology::{Node, default_actuate, default_observe};
2489 let tmp = tempfile::tempdir().unwrap();
2490
2491 // Two local nodes, each pre-seeded as if a promote had flipped them to
2492 // 2.0.0 (current -> releases/2.0.0), with the prior 1.0.0 still on disk.
2493 let mut nodes = Vec::new();
2494 for name in ["n1", "n2"] {
2495 let rr = tmp.path().join(name);
2496 for v in ["1.0.0", "2.0.0"] {
2497 tokio::fs::create_dir_all(rr.join("releases").join(v))
2498 .await
2499 .unwrap();
2500 }
2501 tokio::fs::symlink("releases/2.0.0", rr.join("current"))
2502 .await
2503 .unwrap();
2504 nodes.push(Node {
2505 platform: None,
2506 base_image: None,
2507 libc: None,
2508 name: name.into(),
2509 ssh_target: "local".into(),
2510 release_root: rr.to_string_lossy().into_owned(),
2511 service_name: "x.service".into(),
2512 health_url: None,
2513 config_check_env_file: None,
2514 actuate: default_actuate(),
2515 observe: default_observe(),
2516 companions: Vec::new(),
2517 });
2518 }
2519
2520 let mut state = test_state().await;
2521 // The rollback target needs a versions row. The release dir name comes
2522 // from the artifact_path's parent (legacy layout: releases/<version>).
2523 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')")
2524 .execute(&state.pool).await.unwrap();
2525 let execs: crate::state::ExecutorMap = nodes
2526 .iter()
2527 .map(|n| (n.name.clone(), crate::state::build_executor(n)))
2528 .collect();
2529 state.executors = std::sync::Arc::new(execs);
2530
2531 let refs: Vec<&Node> = nodes.iter().collect();
2532 let report = rollback_deployed_nodes(&state, &tid("a"), &refs, "1.0.0").await;
2533 assert_eq!(report.restored, 2, "both deployed nodes should be restored");
2534 assert!(
2535 report.is_consistent(),
2536 "nothing should be indeterminate: {report:?}"
2537 );
2538
2539 for n in &nodes {
2540 let cur = tokio::fs::read_link(std::path::Path::new(&n.release_root).join("current"))
2541 .await
2542 .unwrap();
2543 assert_eq!(
2544 cur.to_string_lossy(),
2545 "releases/1.0.0",
2546 "node {} rolled back",
2547 n.name
2548 );
2549 }
2550 }
2551
2552 #[tokio::test]
2553 async fn canary_rollback_is_noop_without_a_previous_artifact() {
2554 use crate::topology::{Node, default_actuate, default_observe};
2555 let tmp = tempfile::tempdir().unwrap();
2556 let rr = tmp.path().join("n1");
2557 tokio::fs::create_dir_all(&rr).await.unwrap();
2558 let node = Node {
2559 platform: None,
2560 base_image: None,
2561 libc: None,
2562 name: "n1".into(),
2563 ssh_target: "local".into(),
2564 release_root: rr.to_string_lossy().into_owned(),
2565 service_name: "x.service".into(),
2566 health_url: None,
2567 config_check_env_file: None,
2568 actuate: default_actuate(),
2569 observe: default_observe(),
2570 companions: Vec::new(),
2571 };
2572 let state = test_state().await; // no versions row for "9.9.9"
2573 let report = rollback_deployed_nodes(&state, &tid("a"), &[&node], "9.9.9").await;
2574 assert_eq!(
2575 report.restored, 0,
2576 "no artifact to roll back to -> nothing restored, no panic"
2577 );
2578 assert_eq!(
2579 report.touched(),
2580 1,
2581 "the node must be accounted for somewhere"
2582 );
2583 // No rollback could even be attempted, so the node's version is not
2584 // knowable here. That must read as indeterminate, not as safe.
2585 assert_eq!(report.indeterminate, 1);
2586 assert!(!report.is_consistent());
2587 }
2588
2589 /// The 2026-08-01 prod incident, as a unit test on the reporting shape.
2590 ///
2591 /// A rollback that fails before the symlink swap leaves the node on the
2592 /// version it was already running — the one being rolled back to. The old
2593 /// code counted that as "not restored" and reported it as `restored=0 of=1`
2594 /// plus "it remains on the new version — manual intervention needed", which
2595 /// sent an operator to inspect a production box that was entirely fine.
2596 #[test]
2597 fn a_rollback_that_failed_before_the_swap_is_not_an_incident() {
2598 let report = RollbackReport {
2599 restored: 0,
2600 already_on_previous: 1,
2601 indeterminate: 0,
2602 };
2603 assert_eq!(report.touched(), 1);
2604 assert!(
2605 report.is_consistent(),
2606 "a node that never left the previous version is not split-brain"
2607 );
2608
2609 // Contrast: the same zero restored, but the swap had run. This one does
2610 // warrant a human, and the two must not report the same way.
2611 let real = RollbackReport {
2612 restored: 0,
2613 already_on_previous: 0,
2614 indeterminate: 1,
2615 };
2616 assert_eq!(real.touched(), 1);
2617 assert!(!real.is_consistent());
2618 }
2619
2620 /// A genuine split-brain still reports as one: some nodes back on the old
2621 /// version, one stranded.
2622 #[test]
2623 fn a_mixed_outcome_is_inconsistent_if_any_node_is_unknown() {
2624 let report = RollbackReport {
2625 restored: 2,
2626 already_on_previous: 1,
2627 indeterminate: 1,
2628 };
2629 assert_eq!(report.touched(), 4);
2630 assert!(!report.is_consistent());
2631 }
2632
2633 // ---- FleetFake: a multi-node promote across recorded fake executors ----
2634 //
2635 // The route-level promote tests until now ran a single local node against a
2636 // real LocalExec, so the sequential-canary fan-out, the cross-node deploy
2637 // ordering, and the mid-canary rollback of already-flipped nodes had no
2638 // coverage. FleetFake records every deploy op (tagged by node) into one
2639 // shared log so ordering is visible across the fleet, and can fail any op
2640 // whose shell script or rsync target contains a marker — used to fail a
2641 // node's forward deploy of the new version while its rollback to the prior
2642 // version (a different `releases/<v>` path) still succeeds.
2643
2644 struct FleetFake {
2645 tag: String,
2646 caps: CapabilitySet,
2647 log: Arc<StdMutex<Vec<String>>>,
2648 fail_if_contains: Option<String>,
2649 }
2650
2651 impl FleetFake {
2652 fn fails(&self, text: &str) -> bool {
2653 self.fail_if_contains
2654 .as_deref()
2655 .is_some_and(|m| text.contains(m))
2656 }
2657 }
2658
2659 #[async_trait]
2660 impl Executor for FleetFake {
2661 async fn run_streaming(
2662 &self,
2663 step: &Step,
2664 _sink: &mut dyn LogSink,
2665 ) -> anyhow::Result<RunOutput> {
2666 let script = step.argv.last().cloned().unwrap_or_default();
2667 self.log
2668 .lock()
2669 .unwrap()
2670 .push(format!("{}:run:{script}", self.tag));
2671 Ok(RunOutput {
2672 status: std::process::ExitStatus::from_raw(if self.fails(&script) {
2673 1 << 8
2674 } else {
2675 0
2676 }),
2677 stdout: Vec::new(),
2678 stderr: Vec::new(),
2679 })
2680 }
2681 async fn pull_file(
2682 &self,
2683 _r: &std::path::Path,
2684 _l: &std::path::Path,
2685 _o: &SyncOpts,
2686 ) -> anyhow::Result<()> {
2687 Ok(())
2688 }
2689 async fn pull_dir(
2690 &self,
2691 _r: &std::path::Path,
2692 _l: &std::path::Path,
2693 _o: &SyncOpts,
2694 ) -> anyhow::Result<()> {
2695 Ok(())
2696 }
2697 async fn pull_glob(
2698 &self,
2699 _g: &str,
2700 _l: &std::path::Path,
2701 _o: &SyncOpts,
2702 ) -> anyhow::Result<()> {
2703 Ok(())
2704 }
2705 async fn push_dir(
2706 &self,
2707 _local: &std::path::Path,
2708 remote: &std::path::Path,
2709 _o: &SyncOpts,
2710 ) -> anyhow::Result<()> {
2711 let dst = remote.display().to_string();
2712 self.log
2713 .lock()
2714 .unwrap()
2715 .push(format!("{}:push:{dst}", self.tag));
2716 if self.fails(&dst) {
2717 anyhow::bail!("fake rsync failure on {}", self.tag);
2718 }
2719 Ok(())
2720 }
2721 fn capabilities(&self) -> &CapabilitySet {
2722 &self.caps
2723 }
2724 }
2725
2726 /// Rebuild tier "a" with `names` as remote fake nodes sharing one op log,
2727 /// seed the version/tier_state prerequisites for a promote of 3.0.0 up from
2728 /// `host` (tier "a" starts on 2.0.0 with 1.0.0 behind it), and optionally
2729 /// make `fail_node` fail any op containing `fail_marker`. Returns the state
2730 /// and the shared log.
2731 async fn fleet_fixture(
2732 names: &[&str],
2733 fail_node: Option<&str>,
2734 fail_marker: &str,
2735 ) -> (AppState, Arc<StdMutex<Vec<String>>>) {
2736 use crate::topology::{default_actuate, default_observe};
2737 let mut state = test_state().await;
2738 let log = Arc::new(StdMutex::new(Vec::<String>::new()));
2739
2740 let nodes: Vec<Node> = names
2741 .iter()
2742 .map(|name| Node {
2743 platform: None,
2744 base_image: None,
2745 libc: None,
2746 name: (*name).into(),
2747 ssh_target: format!("deploy@{name}"),
2748 release_root: format!("/tmp/fleet/{name}"),
2749 service_name: "makenotwork.service".into(),
2750 health_url: None,
2751 config_check_env_file: None,
2752 actuate: default_actuate(),
2753 observe: default_observe(),
2754 companions: Vec::new(),
2755 })
2756 .collect();
2757
2758 let mut topo = (*state.topo).clone();
2759 topo.tiers[1].nodes = nodes.clone();
2760 // Drop the tier's post-deploy gate: the deploy fan-out is the subject
2761 // here, and node_health would need its own probe wiring.
2762 topo.tiers[1].gates = vec![];
2763 state.topo = Arc::new(topo);
2764
2765 let execs: crate::state::ExecutorMap = nodes
2766 .iter()
2767 .map(|n| {
2768 let nm = n.name.to_string();
2769 let fail = fail_node == Some(nm.as_str());
2770 let exec: Arc<dyn Executor> = Arc::new(FleetFake {
2771 tag: nm,
2772 caps: CapabilitySet::from_tokens(["deploy", "restart"], ["health"]),
2773 log: log.clone(),
2774 fail_if_contains: fail.then(|| fail_marker.to_string()),
2775 });
2776 (n.name.clone(), exec)
2777 })
2778 .collect();
2779 state.executors = Arc::new(execs);
2780
2781 for n in &nodes {
2782 // deploys.node FKs into `nodes`.
2783 sqlx::query(
2784 "INSERT INTO nodes (name, tier, ssh_target, release_root) VALUES (?, 'a', ?, ?)",
2785 )
2786 .bind(&n.name)
2787 .bind(&n.ssh_target)
2788 .bind(&n.release_root)
2789 .execute(&state.pool)
2790 .await
2791 .unwrap();
2792 }
2793 for v in ["1.0.0", "2.0.0", "3.0.0"] {
2794 // Legacy (pre-identity) artifact_path is `releases/<version>/<bin>`,
2795 // so the release dir the node mirrors is named for the version. The
2796 // fixture reflects that real layout (parent basename == version).
2797 sqlx::query("INSERT INTO versions (version, git_sha, built_at, artifact_path) VALUES (?, 'sha', datetime('now'), ?)")
2798 .bind(v).bind(format!("/tmp/staged/releases/{v}/makenotwork")).execute(&state.pool).await.unwrap();
2799 }
2800 sqlx::query("UPDATE tier_state SET current_version = '3.0.0' WHERE tier = 'host'")
2801 .execute(&state.pool)
2802 .await
2803 .unwrap();
2804 sqlx::query(
2805 "UPDATE tier_state SET current_version = '2.0.0', previous_version = '1.0.0' WHERE tier = 'a'",
2806 )
2807 .execute(&state.pool)
2808 .await
2809 .unwrap();
2810 (state, log)
2811 }
2812
2813 /// Index of the first logged op belonging to `node` (panics if the node was
2814 /// never touched — the message names it).
2815 fn first_touch(log: &[String], node: &str) -> usize {
2816 let prefix = format!("{node}:");
2817 log.iter()
2818 .position(|e| e.starts_with(&prefix))
2819 .unwrap_or_else(|| panic!("node {node:?} was never deployed to; log: {log:#?}"))
2820 }
2821
2822 #[tokio::test]
2823 async fn promote_deploys_every_node_in_tier_order_and_advances() {
2824 let (state, log) = fleet_fixture(&["a1", "a2", "a3"], None, "").await;
2825 let pool = state.pool.clone();
2826
2827 let Json(body) = promote_inner(state, "a".into(), PromoteBody::default())
2828 .await
2829 .expect("all nodes deploy, so the promote succeeds");
2830 assert_eq!(
2831 body["nodes_deployed"],
2832 serde_json::json!(["a1", "a2", "a3"]),
2833 "the response names every node the promote reached",
2834 );
2835
2836 let (cur, prev) = tier_versions(&pool, "a").await;
2837 assert_eq!(cur.as_deref(), Some("3.0.0"));
2838 assert_eq!(prev.as_deref(), Some("2.0.0"));
2839
2840 // Sequential canary: a1 is fully touched before a2, a2 before a3.
2841 let log = log.lock().unwrap().clone();
2842 assert!(
2843 first_touch(&log, "a1") < first_touch(&log, "a2")
2844 && first_touch(&log, "a2") < first_touch(&log, "a3"),
2845 "nodes must deploy in tier order: {log:#?}",
2846 );
2847
2848 // Every node has a green deploy row for the promoted version.
2849 let ok: i64 = sqlx::query_scalar(
2850 "SELECT COUNT(*) FROM deploys WHERE version = '3.0.0' AND outcome = 'ok'",
2851 )
2852 .fetch_one(&pool)
2853 .await
2854 .unwrap();
2855 assert_eq!(ok, 3, "one ok deploy row per node");
2856 }
2857
2858 #[tokio::test]
2859 async fn a_mid_canary_deploy_failure_rolls_touched_nodes_back_and_does_not_advance() {
2860 // a2 fails its forward deploy of 3.0.0; a1 was already flipped, a3 is
2861 // never reached. The touched nodes (a1, a2) roll back to 2.0.0 — their
2862 // rollback ops target `releases/2.0.0`, which the marker does not match —
2863 // and tier_state must NOT advance.
2864 let (state, log) = fleet_fixture(&["a1", "a2", "a3"], Some("a2"), "releases/3.0.0").await;
2865 let pool = state.pool.clone();
2866
2867 let err = promote_inner(state, "a".into(), PromoteBody::default())
2868 .await
2869 .expect_err("a mid-canary deploy failure must fail the promote");
2870 assert!(
2871 matches!(err, crate::error::Error::Other(_)),
2872 "a deploy failure propagates as Other, got: {err:?}",
2873 );
2874
2875 // tier_state untouched: the failure returns before advance_tier.
2876 let (cur, prev) = tier_versions(&pool, "a").await;
2877 assert_eq!(
2878 cur.as_deref(),
2879 Some("2.0.0"),
2880 "a failed rollout must not advance"
2881 );
2882 assert_eq!(prev.as_deref(), Some("1.0.0"));
2883
2884 let log = log.lock().unwrap().clone();
2885 // a3 sits after the failed a2 in the sequence and is never touched.
2886 assert!(
2887 !log.iter().any(|e| e.starts_with("a3:")),
2888 "nodes after the failure must not be deployed to: {log:#?}",
2889 );
2890 // Both touched nodes were rolled back to the prior version.
2891 for n in ["a1", "a2"] {
2892 assert!(
2893 log.iter()
2894 .any(|e| e.starts_with(&format!("{n}:")) && e.contains("releases/2.0.0")),
2895 "touched node {n} must be restored to 2.0.0: {log:#?}",
2896 );
2897 }
2898
2899 // The forward attempt is on the record: a1 ok, a2 failed.
2900 let a1: String = sqlx::query_scalar(
2901 "SELECT outcome FROM deploys WHERE version = '3.0.0' AND node = 'a1'",
2902 )
2903 .fetch_one(&pool)
2904 .await
2905 .unwrap();
2906 assert_eq!(a1, "ok");
2907 let a2: String = sqlx::query_scalar(
2908 "SELECT outcome FROM deploys WHERE version = '3.0.0' AND node = 'a2'",
2909 )
2910 .fetch_one(&pool)
2911 .await
2912 .unwrap();
2913 assert_eq!(a2, "failed");
2914
2915 // Both touched nodes restored => the tier is consistent on 2.0.0, so the
2916 // partial flag is cleared, not set.
2917 let reason: Option<String> =
2918 sqlx::query_scalar("SELECT partial_reason FROM tier_state WHERE tier = 'a'")
2919 .fetch_one(&pool)
2920 .await
2921 .unwrap();
2922 assert_eq!(
2923 reason, None,
2924 "a fully-restored canary leaves the tier consistent, not partial",
2925 );
2926 }
2927
2928 /// Build a one-node tier "a" on a tempdir release root, pre-seeded as if a
2929 /// promote had flipped it to `current` with `prev` still staged on disk.
2930 /// Returns the state (topology rewired to the tempdir node) and the tempdir,
2931 /// which the caller must keep alive.
2932 async fn rollback_fixture(prev: &str, current: &str) -> (AppState, tempfile::TempDir) {
2933 use crate::topology::{Node, default_actuate, default_observe};
2934 let tmp = tempfile::tempdir().unwrap();
2935 let rr = tmp.path().join("a-local");
2936 for v in [prev, current] {
2937 tokio::fs::create_dir_all(rr.join("releases").join(v))
2938 .await
2939 .unwrap();
2940 }
2941 tokio::fs::symlink(format!("releases/{current}"), rr.join("current"))
2942 .await
2943 .unwrap();
2944 let node = Node {
2945 platform: None,
2946 base_image: None,
2947 libc: None,
2948 name: "a-local".into(),
2949 ssh_target: "local".into(),
2950 release_root: rr.to_string_lossy().into_owned(),
2951 service_name: "x.service".into(),
2952 health_url: None,
2953 config_check_env_file: None,
2954 actuate: default_actuate(),
2955 observe: default_observe(),
2956 companions: Vec::new(),
2957 };
2958
2959 let mut state = test_state().await;
2960 let mut topo = (*state.topo).clone();
2961 topo.tiers[1].nodes = vec![node];
2962 state.executors = Arc::new(crate::state::build_executors(&topo));
2963 state.topo = Arc::new(topo);
2964
2965 // deploys.node FKs into `nodes`, so the promote path needs the row.
2966 sqlx::query("INSERT INTO nodes (name, tier, ssh_target, release_root) VALUES ('a-local', 'a', 'local', ?)")
2967 .bind(rr.to_string_lossy().into_owned())
2968 .execute(&state.pool).await.unwrap();
2969 for v in [prev, current] {
2970 sqlx::query("INSERT INTO versions (version, git_sha, built_at, artifact_path) VALUES (?, 'sha', datetime('now'), ?)")
2971 .bind(v).bind(format!("/tmp/staged/releases/{v}/makenotwork")).execute(&state.pool).await.unwrap();
2972 }
2973 sqlx::query(
2974 "UPDATE tier_state SET current_version = ?, previous_version = ? WHERE tier = 'a'",
2975 )
2976 .bind(current)
2977 .bind(prev)
2978 .execute(&state.pool)
2979 .await
2980 .unwrap();
2981 (state, tmp)
2982 }
2983
2984 async fn tier_versions(pool: &SqlitePool, tier: &str) -> (Option<String>, Option<String>) {
2985 sqlx::query_as("SELECT current_version, previous_version FROM tier_state WHERE tier = ?")
2986 .bind(tier)
2987 .fetch_one(pool)
2988 .await
2989 .unwrap()
2990 }
2991
2992 #[tokio::test]
2993 async fn rollback_clears_previous_version_rather_than_swapping_it() {
2994 // The swap bug: writing the version we just rolled OFF into
2995 // previous_version made a second /rollback roll FORWARD onto the broken
2996 // build the operator was escaping. previous_version must go NULL.
2997 let (state, _tmp) = rollback_fixture("1.0.0", "2.0.0").await;
2998 let pool = state.pool.clone();
2999
3000 let _ = rollback(State(state), Path("a".to_string())).await.unwrap();
3001
3002 let (cur, prev) = tier_versions(&pool, "a").await;
3003 assert_eq!(
3004 cur.as_deref(),
3005 Some("1.0.0"),
3006 "rolled back to the previous version"
3007 );
3008 assert_eq!(
3009 prev, None,
3010 "the version we rolled off must NOT become the rollback target"
3011 );
3012 }
3013
3014 #[tokio::test]
3015 async fn second_rollback_refuses_instead_of_rolling_forward() {
3016 let (state, _tmp) = rollback_fixture("1.0.0", "2.0.0").await;
3017 let pool = state.pool.clone();
3018 let release_root = state.topo.tiers[1].nodes[0].release_root.clone();
3019
3020 let _ = rollback(State(state.clone()), Path("a".to_string()))
3021 .await
3022 .unwrap();
3023 let err = rollback(State(state), Path("a".to_string()))
3024 .await
3025 .expect_err("only one step of history is tracked; a second rollback must refuse");
3026 assert!(
3027 matches!(&err, crate::error::Error::GateBlocked(m) if m.contains("no previous_version")),
3028 "expected a loud refusal, got: {err:?}",
3029 );
3030
3031 // The refusal is total: neither the DB nor the node moved back to 2.0.0.
3032 let (cur, _) = tier_versions(&pool, "a").await;
3033 assert_eq!(cur.as_deref(), Some("1.0.0"));
3034 let link = tokio::fs::read_link(std::path::Path::new(&release_root).join("current"))
3035 .await
3036 .unwrap();
3037 assert_eq!(
3038 link.to_string_lossy(),
3039 "releases/1.0.0",
3040 "node stays on the rolled-back version"
3041 );
3042 }
3043
3044 #[tokio::test]
3045 async fn promote_refuses_an_unprovisioned_tier() {
3046 // Every step of a promote to a node-less tier is a silent no-op that
3047 // still reports success: the deploy loop iterates nothing and
3048 // advance_tier records a current_version the tier never received.
3049 let (mut state, _tmp) = rollback_fixture("1.0.0", "2.0.0").await;
3050 let mut topo = (*state.topo).clone();
3051 topo.tiers[1].provisioned = false;
3052 topo.tiers[1].nodes.clear();
3053 state.topo = Arc::new(topo);
3054 let pool = state.pool.clone();
3055 sqlx::query("UPDATE tier_state SET current_version = '2.0.0' WHERE tier = 'host'")
3056 .execute(&pool)
3057 .await
3058 .unwrap();
3059
3060 let err = promote_inner(state, "a".into(), PromoteBody::default())
3061 .await
3062 .expect_err("promoting to an unprovisioned tier must be refused");
3063 assert!(
3064 matches!(&err, crate::error::Error::GateBlocked(m) if m.contains("not provisioned")),
3065 "got: {err:?}",
3066 );
3067
3068 // And nothing was recorded: the tier keeps whatever it had before.
3069 let (cur, _) = tier_versions(&pool, "a").await;
3070 assert_eq!(
3071 cur.as_deref(),
3072 Some("2.0.0"),
3073 "a refused promote must not advance tier_state",
3074 );
3075 }
3076
3077 #[tokio::test]
3078 async fn promote_advances_the_tier_and_flips_the_symlink_when_gates_are_green() {
3079 // The happy path. Every other promote test asserts a refusal or a red
3080 // outcome, so nothing pinned what a *successful* promote actually does:
3081 // deploy reaches the node, the `current` symlink flips, tier_state
3082 // advances with previous_version = the version we came off, any stale
3083 // partial flag clears, and the handler reports the nodes it touched.
3084 use crate::topology::Gate;
3085 let (mut state, _tmp) = rollback_fixture("1.0.0", "2.0.0").await;
3086
3087 // Source tier `host` gates the promote on cargo_test. Satisfying it is
3088 // the point of the test: the sibling case asserts an unsatisfied gate
3089 // blocks, this one asserts a satisfied gate lets the promote through.
3090 let mut topo = (*state.topo).clone();
3091 topo.tiers[0].gates = vec![Gate::CargoTest];
3092 state.topo = Arc::new(topo);
3093 let pool = state.pool.clone();
3094 let release_root = state.topo.tiers[1].nodes[0].release_root.clone();
3095
3096 // 3.0.0 is staged on the build host and green on `host`.
3097 tokio::fs::create_dir_all(
3098 std::path::Path::new(&release_root)
3099 .join("releases")
3100 .join("3.0.0"),
3101 )
3102 .await
3103 .unwrap();
3104 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')")
3105 .execute(&pool).await.unwrap();
3106 sqlx::query("UPDATE tier_state SET current_version = '3.0.0' WHERE tier = 'host'")
3107 .execute(&pool)
3108 .await
3109 .unwrap();
3110 insert_gate(&pool, "host", "3.0.0", "cargo_test", 1).await;
3111
3112 // A stale flag from an earlier incident, which a clean rollout clears.
3113 set_partial(&state, &tid("a"), "left over from a previous canary").await;
3114
3115 let Json(body) = promote_inner(state, "a".into(), PromoteBody::default())
3116 .await
3117 .expect("gates are green and the node deploys, so the promote succeeds");
3118
3119 assert_eq!(body["tier"], "a");
3120 assert_eq!(body["version"], "3.0.0");
3121 assert_eq!(
3122 body["nodes_deployed"],
3123 serde_json::json!(["a-local"]),
3124 "the response names every node the promote reached",
3125 );
3126
3127 let (cur, prev) = tier_versions(&pool, "a").await;
3128 assert_eq!(cur.as_deref(), Some("3.0.0"));
3129 assert_eq!(
3130 prev.as_deref(),
3131 Some("2.0.0"),
3132 "previous_version is the version we came off, so a rollback aims at it",
3133 );
3134
3135 // The node genuinely moved: the promote is not just bookkeeping.
3136 let link = tokio::fs::read_link(std::path::Path::new(&release_root).join("current"))
3137 .await
3138 .unwrap();
3139 assert_eq!(link.to_string_lossy(), "releases/3.0.0");
3140
3141 let reason: Option<String> =
3142 sqlx::query_scalar("SELECT partial_reason FROM tier_state WHERE tier = 'a'")
3143 .fetch_one(&pool)
3144 .await
3145 .unwrap();
3146 assert_eq!(
3147 reason, None,
3148 "a clean full rollout clears a stale partial flag",
3149 );
3150
3151 // The deploy is on the record as having succeeded, which is what the
3152 // next promote's gate check and /state both read.
3153 let (node, outcome): (String, String) =
3154 sqlx::query_as("SELECT node, outcome FROM deploys WHERE version = '3.0.0'")
3155 .fetch_one(&pool)
3156 .await
3157 .unwrap();
3158 assert_eq!(node, "a-local");
3159 assert_eq!(outcome, "ok");
3160 }
3161
3162 #[tokio::test]
3163 async fn promote_fails_and_flags_the_tier_when_post_deploy_gates_are_red() {
3164 // The deploy reached every node, so tier_state advances (a stale
3165 // current_version would aim a later rollback at the wrong artifact), but
3166 // the promote must NOT report success: the tier is flagged partial and
3167 // the handler returns the gate failure.
3168 use crate::topology::Gate;
3169 let (mut state, _tmp) = rollback_fixture("1.0.0", "2.0.0").await;
3170 // node_health is the tier's only gate, and it fails closed with no
3171 // probes — an empty executor map gives it nothing to probe while
3172 // deploy_node still falls back to a built executor and succeeds.
3173 let mut topo = (*state.topo).clone();
3174 topo.tiers[1].gates = vec![Gate::NodeHealth];
3175 state.topo = Arc::new(topo);
3176 state.executors = Arc::new(crate::state::ExecutorMap::new());
3177 let pool = state.pool.clone();
3178 // Promote 3.0.0 up from host, which configures no gates of its own.
3179 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')")
3180 .execute(&pool).await.unwrap();
3181 sqlx::query("UPDATE tier_state SET current_version = '3.0.0' WHERE tier = 'host'")
3182 .execute(&pool)
3183 .await
3184 .unwrap();
3185
3186 let err = promote_inner(state, "a".into(), PromoteBody::default())
3187 .await
3188 .expect_err("red post-deploy gates must fail the promote");
3189 assert!(
3190 matches!(&err, crate::error::Error::GateBlocked(m) if m.contains("node_health")),
3191 "the failure must name the red gate, got: {err:?}",
3192 );
3193
3194 let (cur, _) = tier_versions(&pool, "a").await;
3195 assert_eq!(
3196 cur.as_deref(),
3197 Some("3.0.0"),
3198 "tier_state tracks what the nodes actually run"
3199 );
3200 let reason: Option<String> =
3201 sqlx::query_scalar("SELECT partial_reason FROM tier_state WHERE tier = 'a'")
3202 .fetch_one(&pool)
3203 .await
3204 .unwrap();
3205 assert!(
3206 reason.as_deref().is_some_and(|r| r.contains("node_health")),
3207 "the tier must be flagged for /state and the TUI, got: {reason:?}",
3208 );
3209 }
3210
3211 #[tokio::test]
3212 async fn set_partial_then_clear_roundtrips() {
3213 let state = test_state().await;
3214 let read = || async {
3215 sqlx::query_scalar::<_, Option<String>>(
3216 "SELECT partial_reason FROM tier_state WHERE tier = 'a'",
3217 )
3218 .fetch_one(&state.pool)
3219 .await
3220 .unwrap()
3221 };
3222 assert_eq!(read().await, None, "consistent tier starts clean");
3223 set_partial(&state, &tid("a"), "canary rollback incomplete: 1/2").await;
3224 assert_eq!(
3225 read().await.as_deref(),
3226 Some("canary rollback incomplete: 1/2")
3227 );
3228 clear_partial(&state, &tid("a")).await;
3229 assert_eq!(read().await, None, "clear nulls it back out");
3230 }
3231
3232 #[tokio::test]
3233 async fn state_surfaces_partial_reason() {
3234 use axum::extract::State;
3235 let state = test_state().await;
3236 set_partial(
3237 &state,
3238 &tid("a"),
3239 "first-deploy canary failed: 1 node(s) on 2.0.0",
3240 )
3241 .await;
3242 let Json(view) = get_state(State(state)).await.unwrap();
3243 let a = view.tiers.iter().find(|t| t.name == "a").unwrap();
3244 assert_eq!(
3245 a.partial_reason.as_deref(),
3246 Some("first-deploy canary failed: 1 node(s) on 2.0.0"),
3247 );
3248 let host = view.tiers.iter().find(|t| t.name == "host").unwrap();
3249 assert_eq!(
3250 host.partial_reason, None,
3251 "untouched tier stays clean in /state"
3252 );
3253 }
3254
3255 #[tokio::test]
3256 async fn state_build_is_null_until_first_rebuild_then_surfaces_latest() {
3257 use axum::extract::State;
3258 let state = test_state().await;
3259 // No build runs yet → build is null, so /state doesn't pretend a build
3260 // is happening.
3261 let Json(view) = get_state(State(state.clone())).await.unwrap();
3262 assert!(view.build.is_none());
3263
3264 // A failed run must surface its cause in /state, not just in /runs.
3265 let run_id = crate::runs::create(&state.pool, &state.cfg.id, "deadbeef")
3266 .await
3267 .unwrap();
3268 crate::runs::mark_failed(&state.pool, run_id, "cargo_test: 3 test(s) failed")
3269 .await
3270 .unwrap();
3271 let Json(view) = get_state(State(state)).await.unwrap();
3272 let b = view.build.expect("build surfaced");
3273 assert_eq!(b.run_id, run_id.0);
3274 assert_eq!(b.result, "failed");
3275 assert_eq!(
3276 b.failure_summary.as_deref(),
3277 Some("cargo_test: 3 test(s) failed")
3278 );
3279 }
3280
3281 #[tokio::test]
3282 async fn status_json_serves_the_shared_payload_over_the_real_router() {
3283 // The mapping itself is tested in `crate::status`. This asserts the
3284 // route is wired, serves valid JSON, and stays internally consistent
3285 // (no dangling child or action references) against a real topology
3286 // rather than a hand-built fixture.
3287 let state = test_state().await;
3288 set_partial(&state, &tid("a"), "canary rollback incomplete: 1/2").await;
3289
3290 let resp = router(state)
3291 .oneshot(
3292 Request::builder()
3293 .uri("/status.json")
3294 .body(Body::empty())
3295 .unwrap(),
3296 )
3297 .await
3298 .unwrap();
3299 assert_eq!(resp.status(), StatusCode::OK);
3300
3301 let body = http_body_util::BodyExt::collect(resp.into_body())
3302 .await
3303 .unwrap()
3304 .to_bytes();
3305 let payload: ops_status::Payload = serde_json::from_slice(&body).unwrap();
3306
3307 assert_eq!(payload.source, crate::status::SOURCE);
3308 assert_eq!(payload.schema, ops_status::SCHEMA_VERSION);
3309 assert_eq!(payload.validate(), Ok(()));
3310 assert_eq!(
3311 payload.node("tier:a").unwrap().status,
3312 ops_status::Status::Failed,
3313 "a partial tier must surface as failed over the wire"
3314 );
3315 assert_eq!(payload.worst_status(), ops_status::Status::Failed);
3316 }
3317
3318 #[tokio::test]
3319 async fn promote_with_explicit_version_but_missing_artifact_404s() {
3320 // Explicit version supplied, gates trivially pass (mm has none in
3321 // test_topo), but `versions` table has no row → 404.
3322 let state = test_state().await;
3323 let app = router(state);
3324 let resp = app
3325 .oneshot(
3326 Request::builder()
3327 .method("POST")
3328 .uri("/promote/a")
3329 .header("content-type", "application/json")
3330 .body(Body::from(r#"{"version":"9.9.9"}"#))
3331 .unwrap(),
3332 )
3333 .await
3334 .unwrap();
3335 assert_eq!(resp.status(), StatusCode::NOT_FOUND);
3336 }
3337
3338 // ---- GET /logs/{version}/{gate} ----
3339
3340 async fn state_with_logs_root(logs_root: PathBuf) -> AppState {
3341 let mut s = test_state().await;
3342 let mut cfg = (*s.cfg).clone();
3343 cfg.logs_root = logs_root;
3344 s.cfg = Arc::new(cfg);
3345 s
3346 }
3347
3348 #[tokio::test]
3349 async fn get_gate_log_returns_file_contents() {
3350 let tmp = tempfile::tempdir().unwrap();
3351 let dir = tmp.path().join("0.9.5");
3352 tokio::fs::create_dir_all(&dir).await.unwrap();
3353 tokio::fs::write(dir.join("cargo_test.log"), b"hello sandod\n")
3354 .await
3355 .unwrap();
3356
3357 let state = state_with_logs_root(tmp.path().to_path_buf()).await;
3358 let app = router(state);
3359 let resp = app
3360 .oneshot(
3361 Request::builder()
3362 .uri("/logs/0.9.5/cargo_test")
3363 .body(Body::empty())
3364 .unwrap(),
3365 )
3366 .await
3367 .unwrap();
3368 assert_eq!(resp.status(), StatusCode::OK);
3369 assert_eq!(body_string(resp).await, "hello sandod\n");
3370 }
3371
3372 #[tokio::test]
3373 async fn get_gate_log_accepts_a_log_ref_verbatim() {
3374 // `log_ref` on a gate row is `<build_id>/<gate>.log`; appending it to
3375 // `/logs/` must work as-is, so following the ref is the path of least
3376 // effort. Guessing from the version is what returns another run's output.
3377 let tmp = tempfile::tempdir().unwrap();
3378 tokio::fs::create_dir_all(tmp.path().join("62"))
3379 .await
3380 .unwrap();
3381 tokio::fs::write(tmp.path().join("62/cargo_deny.log"), b"run 62 only")
3382 .await
3383 .unwrap();
3384 let state = state_with_logs_root(tmp.path().to_path_buf()).await;
3385 let app = router(state);
3386 let resp = app
3387 .oneshot(
3388 Request::builder()
3389 .uri("/logs/62/cargo_deny.log")
3390 .body(Body::empty())
3391 .unwrap(),
3392 )
3393 .await
3394 .unwrap();
3395 assert_eq!(resp.status(), StatusCode::OK);
3396 assert_eq!(body_string(resp).await, "run 62 only");
3397 }
3398
3399 #[tokio::test]
3400 async fn get_gate_log_404s_when_missing() {
3401 let tmp = tempfile::tempdir().unwrap();
3402 let state = state_with_logs_root(tmp.path().to_path_buf()).await;
3403 let app = router(state);
3404 let resp = app
3405 .oneshot(
3406 Request::builder()
3407 .uri("/logs/0.9.5/cargo_test")
3408 .body(Body::empty())
3409 .unwrap(),
3410 )
3411 .await
3412 .unwrap();
3413 assert_eq!(resp.status(), StatusCode::NOT_FOUND);
3414 }
3415
3416 // ---- CF2: bearer-token auth on deploy mutators ----
3417
3418 #[tokio::test]
3419 async fn mutating_route_requires_bearer_when_token_set() {
3420 let mut state = test_state().await;
3421 state.api_token = Some(std::sync::Arc::from("s3cr3t"));
3422 let app = router(state);
3423
3424 // No Authorization header -> 401, before any deploy logic runs.
3425 let resp = app
3426 .clone()
3427 .oneshot(
3428 Request::builder()
3429 .method("POST")
3430 .uri("/promote/a")
3431 .body(Body::empty())
3432 .unwrap(),
3433 )
3434 .await
3435 .unwrap();
3436 assert_eq!(resp.status(), StatusCode::UNAUTHORIZED);
3437
3438 // Wrong token -> 401.
3439 let resp = app
3440 .clone()
3441 .oneshot(
3442 Request::builder()
3443 .method("POST")
3444 .uri("/promote/a")
3445 .header("authorization", "Bearer nope")
3446 .body(Body::empty())
3447 .unwrap(),
3448 )
3449 .await
3450 .unwrap();
3451 assert_eq!(resp.status(), StatusCode::UNAUTHORIZED);
3452
3453 // Correct token -> passes auth (then blocked downstream by gates /
3454 // missing predecessor version, but specifically NOT 401).
3455 let resp = app
3456 .clone()
3457 .oneshot(
3458 Request::builder()
3459 .method("POST")
3460 .uri("/promote/a")
3461 .header("authorization", "Bearer s3cr3t")
3462 .body(Body::empty())
3463 .unwrap(),
3464 )
3465 .await
3466 .unwrap();
3467 assert_ne!(resp.status(), StatusCode::UNAUTHORIZED);
3468 }
3469
3470 #[tokio::test]
3471 async fn read_routes_require_token_when_set() {
3472 // Reads expose prod state (versions, SHAs, gate logs, the event stream),
3473 // so they are bearer-gated too — not just the mutators. A tailnet peer
3474 // without the token gets 401; the TUI presents the token and gets 200.
3475 let mut state = test_state().await;
3476 state.api_token = Some(std::sync::Arc::from("s3cr3t"));
3477 let app = router(state);
3478
3479 // No token -> 401 on a read.
3480 let resp = app
3481 .clone()
3482 .oneshot(
3483 Request::builder()
3484 .uri("/state")
3485 .body(Body::empty())
3486 .unwrap(),
3487 )
3488 .await
3489 .unwrap();
3490 assert_eq!(resp.status(), StatusCode::UNAUTHORIZED);
3491
3492 // Correct token -> 200.
3493 let resp = app
3494 .oneshot(
3495 Request::builder()
3496 .uri("/state")
3497 .header("authorization", "Bearer s3cr3t")
3498 .body(Body::empty())
3499 .unwrap(),
3500 )
3501 .await
3502 .unwrap();
3503 assert_eq!(resp.status(), StatusCode::OK);
3504 }
3505
3506 #[tokio::test]
3507 async fn read_routes_open_without_a_token() {
3508 // The loopback/dev posture: no token configured, reads pass through so a
3509 // local TUI needs no credential.
3510 let state = test_state().await;
3511 let app = router(state);
3512 let resp = app
3513 .oneshot(
3514 Request::builder()
3515 .uri("/state")
3516 .body(Body::empty())
3517 .unwrap(),
3518 )
3519 .await
3520 .unwrap();
3521 assert_eq!(resp.status(), StatusCode::OK);
3522 }
3523
3524 #[tokio::test]
3525 async fn self_update_malformed_body_is_400_not_422() {
3526 // TypedBody funnels a JSON deserialize failure through the Error envelope
3527 // (400), not axum's raw 422 — keeping every mutator on one error contract.
3528 let state = test_state().await; // no token -> auth passes, body is the gate
3529 let app = router(state);
3530 let resp = app
3531 .oneshot(
3532 Request::builder()
3533 .method("POST")
3534 .uri("/self-update")
3535 .header("content-type", "application/json")
3536 .body(Body::from("{ not valid json"))
3537 .unwrap(),
3538 )
3539 .await
3540 .unwrap();
3541 assert_eq!(resp.status(), StatusCode::BAD_REQUEST);
3542 }
3543
3544 #[tokio::test]
3545 async fn gate_log_rejects_unknown_gate_kind() {
3546 // The gate segment is an allowlisted GateKind, not a free-form filename:
3547 // an unknown kind is a 404, so `*.log` basenames can't be probed.
3548 let state = test_state().await;
3549 let app = router(state);
3550 let resp = app
3551 .oneshot(
3552 Request::builder()
3553 .uri("/logs/0.9.6/passwd")
3554 .body(Body::empty())
3555 .unwrap(),
3556 )
3557 .await
3558 .unwrap();
3559 assert_eq!(resp.status(), StatusCode::NOT_FOUND);
3560 }
3561
3562 /// Path-traversal guard: a `..` segment must not escape logs_root.
3563 /// axum's `{name}` param already rejects a literal `/` in the value, but
3564 /// `..` as a whole segment is structurally valid and must be blocked at
3565 /// the handler.
3566 #[tokio::test]
3567 async fn get_gate_log_rejects_dotdot_segments() {
3568 let tmp = tempfile::tempdir().unwrap();
3569 let state = state_with_logs_root(tmp.path().to_path_buf()).await;
3570 let app = router(state);
3571 let resp = app
3572 .oneshot(
3573 Request::builder()
3574 .uri("/logs/../etc/passwd")
3575 .body(Body::empty())
3576 .unwrap(),
3577 )
3578 .await
3579 .unwrap();
3580 assert_eq!(resp.status(), StatusCode::NOT_FOUND);
3581 }
3582
3583 // ---- build-identity path (wiki release-artifact-identity) ----
3584
3585 /// Insert a settled (passed) build_runs row with content identity and return
3586 /// its id. `staged_path` is the content-addressed release dir the bundle was
3587 /// published to (`releases/<digest16>`).
3588 async fn seed_build(pool: &SqlitePool, sha: &str, version: &str, staged_path: &str) -> i64 {
3589 sqlx::query_scalar(
3590 "INSERT INTO build_runs (sha, version, phase, result, started_at, bundle_digest, staged_path)
3591 VALUES (?, ?, 'done', 'passed', datetime('now'), ?, ?) RETURNING id",
3592 )
3593 .bind(sha)
3594 .bind(version)
3595 .bind(format!("{sha}-digest"))
3596 .bind(staged_path)
3597 .fetch_one(pool)
3598 .await
3599 .unwrap()
3600 }
3601
3602 async fn insert_gate_build(
3603 pool: &SqlitePool,
3604 tier: &str,
3605 version: &str,
3606 kind: &str,
3607 passed: i64,
3608 build_id: i64,
3609 ) {
3610 let status = if passed == 1 { "passed" } else { "failed" };
3611 sqlx::query(
3612 "INSERT INTO gate_runs (version, tier, gate_kind, started_at, finished_at, status, build_id) \
3613 VALUES (?, ?, ?, datetime('now'), datetime('now'), ?, ?)",
3614 )
3615 .bind(version)
3616 .bind(tier)
3617 .bind(kind)
3618 .bind(status)
3619 .bind(build_id)
3620 .execute(pool)
3621 .await
3622 .unwrap();
3623 }
3624
3625 #[tokio::test]
3626 async fn unsatisfied_gates_keys_build_evidence_on_build_id() {
3627 let pool = fresh_pool().await;
3628 seed(&pool, "a", "3.0.0").await;
3629 // Two builds of the SAME version string. Only b1's gate ran.
3630 let b1 = seed_build(&pool, "sha1", "3.0.0", "/rel/1111111111111111").await;
3631 let b2 = seed_build(&pool, "sha2", "3.0.0", "/rel/2222222222222222").await;
3632 insert_gate_build(&pool, "a", "3.0.0", "cargo_test", 1, b1).await;
3633
3634 // b1's own evidence satisfies.
3635 let ok = unsatisfied_gates(
3636 &pool,
3637 &crate::domain::AppId::default(),
3638 &tid("a"),
3639 &[Gate::CargoTest],
3640 &Evidence {
3641 version: "3.0.0",
3642 builds: &[PromotedBuild {
3643 platform: None,
3644 build_id: Some(b1),
3645 }],
3646 tier_build: None,
3647 },
3648 false,
3649 )
3650 .await
3651 .unwrap();
3652 assert!(ok.is_empty(), "the build that passed its gate is satisfied");
3653
3654 // b2 shares the version but has no gate row of its own: fail closed. This
3655 // is the hole — a rebuild reusing the version must not ride b1's evidence.
3656 let bad = unsatisfied_gates(
3657 &pool,
3658 &crate::domain::AppId::default(),
3659 &tid("a"),
3660 &[Gate::CargoTest],
3661 &Evidence {
3662 version: "3.0.0",
3663 builds: &[PromotedBuild {
3664 platform: None,
3665 build_id: Some(b2),
3666 }],
3667 tier_build: None,
3668 },
3669 false,
3670 )
3671 .await
3672 .unwrap();
3673 assert_eq!(
3674 bad,
3675 vec!["cargo_test".to_string()],
3676 "a different build of the same version does not inherit the evidence"
3677 );
3678
3679 // Legacy (pre-identity) callers still resolve by version string.
3680 let legacy = unsatisfied_gates(
3681 &pool,
3682 &crate::domain::AppId::default(),
3683 &tid("a"),
3684 &[Gate::CargoTest],
3685 &Evidence {
3686 version: "3.0.0",
3687 builds: &[],
3688 tier_build: None,
3689 },
3690 false,
3691 )
3692 .await
3693 .unwrap();
3694 assert!(legacy.is_empty(), "version-keyed legacy path unchanged");
3695 }
3696
3697 fn plat(s: &str) -> crate::domain::Platform {
3698 crate::domain::Platform::parse(s).unwrap()
3699 }
3700
3701 /// The structural block that made a cross-architecture promote impossible.
3702 ///
3703 /// pom 0.4.3 astra -> hetzner ships the x86_64 bundle. astra is aarch64 and
3704 /// has only ever run the aarch64 one, so `node_health` looked up against the
3705 /// x86_64 build found nothing and fail-closed refused — with `/state`
3706 /// reporting astra's node_health passed, correctly, off astra's own row. No
3707 /// amount of re-running or re-confirming could clear it: astra will never run
3708 /// that bundle. Measured 2026-08-19, promote refused twice.
3709 ///
3710 /// node_health is evidence about the tier, so it is keyed on the tier's own
3711 /// build. cargo_test is evidence about the bytes, so it stays per shipped
3712 /// build — the assertion below holds both halves at once.
3713 #[tokio::test]
3714 async fn tier_gates_key_on_the_tier_s_build_not_on_what_it_ships_onward() {
3715 let pool = fresh_pool().await;
3716 seed(&pool, "astra", "0.4.3").await;
3717 let arm = seed_build(&pool, "sha-arm", "0.4.3", "/rel/aaaaaaaaaaaaaaaa").await;
3718 let x86 = seed_build(&pool, "sha-x86", "0.4.3", "/rel/bbbbbbbbbbbbbbbb").await;
3719 // What astra actually has: its own node_health, and each bundle's own
3720 // artifact evidence from its own intake.
3721 insert_gate_build(&pool, "astra", "0.4.3", "node_health", 1, arm).await;
3722 insert_gate_build(&pool, "astra", "0.4.3", "cargo_test", 1, arm).await;
3723 insert_gate_build(&pool, "astra", "0.4.3", "cargo_test", 1, x86).await;
3724
3725 let ships_x86 = [PromotedBuild {
3726 platform: Some(plat("linux/x86_64")),
3727 build_id: Some(x86),
3728 }];
3729 let pending = unsatisfied_gates(
3730 &pool,
3731 &crate::domain::AppId::default(),
3732 &tid("astra"),
3733 &[Gate::NodeHealth, Gate::CargoTest],
3734 &Evidence {
3735 version: "0.4.3",
3736 builds: &ships_x86,
3737 tier_build: Some(arm),
3738 }, // astra's own build
3739 false,
3740 )
3741 .await
3742 .unwrap();
3743 assert!(
3744 pending.is_empty(),
3745 "astra's node_health vouches for astra, not for the bundle leaving it: {pending:?}"
3746 );
3747
3748 // And it is still a real gate: a tier whose own node_health never passed
3749 // is refused, however green the bundle it is shipping.
3750 seed(&pool, "hetzner", "0.4.3").await;
3751 insert_gate_build(&pool, "hetzner", "0.4.3", "cargo_test", 1, x86).await;
3752 let pending = unsatisfied_gates(
3753 &pool,
3754 &crate::domain::AppId::default(),
3755 &tid("hetzner"),
3756 &[Gate::NodeHealth, Gate::CargoTest],
3757 &Evidence {
3758 version: "0.4.3",
3759 builds: &ships_x86,
3760 tier_build: Some(x86),
3761 },
3762 false,
3763 )
3764 .await
3765 .unwrap();
3766 assert_eq!(pending, vec!["node_health".to_string()]);
3767 }
3768
3769 /// The looseness this closes: one pom version is two bundles with two
3770 /// digests, each accepted through its own intake. Checking only the build the
3771 /// source tier points at let the sibling ship on gate rows nobody read.
3772 /// Every build the promote will ship must show its own passed row.
3773 #[tokio::test]
3774 async fn every_shipped_build_must_show_its_own_gate_evidence() {
3775 let pool = fresh_pool().await;
3776 seed(&pool, "a", "4.0.0").await;
3777 let arm = seed_build(&pool, "sha-arm", "4.0.0", "/rel/aaaaaaaaaaaaaaaa").await;
3778 let x86 = seed_build(&pool, "sha-x86", "4.0.0", "/rel/bbbbbbbbbbbbbbbb").await;
3779 // Only the aarch64 half was gated. This is exactly the state a
3780 // two-architecture release passes through while the second build runs.
3781 insert_gate_build(&pool, "a", "4.0.0", "cargo_test", 1, arm).await;
3782
3783 let both = [
3784 PromotedBuild {
3785 platform: Some(plat("linux/aarch64")),
3786 build_id: Some(arm),
3787 },
3788 PromotedBuild {
3789 platform: Some(plat("linux/x86_64")),
3790 build_id: Some(x86),
3791 },
3792 ];
3793 let pending = unsatisfied_gates(
3794 &pool,
3795 &crate::domain::AppId::default(),
3796 &tid("a"),
3797 &[Gate::CargoTest],
3798 &Evidence {
3799 version: "4.0.0",
3800 builds: &both,
3801 tier_build: None,
3802 },
3803 false,
3804 )
3805 .await
3806 .unwrap();
3807 assert_eq!(
3808 pending,
3809 vec!["cargo_test (linux/x86_64)".to_string()],
3810 "the ungated half blocks the promote, and the message says which half"
3811 );
3812
3813 // Gate the sibling and the promote clears. Each architecture stands on
3814 // its own evidence; neither inherits the other's.
3815 insert_gate_build(&pool, "a", "4.0.0", "cargo_test", 1, x86).await;
3816 let pending = unsatisfied_gates(
3817 &pool,
3818 &crate::domain::AppId::default(),
3819 &tid("a"),
3820 &[Gate::CargoTest],
3821 &Evidence {
3822 version: "4.0.0",
3823 builds: &both,
3824 tier_build: None,
3825 },
3826 false,
3827 )
3828 .await
3829 .unwrap();
3830 assert!(pending.is_empty(), "both halves gated: {pending:?}");
3831 }
3832
3833 /// A single-platform product's error message is exactly what it always was.
3834 /// The platform qualifier is for telling two halves apart, so adding it to a
3835 /// product that has one half would be noise in the one message an operator
3836 /// reads under pressure.
3837 #[tokio::test]
3838 async fn one_shipped_build_reports_an_unqualified_gate_name() {
3839 let pool = fresh_pool().await;
3840 seed(&pool, "a", "5.0.0").await;
3841 let only = seed_build(&pool, "sha-one", "5.0.0", "/rel/cccccccccccccccc").await;
3842
3843 let pending = unsatisfied_gates(
3844 &pool,
3845 &crate::domain::AppId::default(),
3846 &tid("a"),
3847 &[Gate::CargoTest],
3848 &Evidence {
3849 version: "5.0.0",
3850 builds: &[PromotedBuild {
3851 platform: Some(plat("linux/x86_64")),
3852 build_id: Some(only),
3853 }],
3854 tier_build: None,
3855 },
3856 false,
3857 )
3858 .await
3859 .unwrap();
3860 assert_eq!(pending, vec!["cargo_test".to_string()]);
3861 }
3862
3863 /// `burn_in` is keyed on the tier's clock, not on a build, so a
3864 /// two-architecture promote must ask about it once rather than name it twice
3865 /// in the failure.
3866 #[tokio::test]
3867 async fn a_tier_scoped_gate_is_reported_once_across_several_builds() {
3868 let pool = fresh_pool().await;
3869 seed(&pool, "a", "6.0.0").await;
3870 let arm = seed_build(&pool, "sha-arm6", "6.0.0", "/rel/dddddddddddddddd").await;
3871 let x86 = seed_build(&pool, "sha-x866", "6.0.0", "/rel/eeeeeeeeeeeeeeee").await;
3872
3873 let pending = unsatisfied_gates(
3874 &pool,
3875 &crate::domain::AppId::default(),
3876 &tid("a"),
3877 &[Gate::BurnIn { hours: 48 }],
3878 &Evidence {
3879 version: "6.0.0",
3880 builds: &[
3881 PromotedBuild {
3882 platform: Some(plat("linux/aarch64")),
3883 build_id: Some(arm),
3884 },
3885 PromotedBuild {
3886 platform: Some(plat("linux/x86_64")),
3887 build_id: Some(x86),
3888 },
3889 ],
3890 tier_build: None,
3891 },
3892 false,
3893 )
3894 .await
3895 .unwrap();
3896 assert_eq!(
3897 pending,
3898 vec!["burn_in".to_string()],
3899 "a tier-scoped gate belongs to the tier, not to each build"
3900 );
3901 }
3902
3903 /// A tier is usually several nodes on one architecture. Deduplicating means
3904 /// three x86_64 nodes ask about one build once, rather than repeating the
3905 /// same gate name three times in the error.
3906 #[test]
3907 fn distinct_builds_collapses_nodes_that_share_a_build() {
3908 use super::promotion::distinct_builds;
3909 let node = Node {
3910 platform: None,
3911 base_image: None,
3912 libc: None,
3913 name: "n1".into(),
3914 ssh_target: "local".into(),
3915 release_root: "/tmp/n1".into(),
3916 service_name: "makenotwork.service".into(),
3917 health_url: None,
3918 config_check_env_file: None,
3919 actuate: crate::topology::default_actuate(),
3920 observe: crate::topology::default_observe(),
3921 companions: Vec::new(),
3922 };
3923 let bundles = vec![
3924 (
3925 &node,
3926 std::path::PathBuf::from("/rel/a"),
3927 Some(plat("linux/x86_64")),
3928 Some(7),
3929 ),
3930 (
3931 &node,
3932 std::path::PathBuf::from("/rel/a"),
3933 Some(plat("linux/x86_64")),
3934 Some(7),
3935 ),
3936 (
3937 &node,
3938 std::path::PathBuf::from("/rel/b"),
3939 Some(plat("linux/aarch64")),
3940 Some(8),
3941 ),
3942 ];
3943 let builds = distinct_builds(&bundles);
3944 assert_eq!(builds.len(), 2);
3945 assert_eq!(builds[0].build_id, Some(7));
3946 assert_eq!(builds[1].build_id, Some(8));
3947 }
3948
3949 #[tokio::test]
3950 async fn promote_rejects_an_explicit_version_that_is_not_the_source_build() {
3951 // The burn-in hole: `promote --version Y` used to check the SOURCE tier's
3952 // clock/evidence (which belong to whatever is current there), letting Y
3953 // inherit another build's 48h. Now promote resolves the source's current
3954 // build and refuses an explicit version that isn't it.
3955 let state = test_state().await;
3956 seed_version(&state.pool, "3.0.0").await;
3957 let b = seed_build(&state.pool, "shaB", "3.0.0", "/rel/deadbeefdeadbeef").await;
3958 sqlx::query(
3959 "UPDATE tier_state SET current_version='3.0.0', current_build_id=? WHERE tier='host'",
3960 )
3961 .bind(b)
3962 .execute(&state.pool)
3963 .await
3964 .unwrap();
3965
3966 let err = promote_inner(
3967 state,
3968 "a".into(),
3969 PromoteBody {
3970 version: Some("2.0.0".into()),
3971 ..Default::default()
3972 },
3973 )
3974 .await
3975 .expect_err("promoting a version other than the source build must be refused");
3976 assert!(
3977 matches!(&err, crate::error::Error::GateBlocked(m) if m.contains("vouched for")),
3978 "the refusal must name the mismatch, got: {err:?}",
3979 );
3980 }
3981
3982 #[tokio::test]
3983 async fn promote_identity_path_deploys_the_source_build_and_advances_build_id() {
3984 // Full promote through the identity path: source tier points at a build,
3985 // promote resolves the artifact through it, deploys build_runs.staged_path
3986 // (content-addressed), and advances the target's current_build_id.
3987 let mut state = test_state().await;
3988 let node_root = tempfile::tempdir().unwrap();
3989 // Point the a-local node at a real tempdir so the symlink swap is checkable.
3990 let mut topo = (*state.topo).clone();
3991 topo.tiers[1].nodes[0].release_root = node_root.path().to_string_lossy().into_owned();
3992 topo.tiers[1].gates = vec![]; // isolate the deploy fan-out
3993 state.topo = Arc::new(topo);
3994 state.executors = Arc::new(crate::state::build_executors(&state.topo));
3995
3996 // deploys.node FKs into `nodes`, so the promote path needs the row.
3997 sqlx::query("INSERT INTO nodes (name, tier, ssh_target, release_root) VALUES ('a-local', 'a', 'local', ?)")
3998 .bind(node_root.path().to_string_lossy().into_owned())
3999 .execute(&state.pool)
4000 .await
4001 .unwrap();
4002
4003 seed_version(&state.pool, "3.0.0").await;
4004 let staged = format!(
4005 "{}/releases/abc123abc123abc1",
4006 node_root.path().to_string_lossy()
4007 );
4008 let b = seed_build(&state.pool, "shaB", "3.0.0", &staged).await;
4009 sqlx::query(
4010 "UPDATE tier_state SET current_version='3.0.0', current_build_id=? WHERE tier='host'",
4011 )
4012 .bind(b)
4013 .execute(&state.pool)
4014 .await
4015 .unwrap();
4016
4017 let pool = state.pool.clone();
4018 let Json(body) = promote_inner(state, "a".into(), PromoteBody::default())
4019 .await
4020 .expect("identity-path promote succeeds");
4021 assert_eq!(body["version"], "3.0.0");
4022
4023 // Target tier advanced with the build identity, not just the version.
4024 let (cur_v, cur_b): (Option<String>, Option<i64>) = sqlx::query_as(
4025 "SELECT current_version, current_build_id FROM tier_state WHERE tier = 'a'",
4026 )
4027 .fetch_one(&pool)
4028 .await
4029 .unwrap();
4030 assert_eq!(cur_v.as_deref(), Some("3.0.0"));
4031 assert_eq!(cur_b, Some(b), "target records the promoted build id");
4032
4033 // The deploy row is attributed to the build.
4034 let deploy_b: Option<i64> = sqlx::query_scalar(
4035 "SELECT build_id FROM deploys WHERE tier = 'a' ORDER BY id DESC LIMIT 1",
4036 )
4037 .fetch_one(&pool)
4038 .await
4039 .unwrap();
4040 assert_eq!(deploy_b, Some(b));
4041
4042 // The node's `current` points at the content-addressed release dir, whose
4043 // name is the staged_path's basename — not the version.
4044 let link = tokio::fs::read_link(node_root.path().join("current"))
4045 .await
4046 .unwrap();
4047 assert_eq!(link.to_string_lossy(), "releases/abc123abc123abc1");
4048 }
4049
4050 /// Insert a bare `versions` row (FK target for tier_state.current_version).
4051 async fn seed_version(pool: &SqlitePool, version: &str) {
4052 sqlx::query("INSERT INTO versions (version, git_sha, built_at, artifact_path) VALUES (?, 'sha', datetime('now'), '/tmp/x') ON CONFLICT DO NOTHING")
4053 .bind(version).execute(pool).await.unwrap();
4054 }
4055
4056 // ---- per-node bundle resolution (wiki release-artifact-identity) ----
4057
4058 /// A settled build of `version` recorded as `platform`'s half of it.
4059 async fn seed_platform_build(
4060 pool: &SqlitePool,
4061 sha: &str,
4062 version: &str,
4063 platform: &str,
4064 staged_path: &str,
4065 ) -> i64 {
4066 sqlx::query_scalar(
4067 "INSERT INTO build_runs (sha, version, phase, result, started_at, bundle_digest, staged_path, platform)
4068 VALUES (?, ?, 'done', 'passed', datetime('now'), ?, ?, ?) RETURNING id",
4069 )
4070 .bind(sha)
4071 .bind(version)
4072 .bind(format!("{sha}-digest"))
4073 .bind(staged_path)
4074 .bind(platform)
4075 .fetch_one(pool)
4076 .await
4077 .unwrap()
4078 }
4079
4080 fn node_on(name: &str, platform: Option<&str>) -> Node {
4081 Node {
4082 platform: platform.map(plat),
4083 base_image: None,
4084 libc: None,
4085 name: name.into(),
4086 ssh_target: "local".into(),
4087 release_root: format!("/tmp/{name}"),
4088 service_name: "makenotwork.service".into(),
4089 health_url: None,
4090 config_check_env_file: None,
4091 actuate: crate::topology::default_actuate(),
4092 observe: crate::topology::default_observe(),
4093 companions: Vec::new(),
4094 }
4095 }
4096
4097 /// One pom version is two bundles with two digests. The node states which
4098 /// architecture it can run, and the sibling bundle is resolved out of
4099 /// `build_runs` rather than the caller's own half being shipped everywhere.
4100 #[tokio::test]
4101 async fn a_node_on_the_other_architecture_gets_its_own_bundle() {
4102 let state = test_state().await;
4103 seed_version(&state.pool, "5.0.0").await;
4104 let arm = seed_platform_build(
4105 &state.pool,
4106 "sha-arm",
4107 "5.0.0",
4108 "linux/aarch64",
4109 "/rel/aaaaaaaaaaaaaaaa",
4110 )
4111 .await;
4112 let x86 = seed_platform_build(
4113 &state.pool,
4114 "sha-x86",
4115 "5.0.0",
4116 "linux/x86_64",
4117 "/rel/bbbbbbbbbbbbbbbb",
4118 )
4119 .await;
4120
4121 let n_arm = node_on("astra", Some("linux/aarch64"));
4122 let n_x86 = node_on("hetzner", Some("linux/x86_64"));
4123 let bundles = super::promotion::bundles_for_nodes(
4124 &state,
4125 "5.0.0",
4126 &[&n_arm, &n_x86],
4127 std::path::Path::new("/rel/aaaaaaaaaaaaaaaa"),
4128 Some(&plat("linux/aarch64")),
4129 Some(arm),
4130 )
4131 .await
4132 .expect("both architectures have a green bundle at this version");
4133
4134 assert_eq!(bundles.len(), 2);
4135 assert_eq!(
4136 bundles[0].1,
4137 std::path::PathBuf::from("/rel/aaaaaaaaaaaaaaaa")
4138 );
4139 assert_eq!(bundles[0].2, Some(plat("linux/aarch64")));
4140 assert_eq!(bundles[0].3, Some(arm));
4141 // The one the caller never held: resolved by platform, and it carries
4142 // the sibling's own build id so its own gate evidence is what gets
4143 // checked.
4144 assert_eq!(
4145 bundles[1].1,
4146 std::path::PathBuf::from("/rel/bbbbbbbbbbbbbbbb")
4147 );
4148 assert_eq!(bundles[1].2, Some(plat("linux/x86_64")));
4149 assert_eq!(bundles[1].3, Some(x86));
4150 }
4151
4152 /// A version with no green bundle for the node's architecture fails the
4153 /// whole resolution, before any node is touched.
4154 #[tokio::test]
4155 async fn a_missing_architecture_half_refuses_the_promote_rather_than_defaulting() {
4156 let state = test_state().await;
4157 seed_version(&state.pool, "5.0.0").await;
4158 seed_platform_build(
4159 &state.pool,
4160 "sha-arm",
4161 "5.0.0",
4162 "linux/aarch64",
4163 "/rel/aaaaaaaaaaaaaaaa",
4164 )
4165 .await;
4166 let n_x86 = node_on("hetzner", Some("linux/x86_64"));
4167 let err = super::promotion::bundles_for_nodes(
4168 &state,
4169 "5.0.0",
4170 &[&n_x86],
4171 std::path::Path::new("/rel/aaaaaaaaaaaaaaaa"),
4172 Some(&plat("linux/aarch64")),
4173 None,
4174 )
4175 .await
4176 .expect_err("the x86_64 half was never built, so there is nothing to ship");
4177 assert!(
4178 format!("{err:?}").contains("no green linux/x86_64 bundle"),
4179 "{err:?}"
4180 );
4181 }
4182
4183 /// When the node and the caller name the same platform, the caller's bundle
4184 /// is the answer and no lookup happens. A single-platform product whose
4185 /// nodes have started stating a platform still has no `build_runs.platform`
4186 /// row to find, so a lookup here would refuse a promote that is fine.
4187 #[tokio::test]
4188 async fn a_node_that_agrees_with_the_caller_takes_the_callers_bundle() {
4189 let state = test_state().await;
4190 seed_version(&state.pool, "5.0.0").await;
4191 let node = node_on("hetzner", Some("linux/x86_64"));
4192 let bundles = super::promotion::bundles_for_nodes(
4193 &state,
4194 "5.0.0",
4195 &[&node],
4196 std::path::Path::new("/rel/legacy"),
4197 Some(&plat("linux/x86_64")),
4198 None,
4199 )
4200 .await
4201 .expect("the caller already holds the bundle this node wants");
4202 assert_eq!(bundles[0].1, std::path::PathBuf::from("/rel/legacy"));
4203 assert_eq!(bundles[0].2, Some(plat("linux/x86_64")));
4204 assert_eq!(bundles[0].3, None);
4205 }
4206
4207 /// The previous version is two bundles too. If the one this node runs
4208 /// cannot be resolved, nothing is attempted anywhere and every touched node
4209 /// is indeterminate — which is the truth, not a default.
4210 #[tokio::test]
4211 async fn a_rollback_that_cannot_resolve_a_bundle_reports_every_node_indeterminate() {
4212 let state = test_state().await;
4213 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')")
4214 .execute(&state.pool).await.unwrap();
4215 // No build_runs row for either node's architecture at 1.0.0.
4216 let n1 = node_on("n1", Some("linux/aarch64"));
4217 let n2 = node_on("n2", Some("linux/aarch64"));
4218 let report = rollback_deployed_nodes(&state, &tid("a"), &[&n1, &n2], "1.0.0").await;
4219 assert_eq!(report.restored, 0);
4220 assert_eq!(
4221 report.indeterminate, 2,
4222 "one per node the rollback never reached: {report:?}"
4223 );
4224 assert!(!report.is_consistent());
4225 }
4226
4227 /// A rollback that fails at the symlink swap on one node of several leaves
4228 /// that node unknown and the rest restored. The count has to be per node:
4229 /// a tier reported wholesale indeterminate sends an operator to inspect
4230 /// boxes that are fine, and a tier reported wholesale restored hides the
4231 /// one that is not.
4232 #[tokio::test]
4233 async fn a_rollback_failing_on_one_node_counts_only_that_node_indeterminate() {
4234 // The marker matches the swap-and-restart script, which is the only op
4235 // annotated AtOrAfterSwap; a1's rollback therefore lands in the
4236 // indeterminate arm rather than the already-on-previous one.
4237 let (state, _log) = fleet_fixture(&["a1", "a2"], Some("a1"), "reload-or-restart").await;
4238 let nodes: Vec<Node> = state.topo.tiers[1].nodes.clone();
4239 let refs: Vec<&Node> = nodes.iter().collect();
4240
4241 let report = rollback_deployed_nodes(&state, &tid("a"), &refs, "1.0.0").await;
4242 assert_eq!(report.restored, 1, "a2 came back: {report:?}");
4243 assert_eq!(
4244 report.indeterminate, 1,
4245 "only the node whose swap failed is unknown: {report:?}"
4246 );
4247 assert_eq!(report.already_on_previous, 0);
4248 assert_eq!(report.touched(), 2);
4249 assert!(!report.is_consistent());
4250 }
4251 }
4252