Skip to main content

max / makenotwork

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