Skip to main content

max / makenotwork

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