Skip to main content

max / makenotwork

109.7 KB · 2894 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("/self-update", post(self_update))
30 .route("/confirm/{tier}", post(confirm))
31 .route("/backup/fetch", post(backup_fetch))
32 // reads — prod state, now also gated
33 .route("/state", get(get_state))
34 .route("/status.json", get(get_status_json))
35 .route("/runs/{id}", get(get_run))
36 .route("/runs/{id}/wait", get(get_run_wait))
37 .route("/logs/{version}/{gate}", get(get_gate_log))
38 .route("/events", get(events_ws))
39 .route_layer(axum::middleware::from_fn(move |req, next| {
40 require_bearer(token.clone(), req, next)
41 }));
42
43 Router::new().merge(protected).with_state(state)
44 }
45
46 /// Bearer-token gate for the deploy mutators. When no token is configured the
47 /// request passes (main() only allows that on a loopback bind). Comparison is
48 /// constant-time to avoid leaking the token via timing.
49 async fn require_bearer(
50 token: Option<std::sync::Arc<str>>,
51 req: axum::extract::Request,
52 next: axum::middleware::Next,
53 ) -> axum::response::Response {
54 let Some(expected) = token.as_deref() else {
55 return next.run(req).await;
56 };
57 let path = req.uri().path().to_string();
58 let method = req.method().clone();
59 let ok = req
60 .headers()
61 .get(axum::http::header::AUTHORIZATION)
62 .and_then(|h| h.to_str().ok())
63 .and_then(|h| h.strip_prefix("Bearer "))
64 .is_some_and(|t| ops_core::daemon::ct_eq(t, expected));
65 if ok {
66 // Attribute mutations (POST) to a caller. Reads are GET and poll every
67 // few seconds, so logging them would drown the journal; the audit value
68 // is in *who shipped/rolled back prod*, which is always a POST. The peer
69 // address is the closest identity we have without per-operator tokens
70 // (`tailscale whois` is a future refinement); a shared token at least
71 // gets a source host into the trail.
72 if method == axum::http::Method::POST {
73 let peer = req
74 .extensions()
75 .get::<axum::extract::ConnectInfo<std::net::SocketAddr>>()
76 .map_or_else(|| "unknown".into(), |ci| ci.0.to_string());
77 tracing::info!(path = %path, peer = %peer, "authenticated deploy mutation");
78 }
79 next.run(req).await
80 } else {
81 tracing::warn!(path = %path, "rejected unauthenticated request");
82 (
83 axum::http::StatusCode::UNAUTHORIZED,
84 "missing or invalid bearer token\n",
85 )
86 .into_response()
87 }
88 }
89
90 #[derive(Serialize)]
91 pub(crate) struct StateView {
92 /// The running sandod's own package version. Lets a self-update caller
93 /// confirm the new binary is live after the restart (the tier versions are
94 /// the *deployed product*, not the controller).
95 pub(crate) sandod_version: &'static str,
96 pub(crate) tiers: Vec<TierView>,
97 /// The most recent build run (the resource `GET /runs/{id}` exposes in
98 /// full). Surfaced here so a `/state` poller sees an in-flight or failed
99 /// build — the tier versions only ever reflect the last *success*, so
100 /// without this `/state` looks frozen for the whole build. `null` until
101 /// the first `/rebuild`.
102 pub(crate) build: Option<crate::runs::BuildSummary>,
103 }
104
105 #[derive(Serialize)]
106 pub(crate) struct TierView {
107 pub(crate) name: String,
108 pub(crate) ord: i64,
109 pub(crate) provisioned: bool,
110 pub(crate) canary: String,
111 pub(crate) current_version: Option<String>,
112 pub(crate) previous_version: Option<String>,
113 pub(crate) burn_in_started_at: Option<String>,
114 /// Non-null when the tier was left in a partial / mixed-version state by a
115 /// failed promote or rollback whose compensation could not fully restore
116 /// consistency. NULL when the tier is consistent. The TUI flags it red.
117 pub(crate) partial_reason: Option<String>,
118 pub(crate) nodes: Vec<String>,
119 pub(crate) gates: Vec<GateView>,
120 }
121
122 #[derive(Serialize)]
123 pub(crate) struct GateView {
124 pub(crate) kind: String,
125 pub(crate) finished_at: Option<String>,
126 /// `'passed' | 'failed' | 'blocked'` or NULL while in-flight. The TUI
127 /// uses this to choose green/red/yellow rendering.
128 pub(crate) status: Option<String>,
129 /// Full typed `GateOutcome` as a JSON object, when present.
130 /// Deserialized lazily by the consumer; sandod doesn't re-parse it.
131 pub(crate) outcome: Option<serde_json::Value>,
132 /// Relative path under `cfg.logs_root` to the persisted stdout/stderr.
133 pub(crate) log_ref: Option<String>,
134 }
135
136 async fn get_state(State(s): State<AppState>) -> Result<Json<StateView>> {
137 Ok(Json(state_view(&s).await?))
138 }
139
140 /// `GET /status.json` — the same state, in the shared cross-service payload
141 /// every operator surface renders. See `crate::status`.
142 async fn get_status_json(State(s): State<AppState>) -> Result<Json<ops_status::Payload>> {
143 let view = state_view(&s).await?;
144 Ok(Json(crate::status::payload(&view, chrono::Utc::now())))
145 }
146
147 /// The shared read behind `/state` and `/status.json`.
148 pub(crate) async fn state_view(s: &AppState) -> Result<StateView> {
149 let rows = sqlx::query(
150 "SELECT t.name, t.ord, t.provisioned, t.canary,
151 ts.current_version, ts.previous_version, ts.burn_in_started_at,
152 ts.partial_reason
153 FROM tiers t
154 LEFT JOIN tier_state ts ON ts.tier = t.name
155 ORDER BY t.ord",
156 )
157 .fetch_all(&s.pool)
158 .await?;
159
160 let mut tiers: Vec<TierView> = Vec::with_capacity(rows.len());
161 for r in rows {
162 let name: String = r.get("name");
163 let current_version: Option<String> = r.get("current_version");
164
165 let nodes: Vec<String> =
166 sqlx::query_scalar("SELECT name FROM nodes WHERE tier = ? ORDER BY name")
167 .bind(&name)
168 .fetch_all(&s.pool)
169 .await?;
170
171 // Surface gates for current_version when set, otherwise for the most
172 // recently attempted version on this tier. Without the fallback, a
173 // tier that has never gone green (MM after a build failure, B before
174 // first deploy) exposes no gate detail via /state — debugging required
175 // SSH and direct SQLite access. See sando todo: gate observability.
176 let gate_version: Option<String> = if current_version.is_some() {
177 current_version.clone()
178 } else {
179 sqlx::query_scalar(
180 "SELECT version FROM gate_runs WHERE tier = ?
181 ORDER BY id DESC LIMIT 1",
182 )
183 .bind(&name)
184 .fetch_optional(&s.pool)
185 .await?
186 };
187
188 let gates: Vec<GateView> = if let Some(ver) = gate_version.as_ref() {
189 // Most recent gate_runs row per gate_kind for (tier, ver).
190 sqlx::query(
191 "SELECT gate_kind, finished_at, status, outcome_json, log_ref
192 FROM gate_runs g
193 WHERE tier = ?1 AND version = ?2
194 AND id = (SELECT MAX(id) FROM gate_runs
195 WHERE tier = ?1 AND version = ?2 AND gate_kind = g.gate_kind)
196 ORDER BY gate_kind",
197 )
198 .bind(&name)
199 .bind(ver)
200 .fetch_all(&s.pool)
201 .await?
202 .into_iter()
203 .map(|gr| GateView {
204 kind: gr.get("gate_kind"),
205 finished_at: gr.get("finished_at"),
206 status: gr.get("status"),
207 outcome: gr
208 .get::<Option<String>, _>("outcome_json")
209 .and_then(|s| serde_json::from_str(&s).ok()),
210 log_ref: gr.get("log_ref"),
211 })
212 .collect()
213 } else {
214 Vec::new()
215 };
216
217 tiers.push(TierView {
218 name,
219 ord: r.get("ord"),
220 provisioned: r.get::<i64, _>("provisioned") != 0,
221 canary: r.get("canary"),
222 current_version,
223 previous_version: r.get("previous_version"),
224 burn_in_started_at: r.get("burn_in_started_at"),
225 partial_reason: r.get("partial_reason"),
226 nodes,
227 gates,
228 });
229 }
230
231 let build = crate::runs::latest_summary(&s.pool).await?;
232 Ok(StateView {
233 sandod_version: env!("CARGO_PKG_VERSION"),
234 tiers,
235 build,
236 })
237 }
238
239 #[derive(Deserialize, Default)]
240 struct PromoteBody {
241 /// Optional. If absent, defaults to the predecessor tier's `current_version`
242 /// (i.e. promote whatever just finished baking on the previous tier).
243 #[serde(default)]
244 version: Option<String>,
245 #[serde(default)]
246 hotfix: bool,
247 #[serde(default)]
248 reset_burn_in: bool,
249 /// Set when the release carries a database migration. Rollback restores the
250 /// binary and release_contents only, never the schema (MNW migrates forward
251 /// on boot and has no down path), so a migration-bearing promote is one-way.
252 /// This forces a fresh `manual_confirm` on the predecessor tier even if the
253 /// tier does not configure one, so the operator consciously acknowledges the
254 /// rollback contract. `hotfix` does not suppress it. See
255 /// `deploy/README.md` "Rollback contract".
256 #[serde(default)]
257 bears_migration: bool,
258 }
259
260 async fn promote(
261 State(s): State<AppState>,
262 Path(tier): Path<String>,
263 body: Option<Json<PromoteBody>>,
264 ) -> Result<Json<serde_json::Value>> {
265 let body = body.map(|Json(b)| b).unwrap_or_default();
266 // Run the deploy in a detached task, not inline in the request future.
267 // A promote's rsync + restart can outlast a client's socket timeout; if the
268 // client disconnects, axum drops the handler future — and if the deploy ran
269 // inline that would abandon the rollout mid-flight (symlink half-swapped,
270 // tier_state not advanced, and — before the kill_on_drop fix — an orphaned
271 // rsync). Spawning detaches the work from the connection: the task owns the
272 // deploy lock and runs to completion regardless, while a still-connected
273 // client still receives the result. Mirrors how /rebuild already runs its
274 // build off-request.
275 tokio::spawn(promote_inner(s, tier, body))
276 .await
277 .map_err(|e| {
278 crate::error::Error::Other(anyhow::anyhow!("promote task failed to join: {e}"))
279 })?
280 }
281
282 async fn rollback(
283 State(s): State<AppState>,
284 Path(tier): Path<String>,
285 ) -> Result<Json<serde_json::Value>> {
286 // Serialize against any concurrent promote/rollback (CF3).
287 let _deploy_guard = s.deploy_lock.lock().await;
288 let tier = crate::domain::TierId::new(tier);
289 let target = s
290 .topo
291 .tiers
292 .iter()
293 .find(|t| t.name == tier)
294 .ok_or(crate::error::Error::NotFound)?;
295
296 let row: Option<(Option<String>, Option<String>)> =
297 sqlx::query_as("SELECT current_version, previous_version FROM tier_state WHERE tier = ?")
298 .bind(&tier)
299 .fetch_optional(&s.pool)
300 .await
301 .map_err(crate::error::Error::Db)?;
302 let (Some(current_str), Some(previous_str)) = row.unwrap_or((None, None)) else {
303 return Err(crate::error::Error::GateBlocked(
304 "no previous_version to roll back to".into(),
305 ));
306 };
307 let current = crate::domain::Version::parse(&current_str)
308 .map_err(|e| crate::error::Error::Other(anyhow::anyhow!(e)))?;
309 let previous = crate::domain::Version::parse(&previous_str)
310 .map_err(|e| crate::error::Error::Other(anyhow::anyhow!(e)))?;
311
312 let bin: Option<(String,)> =
313 sqlx::query_as("SELECT artifact_path FROM versions WHERE version = ?")
314 .bind(&previous)
315 .fetch_optional(&s.pool)
316 .await
317 .map_err(crate::error::Error::Db)?;
318 let Some((bin,)) = bin else {
319 return Err(crate::error::Error::GateBlocked(format!(
320 "previous version {previous} has no artifact_path; rollback impossible"
321 )));
322 };
323 let bin_path = std::path::PathBuf::from(bin);
324 let staged_dir = bin_path
325 .parent()
326 .ok_or_else(|| crate::error::Error::Other(anyhow::anyhow!("artifact_path has no parent")))?
327 .to_path_buf();
328
329 // Deploy `previous` to every node. A failure partway through leaves the fleet
330 // split-brain (some on `previous`, some still on `current`); record exactly how
331 // far we got so /state surfaces it rather than reporting a clean rollback.
332 let total = target.nodes.len();
333 for (i, node) in target.nodes.iter().enumerate() {
334 let executor = s
335 .executors
336 .get(&node.name)
337 .cloned()
338 .unwrap_or_else(|| crate::state::build_executor(node));
339 if let Err(e) = crate::deploy::deploy_node(
340 executor.as_ref(),
341 node,
342 &previous_str,
343 &staged_dir,
344 s.cfg.primary_bin(),
345 )
346 .await
347 {
348 set_partial(
349 &s,
350 &tier,
351 &format!(
352 "rollback to {previous} incomplete: {i}/{total} nodes rolled back, \
353 node {} failed; remaining nodes still on {current} — manual check needed",
354 node.name,
355 ),
356 )
357 .await;
358 tracing::error!(tier = %tier, node = %node.name, done = i, of = total,
359 to = %previous, "rollback failed mid-fleet; tier left partial");
360 return Err(crate::error::Error::Other(e));
361 }
362 }
363
364 // `previous_version` becomes NULL, not `current`. Sando tracks exactly one
365 // step of history, so writing the version we just rolled *off* into the
366 // rollback slot makes a second /rollback roll FORWARD onto the build the
367 // operator was escaping. NULL makes the second call fail loudly with "no
368 // previous_version to roll back to" — honest about the depth we keep.
369 // Stepping back further is the deploys-table walk we deliberately don't do.
370 //
371 // `advanced_at = now` even though burn_in_started_at is nulled: a rollback IS
372 // a change to the deployed identity, so the startup reconcile must treat the
373 // roll-off `deploys` row as predating this state — otherwise a cleanly
374 // rolled-back tier would look like an unrecorded deploy (migration 009 /
375 // [`crate::reconcile`]).
376 let now = chrono::Utc::now().to_rfc3339();
377 sqlx::query(
378 "UPDATE tier_state
379 SET current_version = ?, previous_version = NULL,
380 burn_in_started_at = NULL, advanced_at = ?
381 WHERE tier = ?",
382 )
383 .bind(&previous)
384 .bind(&now)
385 .bind(&tier)
386 .execute(&s.pool)
387 .await
388 .map_err(crate::error::Error::Db)?;
389
390 // Full rollback succeeded on every node: the tier is consistent again.
391 clear_partial(&s, &tier).await;
392
393 tracing::warn!(tier = %tier, from = %current, to = %previous, "rollback complete");
394 crate::events::emit(
395 &s.events,
396 crate::events::Event::Rollback {
397 tier: tier.clone(),
398 from: current.clone(),
399 to: previous.clone(),
400 },
401 );
402
403 Ok(Json(serde_json::json!({
404 "tier": tier,
405 "rolled_back_from": current,
406 "now_running": previous,
407 })))
408 }
409
410 #[derive(Deserialize, Default)]
411 struct RebuildBody {
412 /// Specific sha to build. If absent, resolve `topo.repo.branch` from the bare repo.
413 #[serde(default)]
414 sha: Option<String>,
415 }
416
417 async fn rebuild(
418 State(s): State<AppState>,
419 body: Option<Json<RebuildBody>>,
420 ) -> Result<Json<serde_json::Value>> {
421 let body = body.map(|Json(b)| b).unwrap_or_default();
422 let sha = match body.sha {
423 Some(sha) => sha,
424 None => {
425 // Omitted sha = "build the deploy branch's tip". Fetch upstream
426 // first so we resolve the *upstream* HEAD, not a possibly-stale
427 // local branch ref — the build task fetches too, but only after the
428 // sha is already chosen, so without this `/rebuild {}` could build
429 // an old commit. A fetch failure is non-fatal: fall back to the
430 // current bare-repo tip (same policy as the build task).
431 let bare = std::path::Path::new(&s.topo.repo.bare_path);
432 if let Some(upstream) = s.topo.repo.upstream.as_deref()
433 && let Err(e) =
434 crate::git::fetch_upstream(bare, upstream, &s.topo.repo.branch).await
435 {
436 tracing::warn!(error = %e, "pre-resolve upstream fetch failed; resolving current bare-repo branch tip");
437 }
438 crate::git::resolve_ref(bare, &s.topo.repo.branch)
439 .await
440 .map_err(crate::error::Error::Other)?
441 }
442 };
443
444 // Boundary parse: a sha entering Sando must be hex of plausible length.
445 // The build pipeline downstream only ever sees `GitSha`.
446 let sha = crate::domain::GitSha::parse(&sha)
447 .map_err(|e| crate::error::Error::Other(anyhow::anyhow!(e)))?;
448
449 tracing::info!(sha = %sha, "rebuild requested");
450 crate::events::emit(
451 &s.events,
452 crate::events::Event::RebuildRequested { sha: sha.clone() },
453 );
454
455 // One pollable resource per triggered build. Created before the spawn so
456 // the run id is in the response even if the task is aborted milliseconds
457 // later by a still-newer /rebuild.
458 let run_id = crate::runs::create(&s.pool, sha.as_str())
459 .await
460 .map_err(crate::error::Error::Other)?;
461
462 // Latest /rebuild wins: abort any in-flight build before spawning a new
463 // one. Aborting drops the spawned task's future, which drops any
464 // tokio::process::Child it owns; with `kill_on_drop(true)` set on the
465 // cargo Command, SIGKILL propagates to cargo + its rustc children.
466 let mut slot = s.active_build.lock().await;
467 if let Some(prev) = slot.take()
468 && !prev.handle.is_finished()
469 {
470 tracing::warn!("aborting in-flight build for newer /rebuild request");
471 crate::events::emit(
472 &s.events,
473 crate::events::Event::BuildAborted {
474 sha_aborted: sha.clone(),
475 },
476 );
477 prev.handle.abort();
478 // Aborting drops the task before it can settle its own row, so
479 // record the supersession here.
480 crate::runs::mark_aborted(&s.pool, prev.run_id).await.ok();
481 }
482
483 let pool = s.pool.clone();
484 let cfg = s.cfg.clone();
485 let topo = s.topo.clone();
486 let events_for_task = s.events.clone();
487 let sha_for_task = sha.clone();
488 let sha_response = sha.to_string();
489 let pool_for_task = s.pool.clone();
490 let deploy_lock = s.deploy_lock.clone();
491 let handle = tokio::spawn(async move {
492 if let Err(e) = crate::build::build_and_run_host(
493 pool,
494 cfg,
495 topo,
496 sha_for_task.clone(),
497 events_for_task,
498 run_id,
499 deploy_lock,
500 )
501 .await
502 {
503 tracing::error!(sha = %sha_for_task, error = %e, "rebuild pipeline failed");
504 // Pre-gate bails (fetch/checkout/version/scratch) don't settle the
505 // run themselves; the build-step compile error already did. First
506 // terminal write wins, so this is a safety net for the rest.
507 crate::runs::mark_failed(&pool_for_task, run_id, &format!("{e:#}"))
508 .await
509 .ok();
510 }
511 });
512 *slot = Some(crate::state::ActiveBuild {
513 handle: handle.abort_handle(),
514 run_id,
515 });
516
517 Ok(Json(
518 serde_json::json!({ "accepted": true, "sha": sha_response, "run_id": run_id.0 }),
519 ))
520 }
521
522 /// `GET /runs/{id}` — the build-status resource a non-TUI driver polls after
523 /// `/rebuild`. Open (read-only) like `/state` and `/logs`.
524 async fn get_run(
525 State(s): State<AppState>,
526 Path(id): Path<i64>,
527 ) -> Result<Json<crate::runs::RunView>> {
528 crate::runs::get(&s.pool, crate::domain::RunId(id))
529 .await
530 .map_err(crate::error::Error::Other)?
531 .map(Json)
532 .ok_or(crate::error::Error::NotFound)
533 }
534
535 #[derive(Deserialize)]
536 struct WaitParams {
537 /// How long to hold the request open before returning a still-building
538 /// run. Default 30s, capped at 120s.
539 #[serde(default)]
540 timeout_ms: Option<u64>,
541 }
542
543 /// `GET /runs/{id}/wait` — long-poll: hold the request open until the run
544 /// settles (`result != building`) or the timeout elapses, then return the
545 /// current `RunView`. Removes polling-cadence guessing for a headless driver
546 /// (fire `/rebuild`, block on `/wait`). On timeout the run is returned
547 /// still-building (200) and the caller re-issues `/wait`. 404 if unknown.
548 async fn get_run_wait(
549 State(s): State<AppState>,
550 Path(id): Path<i64>,
551 Query(p): Query<WaitParams>,
552 ) -> Result<Json<crate::runs::RunView>> {
553 let run_id = crate::domain::RunId(id);
554 let timeout = std::time::Duration::from_millis(p.timeout_ms.unwrap_or(30_000).min(120_000));
555 let deadline = tokio::time::Instant::now() + timeout;
556 // Poll the row rather than wiring a per-run notifier: a build settles on
557 // the minute scale, so a sub-second tick is plenty responsive and the
558 // query is a single indexed read. The request releases its pool handle
559 // between ticks.
560 let tick = std::time::Duration::from_millis(750);
561 loop {
562 let view = crate::runs::get(&s.pool, run_id)
563 .await
564 .map_err(crate::error::Error::Other)?
565 .ok_or(crate::error::Error::NotFound)?;
566 let now = tokio::time::Instant::now();
567 if view.result != "building" || now >= deadline {
568 return Ok(Json(view));
569 }
570 tokio::time::sleep((deadline - now).min(tick)).await;
571 }
572 }
573
574 #[derive(Deserialize)]
575 struct SelfUpdateBody {
576 /// The commit to rebuild sandod from. Must already be on the canonical
577 /// remote (the updater `git fetch`es it).
578 sha: String,
579 }
580
581 /// The privileged updater unit instance for `sha`. A git sha is hex-only, so it
582 /// is a safe systemd instance name with no escaping needed.
583 fn self_update_unit(sha: &crate::domain::GitSha) -> String {
584 format!("sando-update@{sha}.service")
585 }
586
587 /// Trigger a rebuild + restart of sandod *itself* to `sha`. sandod runs
588 /// unprivileged (User=sando, NoNewPrivileges, ProtectSystem=strict) and cannot
589 /// write `/usr/local/bin/sandod` or restart its own service — so it only
590 /// *triggers* the root oneshot `sando-update@<sha>.service` (which the sando
591 /// user is authorized to start by a scoped polkit rule). That unit builds
592 /// `sando/daemon` as the sando user in a dedicated checkout, installs the new
593 /// binary, and restarts sandod. Bearer-gated like the other mutators; the new
594 /// version shows up in `/state`'s `sandod_version` once the restart lands.
595 async fn self_update(
596 State(s): State<AppState>,
597 crate::error::TypedBody(body): crate::error::TypedBody<SelfUpdateBody>,
598 ) -> Result<Json<serde_json::Value>> {
599 let sha = crate::domain::GitSha::parse(&body.sha)
600 .map_err(|e| crate::error::Error::BadRequest(format!("invalid sha: {e}")))?;
601
602 let unit = self_update_unit(&sha);
603
604 // Don't restart the controller out from under an in-flight server build — the
605 // restart would SIGKILL it mid-deploy. Hold the active_build slot across the
606 // trigger, not just the check: releasing it before triggering let a /rebuild
607 // claim the slot in the gap and start a build the updater then killed (the
608 // Run 2 TOCTOU). With the guard held, no build can start until the updater is
609 // enqueued.
610 {
611 let slot = s.active_build.lock().await;
612 if slot.as_ref().is_some_and(|b| !b.handle.is_finished()) {
613 return Err(crate::error::Error::GateBlocked(
614 "a server build is in flight; retry /self-update once it settles".into(),
615 ));
616 }
617 // Also refuse during an in-flight promote/rollback: the updater's restart
618 // would SIGKILL the daemon mid-deploy. try_lock (not lock) so we reject
619 // rather than queue behind a long deploy; the guard is dropped immediately
620 // — it only needs to observe that no deploy holds it right now.
621 if s.deploy_lock.try_lock().is_err() {
622 return Err(crate::error::Error::GateBlocked(
623 "a promote/rollback is in flight; retry /self-update once it settles".into(),
624 ));
625 }
626
627 tracing::warn!(sha = %sha, unit = %unit, "self-update requested; triggering privileged updater");
628 // `--no-block`: return as soon as the job is enqueued. The build+restart
629 // outcome lands in `journalctl -u <unit>`; sandod is restarted out from
630 // under this request, so there is nothing more to await here.
631 let status = tokio::process::Command::new("systemctl")
632 .args(["start", "--no-block", &unit])
633 .status()
634 .await
635 .map_err(|e| crate::error::Error::Other(anyhow::anyhow!("spawning systemctl: {e}")))?;
636 if !status.success() {
637 return Err(crate::error::Error::Other(anyhow::anyhow!(
638 "systemctl start {unit} exited {status}; is sando-update@.service installed and the sando-user polkit rule in place?"
639 )));
640 }
641 }
642 Ok(Json(
643 serde_json::json!({ "accepted": true, "sha": sha.to_string(), "unit": unit }),
644 ))
645 }
646
647 async fn confirm(
648 State(s): State<AppState>,
649 Path(tier): Path<String>,
650 ) -> Result<Json<serde_json::Value>> {
651 // Operator-driven satisfaction of a `manual_confirm` gate. Looks up the
652 // pending version (current MM version, or the tier's own if non-mm) and
653 // inserts a passing gate_runs row so /promote can advance.
654 //
655 // Hold deploy_lock so the version we read and the row we insert reference the
656 // same tier state — a concurrent rollback can't change current_version between
657 // the read and the insert (ultra-fuzz Run 2, F3).
658 let _deploy_guard = s.deploy_lock.lock().await;
659 let tier = crate::domain::TierId::new(tier);
660 let target = s
661 .topo
662 .tiers
663 .iter()
664 .find(|t| t.name == tier)
665 .ok_or(crate::error::Error::NotFound)?;
666
667 let version_str: Option<String> =
668 sqlx::query_scalar("SELECT current_version FROM tier_state WHERE tier = ?")
669 .bind(&target.name)
670 .fetch_optional(&s.pool)
671 .await
672 .map_err(crate::error::Error::Db)?
673 .flatten();
674 let version_str = version_str.ok_or_else(|| {
675 crate::error::Error::GateBlocked(format!(
676 "tier {tier} has no current_version; nothing to confirm"
677 ))
678 })?;
679 let version = crate::domain::Version::parse(&version_str)
680 .map_err(|e| crate::error::Error::Other(anyhow::anyhow!(e)))?;
681
682 let now = chrono::Utc::now().to_rfc3339();
683 let outcome =
684 crate::outcome::GateOutcome::passed(crate::outcome::PassNote::OperatorConfirmed {
685 at: chrono::Utc::now(),
686 });
687 let outcome_json = serde_json::to_string(&outcome)
688 .map_err(|e| crate::error::Error::Other(anyhow::anyhow!(e)))?;
689 sqlx::query(
690 "INSERT INTO gate_runs (version, tier, gate_kind, started_at, finished_at, status, outcome_json)
691 VALUES (?, ?, 'manual_confirm', ?, ?, 'passed', ?)",
692 )
693 .bind(&version).bind(&target.name).bind(&now).bind(&now).bind(&outcome_json)
694 .execute(&s.pool).await.map_err(crate::error::Error::Db)?;
695
696 tracing::info!(tier = %tier, version = %version, "manual_confirm recorded");
697 crate::events::emit(
698 &s.events,
699 crate::events::Event::ManualConfirm {
700 tier: tier.clone(),
701 version: version.clone(),
702 },
703 );
704
705 Ok(Json(
706 serde_json::json!({ "tier": tier, "version": version }),
707 ))
708 }
709
710 async fn backup_fetch(State(s): State<AppState>) -> Result<Json<serde_json::Value>> {
711 let fb = crate::backup::fetch(&s.pool, &s.cfg, &s.topo)
712 .await
713 .map_err(crate::error::Error::Other)?;
714 crate::events::emit(
715 &s.events,
716 crate::events::Event::BackupFetched {
717 source: fb.source.clone(),
718 byte_size: fb.byte_size.unwrap_or(0),
719 },
720 );
721 Ok(Json(serde_json::json!({
722 "source": fb.source,
723 "local_path": fb.local_path,
724 "byte_size": fb.byte_size,
725 })))
726 }
727
728 async fn get_gate_log(
729 State(s): State<AppState>,
730 Path((version, gate)): Path<(String, String)>,
731 ) -> Result<axum::response::Response> {
732 // Guard against `..` / absolute paths — the version segment must be a single
733 // safe component. Without this, `GET /logs/..%2Fetc/passwd` would escape
734 // logs_root.
735 fn safe(seg: &str) -> bool {
736 !seg.is_empty() && !seg.contains('/') && !seg.contains('\\') && seg != "." && seg != ".."
737 }
738 if !safe(&version) {
739 return Err(crate::error::Error::NotFound);
740 }
741 // The gate segment is an allowlisted kind, not a free-form filename: parse it
742 // to `GateKind` so only the five known `<kind>.log` files are reachable. This
743 // closes `*.log` filename probing within a version dir (traversal was already
744 // blocked; this bounds the basename to the known set).
745 let kind: crate::domain::GateKind = gate.parse().map_err(|_| crate::error::Error::NotFound)?;
746 let path = s
747 .cfg
748 .logs_root
749 .join(&version)
750 .join(format!("{}.log", kind.as_str()));
751 // Bound how much a single request can pull into the daemon's memory: a
752 // runaway gate log shouldn't be read whole. Past the cap we return the tail
753 // (the recent, relevant output) behind a truncation marker.
754 const MAX_LOG_BYTES: u64 = 8 * 1024 * 1024;
755 use tokio::io::{AsyncReadExt, AsyncSeekExt};
756 let mut file = match tokio::fs::File::open(&path).await {
757 Ok(f) => f,
758 Err(e) if e.kind() == std::io::ErrorKind::NotFound => {
759 return Err(crate::error::Error::NotFound);
760 }
761 Err(e) => return Err(crate::error::Error::Other(e.into())),
762 };
763 let len = file
764 .metadata()
765 .await
766 .map_err(|e| crate::error::Error::Other(e.into()))?
767 .len();
768 let body: Vec<u8> = if len <= MAX_LOG_BYTES {
769 let mut buf = Vec::with_capacity(len as usize);
770 file.read_to_end(&mut buf)
771 .await
772 .map_err(|e| crate::error::Error::Other(e.into()))?;
773 buf
774 } else {
775 file.seek(std::io::SeekFrom::End(-(MAX_LOG_BYTES as i64)))
776 .await
777 .map_err(|e| crate::error::Error::Other(e.into()))?;
778 let mut buf = format!(
779 "[log truncated: showing the last {} MiB of {len} bytes]\n",
780 MAX_LOG_BYTES / 1024 / 1024,
781 )
782 .into_bytes();
783 file.read_to_end(&mut buf)
784 .await
785 .map_err(|e| crate::error::Error::Other(e.into()))?;
786 buf
787 };
788 Ok((
789 [(
790 axum::http::header::CONTENT_TYPE,
791 "text/plain; charset=utf-8",
792 )],
793 body,
794 )
795 .into_response())
796 }
797
798 async fn events_ws(ws: WebSocketUpgrade, State(s): State<AppState>) -> impl IntoResponse {
799 use axum::extract::ws::Message;
800 use tokio::sync::broadcast::error::RecvError;
801
802 ws.on_upgrade(move |mut socket| async move {
803 // Subscribe to both channels and merge them: a lag on the high-rate log
804 // stream emits its own `lagged` frame without dropping anything on the
805 // status stream (and vice versa), so a busy gate's chunk firehose can't
806 // evict a PromoteComplete/GateDone the operator needs to see.
807 let mut status_rx = s.events.subscribe_status();
808 let mut logs_rx = s.events.subscribe_logs();
809 loop {
810 let recv = tokio::select! {
811 r = status_rx.recv() => r,
812 r = logs_rx.recv() => r,
813 };
814 match recv {
815 Ok(env) => {
816 let json = match serde_json::to_string(&env) {
817 Ok(s) => s,
818 Err(e) => {
819 tracing::warn!(error = %e, "events ws: serialize failed");
820 continue;
821 }
822 };
823 if socket.send(Message::Text(json.into())).await.is_err() {
824 break;
825 }
826 }
827 Err(RecvError::Lagged(n)) => {
828 let _ = socket
829 .send(Message::Text(
830 format!(r#"{{"kind":"lagged","skipped":{n}}}"#).into(),
831 ))
832 .await;
833 }
834 Err(RecvError::Closed) => break,
835 }
836 }
837 })
838 }
839
840 #[cfg(test)]
841 mod tests {
842 use super::promotion::{rollback_deployed_nodes, unsatisfied_gates};
843 use super::*;
844 use crate::config::Config;
845 use crate::topology::{BackupConfig, CanaryPolicy, Gate, Node, RepoConfig, Tier, Topology};
846 use async_trait::async_trait;
847 use axum::body::Body;
848 use axum::http::{Request, StatusCode};
849 use http_body_util::BodyExt;
850 use ops_exec::{CapabilitySet, Executor, LogSink, RunOutput, Step, SyncOpts};
851 use sqlx::SqlitePool;
852 use sqlx::sqlite::SqlitePoolOptions;
853 use std::os::unix::process::ExitStatusExt;
854 use std::path::PathBuf;
855 use std::sync::{Arc, Mutex as StdMutex};
856 use tower::ServiceExt;
857
858 async fn fresh_pool() -> SqlitePool {
859 let pool = SqlitePoolOptions::new()
860 .max_connections(1)
861 .connect("sqlite::memory:")
862 .await
863 .unwrap();
864 sqlx::migrate!("./migrations").run(&pool).await.unwrap();
865 pool
866 }
867
868 /// Two-tier topology used by the route tests: mm (provisioned, no nodes)
869 /// → a (provisioned, one local node). Mirrors the production shape
870 /// without involving real ssh / postgres.
871 fn test_topo() -> Topology {
872 Topology {
873 repo: RepoConfig {
874 bare_path: "/tmp/test.git".into(),
875 branch: "main".into(),
876 upstream: None,
877 },
878 backup: BackupConfig {
879 source: "file:///tmp/test-backup.sql".into(),
880 local_path: "/tmp/local-backup.sql".into(),
881 },
882 tiers: vec![
883 Tier {
884 name: "host".into(),
885 provisioned: true,
886 gates: vec![],
887 canary: CanaryPolicy::Sequential,
888 nodes: vec![],
889 },
890 Tier {
891 name: "a".into(),
892 provisioned: true,
893 gates: vec![Gate::BootSmoke],
894 canary: CanaryPolicy::Sequential,
895 nodes: vec![Node {
896 name: "a-local".into(),
897 ssh_target: "local".into(),
898 release_root: "/tmp/a-node".into(),
899 service_name: "makenotwork.service".into(),
900 health_url: None,
901 config_check_env_file: None,
902 actuate: crate::topology::default_actuate(),
903 observe: crate::topology::default_observe(),
904 companions: Vec::new(),
905 }],
906 },
907 ],
908 }
909 }
910
911 fn test_cfg() -> Config {
912 Config {
913 listen: "127.0.0.1:0".into(),
914 db_path: PathBuf::from(":memory:"),
915 topology_path: PathBuf::from("/tmp/test-sando.toml"),
916 build_host: "test-host".into(),
917 workdir: PathBuf::from("/tmp/sando-work"),
918 release_root: PathBuf::from("/tmp/sando-releases"),
919 scratch_db_url: None,
920 scratch_owner_role: "makenotwork".into(),
921 boot_smoke_port: 18181,
922 code_smoke_port: 18182,
923 bin_names: vec!["makenotwork".into()],
924 logs_root: PathBuf::from("/tmp/sando-logs"),
925 release_contents: vec![],
926 cargo_target_dir: None,
927 gate_timeout_secs: 2400,
928 companions: Vec::new(),
929 test_targets: vec![crate::config::TestTarget {
930 dir: PathBuf::from("server"),
931 features: vec!["fast-tests".into()],
932 all_features: false,
933 scratch_db: true,
934 }],
935 }
936 }
937
938 async fn test_state() -> AppState {
939 let pool = fresh_pool().await;
940 // Seed tier rows so FKs on tier_state / gate_runs are satisfied.
941 for (i, name) in ["host", "a"].iter().enumerate() {
942 sqlx::query(
943 "INSERT INTO tiers (name, ord, provisioned, canary) VALUES (?, ?, 1, 'sequential')",
944 )
945 .bind(name)
946 .bind(i as i64)
947 .execute(&pool)
948 .await
949 .unwrap();
950 sqlx::query("INSERT INTO tier_state (tier) VALUES (?)")
951 .bind(name)
952 .execute(&pool)
953 .await
954 .unwrap();
955 }
956 // Don't call install_recorder in tests — it touches a process-global
957 // and conflicts when tests run in parallel.
958 let topo = test_topo();
959 let executors = Arc::new(crate::state::build_executors(&topo));
960 AppState {
961 pool,
962 topo: Arc::new(topo),
963 cfg: Arc::new(test_cfg()),
964 active_build: Arc::new(tokio::sync::Mutex::new(None)),
965 deploy_lock: Arc::new(tokio::sync::Mutex::new(())),
966 events: crate::events::channel(),
967 executors,
968 api_token: None,
969 }
970 }
971
972 async fn body_string(resp: axum::response::Response) -> String {
973 let bytes = resp.into_body().collect().await.unwrap().to_bytes();
974 String::from_utf8(bytes.to_vec()).unwrap()
975 }
976
977 /// Insert the FK prerequisites for inserting gate_runs/tier_state rows.
978 async fn seed(pool: &SqlitePool, tier: &str, version: &str) {
979 sqlx::query("INSERT INTO tiers (name, ord, provisioned, canary) VALUES (?, 0, 1, 'sequential') ON CONFLICT DO NOTHING")
980 .bind(tier).execute(pool).await.unwrap();
981 sqlx::query("INSERT INTO versions (version, git_sha, built_at, artifact_path) VALUES (?, 'sha', datetime('now'), '/tmp/x') ON CONFLICT DO NOTHING")
982 .bind(version).execute(pool).await.unwrap();
983 sqlx::query("INSERT INTO tier_state (tier, current_version) VALUES (?, NULL) ON CONFLICT DO NOTHING")
984 .bind(tier).execute(pool).await.unwrap();
985 }
986
987 async fn insert_gate(pool: &SqlitePool, tier: &str, version: &str, kind: &str, passed: i64) {
988 let status = if passed == 1 { "passed" } else { "failed" };
989 sqlx::query(
990 "INSERT INTO gate_runs (version, tier, gate_kind, started_at, finished_at, status) \
991 VALUES (?, ?, ?, datetime('now'), datetime('now'), ?)",
992 )
993 .bind(version)
994 .bind(tier)
995 .bind(kind)
996 .bind(status)
997 .execute(pool)
998 .await
999 .unwrap();
1000 }
1001
1002 // ---- unsatisfied_gates ----
1003
1004 fn tid(s: &str) -> crate::domain::TierId {
1005 crate::domain::TierId::new(s)
1006 }
1007
1008 #[tokio::test]
1009 async fn unsatisfied_gates_empty_when_no_configured_gates() {
1010 // A tier that configures no gates has nothing to satisfy.
1011 let pool = fresh_pool().await;
1012 seed(&pool, "host", "0.8.12").await;
1013 let pending = unsatisfied_gates(&pool, &tid("host"), &[], "0.8.12", None, false)
1014 .await
1015 .unwrap();
1016 assert_eq!(pending, Vec::<String>::new());
1017 }
1018
1019 #[tokio::test]
1020 async fn unsatisfied_gates_flags_configured_gate_that_never_ran() {
1021 // THE CF1 FIX: a configured gate with no gate_runs row is unsatisfied
1022 // (fail closed), NOT silently treated as green. Before this, an A tier
1023 // whose boot_smoke never executed exposed zero rows and waved promotion
1024 // straight through to prod.
1025 let pool = fresh_pool().await;
1026 seed(&pool, "a", "0.8.12").await;
1027 let pending =
1028 unsatisfied_gates(&pool, &tid("a"), &[Gate::BootSmoke], "0.8.12", None, false)
1029 .await
1030 .unwrap();
1031 assert_eq!(pending, vec!["boot_smoke".to_string()]);
1032 }
1033
1034 #[tokio::test]
1035 async fn unsatisfied_gates_flags_failed_kind() {
1036 let pool = fresh_pool().await;
1037 seed(&pool, "host", "0.8.12").await;
1038 insert_gate(&pool, "host", "0.8.12", "cargo_test", 0).await;
1039 insert_gate(&pool, "host", "0.8.12", "boot_smoke", 1).await;
1040 let pending = unsatisfied_gates(
1041 &pool,
1042 &tid("host"),
1043 &[Gate::CargoTest, Gate::BootSmoke],
1044 "0.8.12",
1045 None,
1046 false,
1047 )
1048 .await
1049 .unwrap();
1050 assert_eq!(pending, vec!["cargo_test".to_string()]);
1051 }
1052
1053 #[tokio::test]
1054 async fn unsatisfied_gates_latest_row_wins() {
1055 // Two runs of the same gate; only the latest counts. A flap from
1056 // red to green should clear the pending entry.
1057 let pool = fresh_pool().await;
1058 seed(&pool, "host", "0.8.12").await;
1059 insert_gate(&pool, "host", "0.8.12", "cargo_test", 0).await;
1060 insert_gate(&pool, "host", "0.8.12", "cargo_test", 1).await;
1061 let pending = unsatisfied_gates(
1062 &pool,
1063 &tid("host"),
1064 &[Gate::CargoTest],
1065 "0.8.12",
1066 None,
1067 false,
1068 )
1069 .await
1070 .unwrap();
1071 assert!(pending.is_empty());
1072 }
1073
1074 async fn insert_confirm(
1075 pool: &SqlitePool,
1076 tier: &str,
1077 version: &str,
1078 at: chrono::DateTime<chrono::Utc>,
1079 ) {
1080 sqlx::query(
1081 "INSERT INTO gate_runs (version, tier, gate_kind, started_at, finished_at, status) \
1082 VALUES (?, ?, 'manual_confirm', ?, ?, 'passed')",
1083 )
1084 .bind(version)
1085 .bind(tier)
1086 .bind(at.to_rfc3339())
1087 .bind(at.to_rfc3339())
1088 .execute(pool)
1089 .await
1090 .unwrap();
1091 }
1092
1093 #[tokio::test]
1094 async fn unsatisfied_gates_manual_confirm_requires_fresh_confirmation() {
1095 // A confirmation only satisfies the gate if it post-dates the version's
1096 // current landing on the tier (burn_in_started_at). A stale confirm left
1097 // over from before a rollback + rollback-forward must NOT wave it through.
1098 let pool = fresh_pool().await;
1099 seed(&pool, "a", "0.8.12").await;
1100 let landed = chrono::Utc::now();
1101 sqlx::query("UPDATE tier_state SET burn_in_started_at = ? WHERE tier = 'a'")
1102 .bind(landed.to_rfc3339())
1103 .execute(&pool)
1104 .await
1105 .unwrap();
1106
1107 // Stale confirmation (recorded before this landing) -> unsatisfied.
1108 insert_confirm(&pool, "a", "0.8.12", landed - chrono::Duration::hours(1)).await;
1109 let pending = unsatisfied_gates(
1110 &pool,
1111 &tid("a"),
1112 &[Gate::ManualConfirm],
1113 "0.8.12",
1114 None,
1115 false,
1116 )
1117 .await
1118 .unwrap();
1119 assert_eq!(
1120 pending,
1121 vec!["manual_confirm".to_string()],
1122 "stale confirm must not satisfy"
1123 );
1124
1125 // Fresh confirmation (after this landing) -> satisfied.
1126 insert_confirm(&pool, "a", "0.8.12", landed + chrono::Duration::minutes(5)).await;
1127 let pending = unsatisfied_gates(
1128 &pool,
1129 &tid("a"),
1130 &[Gate::ManualConfirm],
1131 "0.8.12",
1132 None,
1133 false,
1134 )
1135 .await
1136 .unwrap();
1137 assert!(pending.is_empty(), "fresh confirm satisfies");
1138 }
1139
1140 #[tokio::test]
1141 async fn unsatisfied_gates_manual_confirm_fails_closed_without_baseline() {
1142 // No landing clock (burn_in_started_at NULL) -> a passed confirm row is
1143 // not provably fresh, so fail closed and require a new confirmation.
1144 let pool = fresh_pool().await;
1145 seed(&pool, "a", "0.8.12").await; // leaves burn_in_started_at NULL
1146 insert_confirm(&pool, "a", "0.8.12", chrono::Utc::now()).await;
1147 let pending = unsatisfied_gates(
1148 &pool,
1149 &tid("a"),
1150 &[Gate::ManualConfirm],
1151 "0.8.12",
1152 None,
1153 false,
1154 )
1155 .await
1156 .unwrap();
1157 assert_eq!(
1158 pending,
1159 vec!["manual_confirm".to_string()],
1160 "no baseline -> fail closed"
1161 );
1162 }
1163
1164 #[tokio::test]
1165 async fn unsatisfied_gates_hotfix_skips_only_burn_in() {
1166 // burn_in is evaluated live (no clock started -> not elapsed); cargo_test
1167 // has a failing row. Normal: both unsatisfied, in configured order.
1168 // hotfix: burn_in suppressed, cargo_test still flagged. Lock the semantic
1169 // so a future change doesn't widen the hotfix bypass.
1170 let pool = fresh_pool().await;
1171 seed(&pool, "a", "0.8.12").await;
1172 insert_gate(&pool, "a", "0.8.12", "cargo_test", 0).await;
1173 let gates = [Gate::BurnIn { hours: 48 }, Gate::CargoTest];
1174
1175 let normal = unsatisfied_gates(&pool, &tid("a"), &gates, "0.8.12", None, false)
1176 .await
1177 .unwrap();
1178 assert_eq!(
1179 normal,
1180 vec!["burn_in".to_string(), "cargo_test".to_string()]
1181 );
1182
1183 let with_hotfix = unsatisfied_gates(&pool, &tid("a"), &gates, "0.8.12", None, true)
1184 .await
1185 .unwrap();
1186 assert_eq!(with_hotfix, vec!["cargo_test".to_string()]);
1187 }
1188
1189 #[tokio::test]
1190 async fn unsatisfied_gates_burn_in_passes_when_window_elapsed() {
1191 // A burn-in clock started far enough in the past satisfies the gate
1192 // live — no gate_runs row needed.
1193 let pool = fresh_pool().await;
1194 seed(&pool, "a", "0.8.12").await;
1195 sqlx::query("UPDATE tier_state SET burn_in_started_at = ? WHERE tier = 'a'")
1196 .bind((chrono::Utc::now() - chrono::Duration::hours(50)).to_rfc3339())
1197 .execute(&pool)
1198 .await
1199 .unwrap();
1200 let pending = unsatisfied_gates(
1201 &pool,
1202 &tid("a"),
1203 &[Gate::BurnIn { hours: 48 }],
1204 "0.8.12",
1205 None,
1206 false,
1207 )
1208 .await
1209 .unwrap();
1210 assert!(pending.is_empty(), "50h elapsed satisfies a 48h burn-in");
1211 }
1212
1213 #[tokio::test]
1214 async fn unsatisfied_gates_ignores_other_tiers_and_versions() {
1215 let pool = fresh_pool().await;
1216 seed(&pool, "host", "0.8.12").await;
1217 seed(&pool, "host", "0.8.11").await;
1218 seed(&pool, "a", "0.8.12").await;
1219 // Mark host/0.8.12 cargo_test failing, but unrelated tiers/versions
1220 // shouldn't pollute the query.
1221 insert_gate(&pool, "host", "0.8.12", "cargo_test", 0).await;
1222 insert_gate(&pool, "a", "0.8.12", "cargo_test", 0).await;
1223 insert_gate(&pool, "host", "0.8.11", "cargo_test", 0).await;
1224
1225 let pending = unsatisfied_gates(
1226 &pool,
1227 &tid("host"),
1228 &[Gate::CargoTest],
1229 "0.8.12",
1230 None,
1231 false,
1232 )
1233 .await
1234 .unwrap();
1235 assert_eq!(pending, vec!["cargo_test".to_string()]);
1236 }
1237
1238 #[tokio::test]
1239 async fn unsatisfied_gates_null_status_is_treated_as_failing() {
1240 // An in-flight gate (started_at set, finished_at + status NULL)
1241 // should NOT be treated as green. Otherwise a race could promote
1242 // before the gate concludes.
1243 let pool = fresh_pool().await;
1244 seed(&pool, "host", "0.8.12").await;
1245 sqlx::query(
1246 "INSERT INTO gate_runs (version, tier, gate_kind, started_at) \
1247 VALUES ('0.8.12', 'host', 'cargo_test', datetime('now'))",
1248 )
1249 .execute(&pool)
1250 .await
1251 .unwrap();
1252
1253 let pending = unsatisfied_gates(
1254 &pool,
1255 &tid("host"),
1256 &[Gate::CargoTest],
1257 "0.8.12",
1258 None,
1259 false,
1260 )
1261 .await
1262 .unwrap();
1263 assert_eq!(pending, vec!["cargo_test".to_string()]);
1264 }
1265
1266 // ---- /confirm/{tier} ----
1267
1268 #[tokio::test]
1269 async fn confirm_rejects_when_tier_has_no_current_version() {
1270 // tier_state.a.current_version is NULL by default. /confirm has
1271 // nothing to confirm against → GateBlocked (400).
1272 let state = test_state().await;
1273 let app = router(state.clone());
1274 let resp = app
1275 .oneshot(
1276 Request::builder()
1277 .method("POST")
1278 .uri("/confirm/a")
1279 .body(Body::empty())
1280 .unwrap(),
1281 )
1282 .await
1283 .unwrap();
1284 assert_eq!(resp.status(), StatusCode::CONFLICT);
1285 let body = body_string(resp).await;
1286 assert!(body.contains("no current_version"), "got: {body}");
1287 }
1288
1289 #[tokio::test]
1290 async fn confirm_accepts_when_current_version_set_and_inserts_row() {
1291 let state = test_state().await;
1292 // Seed a version + advance tier a's state to it.
1293 sqlx::query("INSERT INTO versions (version, git_sha, built_at, artifact_path) VALUES ('0.8.12','sha',datetime('now'),'/tmp/x')")
1294 .execute(&state.pool).await.unwrap();
1295 sqlx::query("UPDATE tier_state SET current_version = '0.8.12' WHERE tier = 'a'")
1296 .execute(&state.pool)
1297 .await
1298 .unwrap();
1299
1300 let app = router(state.clone());
1301 let resp = app
1302 .oneshot(
1303 Request::builder()
1304 .method("POST")
1305 .uri("/confirm/a")
1306 .body(Body::empty())
1307 .unwrap(),
1308 )
1309 .await
1310 .unwrap();
1311 assert_eq!(resp.status(), StatusCode::OK);
1312 let body = body_string(resp).await;
1313 assert!(body.contains("\"tier\":\"a\""));
1314 assert!(body.contains("\"version\":\"0.8.12\""));
1315
1316 // A passing gate_runs row was inserted.
1317 let count: (i64,) = sqlx::query_as(
1318 "SELECT COUNT(*) FROM gate_runs WHERE tier='a' AND gate_kind='manual_confirm' AND status='passed'",
1319 )
1320 .fetch_one(&state.pool)
1321 .await
1322 .unwrap();
1323 assert_eq!(count.0, 1);
1324 }
1325
1326 #[tokio::test]
1327 async fn confirm_404s_for_unknown_tier() {
1328 let state = test_state().await;
1329 let app = router(state);
1330 let resp = app
1331 .oneshot(
1332 Request::builder()
1333 .method("POST")
1334 .uri("/confirm/zzzz")
1335 .body(Body::empty())
1336 .unwrap(),
1337 )
1338 .await
1339 .unwrap();
1340 assert_eq!(resp.status(), StatusCode::NOT_FOUND);
1341 }
1342
1343 #[tokio::test]
1344 async fn get_run_404s_for_unknown_id() {
1345 let state = test_state().await;
1346 let app = router(state);
1347 let resp = app
1348 .oneshot(
1349 Request::builder()
1350 .uri("/runs/999")
1351 .body(Body::empty())
1352 .unwrap(),
1353 )
1354 .await
1355 .unwrap();
1356 assert_eq!(resp.status(), StatusCode::NOT_FOUND);
1357 }
1358
1359 #[tokio::test]
1360 async fn get_run_returns_view_with_gates() {
1361 let state = test_state().await;
1362 // A run that reached version 0.10.2 and ran two host gates (one red).
1363 let run_id = crate::runs::create(&state.pool, "abc1234def")
1364 .await
1365 .unwrap();
1366 let ver: crate::domain::Version = "0.10.2".parse().unwrap();
1367 seed(&state.pool, "host", "0.10.2").await;
1368 crate::runs::set_version(&state.pool, run_id, &ver)
1369 .await
1370 .unwrap();
1371 insert_gate(&state.pool, "host", "0.10.2", "cargo_test", 0).await;
1372 insert_gate(&state.pool, "host", "0.10.2", "boot_smoke", 1).await;
1373
1374 let app = router(state);
1375 let resp = app
1376 .oneshot(
1377 Request::builder()
1378 .uri(format!("/runs/{}", run_id.0))
1379 .body(Body::empty())
1380 .unwrap(),
1381 )
1382 .await
1383 .unwrap();
1384 assert_eq!(resp.status(), StatusCode::OK);
1385 let v: serde_json::Value = serde_json::from_str(&body_string(resp).await).unwrap();
1386 assert_eq!(v["run_id"], run_id.0);
1387 assert_eq!(v["sha"], "abc1234def");
1388 assert_eq!(v["version"], "0.10.2");
1389 assert_eq!(v["result"], "building");
1390 // Both host gates surface, latest-per-kind, alphabetized by kind.
1391 assert_eq!(v["gates"].as_array().unwrap().len(), 2);
1392 assert_eq!(v["gates"][0]["kind"], "boot_smoke");
1393 assert_eq!(v["gates"][0]["status"], "passed");
1394 assert_eq!(v["gates"][1]["kind"], "cargo_test");
1395 assert_eq!(v["gates"][1]["status"], "failed");
1396 }
1397
1398 #[tokio::test]
1399 async fn get_run_wait_returns_immediately_when_settled() {
1400 let state = test_state().await;
1401 let run_id = crate::runs::create(&state.pool, "abc1234def")
1402 .await
1403 .unwrap();
1404 crate::runs::mark_passed(&state.pool, run_id).await.unwrap();
1405
1406 let app = router(state);
1407 // Generous timeout, but an already-settled run must not wait for it.
1408 let resp = app
1409 .oneshot(
1410 Request::builder()
1411 .uri(format!("/runs/{}/wait?timeout_ms=60000", run_id.0))
1412 .body(Body::empty())
1413 .unwrap(),
1414 )
1415 .await
1416 .unwrap();
1417 assert_eq!(resp.status(), StatusCode::OK);
1418 let v: serde_json::Value = serde_json::from_str(&body_string(resp).await).unwrap();
1419 assert_eq!(v["result"], "passed");
1420 }
1421
1422 #[tokio::test]
1423 async fn get_run_wait_returns_building_at_timeout() {
1424 let state = test_state().await;
1425 let run_id = crate::runs::create(&state.pool, "abc1234def")
1426 .await
1427 .unwrap();
1428
1429 let app = router(state);
1430 // timeout_ms=0 → deadline is now → the first poll returns the
1431 // still-building run rather than blocking.
1432 let resp = app
1433 .oneshot(
1434 Request::builder()
1435 .uri(format!("/runs/{}/wait?timeout_ms=0", run_id.0))
1436 .body(Body::empty())
1437 .unwrap(),
1438 )
1439 .await
1440 .unwrap();
1441 assert_eq!(resp.status(), StatusCode::OK);
1442 let v: serde_json::Value = serde_json::from_str(&body_string(resp).await).unwrap();
1443 assert_eq!(v["result"], "building");
1444 }
1445
1446 #[tokio::test]
1447 async fn get_run_wait_404s_for_unknown_id() {
1448 let state = test_state().await;
1449 let app = router(state);
1450 let resp = app
1451 .oneshot(
1452 Request::builder()
1453 .uri("/runs/999/wait?timeout_ms=0")
1454 .body(Body::empty())
1455 .unwrap(),
1456 )
1457 .await
1458 .unwrap();
1459 assert_eq!(resp.status(), StatusCode::NOT_FOUND);
1460 }
1461
1462 #[test]
1463 fn self_update_unit_maps_sha_to_instance() {
1464 let sha = crate::domain::GitSha::parse("abc1234def5678").unwrap();
1465 assert_eq!(
1466 self_update_unit(&sha),
1467 "sando-update@abc1234def5678.service"
1468 );
1469 }
1470
1471 #[tokio::test]
1472 async fn self_update_rejects_bad_sha_with_400() {
1473 // A malformed sha is a client error and must be rejected *before* any
1474 // privileged unit is triggered (so this test never shells out).
1475 let state = test_state().await;
1476 let app = router(state);
1477 let resp = app
1478 .oneshot(
1479 Request::builder()
1480 .method("POST")
1481 .uri("/self-update")
1482 .header("Content-Type", "application/json")
1483 .body(Body::from(r#"{"sha":"not-a-sha!"}"#))
1484 .unwrap(),
1485 )
1486 .await
1487 .unwrap();
1488 assert_eq!(resp.status(), StatusCode::BAD_REQUEST);
1489 }
1490
1491 // ---- /promote/{tier} default-version resolution ----
1492
1493 #[tokio::test]
1494 async fn promote_to_first_tier_is_rejected() {
1495 // tier 0 is host — you /rebuild, not /promote.
1496 let state = test_state().await;
1497 let app = router(state);
1498 let resp = app
1499 .oneshot(
1500 Request::builder()
1501 .method("POST")
1502 .uri("/promote/host")
1503 .body(Body::empty())
1504 .unwrap(),
1505 )
1506 .await
1507 .unwrap();
1508 assert_eq!(resp.status(), StatusCode::CONFLICT);
1509 let body = body_string(resp).await;
1510 assert!(
1511 body.contains("cannot /promote to the first tier"),
1512 "got: {body}"
1513 );
1514 }
1515
1516 #[tokio::test]
1517 async fn promote_without_body_and_no_predecessor_version_errors() {
1518 // tier a has no body version supplied AND its predecessor mm has
1519 // current_version=NULL. Should fail before any deploy.
1520 let state = test_state().await;
1521 let app = router(state);
1522 let resp = app
1523 .oneshot(
1524 Request::builder()
1525 .method("POST")
1526 .uri("/promote/a")
1527 .body(Body::empty())
1528 .unwrap(),
1529 )
1530 .await
1531 .unwrap();
1532 assert_eq!(resp.status(), StatusCode::CONFLICT);
1533 let body = body_string(resp).await;
1534 assert!(
1535 body.contains("no version specified") || body.contains("no current_version"),
1536 "got: {body}"
1537 );
1538 }
1539
1540 #[tokio::test]
1541 async fn promote_blocked_when_predecessor_gate_never_ran() {
1542 // End-to-end CF1: the host tier configures boot_smoke but it never ran
1543 // (no gate_runs row). Promoting host -> a must be GateBlocked, citing the
1544 // unsatisfied gate, instead of waving through on zero evidence. A real
1545 // `versions` row is present so the ONLY thing that can block is the gate.
1546 let pool = fresh_pool().await;
1547 for (i, name) in ["host", "a"].iter().enumerate() {
1548 sqlx::query(
1549 "INSERT INTO tiers (name, ord, provisioned, canary) VALUES (?, ?, 1, 'sequential')",
1550 )
1551 .bind(name)
1552 .bind(i as i64)
1553 .execute(&pool)
1554 .await
1555 .unwrap();
1556 sqlx::query("INSERT INTO tier_state (tier) VALUES (?)")
1557 .bind(name)
1558 .execute(&pool)
1559 .await
1560 .unwrap();
1561 }
1562 let mut topo = test_topo();
1563 topo.tiers[0].gates = vec![Gate::BootSmoke]; // host configures a gate...
1564 let executors = Arc::new(crate::state::build_executors(&topo));
1565 let state = AppState {
1566 pool,
1567 topo: Arc::new(topo),
1568 cfg: Arc::new(test_cfg()),
1569 active_build: Arc::new(tokio::sync::Mutex::new(None)),
1570 deploy_lock: Arc::new(tokio::sync::Mutex::new(())),
1571 events: crate::events::channel(),
1572 executors,
1573 api_token: None,
1574 };
1575 sqlx::query("INSERT INTO versions (version, git_sha, built_at, artifact_path) VALUES ('0.8.12','sha',datetime('now'),'/tmp/x')")
1576 .execute(&state.pool).await.unwrap();
1577 sqlx::query("UPDATE tier_state SET current_version = '0.8.12' WHERE tier = 'host'")
1578 .execute(&state.pool)
1579 .await
1580 .unwrap();
1581
1582 let app = router(state);
1583 let resp = app
1584 .oneshot(
1585 Request::builder()
1586 .method("POST")
1587 .uri("/promote/a")
1588 .body(Body::empty())
1589 .unwrap(),
1590 )
1591 .await
1592 .unwrap();
1593 assert_eq!(resp.status(), StatusCode::CONFLICT);
1594 let body = body_string(resp).await;
1595 assert!(
1596 body.contains("boot_smoke"),
1597 "expected boot_smoke to block; got: {body}"
1598 );
1599 }
1600
1601 #[tokio::test]
1602 async fn migration_bearing_promote_requires_fresh_confirm() {
1603 // The predecessor (host) configures NO gates, so nothing would normally
1604 // block host -> a. A `bears_migration` promote must still be blocked on a
1605 // fresh `manual_confirm` it does not have: rollback restores the binary
1606 // only, so the one-way advance needs a conscious operator sign-off.
1607 let pool = fresh_pool().await;
1608 for (i, name) in ["host", "a"].iter().enumerate() {
1609 sqlx::query(
1610 "INSERT INTO tiers (name, ord, provisioned, canary) VALUES (?, ?, 1, 'sequential')",
1611 )
1612 .bind(name)
1613 .bind(i as i64)
1614 .execute(&pool)
1615 .await
1616 .unwrap();
1617 sqlx::query("INSERT INTO tier_state (tier) VALUES (?)")
1618 .bind(name)
1619 .execute(&pool)
1620 .await
1621 .unwrap();
1622 }
1623 let mut topo = test_topo();
1624 topo.tiers[0].gates = vec![]; // host configures no gates at all
1625 let executors = Arc::new(crate::state::build_executors(&topo));
1626 let state = AppState {
1627 pool,
1628 topo: Arc::new(topo),
1629 cfg: Arc::new(test_cfg()),
1630 active_build: Arc::new(tokio::sync::Mutex::new(None)),
1631 deploy_lock: Arc::new(tokio::sync::Mutex::new(())),
1632 events: crate::events::channel(),
1633 executors,
1634 api_token: None,
1635 };
1636 sqlx::query("INSERT INTO versions (version, git_sha, built_at, artifact_path) VALUES ('0.8.12','sha',datetime('now'),'/tmp/x')")
1637 .execute(&state.pool).await.unwrap();
1638 sqlx::query("UPDATE tier_state SET current_version = '0.8.12' WHERE tier = 'host'")
1639 .execute(&state.pool)
1640 .await
1641 .unwrap();
1642
1643 let app = router(state);
1644 let resp = app
1645 .oneshot(
1646 Request::builder()
1647 .method("POST")
1648 .uri("/promote/a")
1649 .header("content-type", "application/json")
1650 .body(Body::from(r#"{"bears_migration": true}"#))
1651 .unwrap(),
1652 )
1653 .await
1654 .unwrap();
1655 assert_eq!(resp.status(), StatusCode::CONFLICT);
1656 let body = body_string(resp).await;
1657 assert!(
1658 body.contains("manual_confirm"),
1659 "expected the migration promote to block on manual_confirm; got: {body}"
1660 );
1661 }
1662
1663 #[tokio::test]
1664 async fn tier_state_advance_is_atomic_previous_from_old_current() {
1665 // CF3: the promote advance is a single UPDATE where previous_version is
1666 // set from the row's *old* current_version (SQLite evaluates RHS against
1667 // the original row). No read-modify-write to lose under concurrency.
1668 let pool = fresh_pool().await;
1669 seed(&pool, "a", "1.0.0").await;
1670 // current_version FKs into versions, so the target must exist too.
1671 sqlx::query("INSERT INTO versions (version, git_sha, built_at, artifact_path) VALUES ('2.0.0','sha',datetime('now'),'/tmp/x')")
1672 .execute(&pool).await.unwrap();
1673 sqlx::query("UPDATE tier_state SET current_version = '1.0.0' WHERE tier = 'a'")
1674 .execute(&pool)
1675 .await
1676 .unwrap();
1677
1678 // Exercise the sealed forward-advance primitive itself — the same op
1679 // /promote and the host build path both call (S1), not a copy of its SQL.
1680 let v = crate::domain::Version::parse("2.0.0").unwrap();
1681 crate::runs::advance_tier(&pool, "a", &v, None)
1682 .await
1683 .unwrap();
1684
1685 let (cur, prev): (Option<String>, Option<String>) = sqlx::query_as(
1686 "SELECT current_version, previous_version FROM tier_state WHERE tier = 'a'",
1687 )
1688 .fetch_one(&pool)
1689 .await
1690 .unwrap();
1691 assert_eq!(cur.as_deref(), Some("2.0.0"));
1692 assert_eq!(
1693 prev.as_deref(),
1694 Some("1.0.0"),
1695 "previous = the pre-update current, atomically"
1696 );
1697 }
1698
1699 #[tokio::test]
1700 async fn canary_rollback_restores_deployed_nodes_to_previous_version() {
1701 use crate::topology::{Node, default_actuate, default_observe};
1702 let tmp = tempfile::tempdir().unwrap();
1703
1704 // Two local nodes, each pre-seeded as if a promote had flipped them to
1705 // 2.0.0 (current -> releases/2.0.0), with the prior 1.0.0 still on disk.
1706 let mut nodes = Vec::new();
1707 for name in ["n1", "n2"] {
1708 let rr = tmp.path().join(name);
1709 for v in ["1.0.0", "2.0.0"] {
1710 tokio::fs::create_dir_all(rr.join("releases").join(v))
1711 .await
1712 .unwrap();
1713 }
1714 tokio::fs::symlink("releases/2.0.0", rr.join("current"))
1715 .await
1716 .unwrap();
1717 nodes.push(Node {
1718 name: name.into(),
1719 ssh_target: "local".into(),
1720 release_root: rr.to_string_lossy().into_owned(),
1721 service_name: "x.service".into(),
1722 health_url: None,
1723 config_check_env_file: None,
1724 actuate: default_actuate(),
1725 observe: default_observe(),
1726 companions: Vec::new(),
1727 });
1728 }
1729
1730 let mut state = test_state().await;
1731 // The rollback target needs a versions row. The release dir name comes
1732 // from the artifact_path's parent (legacy layout: releases/<version>).
1733 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')")
1734 .execute(&state.pool).await.unwrap();
1735 let execs: crate::state::ExecutorMap = nodes
1736 .iter()
1737 .map(|n| (n.name.clone(), crate::state::build_executor(n)))
1738 .collect();
1739 state.executors = std::sync::Arc::new(execs);
1740
1741 let refs: Vec<&Node> = nodes.iter().collect();
1742 let restored = rollback_deployed_nodes(&state, &tid("a"), &refs, "1.0.0").await;
1743 assert_eq!(restored, 2, "both deployed nodes should be restored");
1744
1745 for n in &nodes {
1746 let cur = tokio::fs::read_link(std::path::Path::new(&n.release_root).join("current"))
1747 .await
1748 .unwrap();
1749 assert_eq!(
1750 cur.to_string_lossy(),
1751 "releases/1.0.0",
1752 "node {} rolled back",
1753 n.name
1754 );
1755 }
1756 }
1757
1758 #[tokio::test]
1759 async fn canary_rollback_is_noop_without_a_previous_artifact() {
1760 use crate::topology::{Node, default_actuate, default_observe};
1761 let tmp = tempfile::tempdir().unwrap();
1762 let rr = tmp.path().join("n1");
1763 tokio::fs::create_dir_all(&rr).await.unwrap();
1764 let node = Node {
1765 name: "n1".into(),
1766 ssh_target: "local".into(),
1767 release_root: rr.to_string_lossy().into_owned(),
1768 service_name: "x.service".into(),
1769 health_url: None,
1770 config_check_env_file: None,
1771 actuate: default_actuate(),
1772 observe: default_observe(),
1773 companions: Vec::new(),
1774 };
1775 let state = test_state().await; // no versions row for "9.9.9"
1776 let restored = rollback_deployed_nodes(&state, &tid("a"), &[&node], "9.9.9").await;
1777 assert_eq!(
1778 restored, 0,
1779 "no artifact to roll back to -> nothing restored, no panic"
1780 );
1781 }
1782
1783 // ---- FleetFake: a multi-node promote across recorded fake executors ----
1784 //
1785 // The route-level promote tests until now ran a single local node against a
1786 // real LocalExec, so the sequential-canary fan-out, the cross-node deploy
1787 // ordering, and the mid-canary rollback of already-flipped nodes had no
1788 // coverage. FleetFake records every deploy op (tagged by node) into one
1789 // shared log so ordering is visible across the fleet, and can fail any op
1790 // whose shell script or rsync target contains a marker — used to fail a
1791 // node's forward deploy of the new version while its rollback to the prior
1792 // version (a different `releases/<v>` path) still succeeds.
1793
1794 struct FleetFake {
1795 tag: String,
1796 caps: CapabilitySet,
1797 log: Arc<StdMutex<Vec<String>>>,
1798 fail_if_contains: Option<String>,
1799 }
1800
1801 impl FleetFake {
1802 fn fails(&self, text: &str) -> bool {
1803 self.fail_if_contains
1804 .as_deref()
1805 .is_some_and(|m| text.contains(m))
1806 }
1807 }
1808
1809 #[async_trait]
1810 impl Executor for FleetFake {
1811 async fn run_streaming(
1812 &self,
1813 step: &Step,
1814 _sink: &mut dyn LogSink,
1815 ) -> anyhow::Result<RunOutput> {
1816 let script = step.argv.last().cloned().unwrap_or_default();
1817 self.log
1818 .lock()
1819 .unwrap()
1820 .push(format!("{}:run:{script}", self.tag));
1821 Ok(RunOutput {
1822 status: std::process::ExitStatus::from_raw(if self.fails(&script) {
1823 1 << 8
1824 } else {
1825 0
1826 }),
1827 stdout: Vec::new(),
1828 stderr: Vec::new(),
1829 })
1830 }
1831 async fn pull_file(
1832 &self,
1833 _r: &std::path::Path,
1834 _l: &std::path::Path,
1835 _o: &SyncOpts,
1836 ) -> anyhow::Result<()> {
1837 Ok(())
1838 }
1839 async fn pull_dir(
1840 &self,
1841 _r: &std::path::Path,
1842 _l: &std::path::Path,
1843 _o: &SyncOpts,
1844 ) -> anyhow::Result<()> {
1845 Ok(())
1846 }
1847 async fn pull_glob(
1848 &self,
1849 _g: &str,
1850 _l: &std::path::Path,
1851 _o: &SyncOpts,
1852 ) -> anyhow::Result<()> {
1853 Ok(())
1854 }
1855 async fn push_dir(
1856 &self,
1857 _local: &std::path::Path,
1858 remote: &std::path::Path,
1859 _o: &SyncOpts,
1860 ) -> anyhow::Result<()> {
1861 let dst = remote.display().to_string();
1862 self.log
1863 .lock()
1864 .unwrap()
1865 .push(format!("{}:push:{dst}", self.tag));
1866 if self.fails(&dst) {
1867 anyhow::bail!("fake rsync failure on {}", self.tag);
1868 }
1869 Ok(())
1870 }
1871 fn capabilities(&self) -> &CapabilitySet {
1872 &self.caps
1873 }
1874 }
1875
1876 /// Rebuild tier "a" with `names` as remote fake nodes sharing one op log,
1877 /// seed the version/tier_state prerequisites for a promote of 3.0.0 up from
1878 /// `host` (tier "a" starts on 2.0.0 with 1.0.0 behind it), and optionally
1879 /// make `fail_node` fail any op containing `fail_marker`. Returns the state
1880 /// and the shared log.
1881 async fn fleet_fixture(
1882 names: &[&str],
1883 fail_node: Option<&str>,
1884 fail_marker: &str,
1885 ) -> (AppState, Arc<StdMutex<Vec<String>>>) {
1886 use crate::topology::{default_actuate, default_observe};
1887 let mut state = test_state().await;
1888 let log = Arc::new(StdMutex::new(Vec::<String>::new()));
1889
1890 let nodes: Vec<Node> = names
1891 .iter()
1892 .map(|name| Node {
1893 name: (*name).into(),
1894 ssh_target: format!("deploy@{name}"),
1895 release_root: format!("/tmp/fleet/{name}"),
1896 service_name: "makenotwork.service".into(),
1897 health_url: None,
1898 config_check_env_file: None,
1899 actuate: default_actuate(),
1900 observe: default_observe(),
1901 companions: Vec::new(),
1902 })
1903 .collect();
1904
1905 let mut topo = (*state.topo).clone();
1906 topo.tiers[1].nodes = nodes.clone();
1907 // Drop the tier's post-deploy gate: the deploy fan-out is the subject
1908 // here, and node_health would need its own probe wiring.
1909 topo.tiers[1].gates = vec![];
1910 state.topo = Arc::new(topo);
1911
1912 let execs: crate::state::ExecutorMap = nodes
1913 .iter()
1914 .map(|n| {
1915 let nm = n.name.to_string();
1916 let fail = fail_node == Some(nm.as_str());
1917 let exec: Arc<dyn Executor> = Arc::new(FleetFake {
1918 tag: nm,
1919 caps: CapabilitySet::from_tokens(["deploy", "restart"], ["health"]),
1920 log: log.clone(),
1921 fail_if_contains: fail.then(|| fail_marker.to_string()),
1922 });
1923 (n.name.clone(), exec)
1924 })
1925 .collect();
1926 state.executors = Arc::new(execs);
1927
1928 for n in &nodes {
1929 // deploys.node FKs into `nodes`.
1930 sqlx::query(
1931 "INSERT INTO nodes (name, tier, ssh_target, release_root) VALUES (?, 'a', ?, ?)",
1932 )
1933 .bind(&n.name)
1934 .bind(&n.ssh_target)
1935 .bind(&n.release_root)
1936 .execute(&state.pool)
1937 .await
1938 .unwrap();
1939 }
1940 for v in ["1.0.0", "2.0.0", "3.0.0"] {
1941 // Legacy (pre-identity) artifact_path is `releases/<version>/<bin>`,
1942 // so the release dir the node mirrors is named for the version. The
1943 // fixture reflects that real layout (parent basename == version).
1944 sqlx::query("INSERT INTO versions (version, git_sha, built_at, artifact_path) VALUES (?, 'sha', datetime('now'), ?)")
1945 .bind(v).bind(format!("/tmp/staged/releases/{v}/makenotwork")).execute(&state.pool).await.unwrap();
1946 }
1947 sqlx::query("UPDATE tier_state SET current_version = '3.0.0' WHERE tier = 'host'")
1948 .execute(&state.pool)
1949 .await
1950 .unwrap();
1951 sqlx::query(
1952 "UPDATE tier_state SET current_version = '2.0.0', previous_version = '1.0.0' WHERE tier = 'a'",
1953 )
1954 .execute(&state.pool)
1955 .await
1956 .unwrap();
1957 (state, log)
1958 }
1959
1960 /// Index of the first logged op belonging to `node` (panics if the node was
1961 /// never touched — the message names it).
1962 fn first_touch(log: &[String], node: &str) -> usize {
1963 let prefix = format!("{node}:");
1964 log.iter()
1965 .position(|e| e.starts_with(&prefix))
1966 .unwrap_or_else(|| panic!("node {node:?} was never deployed to; log: {log:#?}"))
1967 }
1968
1969 #[tokio::test]
1970 async fn promote_deploys_every_node_in_tier_order_and_advances() {
1971 let (state, log) = fleet_fixture(&["a1", "a2", "a3"], None, "").await;
1972 let pool = state.pool.clone();
1973
1974 let Json(body) = promote_inner(state, "a".into(), PromoteBody::default())
1975 .await
1976 .expect("all nodes deploy, so the promote succeeds");
1977 assert_eq!(
1978 body["nodes_deployed"],
1979 serde_json::json!(["a1", "a2", "a3"]),
1980 "the response names every node the promote reached",
1981 );
1982
1983 let (cur, prev) = tier_versions(&pool, "a").await;
1984 assert_eq!(cur.as_deref(), Some("3.0.0"));
1985 assert_eq!(prev.as_deref(), Some("2.0.0"));
1986
1987 // Sequential canary: a1 is fully touched before a2, a2 before a3.
1988 let log = log.lock().unwrap().clone();
1989 assert!(
1990 first_touch(&log, "a1") < first_touch(&log, "a2")
1991 && first_touch(&log, "a2") < first_touch(&log, "a3"),
1992 "nodes must deploy in tier order: {log:#?}",
1993 );
1994
1995 // Every node has a green deploy row for the promoted version.
1996 let ok: i64 = sqlx::query_scalar(
1997 "SELECT COUNT(*) FROM deploys WHERE version = '3.0.0' AND outcome = 'ok'",
1998 )
1999 .fetch_one(&pool)
2000 .await
2001 .unwrap();
2002 assert_eq!(ok, 3, "one ok deploy row per node");
2003 }
2004
2005 #[tokio::test]
2006 async fn a_mid_canary_deploy_failure_rolls_touched_nodes_back_and_does_not_advance() {
2007 // a2 fails its forward deploy of 3.0.0; a1 was already flipped, a3 is
2008 // never reached. The touched nodes (a1, a2) roll back to 2.0.0 — their
2009 // rollback ops target `releases/2.0.0`, which the marker does not match —
2010 // and tier_state must NOT advance.
2011 let (state, log) = fleet_fixture(&["a1", "a2", "a3"], Some("a2"), "releases/3.0.0").await;
2012 let pool = state.pool.clone();
2013
2014 let err = promote_inner(state, "a".into(), PromoteBody::default())
2015 .await
2016 .expect_err("a mid-canary deploy failure must fail the promote");
2017 assert!(
2018 matches!(err, crate::error::Error::Other(_)),
2019 "a deploy failure propagates as Other, got: {err:?}",
2020 );
2021
2022 // tier_state untouched: the failure returns before advance_tier.
2023 let (cur, prev) = tier_versions(&pool, "a").await;
2024 assert_eq!(
2025 cur.as_deref(),
2026 Some("2.0.0"),
2027 "a failed rollout must not advance"
2028 );
2029 assert_eq!(prev.as_deref(), Some("1.0.0"));
2030
2031 let log = log.lock().unwrap().clone();
2032 // a3 sits after the failed a2 in the sequence and is never touched.
2033 assert!(
2034 !log.iter().any(|e| e.starts_with("a3:")),
2035 "nodes after the failure must not be deployed to: {log:#?}",
2036 );
2037 // Both touched nodes were rolled back to the prior version.
2038 for n in ["a1", "a2"] {
2039 assert!(
2040 log.iter()
2041 .any(|e| e.starts_with(&format!("{n}:")) && e.contains("releases/2.0.0")),
2042 "touched node {n} must be restored to 2.0.0: {log:#?}",
2043 );
2044 }
2045
2046 // The forward attempt is on the record: a1 ok, a2 failed.
2047 let a1: String = sqlx::query_scalar(
2048 "SELECT outcome FROM deploys WHERE version = '3.0.0' AND node = 'a1'",
2049 )
2050 .fetch_one(&pool)
2051 .await
2052 .unwrap();
2053 assert_eq!(a1, "ok");
2054 let a2: String = sqlx::query_scalar(
2055 "SELECT outcome FROM deploys WHERE version = '3.0.0' AND node = 'a2'",
2056 )
2057 .fetch_one(&pool)
2058 .await
2059 .unwrap();
2060 assert_eq!(a2, "failed");
2061
2062 // Both touched nodes restored => the tier is consistent on 2.0.0, so the
2063 // partial flag is cleared, not set.
2064 let reason: Option<String> =
2065 sqlx::query_scalar("SELECT partial_reason FROM tier_state WHERE tier = 'a'")
2066 .fetch_one(&pool)
2067 .await
2068 .unwrap();
2069 assert_eq!(
2070 reason, None,
2071 "a fully-restored canary leaves the tier consistent, not partial",
2072 );
2073 }
2074
2075 /// Build a one-node tier "a" on a tempdir release root, pre-seeded as if a
2076 /// promote had flipped it to `current` with `prev` still staged on disk.
2077 /// Returns the state (topology rewired to the tempdir node) and the tempdir,
2078 /// which the caller must keep alive.
2079 async fn rollback_fixture(prev: &str, current: &str) -> (AppState, tempfile::TempDir) {
2080 use crate::topology::{Node, default_actuate, default_observe};
2081 let tmp = tempfile::tempdir().unwrap();
2082 let rr = tmp.path().join("a-local");
2083 for v in [prev, current] {
2084 tokio::fs::create_dir_all(rr.join("releases").join(v))
2085 .await
2086 .unwrap();
2087 }
2088 tokio::fs::symlink(format!("releases/{current}"), rr.join("current"))
2089 .await
2090 .unwrap();
2091 let node = Node {
2092 name: "a-local".into(),
2093 ssh_target: "local".into(),
2094 release_root: rr.to_string_lossy().into_owned(),
2095 service_name: "x.service".into(),
2096 health_url: None,
2097 config_check_env_file: None,
2098 actuate: default_actuate(),
2099 observe: default_observe(),
2100 companions: Vec::new(),
2101 };
2102
2103 let mut state = test_state().await;
2104 let mut topo = (*state.topo).clone();
2105 topo.tiers[1].nodes = vec![node];
2106 state.executors = Arc::new(crate::state::build_executors(&topo));
2107 state.topo = Arc::new(topo);
2108
2109 // deploys.node FKs into `nodes`, so the promote path needs the row.
2110 sqlx::query("INSERT INTO nodes (name, tier, ssh_target, release_root) VALUES ('a-local', 'a', 'local', ?)")
2111 .bind(rr.to_string_lossy().into_owned())
2112 .execute(&state.pool).await.unwrap();
2113 for v in [prev, current] {
2114 sqlx::query("INSERT INTO versions (version, git_sha, built_at, artifact_path) VALUES (?, 'sha', datetime('now'), ?)")
2115 .bind(v).bind(format!("/tmp/staged/releases/{v}/makenotwork")).execute(&state.pool).await.unwrap();
2116 }
2117 sqlx::query(
2118 "UPDATE tier_state SET current_version = ?, previous_version = ? WHERE tier = 'a'",
2119 )
2120 .bind(current)
2121 .bind(prev)
2122 .execute(&state.pool)
2123 .await
2124 .unwrap();
2125 (state, tmp)
2126 }
2127
2128 async fn tier_versions(pool: &SqlitePool, tier: &str) -> (Option<String>, Option<String>) {
2129 sqlx::query_as("SELECT current_version, previous_version FROM tier_state WHERE tier = ?")
2130 .bind(tier)
2131 .fetch_one(pool)
2132 .await
2133 .unwrap()
2134 }
2135
2136 #[tokio::test]
2137 async fn rollback_clears_previous_version_rather_than_swapping_it() {
2138 // The swap bug: writing the version we just rolled OFF into
2139 // previous_version made a second /rollback roll FORWARD onto the broken
2140 // build the operator was escaping. previous_version must go NULL.
2141 let (state, _tmp) = rollback_fixture("1.0.0", "2.0.0").await;
2142 let pool = state.pool.clone();
2143
2144 let _ = rollback(State(state), Path("a".to_string())).await.unwrap();
2145
2146 let (cur, prev) = tier_versions(&pool, "a").await;
2147 assert_eq!(
2148 cur.as_deref(),
2149 Some("1.0.0"),
2150 "rolled back to the previous version"
2151 );
2152 assert_eq!(
2153 prev, None,
2154 "the version we rolled off must NOT become the rollback target"
2155 );
2156 }
2157
2158 #[tokio::test]
2159 async fn second_rollback_refuses_instead_of_rolling_forward() {
2160 let (state, _tmp) = rollback_fixture("1.0.0", "2.0.0").await;
2161 let pool = state.pool.clone();
2162 let release_root = state.topo.tiers[1].nodes[0].release_root.clone();
2163
2164 let _ = rollback(State(state.clone()), Path("a".to_string()))
2165 .await
2166 .unwrap();
2167 let err = rollback(State(state), Path("a".to_string()))
2168 .await
2169 .expect_err("only one step of history is tracked; a second rollback must refuse");
2170 assert!(
2171 matches!(&err, crate::error::Error::GateBlocked(m) if m.contains("no previous_version")),
2172 "expected a loud refusal, got: {err:?}",
2173 );
2174
2175 // The refusal is total: neither the DB nor the node moved back to 2.0.0.
2176 let (cur, _) = tier_versions(&pool, "a").await;
2177 assert_eq!(cur.as_deref(), Some("1.0.0"));
2178 let link = tokio::fs::read_link(std::path::Path::new(&release_root).join("current"))
2179 .await
2180 .unwrap();
2181 assert_eq!(
2182 link.to_string_lossy(),
2183 "releases/1.0.0",
2184 "node stays on the rolled-back version"
2185 );
2186 }
2187
2188 #[tokio::test]
2189 async fn promote_refuses_an_unprovisioned_tier() {
2190 // Every step of a promote to a node-less tier is a silent no-op that
2191 // still reports success: the deploy loop iterates nothing and
2192 // advance_tier records a current_version the tier never received.
2193 let (mut state, _tmp) = rollback_fixture("1.0.0", "2.0.0").await;
2194 let mut topo = (*state.topo).clone();
2195 topo.tiers[1].provisioned = false;
2196 topo.tiers[1].nodes.clear();
2197 state.topo = Arc::new(topo);
2198 let pool = state.pool.clone();
2199 sqlx::query("UPDATE tier_state SET current_version = '2.0.0' WHERE tier = 'host'")
2200 .execute(&pool)
2201 .await
2202 .unwrap();
2203
2204 let err = promote_inner(state, "a".into(), PromoteBody::default())
2205 .await
2206 .expect_err("promoting to an unprovisioned tier must be refused");
2207 assert!(
2208 matches!(&err, crate::error::Error::GateBlocked(m) if m.contains("not provisioned")),
2209 "got: {err:?}",
2210 );
2211
2212 // And nothing was recorded: the tier keeps whatever it had before.
2213 let (cur, _) = tier_versions(&pool, "a").await;
2214 assert_eq!(
2215 cur.as_deref(),
2216 Some("2.0.0"),
2217 "a refused promote must not advance tier_state",
2218 );
2219 }
2220
2221 #[tokio::test]
2222 async fn promote_advances_the_tier_and_flips_the_symlink_when_gates_are_green() {
2223 // The happy path. Every other promote test asserts a refusal or a red
2224 // outcome, so nothing pinned what a *successful* promote actually does:
2225 // deploy reaches the node, the `current` symlink flips, tier_state
2226 // advances with previous_version = the version we came off, any stale
2227 // partial flag clears, and the handler reports the nodes it touched.
2228 use crate::topology::Gate;
2229 let (mut state, _tmp) = rollback_fixture("1.0.0", "2.0.0").await;
2230
2231 // Source tier `host` gates the promote on cargo_test. Satisfying it is
2232 // the point of the test: the sibling case asserts an unsatisfied gate
2233 // blocks, this one asserts a satisfied gate lets the promote through.
2234 let mut topo = (*state.topo).clone();
2235 topo.tiers[0].gates = vec![Gate::CargoTest];
2236 state.topo = Arc::new(topo);
2237 let pool = state.pool.clone();
2238 let release_root = state.topo.tiers[1].nodes[0].release_root.clone();
2239
2240 // 3.0.0 is staged on the build host and green on `host`.
2241 tokio::fs::create_dir_all(
2242 std::path::Path::new(&release_root)
2243 .join("releases")
2244 .join("3.0.0"),
2245 )
2246 .await
2247 .unwrap();
2248 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')")
2249 .execute(&pool).await.unwrap();
2250 sqlx::query("UPDATE tier_state SET current_version = '3.0.0' WHERE tier = 'host'")
2251 .execute(&pool)
2252 .await
2253 .unwrap();
2254 insert_gate(&pool, "host", "3.0.0", "cargo_test", 1).await;
2255
2256 // A stale flag from an earlier incident, which a clean rollout clears.
2257 set_partial(&state, &tid("a"), "left over from a previous canary").await;
2258
2259 let Json(body) = promote_inner(state, "a".into(), PromoteBody::default())
2260 .await
2261 .expect("gates are green and the node deploys, so the promote succeeds");
2262
2263 assert_eq!(body["tier"], "a");
2264 assert_eq!(body["version"], "3.0.0");
2265 assert_eq!(
2266 body["nodes_deployed"],
2267 serde_json::json!(["a-local"]),
2268 "the response names every node the promote reached",
2269 );
2270
2271 let (cur, prev) = tier_versions(&pool, "a").await;
2272 assert_eq!(cur.as_deref(), Some("3.0.0"));
2273 assert_eq!(
2274 prev.as_deref(),
2275 Some("2.0.0"),
2276 "previous_version is the version we came off, so a rollback aims at it",
2277 );
2278
2279 // The node genuinely moved: the promote is not just bookkeeping.
2280 let link = tokio::fs::read_link(std::path::Path::new(&release_root).join("current"))
2281 .await
2282 .unwrap();
2283 assert_eq!(link.to_string_lossy(), "releases/3.0.0");
2284
2285 let reason: Option<String> =
2286 sqlx::query_scalar("SELECT partial_reason FROM tier_state WHERE tier = 'a'")
2287 .fetch_one(&pool)
2288 .await
2289 .unwrap();
2290 assert_eq!(
2291 reason, None,
2292 "a clean full rollout clears a stale partial flag",
2293 );
2294
2295 // The deploy is on the record as having succeeded, which is what the
2296 // next promote's gate check and /state both read.
2297 let (node, outcome): (String, String) =
2298 sqlx::query_as("SELECT node, outcome FROM deploys WHERE version = '3.0.0'")
2299 .fetch_one(&pool)
2300 .await
2301 .unwrap();
2302 assert_eq!(node, "a-local");
2303 assert_eq!(outcome, "ok");
2304 }
2305
2306 #[tokio::test]
2307 async fn promote_fails_and_flags_the_tier_when_post_deploy_gates_are_red() {
2308 // The deploy reached every node, so tier_state advances (a stale
2309 // current_version would aim a later rollback at the wrong artifact), but
2310 // the promote must NOT report success: the tier is flagged partial and
2311 // the handler returns the gate failure.
2312 use crate::topology::Gate;
2313 let (mut state, _tmp) = rollback_fixture("1.0.0", "2.0.0").await;
2314 // node_health is the tier's only gate, and it fails closed with no
2315 // probes — an empty executor map gives it nothing to probe while
2316 // deploy_node still falls back to a built executor and succeeds.
2317 let mut topo = (*state.topo).clone();
2318 topo.tiers[1].gates = vec![Gate::NodeHealth];
2319 state.topo = Arc::new(topo);
2320 state.executors = Arc::new(crate::state::ExecutorMap::new());
2321 let pool = state.pool.clone();
2322 // Promote 3.0.0 up from host, which configures no gates of its own.
2323 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')")
2324 .execute(&pool).await.unwrap();
2325 sqlx::query("UPDATE tier_state SET current_version = '3.0.0' WHERE tier = 'host'")
2326 .execute(&pool)
2327 .await
2328 .unwrap();
2329
2330 let err = promote_inner(state, "a".into(), PromoteBody::default())
2331 .await
2332 .expect_err("red post-deploy gates must fail the promote");
2333 assert!(
2334 matches!(&err, crate::error::Error::GateBlocked(m) if m.contains("node_health")),
2335 "the failure must name the red gate, got: {err:?}",
2336 );
2337
2338 let (cur, _) = tier_versions(&pool, "a").await;
2339 assert_eq!(
2340 cur.as_deref(),
2341 Some("3.0.0"),
2342 "tier_state tracks what the nodes actually run"
2343 );
2344 let reason: Option<String> =
2345 sqlx::query_scalar("SELECT partial_reason FROM tier_state WHERE tier = 'a'")
2346 .fetch_one(&pool)
2347 .await
2348 .unwrap();
2349 assert!(
2350 reason.as_deref().is_some_and(|r| r.contains("node_health")),
2351 "the tier must be flagged for /state and the TUI, got: {reason:?}",
2352 );
2353 }
2354
2355 #[tokio::test]
2356 async fn set_partial_then_clear_roundtrips() {
2357 let state = test_state().await;
2358 let read = || async {
2359 sqlx::query_scalar::<_, Option<String>>(
2360 "SELECT partial_reason FROM tier_state WHERE tier = 'a'",
2361 )
2362 .fetch_one(&state.pool)
2363 .await
2364 .unwrap()
2365 };
2366 assert_eq!(read().await, None, "consistent tier starts clean");
2367 set_partial(&state, &tid("a"), "canary rollback incomplete: 1/2").await;
2368 assert_eq!(
2369 read().await.as_deref(),
2370 Some("canary rollback incomplete: 1/2")
2371 );
2372 clear_partial(&state, &tid("a")).await;
2373 assert_eq!(read().await, None, "clear nulls it back out");
2374 }
2375
2376 #[tokio::test]
2377 async fn state_surfaces_partial_reason() {
2378 use axum::extract::State;
2379 let state = test_state().await;
2380 set_partial(
2381 &state,
2382 &tid("a"),
2383 "first-deploy canary failed: 1 node(s) on 2.0.0",
2384 )
2385 .await;
2386 let Json(view) = get_state(State(state)).await.unwrap();
2387 let a = view.tiers.iter().find(|t| t.name == "a").unwrap();
2388 assert_eq!(
2389 a.partial_reason.as_deref(),
2390 Some("first-deploy canary failed: 1 node(s) on 2.0.0"),
2391 );
2392 let host = view.tiers.iter().find(|t| t.name == "host").unwrap();
2393 assert_eq!(
2394 host.partial_reason, None,
2395 "untouched tier stays clean in /state"
2396 );
2397 }
2398
2399 #[tokio::test]
2400 async fn state_build_is_null_until_first_rebuild_then_surfaces_latest() {
2401 use axum::extract::State;
2402 let state = test_state().await;
2403 // No build runs yet → build is null, so /state doesn't pretend a build
2404 // is happening.
2405 let Json(view) = get_state(State(state.clone())).await.unwrap();
2406 assert!(view.build.is_none());
2407
2408 // A failed run must surface its cause in /state, not just in /runs.
2409 let run_id = crate::runs::create(&state.pool, "deadbeef").await.unwrap();
2410 crate::runs::mark_failed(&state.pool, run_id, "cargo_test: 3 test(s) failed")
2411 .await
2412 .unwrap();
2413 let Json(view) = get_state(State(state)).await.unwrap();
2414 let b = view.build.expect("build surfaced");
2415 assert_eq!(b.run_id, run_id.0);
2416 assert_eq!(b.result, "failed");
2417 assert_eq!(
2418 b.failure_summary.as_deref(),
2419 Some("cargo_test: 3 test(s) failed")
2420 );
2421 }
2422
2423 #[tokio::test]
2424 async fn status_json_serves_the_shared_payload_over_the_real_router() {
2425 // The mapping itself is tested in `crate::status`. This asserts the
2426 // route is wired, serves valid JSON, and stays internally consistent
2427 // (no dangling child or action references) against a real topology
2428 // rather than a hand-built fixture.
2429 let state = test_state().await;
2430 set_partial(&state, &tid("a"), "canary rollback incomplete: 1/2").await;
2431
2432 let resp = router(state)
2433 .oneshot(
2434 Request::builder()
2435 .uri("/status.json")
2436 .body(Body::empty())
2437 .unwrap(),
2438 )
2439 .await
2440 .unwrap();
2441 assert_eq!(resp.status(), StatusCode::OK);
2442
2443 let body = http_body_util::BodyExt::collect(resp.into_body())
2444 .await
2445 .unwrap()
2446 .to_bytes();
2447 let payload: ops_status::Payload = serde_json::from_slice(&body).unwrap();
2448
2449 assert_eq!(payload.source, crate::status::SOURCE);
2450 assert_eq!(payload.schema, ops_status::SCHEMA_VERSION);
2451 assert_eq!(payload.validate(), Ok(()));
2452 assert_eq!(
2453 payload.node("tier:a").unwrap().status,
2454 ops_status::Status::Failed,
2455 "a partial tier must surface as failed over the wire"
2456 );
2457 assert_eq!(payload.worst_status(), ops_status::Status::Failed);
2458 }
2459
2460 #[tokio::test]
2461 async fn promote_with_explicit_version_but_missing_artifact_404s() {
2462 // Explicit version supplied, gates trivially pass (mm has none in
2463 // test_topo), but `versions` table has no row → 404.
2464 let state = test_state().await;
2465 let app = router(state);
2466 let resp = app
2467 .oneshot(
2468 Request::builder()
2469 .method("POST")
2470 .uri("/promote/a")
2471 .header("content-type", "application/json")
2472 .body(Body::from(r#"{"version":"9.9.9"}"#))
2473 .unwrap(),
2474 )
2475 .await
2476 .unwrap();
2477 assert_eq!(resp.status(), StatusCode::NOT_FOUND);
2478 }
2479
2480 // ---- GET /logs/{version}/{gate} ----
2481
2482 async fn state_with_logs_root(logs_root: PathBuf) -> AppState {
2483 let mut s = test_state().await;
2484 let mut cfg = (*s.cfg).clone();
2485 cfg.logs_root = logs_root;
2486 s.cfg = Arc::new(cfg);
2487 s
2488 }
2489
2490 #[tokio::test]
2491 async fn get_gate_log_returns_file_contents() {
2492 let tmp = tempfile::tempdir().unwrap();
2493 let dir = tmp.path().join("0.9.5");
2494 tokio::fs::create_dir_all(&dir).await.unwrap();
2495 tokio::fs::write(dir.join("cargo_test.log"), b"hello sandod\n")
2496 .await
2497 .unwrap();
2498
2499 let state = state_with_logs_root(tmp.path().to_path_buf()).await;
2500 let app = router(state);
2501 let resp = app
2502 .oneshot(
2503 Request::builder()
2504 .uri("/logs/0.9.5/cargo_test")
2505 .body(Body::empty())
2506 .unwrap(),
2507 )
2508 .await
2509 .unwrap();
2510 assert_eq!(resp.status(), StatusCode::OK);
2511 assert_eq!(body_string(resp).await, "hello sandod\n");
2512 }
2513
2514 #[tokio::test]
2515 async fn get_gate_log_404s_when_missing() {
2516 let tmp = tempfile::tempdir().unwrap();
2517 let state = state_with_logs_root(tmp.path().to_path_buf()).await;
2518 let app = router(state);
2519 let resp = app
2520 .oneshot(
2521 Request::builder()
2522 .uri("/logs/0.9.5/cargo_test")
2523 .body(Body::empty())
2524 .unwrap(),
2525 )
2526 .await
2527 .unwrap();
2528 assert_eq!(resp.status(), StatusCode::NOT_FOUND);
2529 }
2530
2531 // ---- CF2: bearer-token auth on deploy mutators ----
2532
2533 #[tokio::test]
2534 async fn mutating_route_requires_bearer_when_token_set() {
2535 let mut state = test_state().await;
2536 state.api_token = Some(std::sync::Arc::from("s3cr3t"));
2537 let app = router(state);
2538
2539 // No Authorization header -> 401, before any deploy logic runs.
2540 let resp = app
2541 .clone()
2542 .oneshot(
2543 Request::builder()
2544 .method("POST")
2545 .uri("/promote/a")
2546 .body(Body::empty())
2547 .unwrap(),
2548 )
2549 .await
2550 .unwrap();
2551 assert_eq!(resp.status(), StatusCode::UNAUTHORIZED);
2552
2553 // Wrong token -> 401.
2554 let resp = app
2555 .clone()
2556 .oneshot(
2557 Request::builder()
2558 .method("POST")
2559 .uri("/promote/a")
2560 .header("authorization", "Bearer nope")
2561 .body(Body::empty())
2562 .unwrap(),
2563 )
2564 .await
2565 .unwrap();
2566 assert_eq!(resp.status(), StatusCode::UNAUTHORIZED);
2567
2568 // Correct token -> passes auth (then blocked downstream by gates /
2569 // missing predecessor version, but specifically NOT 401).
2570 let resp = app
2571 .clone()
2572 .oneshot(
2573 Request::builder()
2574 .method("POST")
2575 .uri("/promote/a")
2576 .header("authorization", "Bearer s3cr3t")
2577 .body(Body::empty())
2578 .unwrap(),
2579 )
2580 .await
2581 .unwrap();
2582 assert_ne!(resp.status(), StatusCode::UNAUTHORIZED);
2583 }
2584
2585 #[tokio::test]
2586 async fn read_routes_require_token_when_set() {
2587 // Reads expose prod state (versions, SHAs, gate logs, the event stream),
2588 // so they are bearer-gated too — not just the mutators. A tailnet peer
2589 // without the token gets 401; the TUI presents the token and gets 200.
2590 let mut state = test_state().await;
2591 state.api_token = Some(std::sync::Arc::from("s3cr3t"));
2592 let app = router(state);
2593
2594 // No token -> 401 on a read.
2595 let resp = app
2596 .clone()
2597 .oneshot(
2598 Request::builder()
2599 .uri("/state")
2600 .body(Body::empty())
2601 .unwrap(),
2602 )
2603 .await
2604 .unwrap();
2605 assert_eq!(resp.status(), StatusCode::UNAUTHORIZED);
2606
2607 // Correct token -> 200.
2608 let resp = app
2609 .oneshot(
2610 Request::builder()
2611 .uri("/state")
2612 .header("authorization", "Bearer s3cr3t")
2613 .body(Body::empty())
2614 .unwrap(),
2615 )
2616 .await
2617 .unwrap();
2618 assert_eq!(resp.status(), StatusCode::OK);
2619 }
2620
2621 #[tokio::test]
2622 async fn read_routes_open_without_a_token() {
2623 // The loopback/dev posture: no token configured, reads pass through so a
2624 // local TUI needs no credential.
2625 let state = test_state().await;
2626 let app = router(state);
2627 let resp = app
2628 .oneshot(
2629 Request::builder()
2630 .uri("/state")
2631 .body(Body::empty())
2632 .unwrap(),
2633 )
2634 .await
2635 .unwrap();
2636 assert_eq!(resp.status(), StatusCode::OK);
2637 }
2638
2639 #[tokio::test]
2640 async fn self_update_malformed_body_is_400_not_422() {
2641 // TypedBody funnels a JSON deserialize failure through the Error envelope
2642 // (400), not axum's raw 422 — keeping every mutator on one error contract.
2643 let state = test_state().await; // no token -> auth passes, body is the gate
2644 let app = router(state);
2645 let resp = app
2646 .oneshot(
2647 Request::builder()
2648 .method("POST")
2649 .uri("/self-update")
2650 .header("content-type", "application/json")
2651 .body(Body::from("{ not valid json"))
2652 .unwrap(),
2653 )
2654 .await
2655 .unwrap();
2656 assert_eq!(resp.status(), StatusCode::BAD_REQUEST);
2657 }
2658
2659 #[tokio::test]
2660 async fn gate_log_rejects_unknown_gate_kind() {
2661 // The gate segment is an allowlisted GateKind, not a free-form filename:
2662 // an unknown kind is a 404, so `*.log` basenames can't be probed.
2663 let state = test_state().await;
2664 let app = router(state);
2665 let resp = app
2666 .oneshot(
2667 Request::builder()
2668 .uri("/logs/0.9.6/passwd")
2669 .body(Body::empty())
2670 .unwrap(),
2671 )
2672 .await
2673 .unwrap();
2674 assert_eq!(resp.status(), StatusCode::NOT_FOUND);
2675 }
2676
2677 /// Path-traversal guard: a `..` segment must not escape logs_root.
2678 /// axum's `{name}` param already rejects a literal `/` in the value, but
2679 /// `..` as a whole segment is structurally valid and must be blocked at
2680 /// the handler.
2681 #[tokio::test]
2682 async fn get_gate_log_rejects_dotdot_segments() {
2683 let tmp = tempfile::tempdir().unwrap();
2684 let state = state_with_logs_root(tmp.path().to_path_buf()).await;
2685 let app = router(state);
2686 let resp = app
2687 .oneshot(
2688 Request::builder()
2689 .uri("/logs/../etc/passwd")
2690 .body(Body::empty())
2691 .unwrap(),
2692 )
2693 .await
2694 .unwrap();
2695 assert_eq!(resp.status(), StatusCode::NOT_FOUND);
2696 }
2697
2698 // ---- build-identity path (wiki release-artifact-identity) ----
2699
2700 /// Insert a settled (passed) build_runs row with content identity and return
2701 /// its id. `staged_path` is the content-addressed release dir the bundle was
2702 /// published to (`releases/<digest16>`).
2703 async fn seed_build(pool: &SqlitePool, sha: &str, version: &str, staged_path: &str) -> i64 {
2704 sqlx::query_scalar(
2705 "INSERT INTO build_runs (sha, version, phase, result, started_at, bundle_digest, staged_path)
2706 VALUES (?, ?, 'done', 'passed', datetime('now'), ?, ?) RETURNING id",
2707 )
2708 .bind(sha)
2709 .bind(version)
2710 .bind(format!("{sha}-digest"))
2711 .bind(staged_path)
2712 .fetch_one(pool)
2713 .await
2714 .unwrap()
2715 }
2716
2717 async fn insert_gate_build(
2718 pool: &SqlitePool,
2719 tier: &str,
2720 version: &str,
2721 kind: &str,
2722 passed: i64,
2723 build_id: i64,
2724 ) {
2725 let status = if passed == 1 { "passed" } else { "failed" };
2726 sqlx::query(
2727 "INSERT INTO gate_runs (version, tier, gate_kind, started_at, finished_at, status, build_id) \
2728 VALUES (?, ?, ?, datetime('now'), datetime('now'), ?, ?)",
2729 )
2730 .bind(version)
2731 .bind(tier)
2732 .bind(kind)
2733 .bind(status)
2734 .bind(build_id)
2735 .execute(pool)
2736 .await
2737 .unwrap();
2738 }
2739
2740 #[tokio::test]
2741 async fn unsatisfied_gates_keys_build_evidence_on_build_id() {
2742 let pool = fresh_pool().await;
2743 seed(&pool, "a", "3.0.0").await;
2744 // Two builds of the SAME version string. Only b1's gate ran.
2745 let b1 = seed_build(&pool, "sha1", "3.0.0", "/rel/1111111111111111").await;
2746 let b2 = seed_build(&pool, "sha2", "3.0.0", "/rel/2222222222222222").await;
2747 insert_gate_build(&pool, "a", "3.0.0", "cargo_test", 1, b1).await;
2748
2749 // b1's own evidence satisfies.
2750 let ok = unsatisfied_gates(
2751 &pool,
2752 &tid("a"),
2753 &[Gate::CargoTest],
2754 "3.0.0",
2755 Some(b1),
2756 false,
2757 )
2758 .await
2759 .unwrap();
2760 assert!(ok.is_empty(), "the build that passed its gate is satisfied");
2761
2762 // b2 shares the version but has no gate row of its own: fail closed. This
2763 // is the hole — a rebuild reusing the version must not ride b1's evidence.
2764 let bad = unsatisfied_gates(
2765 &pool,
2766 &tid("a"),
2767 &[Gate::CargoTest],
2768 "3.0.0",
2769 Some(b2),
2770 false,
2771 )
2772 .await
2773 .unwrap();
2774 assert_eq!(
2775 bad,
2776 vec!["cargo_test".to_string()],
2777 "a different build of the same version does not inherit the evidence"
2778 );
2779
2780 // Legacy (pre-identity) callers still resolve by version string.
2781 let legacy = unsatisfied_gates(&pool, &tid("a"), &[Gate::CargoTest], "3.0.0", None, false)
2782 .await
2783 .unwrap();
2784 assert!(legacy.is_empty(), "version-keyed legacy path unchanged");
2785 }
2786
2787 #[tokio::test]
2788 async fn promote_rejects_an_explicit_version_that_is_not_the_source_build() {
2789 // The burn-in hole: `promote --version Y` used to check the SOURCE tier's
2790 // clock/evidence (which belong to whatever is current there), letting Y
2791 // inherit another build's 48h. Now promote resolves the source's current
2792 // build and refuses an explicit version that isn't it.
2793 let state = test_state().await;
2794 seed_version(&state.pool, "3.0.0").await;
2795 let b = seed_build(&state.pool, "shaB", "3.0.0", "/rel/deadbeefdeadbeef").await;
2796 sqlx::query(
2797 "UPDATE tier_state SET current_version='3.0.0', current_build_id=? WHERE tier='host'",
2798 )
2799 .bind(b)
2800 .execute(&state.pool)
2801 .await
2802 .unwrap();
2803
2804 let err = promote_inner(
2805 state,
2806 "a".into(),
2807 PromoteBody {
2808 version: Some("2.0.0".into()),
2809 ..Default::default()
2810 },
2811 )
2812 .await
2813 .expect_err("promoting a version other than the source build must be refused");
2814 assert!(
2815 matches!(&err, crate::error::Error::GateBlocked(m) if m.contains("vouched for")),
2816 "the refusal must name the mismatch, got: {err:?}",
2817 );
2818 }
2819
2820 #[tokio::test]
2821 async fn promote_identity_path_deploys_the_source_build_and_advances_build_id() {
2822 // Full promote through the identity path: source tier points at a build,
2823 // promote resolves the artifact through it, deploys build_runs.staged_path
2824 // (content-addressed), and advances the target's current_build_id.
2825 let mut state = test_state().await;
2826 let node_root = tempfile::tempdir().unwrap();
2827 // Point the a-local node at a real tempdir so the symlink swap is checkable.
2828 let mut topo = (*state.topo).clone();
2829 topo.tiers[1].nodes[0].release_root = node_root.path().to_string_lossy().into_owned();
2830 topo.tiers[1].gates = vec![]; // isolate the deploy fan-out
2831 state.topo = Arc::new(topo);
2832 state.executors = Arc::new(crate::state::build_executors(&state.topo));
2833
2834 // deploys.node FKs into `nodes`, so the promote path needs the row.
2835 sqlx::query("INSERT INTO nodes (name, tier, ssh_target, release_root) VALUES ('a-local', 'a', 'local', ?)")
2836 .bind(node_root.path().to_string_lossy().into_owned())
2837 .execute(&state.pool)
2838 .await
2839 .unwrap();
2840
2841 seed_version(&state.pool, "3.0.0").await;
2842 let staged = format!(
2843 "{}/releases/abc123abc123abc1",
2844 node_root.path().to_string_lossy()
2845 );
2846 let b = seed_build(&state.pool, "shaB", "3.0.0", &staged).await;
2847 sqlx::query(
2848 "UPDATE tier_state SET current_version='3.0.0', current_build_id=? WHERE tier='host'",
2849 )
2850 .bind(b)
2851 .execute(&state.pool)
2852 .await
2853 .unwrap();
2854
2855 let pool = state.pool.clone();
2856 let Json(body) = promote_inner(state, "a".into(), PromoteBody::default())
2857 .await
2858 .expect("identity-path promote succeeds");
2859 assert_eq!(body["version"], "3.0.0");
2860
2861 // Target tier advanced with the build identity, not just the version.
2862 let (cur_v, cur_b): (Option<String>, Option<i64>) = sqlx::query_as(
2863 "SELECT current_version, current_build_id FROM tier_state WHERE tier = 'a'",
2864 )
2865 .fetch_one(&pool)
2866 .await
2867 .unwrap();
2868 assert_eq!(cur_v.as_deref(), Some("3.0.0"));
2869 assert_eq!(cur_b, Some(b), "target records the promoted build id");
2870
2871 // The deploy row is attributed to the build.
2872 let deploy_b: Option<i64> = sqlx::query_scalar(
2873 "SELECT build_id FROM deploys WHERE tier = 'a' ORDER BY id DESC LIMIT 1",
2874 )
2875 .fetch_one(&pool)
2876 .await
2877 .unwrap();
2878 assert_eq!(deploy_b, Some(b));
2879
2880 // The node's `current` points at the content-addressed release dir, whose
2881 // name is the staged_path's basename — not the version.
2882 let link = tokio::fs::read_link(node_root.path().join("current"))
2883 .await
2884 .unwrap();
2885 assert_eq!(link.to_string_lossy(), "releases/abc123abc123abc1");
2886 }
2887
2888 /// Insert a bare `versions` row (FK target for tier_state.current_version).
2889 async fn seed_version(pool: &SqlitePool, version: &str) {
2890 sqlx::query("INSERT INTO versions (version, git_sha, built_at, artifact_path) VALUES (?, 'sha', datetime('now'), '/tmp/x') ON CONFLICT DO NOTHING")
2891 .bind(version).execute(pool).await.unwrap();
2892 }
2893 }
2894