max / makenotwork

110.3 KB · 2911 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	#[derive(Deserialize, Default)]
711	struct BackupFetchBody {
712	/// Accept a dump that falls below the plausibility floor, re-baselining it to
713	/// this fetch's size. For the case where the source legitimately shrank and
714	/// the floor has wedged itself; the `gzip -t` integrity check still applies.
715	/// Operator-only — never set this from a timer.
716	#[serde(default)]
717	force: bool,
718	}
719
720	async fn backup_fetch(
721	State(s): State<AppState>,
722	body: Option<Json<BackupFetchBody>>,
723	) -> Result<Json<serde_json::Value>> {
724	let body = body.map(\|Json(b)\| b).unwrap_or_default();
725	let fb = crate::backup::fetch(&s.pool, &s.cfg, &s.topo, body.force)
726	.await
727	.map_err(crate::error::Error::Other)?;
728	crate::events::emit(
729	&s.events,
730	crate::events::Event::BackupFetched {
731	source: fb.source.clone(),
732	byte_size: fb.byte_size.unwrap_or(0),
733	},
734	);
735	Ok(Json(serde_json::json!({
736	"source": fb.source,
737	"local_path": fb.local_path,
738	"byte_size": fb.byte_size,
739	})))
740	}
741
742	async fn get_gate_log(
743	State(s): State<AppState>,
744	Path((version, gate)): Path<(String, String)>,
745	) -> Result<axum::response::Response> {
746	// Guard against `..` / absolute paths — the version segment must be a single
747	// safe component. Without this, `GET /logs/..%2Fetc/passwd` would escape
748	// logs_root.
749	fn safe(seg: &str) -> bool {
750	!seg.is_empty() && !seg.contains('/') && !seg.contains('\\') && seg != "." && seg != ".."
751	}
752	if !safe(&version) {
753	return Err(crate::error::Error::NotFound);
754	}
755	// The gate segment is an allowlisted kind, not a free-form filename: parse it
756	// to `GateKind` so only the five known `<kind>.log` files are reachable. This
757	// closes `*.log` filename probing within a version dir (traversal was already
758	// blocked; this bounds the basename to the known set).
759	let kind: crate::domain::GateKind = gate.parse().map_err(\|_\| crate::error::Error::NotFound)?;
760	let path = s
761	.cfg
762	.logs_root
763	.join(&version)
764	.join(format!("{}.log", kind.as_str()));
765	// Bound how much a single request can pull into the daemon's memory: a
766	// runaway gate log shouldn't be read whole. Past the cap we return the tail
767	// (the recent, relevant output) behind a truncation marker.
768	const MAX_LOG_BYTES: u64 = 8 * 1024 * 1024;
769	use tokio::io::{AsyncReadExt, AsyncSeekExt};
770	let mut file = match tokio::fs::File::open(&path).await {
771	Ok(f) => f,
772	Err(e) if e.kind() == std::io::ErrorKind::NotFound => {
773	return Err(crate::error::Error::NotFound);
774	}
775	Err(e) => return Err(crate::error::Error::Other(e.into())),
776	};
777	let len = file
778	.metadata()
779	.await
780	.map_err(\|e\| crate::error::Error::Other(e.into()))?
781	.len();
782	let body: Vec<u8> = if len <= MAX_LOG_BYTES {
783	let mut buf = Vec::with_capacity(len as usize);
784	file.read_to_end(&mut buf)
785	.await
786	.map_err(\|e\| crate::error::Error::Other(e.into()))?;
787	buf
788	} else {
789	file.seek(std::io::SeekFrom::End(-(MAX_LOG_BYTES as i64)))
790	.await
791	.map_err(\|e\| crate::error::Error::Other(e.into()))?;
792	let mut buf = format!(
793	"[log truncated: showing the last {} MiB of {len} bytes]\n",
794	MAX_LOG_BYTES / 1024 / 1024,
795	)
796	.into_bytes();
797	file.read_to_end(&mut buf)
798	.await
799	.map_err(\|e\| crate::error::Error::Other(e.into()))?;
800	buf
801	};
802	Ok((
803	[(
804	axum::http::header::CONTENT_TYPE,
805	"text/plain; charset=utf-8",
806	)],
807	body,
808	)
809	.into_response())
810	}
811
812	async fn events_ws(ws: WebSocketUpgrade, State(s): State<AppState>) -> impl IntoResponse {
813	use axum::extract::ws::Message;
814	use tokio::sync::broadcast::error::RecvError;
815
816	ws.on_upgrade(move \|mut socket\| async move {
817	// Subscribe to both channels and merge them: a lag on the high-rate log
818	// stream emits its own `lagged` frame without dropping anything on the
819	// status stream (and vice versa), so a busy gate's chunk firehose can't
820	// evict a PromoteComplete/GateDone the operator needs to see.
821	let mut status_rx = s.events.subscribe_status();
822	let mut logs_rx = s.events.subscribe_logs();
823	loop {
824	let recv = tokio::select! {
825	r = status_rx.recv() => r,
826	r = logs_rx.recv() => r,
827	};
828	match recv {
829	Ok(env) => {
830	let json = match serde_json::to_string(&env) {
831	Ok(s) => s,
832	Err(e) => {
833	tracing::warn!(error = %e, "events ws: serialize failed");
834	continue;
835	}
836	};
837	if socket.send(Message::Text(json.into())).await.is_err() {
838	break;
839	}
840	}
841	Err(RecvError::Lagged(n)) => {
842	let _ = socket
843	.send(Message::Text(
844	format!(r#"{{"kind":"lagged","skipped":{n}}}"#).into(),
845	))
846	.await;
847	}
848	Err(RecvError::Closed) => break,
849	}
850	}
851	})
852	}
853
854	#[cfg(test)]
855	mod tests {
856	use super::promotion::{rollback_deployed_nodes, unsatisfied_gates};
857	use super::*;
858	use crate::config::Config;
859	use crate::topology::{BackupConfig, CanaryPolicy, Gate, Node, RepoConfig, Tier, Topology};
860	use async_trait::async_trait;
861	use axum::body::Body;
862	use axum::http::{Request, StatusCode};
863	use http_body_util::BodyExt;
864	use ops_exec::{CapabilitySet, Executor, LogSink, RunOutput, Step, SyncOpts};
865	use sqlx::SqlitePool;
866	use sqlx::sqlite::SqlitePoolOptions;
867	use std::os::unix::process::ExitStatusExt;
868	use std::path::PathBuf;
869	use std::sync::{Arc, Mutex as StdMutex};
870	use tower::ServiceExt;
871
872	async fn fresh_pool() -> SqlitePool {
873	let pool = SqlitePoolOptions::new()
874	.max_connections(1)
875	.connect("sqlite::memory:")
876	.await
877	.unwrap();
878	sqlx::migrate!("./migrations").run(&pool).await.unwrap();
879	pool
880	}
881
882	/// Two-tier topology used by the route tests: mm (provisioned, no nodes)
883	/// → a (provisioned, one local node). Mirrors the production shape
884	/// without involving real ssh / postgres.
885	fn test_topo() -> Topology {
886	Topology {
887	repo: RepoConfig {
888	bare_path: "/tmp/test.git".into(),
889	branch: "main".into(),
890	upstream: None,
891	},
892	backup: BackupConfig {
893	source: "file:///tmp/test-backup.sql".into(),
894	local_path: "/tmp/local-backup.sql".into(),
895	},
896	tiers: vec![
897	Tier {
898	name: "host".into(),
899	provisioned: true,
900	gates: vec![],
901	canary: CanaryPolicy::Sequential,
902	nodes: vec![],
903	},
904	Tier {
905	name: "a".into(),
906	provisioned: true,
907	gates: vec![Gate::BootSmoke],
908	canary: CanaryPolicy::Sequential,
909	nodes: vec![Node {
910	name: "a-local".into(),
911	ssh_target: "local".into(),
912	release_root: "/tmp/a-node".into(),
913	service_name: "makenotwork.service".into(),
914	health_url: None,
915	config_check_env_file: None,
916	actuate: crate::topology::default_actuate(),
917	observe: crate::topology::default_observe(),
918	companions: Vec::new(),
919	}],
920	},
921	],
922	aux_repos: Vec::new(),
923	}
924	}
925
926	fn test_cfg() -> Config {
927	Config {
928	listen: "127.0.0.1:0".into(),
929	db_path: PathBuf::from(":memory:"),
930	topology_path: PathBuf::from("/tmp/test-sando.toml"),
931	build_host: "test-host".into(),
932	workdir: PathBuf::from("/tmp/sando-work"),
933	release_root: PathBuf::from("/tmp/sando-releases"),
934	scratch_db_url: None,
935	scratch_owner_role: "makenotwork".into(),
936	boot_smoke_port: 18181,
937	code_smoke_port: 18182,
938	bin_names: vec!["makenotwork".into()],
939	logs_root: PathBuf::from("/tmp/sando-logs"),
940	release_contents: vec![],
941	cargo_target_dir: None,
942	gate_timeout_secs: 2400,
943	companions: Vec::new(),
944	test_targets: vec![crate::config::TestTarget {
945	dir: PathBuf::from("server"),
946	features: vec!["fast-tests".into()],
947	all_features: false,
948	scratch_db: true,
949	}],
950	frontend_builds: vec![],
951	backup_max_age_hours: 48,
952	}
953	}
954
955	async fn test_state() -> AppState {
956	let pool = fresh_pool().await;
957	// Seed tier rows so FKs on tier_state / gate_runs are satisfied.
958	for (i, name) in ["host", "a"].iter().enumerate() {
959	sqlx::query(
960	"INSERT INTO tiers (name, ord, provisioned, canary) VALUES (?, ?, 1, 'sequential')",
961	)
962	.bind(name)
963	.bind(i as i64)
964	.execute(&pool)
965	.await
966	.unwrap();
967	sqlx::query("INSERT INTO tier_state (tier) VALUES (?)")
968	.bind(name)
969	.execute(&pool)
970	.await
971	.unwrap();
972	}
973	// Don't call install_recorder in tests — it touches a process-global
974	// and conflicts when tests run in parallel.
975	let topo = test_topo();
976	let executors = Arc::new(crate::state::build_executors(&topo));
977	AppState {
978	pool,
979	topo: Arc::new(topo),
980	cfg: Arc::new(test_cfg()),
981	active_build: Arc::new(tokio::sync::Mutex::new(None)),
982	deploy_lock: Arc::new(tokio::sync::Mutex::new(())),
983	events: crate::events::channel(),
984	executors,
985	api_token: None,
986	}
987	}
988
989	async fn body_string(resp: axum::response::Response) -> String {
990	let bytes = resp.into_body().collect().await.unwrap().to_bytes();
991	String::from_utf8(bytes.to_vec()).unwrap()
992	}
993
994	/// Insert the FK prerequisites for inserting gate_runs/tier_state rows.
995	async fn seed(pool: &SqlitePool, tier: &str, version: &str) {
996	sqlx::query("INSERT INTO tiers (name, ord, provisioned, canary) VALUES (?, 0, 1, 'sequential') ON CONFLICT DO NOTHING")
997	.bind(tier).execute(pool).await.unwrap();
998	sqlx::query("INSERT INTO versions (version, git_sha, built_at, artifact_path) VALUES (?, 'sha', datetime('now'), '/tmp/x') ON CONFLICT DO NOTHING")
999	.bind(version).execute(pool).await.unwrap();
1000	sqlx::query("INSERT INTO tier_state (tier, current_version) VALUES (?, NULL) ON CONFLICT DO NOTHING")
1001	.bind(tier).execute(pool).await.unwrap();
1002	}
1003
1004	async fn insert_gate(pool: &SqlitePool, tier: &str, version: &str, kind: &str, passed: i64) {
1005	let status = if passed == 1 { "passed" } else { "failed" };
1006	sqlx::query(
1007	"INSERT INTO gate_runs (version, tier, gate_kind, started_at, finished_at, status) \
1008	VALUES (?, ?, ?, datetime('now'), datetime('now'), ?)",
1009	)
1010	.bind(version)
1011	.bind(tier)
1012	.bind(kind)
1013	.bind(status)
1014	.execute(pool)
1015	.await
1016	.unwrap();
1017	}
1018
1019	// ---- unsatisfied_gates ----
1020
1021	fn tid(s: &str) -> crate::domain::TierId {
1022	crate::domain::TierId::new(s)
1023	}
1024
1025	#[tokio::test]
1026	async fn unsatisfied_gates_empty_when_no_configured_gates() {
1027	// A tier that configures no gates has nothing to satisfy.
1028	let pool = fresh_pool().await;
1029	seed(&pool, "host", "0.8.12").await;
1030	let pending = unsatisfied_gates(&pool, &tid("host"), &[], "0.8.12", None, false)
1031	.await
1032	.unwrap();
1033	assert_eq!(pending, Vec::<String>::new());
1034	}
1035
1036	#[tokio::test]
1037	async fn unsatisfied_gates_flags_configured_gate_that_never_ran() {
1038	// THE CF1 FIX: a configured gate with no gate_runs row is unsatisfied
1039	// (fail closed), NOT silently treated as green. Before this, an A tier
1040	// whose boot_smoke never executed exposed zero rows and waved promotion
1041	// straight through to prod.
1042	let pool = fresh_pool().await;
1043	seed(&pool, "a", "0.8.12").await;
1044	let pending =
1045	unsatisfied_gates(&pool, &tid("a"), &[Gate::BootSmoke], "0.8.12", None, false)
1046	.await
1047	.unwrap();
1048	assert_eq!(pending, vec!["boot_smoke".to_string()]);
1049	}
1050
1051	#[tokio::test]
1052	async fn unsatisfied_gates_flags_failed_kind() {
1053	let pool = fresh_pool().await;
1054	seed(&pool, "host", "0.8.12").await;
1055	insert_gate(&pool, "host", "0.8.12", "cargo_test", 0).await;
1056	insert_gate(&pool, "host", "0.8.12", "boot_smoke", 1).await;
1057	let pending = unsatisfied_gates(
1058	&pool,
1059	&tid("host"),
1060	&[Gate::CargoTest, Gate::BootSmoke],
1061	"0.8.12",
1062	None,
1063	false,
1064	)
1065	.await
1066	.unwrap();
1067	assert_eq!(pending, vec!["cargo_test".to_string()]);
1068	}
1069
1070	#[tokio::test]
1071	async fn unsatisfied_gates_latest_row_wins() {
1072	// Two runs of the same gate; only the latest counts. A flap from
1073	// red to green should clear the pending entry.
1074	let pool = fresh_pool().await;
1075	seed(&pool, "host", "0.8.12").await;
1076	insert_gate(&pool, "host", "0.8.12", "cargo_test", 0).await;
1077	insert_gate(&pool, "host", "0.8.12", "cargo_test", 1).await;
1078	let pending = unsatisfied_gates(
1079	&pool,
1080	&tid("host"),
1081	&[Gate::CargoTest],
1082	"0.8.12",
1083	None,
1084	false,
1085	)
1086	.await
1087	.unwrap();
1088	assert!(pending.is_empty());
1089	}
1090
1091	async fn insert_confirm(
1092	pool: &SqlitePool,
1093	tier: &str,
1094	version: &str,
1095	at: chrono::DateTime<chrono::Utc>,
1096	) {
1097	sqlx::query(
1098	"INSERT INTO gate_runs (version, tier, gate_kind, started_at, finished_at, status) \
1099	VALUES (?, ?, 'manual_confirm', ?, ?, 'passed')",
1100	)
1101	.bind(version)
1102	.bind(tier)
1103	.bind(at.to_rfc3339())
1104	.bind(at.to_rfc3339())
1105	.execute(pool)
1106	.await
1107	.unwrap();
1108	}
1109
1110	#[tokio::test]
1111	async fn unsatisfied_gates_manual_confirm_requires_fresh_confirmation() {
1112	// A confirmation only satisfies the gate if it post-dates the version's
1113	// current landing on the tier (burn_in_started_at). A stale confirm left
1114	// over from before a rollback + rollback-forward must NOT wave it through.
1115	let pool = fresh_pool().await;
1116	seed(&pool, "a", "0.8.12").await;
1117	let landed = chrono::Utc::now();
1118	sqlx::query("UPDATE tier_state SET burn_in_started_at = ? WHERE tier = 'a'")
1119	.bind(landed.to_rfc3339())
1120	.execute(&pool)
1121	.await
1122	.unwrap();
1123
1124	// Stale confirmation (recorded before this landing) -> unsatisfied.
1125	insert_confirm(&pool, "a", "0.8.12", landed - chrono::Duration::hours(1)).await;
1126	let pending = unsatisfied_gates(
1127	&pool,
1128	&tid("a"),
1129	&[Gate::ManualConfirm],
1130	"0.8.12",
1131	None,
1132	false,
1133	)
1134	.await
1135	.unwrap();
1136	assert_eq!(
1137	pending,
1138	vec!["manual_confirm".to_string()],
1139	"stale confirm must not satisfy"
1140	);
1141
1142	// Fresh confirmation (after this landing) -> satisfied.
1143	insert_confirm(&pool, "a", "0.8.12", landed + chrono::Duration::minutes(5)).await;
1144	let pending = unsatisfied_gates(
1145	&pool,
1146	&tid("a"),
1147	&[Gate::ManualConfirm],
1148	"0.8.12",
1149	None,
1150	false,
1151	)
1152	.await
1153	.unwrap();
1154	assert!(pending.is_empty(), "fresh confirm satisfies");
1155	}
1156
1157	#[tokio::test]
1158	async fn unsatisfied_gates_manual_confirm_fails_closed_without_baseline() {
1159	// No landing clock (burn_in_started_at NULL) -> a passed confirm row is
1160	// not provably fresh, so fail closed and require a new confirmation.
1161	let pool = fresh_pool().await;
1162	seed(&pool, "a", "0.8.12").await; // leaves burn_in_started_at NULL
1163	insert_confirm(&pool, "a", "0.8.12", chrono::Utc::now()).await;
1164	let pending = unsatisfied_gates(
1165	&pool,
1166	&tid("a"),
1167	&[Gate::ManualConfirm],
1168	"0.8.12",
1169	None,
1170	false,
1171	)
1172	.await
1173	.unwrap();
1174	assert_eq!(
1175	pending,
1176	vec!["manual_confirm".to_string()],
1177	"no baseline -> fail closed"
1178	);
1179	}
1180
1181	#[tokio::test]
1182	async fn unsatisfied_gates_hotfix_skips_only_burn_in() {
1183	// burn_in is evaluated live (no clock started -> not elapsed); cargo_test
1184	// has a failing row. Normal: both unsatisfied, in configured order.
1185	// hotfix: burn_in suppressed, cargo_test still flagged. Lock the semantic
1186	// so a future change doesn't widen the hotfix bypass.
1187	let pool = fresh_pool().await;
1188	seed(&pool, "a", "0.8.12").await;
1189	insert_gate(&pool, "a", "0.8.12", "cargo_test", 0).await;
1190	let gates = [Gate::BurnIn { hours: 48 }, Gate::CargoTest];
1191
1192	let normal = unsatisfied_gates(&pool, &tid("a"), &gates, "0.8.12", None, false)
1193	.await
1194	.unwrap();
1195	assert_eq!(
1196	normal,
1197	vec!["burn_in".to_string(), "cargo_test".to_string()]
1198	);
1199
1200	let with_hotfix = unsatisfied_gates(&pool, &tid("a"), &gates, "0.8.12", None, true)
1201	.await
1202	.unwrap();
1203	assert_eq!(with_hotfix, vec!["cargo_test".to_string()]);
1204	}
1205
1206	#[tokio::test]
1207	async fn unsatisfied_gates_burn_in_passes_when_window_elapsed() {
1208	// A burn-in clock started far enough in the past satisfies the gate
1209	// live — no gate_runs row needed.
1210	let pool = fresh_pool().await;
1211	seed(&pool, "a", "0.8.12").await;
1212	sqlx::query("UPDATE tier_state SET burn_in_started_at = ? WHERE tier = 'a'")
1213	.bind((chrono::Utc::now() - chrono::Duration::hours(50)).to_rfc3339())
1214	.execute(&pool)
1215	.await
1216	.unwrap();
1217	let pending = unsatisfied_gates(
1218	&pool,
1219	&tid("a"),
1220	&[Gate::BurnIn { hours: 48 }],
1221	"0.8.12",
1222	None,
1223	false,
1224	)
1225	.await
1226	.unwrap();
1227	assert!(pending.is_empty(), "50h elapsed satisfies a 48h burn-in");
1228	}
1229
1230	#[tokio::test]
1231	async fn unsatisfied_gates_ignores_other_tiers_and_versions() {
1232	let pool = fresh_pool().await;
1233	seed(&pool, "host", "0.8.12").await;
1234	seed(&pool, "host", "0.8.11").await;
1235	seed(&pool, "a", "0.8.12").await;
1236	// Mark host/0.8.12 cargo_test failing, but unrelated tiers/versions
1237	// shouldn't pollute the query.
1238	insert_gate(&pool, "host", "0.8.12", "cargo_test", 0).await;
1239	insert_gate(&pool, "a", "0.8.12", "cargo_test", 0).await;
1240	insert_gate(&pool, "host", "0.8.11", "cargo_test", 0).await;
1241
1242	let pending = unsatisfied_gates(
1243	&pool,
1244	&tid("host"),
1245	&[Gate::CargoTest],
1246	"0.8.12",
1247	None,
1248	false,
1249	)
1250	.await
1251	.unwrap();
1252	assert_eq!(pending, vec!["cargo_test".to_string()]);
1253	}
1254
1255	#[tokio::test]
1256	async fn unsatisfied_gates_null_status_is_treated_as_failing() {
1257	// An in-flight gate (started_at set, finished_at + status NULL)
1258	// should NOT be treated as green. Otherwise a race could promote
1259	// before the gate concludes.
1260	let pool = fresh_pool().await;
1261	seed(&pool, "host", "0.8.12").await;
1262	sqlx::query(
1263	"INSERT INTO gate_runs (version, tier, gate_kind, started_at) \
1264	VALUES ('0.8.12', 'host', 'cargo_test', datetime('now'))",
1265	)
1266	.execute(&pool)
1267	.await
1268	.unwrap();
1269
1270	let pending = unsatisfied_gates(
1271	&pool,
1272	&tid("host"),
1273	&[Gate::CargoTest],
1274	"0.8.12",
1275	None,
1276	false,
1277	)
1278	.await
1279	.unwrap();
1280	assert_eq!(pending, vec!["cargo_test".to_string()]);
1281	}
1282
1283	// ---- /confirm/{tier} ----
1284
1285	#[tokio::test]
1286	async fn confirm_rejects_when_tier_has_no_current_version() {
1287	// tier_state.a.current_version is NULL by default. /confirm has
1288	// nothing to confirm against → GateBlocked (400).
1289	let state = test_state().await;
1290	let app = router(state.clone());
1291	let resp = app
1292	.oneshot(
1293	Request::builder()
1294	.method("POST")
1295	.uri("/confirm/a")
1296	.body(Body::empty())
1297	.unwrap(),
1298	)
1299	.await
1300	.unwrap();
1301	assert_eq!(resp.status(), StatusCode::CONFLICT);
1302	let body = body_string(resp).await;
1303	assert!(body.contains("no current_version"), "got: {body}");
1304	}
1305
1306	#[tokio::test]
1307	async fn confirm_accepts_when_current_version_set_and_inserts_row() {
1308	let state = test_state().await;
1309	// Seed a version + advance tier a's state to it.
1310	sqlx::query("INSERT INTO versions (version, git_sha, built_at, artifact_path) VALUES ('0.8.12','sha',datetime('now'),'/tmp/x')")
1311	.execute(&state.pool).await.unwrap();
1312	sqlx::query("UPDATE tier_state SET current_version = '0.8.12' WHERE tier = 'a'")
1313	.execute(&state.pool)
1314	.await
1315	.unwrap();
1316
1317	let app = router(state.clone());
1318	let resp = app
1319	.oneshot(
1320	Request::builder()
1321	.method("POST")
1322	.uri("/confirm/a")
1323	.body(Body::empty())
1324	.unwrap(),
1325	)
1326	.await
1327	.unwrap();
1328	assert_eq!(resp.status(), StatusCode::OK);
1329	let body = body_string(resp).await;
1330	assert!(body.contains("\"tier\":\"a\""));
1331	assert!(body.contains("\"version\":\"0.8.12\""));
1332
1333	// A passing gate_runs row was inserted.
1334	let count: (i64,) = sqlx::query_as(
1335	"SELECT COUNT(*) FROM gate_runs WHERE tier='a' AND gate_kind='manual_confirm' AND status='passed'",
1336	)
1337	.fetch_one(&state.pool)
1338	.await
1339	.unwrap();
1340	assert_eq!(count.0, 1);
1341	}
1342
1343	#[tokio::test]
1344	async fn confirm_404s_for_unknown_tier() {
1345	let state = test_state().await;
1346	let app = router(state);
1347	let resp = app
1348	.oneshot(
1349	Request::builder()
1350	.method("POST")
1351	.uri("/confirm/zzzz")
1352	.body(Body::empty())
1353	.unwrap(),
1354	)
1355	.await
1356	.unwrap();
1357	assert_eq!(resp.status(), StatusCode::NOT_FOUND);
1358	}
1359
1360	#[tokio::test]
1361	async fn get_run_404s_for_unknown_id() {
1362	let state = test_state().await;
1363	let app = router(state);
1364	let resp = app
1365	.oneshot(
1366	Request::builder()
1367	.uri("/runs/999")
1368	.body(Body::empty())
1369	.unwrap(),
1370	)
1371	.await
1372	.unwrap();
1373	assert_eq!(resp.status(), StatusCode::NOT_FOUND);
1374	}
1375
1376	#[tokio::test]
1377	async fn get_run_returns_view_with_gates() {
1378	let state = test_state().await;
1379	// A run that reached version 0.10.2 and ran two host gates (one red).
1380	let run_id = crate::runs::create(&state.pool, "abc1234def")
1381	.await
1382	.unwrap();
1383	let ver: crate::domain::Version = "0.10.2".parse().unwrap();
1384	seed(&state.pool, "host", "0.10.2").await;
1385	crate::runs::set_version(&state.pool, run_id, &ver)
1386	.await
1387	.unwrap();
1388	insert_gate(&state.pool, "host", "0.10.2", "cargo_test", 0).await;
1389	insert_gate(&state.pool, "host", "0.10.2", "boot_smoke", 1).await;
1390
1391	let app = router(state);
1392	let resp = app
1393	.oneshot(
1394	Request::builder()
1395	.uri(format!("/runs/{}", run_id.0))
1396	.body(Body::empty())
1397	.unwrap(),
1398	)
1399	.await
1400	.unwrap();
1401	assert_eq!(resp.status(), StatusCode::OK);
1402	let v: serde_json::Value = serde_json::from_str(&body_string(resp).await).unwrap();
1403	assert_eq!(v["run_id"], run_id.0);
1404	assert_eq!(v["sha"], "abc1234def");
1405	assert_eq!(v["version"], "0.10.2");
1406	assert_eq!(v["result"], "building");
1407	// Both host gates surface, latest-per-kind, alphabetized by kind.
1408	assert_eq!(v["gates"].as_array().unwrap().len(), 2);
1409	assert_eq!(v["gates"][0]["kind"], "boot_smoke");
1410	assert_eq!(v["gates"][0]["status"], "passed");
1411	assert_eq!(v["gates"][1]["kind"], "cargo_test");
1412	assert_eq!(v["gates"][1]["status"], "failed");
1413	}
1414
1415	#[tokio::test]
1416	async fn get_run_wait_returns_immediately_when_settled() {
1417	let state = test_state().await;
1418	let run_id = crate::runs::create(&state.pool, "abc1234def")
1419	.await
1420	.unwrap();
1421	crate::runs::mark_passed(&state.pool, run_id).await.unwrap();
1422
1423	let app = router(state);
1424	// Generous timeout, but an already-settled run must not wait for it.
1425	let resp = app
1426	.oneshot(
1427	Request::builder()
1428	.uri(format!("/runs/{}/wait?timeout_ms=60000", run_id.0))
1429	.body(Body::empty())
1430	.unwrap(),
1431	)
1432	.await
1433	.unwrap();
1434	assert_eq!(resp.status(), StatusCode::OK);
1435	let v: serde_json::Value = serde_json::from_str(&body_string(resp).await).unwrap();
1436	assert_eq!(v["result"], "passed");
1437	}
1438
1439	#[tokio::test]
1440	async fn get_run_wait_returns_building_at_timeout() {
1441	let state = test_state().await;
1442	let run_id = crate::runs::create(&state.pool, "abc1234def")
1443	.await
1444	.unwrap();
1445
1446	let app = router(state);
1447	// timeout_ms=0 → deadline is now → the first poll returns the
1448	// still-building run rather than blocking.
1449	let resp = app
1450	.oneshot(
1451	Request::builder()
1452	.uri(format!("/runs/{}/wait?timeout_ms=0", run_id.0))
1453	.body(Body::empty())
1454	.unwrap(),
1455	)
1456	.await
1457	.unwrap();
1458	assert_eq!(resp.status(), StatusCode::OK);
1459	let v: serde_json::Value = serde_json::from_str(&body_string(resp).await).unwrap();
1460	assert_eq!(v["result"], "building");
1461	}
1462
1463	#[tokio::test]
1464	async fn get_run_wait_404s_for_unknown_id() {
1465	let state = test_state().await;
1466	let app = router(state);
1467	let resp = app
1468	.oneshot(
1469	Request::builder()
1470	.uri("/runs/999/wait?timeout_ms=0")
1471	.body(Body::empty())
1472	.unwrap(),
1473	)
1474	.await
1475	.unwrap();
1476	assert_eq!(resp.status(), StatusCode::NOT_FOUND);
1477	}
1478
1479	#[test]
1480	fn self_update_unit_maps_sha_to_instance() {
1481	let sha = crate::domain::GitSha::parse("abc1234def5678").unwrap();
1482	assert_eq!(
1483	self_update_unit(&sha),
1484	"sando-update@abc1234def5678.service"
1485	);
1486	}
1487
1488	#[tokio::test]
1489	async fn self_update_rejects_bad_sha_with_400() {
1490	// A malformed sha is a client error and must be rejected before any
1491	// privileged unit is triggered (so this test never shells out).
1492	let state = test_state().await;
1493	let app = router(state);
1494	let resp = app
1495	.oneshot(
1496	Request::builder()
1497	.method("POST")
1498	.uri("/self-update")
1499	.header("Content-Type", "application/json")
1500	.body(Body::from(r#"{"sha":"not-a-sha!"}"#))
1501	.unwrap(),
1502	)
1503	.await
1504	.unwrap();
1505	assert_eq!(resp.status(), StatusCode::BAD_REQUEST);
1506	}
1507
1508	// ---- /promote/{tier} default-version resolution ----
1509
1510	#[tokio::test]
1511	async fn promote_to_first_tier_is_rejected() {
1512	// tier 0 is host — you /rebuild, not /promote.
1513	let state = test_state().await;
1514	let app = router(state);
1515	let resp = app
1516	.oneshot(
1517	Request::builder()
1518	.method("POST")
1519	.uri("/promote/host")
1520	.body(Body::empty())
1521	.unwrap(),
1522	)
1523	.await
1524	.unwrap();
1525	assert_eq!(resp.status(), StatusCode::CONFLICT);
1526	let body = body_string(resp).await;
1527	assert!(
1528	body.contains("cannot /promote to the first tier"),
1529	"got: {body}"
1530	);
1531	}
1532
1533	#[tokio::test]
1534	async fn promote_without_body_and_no_predecessor_version_errors() {
1535	// tier a has no body version supplied AND its predecessor mm has
1536	// current_version=NULL. Should fail before any deploy.
1537	let state = test_state().await;
1538	let app = router(state);
1539	let resp = app
1540	.oneshot(
1541	Request::builder()
1542	.method("POST")
1543	.uri("/promote/a")
1544	.body(Body::empty())
1545	.unwrap(),
1546	)
1547	.await
1548	.unwrap();
1549	assert_eq!(resp.status(), StatusCode::CONFLICT);
1550	let body = body_string(resp).await;
1551	assert!(
1552	body.contains("no version specified") \|\| body.contains("no current_version"),
1553	"got: {body}"
1554	);
1555	}
1556
1557	#[tokio::test]
1558	async fn promote_blocked_when_predecessor_gate_never_ran() {
1559	// End-to-end CF1: the host tier configures boot_smoke but it never ran
1560	// (no gate_runs row). Promoting host -> a must be GateBlocked, citing the
1561	// unsatisfied gate, instead of waving through on zero evidence. A real
1562	// `versions` row is present so the ONLY thing that can block is the gate.
1563	let pool = fresh_pool().await;
1564	for (i, name) in ["host", "a"].iter().enumerate() {
1565	sqlx::query(
1566	"INSERT INTO tiers (name, ord, provisioned, canary) VALUES (?, ?, 1, 'sequential')",
1567	)
1568	.bind(name)
1569	.bind(i as i64)
1570	.execute(&pool)
1571	.await
1572	.unwrap();
1573	sqlx::query("INSERT INTO tier_state (tier) VALUES (?)")
1574	.bind(name)
1575	.execute(&pool)
1576	.await
1577	.unwrap();
1578	}
1579	let mut topo = test_topo();
1580	topo.tiers[0].gates = vec![Gate::BootSmoke]; // host configures a gate...
1581	let executors = Arc::new(crate::state::build_executors(&topo));
1582	let state = AppState {
1583	pool,
1584	topo: Arc::new(topo),
1585	cfg: Arc::new(test_cfg()),
1586	active_build: Arc::new(tokio::sync::Mutex::new(None)),
1587	deploy_lock: Arc::new(tokio::sync::Mutex::new(())),
1588	events: crate::events::channel(),
1589	executors,
1590	api_token: None,
1591	};
1592	sqlx::query("INSERT INTO versions (version, git_sha, built_at, artifact_path) VALUES ('0.8.12','sha',datetime('now'),'/tmp/x')")
1593	.execute(&state.pool).await.unwrap();
1594	sqlx::query("UPDATE tier_state SET current_version = '0.8.12' WHERE tier = 'host'")
1595	.execute(&state.pool)
1596	.await
1597	.unwrap();
1598
1599	let app = router(state);
1600	let resp = app
1601	.oneshot(
1602	Request::builder()
1603	.method("POST")
1604	.uri("/promote/a")
1605	.body(Body::empty())
1606	.unwrap(),
1607	)
1608	.await
1609	.unwrap();
1610	assert_eq!(resp.status(), StatusCode::CONFLICT);
1611	let body = body_string(resp).await;
1612	assert!(
1613	body.contains("boot_smoke"),
1614	"expected boot_smoke to block; got: {body}"
1615	);
1616	}
1617
1618	#[tokio::test]
1619	async fn migration_bearing_promote_requires_fresh_confirm() {
1620	// The predecessor (host) configures NO gates, so nothing would normally
1621	// block host -> a. A `bears_migration` promote must still be blocked on a
1622	// fresh `manual_confirm` it does not have: rollback restores the binary
1623	// only, so the one-way advance needs a conscious operator sign-off.
1624	let pool = fresh_pool().await;
1625	for (i, name) in ["host", "a"].iter().enumerate() {
1626	sqlx::query(
1627	"INSERT INTO tiers (name, ord, provisioned, canary) VALUES (?, ?, 1, 'sequential')",
1628	)
1629	.bind(name)
1630	.bind(i as i64)
1631	.execute(&pool)
1632	.await
1633	.unwrap();
1634	sqlx::query("INSERT INTO tier_state (tier) VALUES (?)")
1635	.bind(name)
1636	.execute(&pool)
1637	.await
1638	.unwrap();
1639	}
1640	let mut topo = test_topo();
1641	topo.tiers[0].gates = vec![]; // host configures no gates at all
1642	let executors = Arc::new(crate::state::build_executors(&topo));
1643	let state = AppState {
1644	pool,
1645	topo: Arc::new(topo),
1646	cfg: Arc::new(test_cfg()),
1647	active_build: Arc::new(tokio::sync::Mutex::new(None)),
1648	deploy_lock: Arc::new(tokio::sync::Mutex::new(())),
1649	events: crate::events::channel(),
1650	executors,
1651	api_token: None,
1652	};
1653	sqlx::query("INSERT INTO versions (version, git_sha, built_at, artifact_path) VALUES ('0.8.12','sha',datetime('now'),'/tmp/x')")
1654	.execute(&state.pool).await.unwrap();
1655	sqlx::query("UPDATE tier_state SET current_version = '0.8.12' WHERE tier = 'host'")
1656	.execute(&state.pool)
1657	.await
1658	.unwrap();
1659
1660	let app = router(state);
1661	let resp = app
1662	.oneshot(
1663	Request::builder()
1664	.method("POST")
1665	.uri("/promote/a")
1666	.header("content-type", "application/json")
1667	.body(Body::from(r#"{"bears_migration": true}"#))
1668	.unwrap(),
1669	)
1670	.await
1671	.unwrap();
1672	assert_eq!(resp.status(), StatusCode::CONFLICT);
1673	let body = body_string(resp).await;
1674	assert!(
1675	body.contains("manual_confirm"),
1676	"expected the migration promote to block on manual_confirm; got: {body}"
1677	);
1678	}
1679
1680	#[tokio::test]
1681	async fn tier_state_advance_is_atomic_previous_from_old_current() {
1682	// CF3: the promote advance is a single UPDATE where previous_version is
1683	// set from the row's old current_version (SQLite evaluates RHS against
1684	// the original row). No read-modify-write to lose under concurrency.
1685	let pool = fresh_pool().await;
1686	seed(&pool, "a", "1.0.0").await;
1687	// current_version FKs into versions, so the target must exist too.
1688	sqlx::query("INSERT INTO versions (version, git_sha, built_at, artifact_path) VALUES ('2.0.0','sha',datetime('now'),'/tmp/x')")
1689	.execute(&pool).await.unwrap();
1690	sqlx::query("UPDATE tier_state SET current_version = '1.0.0' WHERE tier = 'a'")
1691	.execute(&pool)
1692	.await
1693	.unwrap();
1694
1695	// Exercise the sealed forward-advance primitive itself — the same op
1696	// /promote and the host build path both call (S1), not a copy of its SQL.
1697	let v = crate::domain::Version::parse("2.0.0").unwrap();
1698	crate::runs::advance_tier(&pool, "a", &v, None)
1699	.await
1700	.unwrap();
1701
1702	let (cur, prev): (Option<String>, Option<String>) = sqlx::query_as(
1703	"SELECT current_version, previous_version FROM tier_state WHERE tier = 'a'",
1704	)
1705	.fetch_one(&pool)
1706	.await
1707	.unwrap();
1708	assert_eq!(cur.as_deref(), Some("2.0.0"));
1709	assert_eq!(
1710	prev.as_deref(),
1711	Some("1.0.0"),
1712	"previous = the pre-update current, atomically"
1713	);
1714	}
1715
1716	#[tokio::test]
1717	async fn canary_rollback_restores_deployed_nodes_to_previous_version() {
1718	use crate::topology::{Node, default_actuate, default_observe};
1719	let tmp = tempfile::tempdir().unwrap();
1720
1721	// Two local nodes, each pre-seeded as if a promote had flipped them to
1722	// 2.0.0 (current -> releases/2.0.0), with the prior 1.0.0 still on disk.
1723	let mut nodes = Vec::new();
1724	for name in ["n1", "n2"] {
1725	let rr = tmp.path().join(name);
1726	for v in ["1.0.0", "2.0.0"] {
1727	tokio::fs::create_dir_all(rr.join("releases").join(v))
1728	.await
1729	.unwrap();
1730	}
1731	tokio::fs::symlink("releases/2.0.0", rr.join("current"))
1732	.await
1733	.unwrap();
1734	nodes.push(Node {
1735	name: name.into(),
1736	ssh_target: "local".into(),
1737	release_root: rr.to_string_lossy().into_owned(),
1738	service_name: "x.service".into(),
1739	health_url: None,
1740	config_check_env_file: None,
1741	actuate: default_actuate(),
1742	observe: default_observe(),
1743	companions: Vec::new(),
1744	});
1745	}
1746
1747	let mut state = test_state().await;
1748	// The rollback target needs a versions row. The release dir name comes
1749	// from the artifact_path's parent (legacy layout: releases/<version>).
1750	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')")
1751	.execute(&state.pool).await.unwrap();
1752	let execs: crate::state::ExecutorMap = nodes
1753	.iter()
1754	.map(\|n\| (n.name.clone(), crate::state::build_executor(n)))
1755	.collect();
1756	state.executors = std::sync::Arc::new(execs);
1757
1758	let refs: Vec<&Node> = nodes.iter().collect();
1759	let restored = rollback_deployed_nodes(&state, &tid("a"), &refs, "1.0.0").await;
1760	assert_eq!(restored, 2, "both deployed nodes should be restored");
1761
1762	for n in &nodes {
1763	let cur = tokio::fs::read_link(std::path::Path::new(&n.release_root).join("current"))
1764	.await
1765	.unwrap();
1766	assert_eq!(
1767	cur.to_string_lossy(),
1768	"releases/1.0.0",
1769	"node {} rolled back",
1770	n.name
1771	);
1772	}
1773	}
1774
1775	#[tokio::test]
1776	async fn canary_rollback_is_noop_without_a_previous_artifact() {
1777	use crate::topology::{Node, default_actuate, default_observe};
1778	let tmp = tempfile::tempdir().unwrap();
1779	let rr = tmp.path().join("n1");
1780	tokio::fs::create_dir_all(&rr).await.unwrap();
1781	let node = Node {
1782	name: "n1".into(),
1783	ssh_target: "local".into(),
1784	release_root: rr.to_string_lossy().into_owned(),
1785	service_name: "x.service".into(),
1786	health_url: None,
1787	config_check_env_file: None,
1788	actuate: default_actuate(),
1789	observe: default_observe(),
1790	companions: Vec::new(),
1791	};
1792	let state = test_state().await; // no versions row for "9.9.9"
1793	let restored = rollback_deployed_nodes(&state, &tid("a"), &[&node], "9.9.9").await;
1794	assert_eq!(
1795	restored, 0,
1796	"no artifact to roll back to -> nothing restored, no panic"
1797	);
1798	}
1799
1800	// ---- FleetFake: a multi-node promote across recorded fake executors ----
1801	//
1802	// The route-level promote tests until now ran a single local node against a
1803	// real LocalExec, so the sequential-canary fan-out, the cross-node deploy
1804	// ordering, and the mid-canary rollback of already-flipped nodes had no
1805	// coverage. FleetFake records every deploy op (tagged by node) into one
1806	// shared log so ordering is visible across the fleet, and can fail any op
1807	// whose shell script or rsync target contains a marker — used to fail a
1808	// node's forward deploy of the new version while its rollback to the prior
1809	// version (a different `releases/<v>` path) still succeeds.
1810
1811	struct FleetFake {
1812	tag: String,
1813	caps: CapabilitySet,
1814	log: Arc<StdMutex<Vec<String>>>,
1815	fail_if_contains: Option<String>,
1816	}
1817
1818	impl FleetFake {
1819	fn fails(&self, text: &str) -> bool {
1820	self.fail_if_contains
1821	.as_deref()
1822	.is_some_and(\|m\| text.contains(m))
1823	}
1824	}
1825
1826	#[async_trait]
1827	impl Executor for FleetFake {
1828	async fn run_streaming(
1829	&self,
1830	step: &Step,
1831	_sink: &mut dyn LogSink,
1832	) -> anyhow::Result<RunOutput> {
1833	let script = step.argv.last().cloned().unwrap_or_default();
1834	self.log
1835	.lock()
1836	.unwrap()
1837	.push(format!("{}:run:{script}", self.tag));
1838	Ok(RunOutput {
1839	status: std::process::ExitStatus::from_raw(if self.fails(&script) {
1840	1 << 8
1841	} else {
1842	0
1843	}),
1844	stdout: Vec::new(),
1845	stderr: Vec::new(),
1846	})
1847	}
1848	async fn pull_file(
1849	&self,
1850	_r: &std::path::Path,
1851	_l: &std::path::Path,
1852	_o: &SyncOpts,
1853	) -> anyhow::Result<()> {
1854	Ok(())
1855	}
1856	async fn pull_dir(
1857	&self,
1858	_r: &std::path::Path,
1859	_l: &std::path::Path,
1860	_o: &SyncOpts,
1861	) -> anyhow::Result<()> {
1862	Ok(())
1863	}
1864	async fn pull_glob(
1865	&self,
1866	_g: &str,
1867	_l: &std::path::Path,
1868	_o: &SyncOpts,
1869	) -> anyhow::Result<()> {
1870	Ok(())
1871	}
1872	async fn push_dir(
1873	&self,
1874	_local: &std::path::Path,
1875	remote: &std::path::Path,
1876	_o: &SyncOpts,
1877	) -> anyhow::Result<()> {
1878	let dst = remote.display().to_string();
1879	self.log
1880	.lock()
1881	.unwrap()
1882	.push(format!("{}:push:{dst}", self.tag));
1883	if self.fails(&dst) {
1884	anyhow::bail!("fake rsync failure on {}", self.tag);
1885	}
1886	Ok(())
1887	}
1888	fn capabilities(&self) -> &CapabilitySet {
1889	&self.caps
1890	}
1891	}
1892
1893	/// Rebuild tier "a" with `names` as remote fake nodes sharing one op log,
1894	/// seed the version/tier_state prerequisites for a promote of 3.0.0 up from
1895	/// `host` (tier "a" starts on 2.0.0 with 1.0.0 behind it), and optionally
1896	/// make `fail_node` fail any op containing `fail_marker`. Returns the state
1897	/// and the shared log.
1898	async fn fleet_fixture(
1899	names: &[&str],
1900	fail_node: Option<&str>,
1901	fail_marker: &str,
1902	) -> (AppState, Arc<StdMutex<Vec<String>>>) {
1903	use crate::topology::{default_actuate, default_observe};
1904	let mut state = test_state().await;
1905	let log = Arc::new(StdMutex::new(Vec::<String>::new()));
1906
1907	let nodes: Vec<Node> = names
1908	.iter()
1909	.map(\|name\| Node {
1910	name: (*name).into(),
1911	ssh_target: format!("deploy@{name}"),
1912	release_root: format!("/tmp/fleet/{name}"),
1913	service_name: "makenotwork.service".into(),
1914	health_url: None,
1915	config_check_env_file: None,
1916	actuate: default_actuate(),
1917	observe: default_observe(),
1918	companions: Vec::new(),
1919	})
1920	.collect();
1921
1922	let mut topo = (*state.topo).clone();
1923	topo.tiers[1].nodes = nodes.clone();
1924	// Drop the tier's post-deploy gate: the deploy fan-out is the subject
1925	// here, and node_health would need its own probe wiring.
1926	topo.tiers[1].gates = vec![];
1927	state.topo = Arc::new(topo);
1928
1929	let execs: crate::state::ExecutorMap = nodes
1930	.iter()
1931	.map(\|n\| {
1932	let nm = n.name.to_string();
1933	let fail = fail_node == Some(nm.as_str());
1934	let exec: Arc<dyn Executor> = Arc::new(FleetFake {
1935	tag: nm,
1936	caps: CapabilitySet::from_tokens(["deploy", "restart"], ["health"]),
1937	log: log.clone(),
1938	fail_if_contains: fail.then(\|\| fail_marker.to_string()),
1939	});
1940	(n.name.clone(), exec)
1941	})
1942	.collect();
1943	state.executors = Arc::new(execs);
1944
1945	for n in &nodes {
1946	// deploys.node FKs into `nodes`.
1947	sqlx::query(
1948	"INSERT INTO nodes (name, tier, ssh_target, release_root) VALUES (?, 'a', ?, ?)",
1949	)
1950	.bind(&n.name)
1951	.bind(&n.ssh_target)
1952	.bind(&n.release_root)
1953	.execute(&state.pool)
1954	.await
1955	.unwrap();
1956	}
1957	for v in ["1.0.0", "2.0.0", "3.0.0"] {
1958	// Legacy (pre-identity) artifact_path is `releases/<version>/<bin>`,
1959	// so the release dir the node mirrors is named for the version. The
1960	// fixture reflects that real layout (parent basename == version).
1961	sqlx::query("INSERT INTO versions (version, git_sha, built_at, artifact_path) VALUES (?, 'sha', datetime('now'), ?)")
1962	.bind(v).bind(format!("/tmp/staged/releases/{v}/makenotwork")).execute(&state.pool).await.unwrap();
1963	}
1964	sqlx::query("UPDATE tier_state SET current_version = '3.0.0' WHERE tier = 'host'")
1965	.execute(&state.pool)
1966	.await
1967	.unwrap();
1968	sqlx::query(
1969	"UPDATE tier_state SET current_version = '2.0.0', previous_version = '1.0.0' WHERE tier = 'a'",
1970	)
1971	.execute(&state.pool)
1972	.await
1973	.unwrap();
1974	(state, log)
1975	}
1976
1977	/// Index of the first logged op belonging to `node` (panics if the node was
1978	/// never touched — the message names it).
1979	fn first_touch(log: &[String], node: &str) -> usize {
1980	let prefix = format!("{node}:");
1981	log.iter()
1982	.position(\|e\| e.starts_with(&prefix))
1983	.unwrap_or_else(\|\| panic!("node {node:?} was never deployed to; log: {log:#?}"))
1984	}
1985
1986	#[tokio::test]
1987	async fn promote_deploys_every_node_in_tier_order_and_advances() {
1988	let (state, log) = fleet_fixture(&["a1", "a2", "a3"], None, "").await;
1989	let pool = state.pool.clone();
1990
1991	let Json(body) = promote_inner(state, "a".into(), PromoteBody::default())
1992	.await
1993	.expect("all nodes deploy, so the promote succeeds");
1994	assert_eq!(
1995	body["nodes_deployed"],
1996	serde_json::json!(["a1", "a2", "a3"]),
1997	"the response names every node the promote reached",
1998	);
1999
2000	let (cur, prev) = tier_versions(&pool, "a").await;
2001	assert_eq!(cur.as_deref(), Some("3.0.0"));
2002	assert_eq!(prev.as_deref(), Some("2.0.0"));
2003
2004	// Sequential canary: a1 is fully touched before a2, a2 before a3.
2005	let log = log.lock().unwrap().clone();
2006	assert!(
2007	first_touch(&log, "a1") < first_touch(&log, "a2")
2008	&& first_touch(&log, "a2") < first_touch(&log, "a3"),
2009	"nodes must deploy in tier order: {log:#?}",
2010	);
2011
2012	// Every node has a green deploy row for the promoted version.
2013	let ok: i64 = sqlx::query_scalar(
2014	"SELECT COUNT(*) FROM deploys WHERE version = '3.0.0' AND outcome = 'ok'",
2015	)
2016	.fetch_one(&pool)
2017	.await
2018	.unwrap();
2019	assert_eq!(ok, 3, "one ok deploy row per node");
2020	}
2021
2022	#[tokio::test]
2023	async fn a_mid_canary_deploy_failure_rolls_touched_nodes_back_and_does_not_advance() {
2024	// a2 fails its forward deploy of 3.0.0; a1 was already flipped, a3 is
2025	// never reached. The touched nodes (a1, a2) roll back to 2.0.0 — their
2026	// rollback ops target `releases/2.0.0`, which the marker does not match —
2027	// and tier_state must NOT advance.
2028	let (state, log) = fleet_fixture(&["a1", "a2", "a3"], Some("a2"), "releases/3.0.0").await;
2029	let pool = state.pool.clone();
2030
2031	let err = promote_inner(state, "a".into(), PromoteBody::default())
2032	.await
2033	.expect_err("a mid-canary deploy failure must fail the promote");
2034	assert!(
2035	matches!(err, crate::error::Error::Other(_)),
2036	"a deploy failure propagates as Other, got: {err:?}",
2037	);
2038
2039	// tier_state untouched: the failure returns before advance_tier.
2040	let (cur, prev) = tier_versions(&pool, "a").await;
2041	assert_eq!(
2042	cur.as_deref(),
2043	Some("2.0.0"),
2044	"a failed rollout must not advance"
2045	);
2046	assert_eq!(prev.as_deref(), Some("1.0.0"));
2047
2048	let log = log.lock().unwrap().clone();
2049	// a3 sits after the failed a2 in the sequence and is never touched.
2050	assert!(
2051	!log.iter().any(\|e\| e.starts_with("a3:")),
2052	"nodes after the failure must not be deployed to: {log:#?}",
2053	);
2054	// Both touched nodes were rolled back to the prior version.
2055	for n in ["a1", "a2"] {
2056	assert!(
2057	log.iter()
2058	.any(\|e\| e.starts_with(&format!("{n}:")) && e.contains("releases/2.0.0")),
2059	"touched node {n} must be restored to 2.0.0: {log:#?}",
2060	);
2061	}
2062
2063	// The forward attempt is on the record: a1 ok, a2 failed.
2064	let a1: String = sqlx::query_scalar(
2065	"SELECT outcome FROM deploys WHERE version = '3.0.0' AND node = 'a1'",
2066	)
2067	.fetch_one(&pool)
2068	.await
2069	.unwrap();
2070	assert_eq!(a1, "ok");
2071	let a2: String = sqlx::query_scalar(
2072	"SELECT outcome FROM deploys WHERE version = '3.0.0' AND node = 'a2'",
2073	)
2074	.fetch_one(&pool)
2075	.await
2076	.unwrap();
2077	assert_eq!(a2, "failed");
2078
2079	// Both touched nodes restored => the tier is consistent on 2.0.0, so the
2080	// partial flag is cleared, not set.
2081	let reason: Option<String> =
2082	sqlx::query_scalar("SELECT partial_reason FROM tier_state WHERE tier = 'a'")
2083	.fetch_one(&pool)
2084	.await
2085	.unwrap();
2086	assert_eq!(
2087	reason, None,
2088	"a fully-restored canary leaves the tier consistent, not partial",
2089	);
2090	}
2091
2092	/// Build a one-node tier "a" on a tempdir release root, pre-seeded as if a
2093	/// promote had flipped it to `current` with `prev` still staged on disk.
2094	/// Returns the state (topology rewired to the tempdir node) and the tempdir,
2095	/// which the caller must keep alive.
2096	async fn rollback_fixture(prev: &str, current: &str) -> (AppState, tempfile::TempDir) {
2097	use crate::topology::{Node, default_actuate, default_observe};
2098	let tmp = tempfile::tempdir().unwrap();
2099	let rr = tmp.path().join("a-local");
2100	for v in [prev, current] {
2101	tokio::fs::create_dir_all(rr.join("releases").join(v))
2102	.await
2103	.unwrap();
2104	}
2105	tokio::fs::symlink(format!("releases/{current}"), rr.join("current"))
2106	.await
2107	.unwrap();
2108	let node = Node {
2109	name: "a-local".into(),
2110	ssh_target: "local".into(),
2111	release_root: rr.to_string_lossy().into_owned(),
2112	service_name: "x.service".into(),
2113	health_url: None,
2114	config_check_env_file: None,
2115	actuate: default_actuate(),
2116	observe: default_observe(),
2117	companions: Vec::new(),
2118	};
2119
2120	let mut state = test_state().await;
2121	let mut topo = (*state.topo).clone();
2122	topo.tiers[1].nodes = vec![node];
2123	state.executors = Arc::new(crate::state::build_executors(&topo));
2124	state.topo = Arc::new(topo);
2125
2126	// deploys.node FKs into `nodes`, so the promote path needs the row.
2127	sqlx::query("INSERT INTO nodes (name, tier, ssh_target, release_root) VALUES ('a-local', 'a', 'local', ?)")
2128	.bind(rr.to_string_lossy().into_owned())
2129	.execute(&state.pool).await.unwrap();
2130	for v in [prev, current] {
2131	sqlx::query("INSERT INTO versions (version, git_sha, built_at, artifact_path) VALUES (?, 'sha', datetime('now'), ?)")
2132	.bind(v).bind(format!("/tmp/staged/releases/{v}/makenotwork")).execute(&state.pool).await.unwrap();
2133	}
2134	sqlx::query(
2135	"UPDATE tier_state SET current_version = ?, previous_version = ? WHERE tier = 'a'",
2136	)
2137	.bind(current)
2138	.bind(prev)
2139	.execute(&state.pool)
2140	.await
2141	.unwrap();
2142	(state, tmp)
2143	}
2144
2145	async fn tier_versions(pool: &SqlitePool, tier: &str) -> (Option<String>, Option<String>) {
2146	sqlx::query_as("SELECT current_version, previous_version FROM tier_state WHERE tier = ?")
2147	.bind(tier)
2148	.fetch_one(pool)
2149	.await
2150	.unwrap()
2151	}
2152
2153	#[tokio::test]
2154	async fn rollback_clears_previous_version_rather_than_swapping_it() {
2155	// The swap bug: writing the version we just rolled OFF into
2156	// previous_version made a second /rollback roll FORWARD onto the broken
2157	// build the operator was escaping. previous_version must go NULL.
2158	let (state, _tmp) = rollback_fixture("1.0.0", "2.0.0").await;
2159	let pool = state.pool.clone();
2160
2161	let _ = rollback(State(state), Path("a".to_string())).await.unwrap();
2162
2163	let (cur, prev) = tier_versions(&pool, "a").await;
2164	assert_eq!(
2165	cur.as_deref(),
2166	Some("1.0.0"),
2167	"rolled back to the previous version"
2168	);
2169	assert_eq!(
2170	prev, None,
2171	"the version we rolled off must NOT become the rollback target"
2172	);
2173	}
2174
2175	#[tokio::test]
2176	async fn second_rollback_refuses_instead_of_rolling_forward() {
2177	let (state, _tmp) = rollback_fixture("1.0.0", "2.0.0").await;
2178	let pool = state.pool.clone();
2179	let release_root = state.topo.tiers[1].nodes[0].release_root.clone();
2180
2181	let _ = rollback(State(state.clone()), Path("a".to_string()))
2182	.await
2183	.unwrap();
2184	let err = rollback(State(state), Path("a".to_string()))
2185	.await
2186	.expect_err("only one step of history is tracked; a second rollback must refuse");
2187	assert!(
2188	matches!(&err, crate::error::Error::GateBlocked(m) if m.contains("no previous_version")),
2189	"expected a loud refusal, got: {err:?}",
2190	);
2191
2192	// The refusal is total: neither the DB nor the node moved back to 2.0.0.
2193	let (cur, _) = tier_versions(&pool, "a").await;
2194	assert_eq!(cur.as_deref(), Some("1.0.0"));
2195	let link = tokio::fs::read_link(std::path::Path::new(&release_root).join("current"))
2196	.await
2197	.unwrap();
2198	assert_eq!(
2199	link.to_string_lossy(),
2200	"releases/1.0.0",
2201	"node stays on the rolled-back version"
2202	);
2203	}
2204
2205	#[tokio::test]
2206	async fn promote_refuses_an_unprovisioned_tier() {
2207	// Every step of a promote to a node-less tier is a silent no-op that
2208	// still reports success: the deploy loop iterates nothing and
2209	// advance_tier records a current_version the tier never received.
2210	let (mut state, _tmp) = rollback_fixture("1.0.0", "2.0.0").await;
2211	let mut topo = (*state.topo).clone();
2212	topo.tiers[1].provisioned = false;
2213	topo.tiers[1].nodes.clear();
2214	state.topo = Arc::new(topo);
2215	let pool = state.pool.clone();
2216	sqlx::query("UPDATE tier_state SET current_version = '2.0.0' WHERE tier = 'host'")
2217	.execute(&pool)
2218	.await
2219	.unwrap();
2220
2221	let err = promote_inner(state, "a".into(), PromoteBody::default())
2222	.await
2223	.expect_err("promoting to an unprovisioned tier must be refused");
2224	assert!(
2225	matches!(&err, crate::error::Error::GateBlocked(m) if m.contains("not provisioned")),
2226	"got: {err:?}",
2227	);
2228
2229	// And nothing was recorded: the tier keeps whatever it had before.
2230	let (cur, _) = tier_versions(&pool, "a").await;
2231	assert_eq!(
2232	cur.as_deref(),
2233	Some("2.0.0"),
2234	"a refused promote must not advance tier_state",
2235	);
2236	}
2237
2238	#[tokio::test]
2239	async fn promote_advances_the_tier_and_flips_the_symlink_when_gates_are_green() {
2240	// The happy path. Every other promote test asserts a refusal or a red
2241	// outcome, so nothing pinned what a successful promote actually does:
2242	// deploy reaches the node, the `current` symlink flips, tier_state
2243	// advances with previous_version = the version we came off, any stale
2244	// partial flag clears, and the handler reports the nodes it touched.
2245	use crate::topology::Gate;
2246	let (mut state, _tmp) = rollback_fixture("1.0.0", "2.0.0").await;
2247
2248	// Source tier `host` gates the promote on cargo_test. Satisfying it is
2249	// the point of the test: the sibling case asserts an unsatisfied gate
2250	// blocks, this one asserts a satisfied gate lets the promote through.
2251	let mut topo = (*state.topo).clone();
2252	topo.tiers[0].gates = vec![Gate::CargoTest];
2253	state.topo = Arc::new(topo);
2254	let pool = state.pool.clone();
2255	let release_root = state.topo.tiers[1].nodes[0].release_root.clone();
2256
2257	// 3.0.0 is staged on the build host and green on `host`.
2258	tokio::fs::create_dir_all(
2259	std::path::Path::new(&release_root)
2260	.join("releases")
2261	.join("3.0.0"),
2262	)
2263	.await
2264	.unwrap();
2265	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')")
2266	.execute(&pool).await.unwrap();
2267	sqlx::query("UPDATE tier_state SET current_version = '3.0.0' WHERE tier = 'host'")
2268	.execute(&pool)
2269	.await
2270	.unwrap();
2271	insert_gate(&pool, "host", "3.0.0", "cargo_test", 1).await;
2272
2273	// A stale flag from an earlier incident, which a clean rollout clears.
2274	set_partial(&state, &tid("a"), "left over from a previous canary").await;
2275
2276	let Json(body) = promote_inner(state, "a".into(), PromoteBody::default())
2277	.await
2278	.expect("gates are green and the node deploys, so the promote succeeds");
2279
2280	assert_eq!(body["tier"], "a");
2281	assert_eq!(body["version"], "3.0.0");
2282	assert_eq!(
2283	body["nodes_deployed"],
2284	serde_json::json!(["a-local"]),
2285	"the response names every node the promote reached",
2286	);
2287
2288	let (cur, prev) = tier_versions(&pool, "a").await;
2289	assert_eq!(cur.as_deref(), Some("3.0.0"));
2290	assert_eq!(
2291	prev.as_deref(),
2292	Some("2.0.0"),
2293	"previous_version is the version we came off, so a rollback aims at it",
2294	);
2295
2296	// The node genuinely moved: the promote is not just bookkeeping.
2297	let link = tokio::fs::read_link(std::path::Path::new(&release_root).join("current"))
2298	.await
2299	.unwrap();
2300	assert_eq!(link.to_string_lossy(), "releases/3.0.0");
2301
2302	let reason: Option<String> =
2303	sqlx::query_scalar("SELECT partial_reason FROM tier_state WHERE tier = 'a'")
2304	.fetch_one(&pool)
2305	.await
2306	.unwrap();
2307	assert_eq!(
2308	reason, None,
2309	"a clean full rollout clears a stale partial flag",
2310	);
2311
2312	// The deploy is on the record as having succeeded, which is what the
2313	// next promote's gate check and /state both read.
2314	let (node, outcome): (String, String) =
2315	sqlx::query_as("SELECT node, outcome FROM deploys WHERE version = '3.0.0'")
2316	.fetch_one(&pool)
2317	.await
2318	.unwrap();
2319	assert_eq!(node, "a-local");
2320	assert_eq!(outcome, "ok");
2321	}
2322
2323	#[tokio::test]
2324	async fn promote_fails_and_flags_the_tier_when_post_deploy_gates_are_red() {
2325	// The deploy reached every node, so tier_state advances (a stale
2326	// current_version would aim a later rollback at the wrong artifact), but
2327	// the promote must NOT report success: the tier is flagged partial and
2328	// the handler returns the gate failure.
2329	use crate::topology::Gate;
2330	let (mut state, _tmp) = rollback_fixture("1.0.0", "2.0.0").await;
2331	// node_health is the tier's only gate, and it fails closed with no
2332	// probes — an empty executor map gives it nothing to probe while
2333	// deploy_node still falls back to a built executor and succeeds.
2334	let mut topo = (*state.topo).clone();
2335	topo.tiers[1].gates = vec![Gate::NodeHealth];
2336	state.topo = Arc::new(topo);
2337	state.executors = Arc::new(crate::state::ExecutorMap::new());
2338	let pool = state.pool.clone();
2339	// Promote 3.0.0 up from host, which configures no gates of its own.
2340	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')")
2341	.execute(&pool).await.unwrap();
2342	sqlx::query("UPDATE tier_state SET current_version = '3.0.0' WHERE tier = 'host'")
2343	.execute(&pool)
2344	.await
2345	.unwrap();
2346
2347	let err = promote_inner(state, "a".into(), PromoteBody::default())
2348	.await
2349	.expect_err("red post-deploy gates must fail the promote");
2350	assert!(
2351	matches!(&err, crate::error::Error::GateBlocked(m) if m.contains("node_health")),
2352	"the failure must name the red gate, got: {err:?}",
2353	);
2354
2355	let (cur, _) = tier_versions(&pool, "a").await;
2356	assert_eq!(
2357	cur.as_deref(),
2358	Some("3.0.0"),
2359	"tier_state tracks what the nodes actually run"
2360	);
2361	let reason: Option<String> =
2362	sqlx::query_scalar("SELECT partial_reason FROM tier_state WHERE tier = 'a'")
2363	.fetch_one(&pool)
2364	.await
2365	.unwrap();
2366	assert!(
2367	reason.as_deref().is_some_and(\|r\| r.contains("node_health")),
2368	"the tier must be flagged for /state and the TUI, got: {reason:?}",
2369	);
2370	}
2371
2372	#[tokio::test]
2373	async fn set_partial_then_clear_roundtrips() {
2374	let state = test_state().await;
2375	let read = \|\| async {
2376	sqlx::query_scalar::<_, Option<String>>(
2377	"SELECT partial_reason FROM tier_state WHERE tier = 'a'",
2378	)
2379	.fetch_one(&state.pool)
2380	.await
2381	.unwrap()
2382	};
2383	assert_eq!(read().await, None, "consistent tier starts clean");
2384	set_partial(&state, &tid("a"), "canary rollback incomplete: 1/2").await;
2385	assert_eq!(
2386	read().await.as_deref(),
2387	Some("canary rollback incomplete: 1/2")
2388	);
2389	clear_partial(&state, &tid("a")).await;
2390	assert_eq!(read().await, None, "clear nulls it back out");
2391	}
2392
2393	#[tokio::test]
2394	async fn state_surfaces_partial_reason() {
2395	use axum::extract::State;
2396	let state = test_state().await;
2397	set_partial(
2398	&state,
2399	&tid("a"),
2400	"first-deploy canary failed: 1 node(s) on 2.0.0",
2401	)
2402	.await;
2403	let Json(view) = get_state(State(state)).await.unwrap();
2404	let a = view.tiers.iter().find(\|t\| t.name == "a").unwrap();
2405	assert_eq!(
2406	a.partial_reason.as_deref(),
2407	Some("first-deploy canary failed: 1 node(s) on 2.0.0"),
2408	);
2409	let host = view.tiers.iter().find(\|t\| t.name == "host").unwrap();
2410	assert_eq!(
2411	host.partial_reason, None,
2412	"untouched tier stays clean in /state"
2413	);
2414	}
2415
2416	#[tokio::test]
2417	async fn state_build_is_null_until_first_rebuild_then_surfaces_latest() {
2418	use axum::extract::State;
2419	let state = test_state().await;
2420	// No build runs yet → build is null, so /state doesn't pretend a build
2421	// is happening.
2422	let Json(view) = get_state(State(state.clone())).await.unwrap();
2423	assert!(view.build.is_none());
2424
2425	// A failed run must surface its cause in /state, not just in /runs.
2426	let run_id = crate::runs::create(&state.pool, "deadbeef").await.unwrap();
2427	crate::runs::mark_failed(&state.pool, run_id, "cargo_test: 3 test(s) failed")
2428	.await
2429	.unwrap();
2430	let Json(view) = get_state(State(state)).await.unwrap();
2431	let b = view.build.expect("build surfaced");
2432	assert_eq!(b.run_id, run_id.0);
2433	assert_eq!(b.result, "failed");
2434	assert_eq!(
2435	b.failure_summary.as_deref(),
2436	Some("cargo_test: 3 test(s) failed")
2437	);
2438	}
2439
2440	#[tokio::test]
2441	async fn status_json_serves_the_shared_payload_over_the_real_router() {
2442	// The mapping itself is tested in `crate::status`. This asserts the
2443	// route is wired, serves valid JSON, and stays internally consistent
2444	// (no dangling child or action references) against a real topology
2445	// rather than a hand-built fixture.
2446	let state = test_state().await;
2447	set_partial(&state, &tid("a"), "canary rollback incomplete: 1/2").await;
2448
2449	let resp = router(state)
2450	.oneshot(
2451	Request::builder()
2452	.uri("/status.json")
2453	.body(Body::empty())
2454	.unwrap(),
2455	)
2456	.await
2457	.unwrap();
2458	assert_eq!(resp.status(), StatusCode::OK);
2459
2460	let body = http_body_util::BodyExt::collect(resp.into_body())
2461	.await
2462	.unwrap()
2463	.to_bytes();
2464	let payload: ops_status::Payload = serde_json::from_slice(&body).unwrap();
2465
2466	assert_eq!(payload.source, crate::status::SOURCE);
2467	assert_eq!(payload.schema, ops_status::SCHEMA_VERSION);
2468	assert_eq!(payload.validate(), Ok(()));
2469	assert_eq!(
2470	payload.node("tier:a").unwrap().status,
2471	ops_status::Status::Failed,
2472	"a partial tier must surface as failed over the wire"
2473	);
2474	assert_eq!(payload.worst_status(), ops_status::Status::Failed);
2475	}
2476
2477	#[tokio::test]
2478	async fn promote_with_explicit_version_but_missing_artifact_404s() {
2479	// Explicit version supplied, gates trivially pass (mm has none in
2480	// test_topo), but `versions` table has no row → 404.
2481	let state = test_state().await;
2482	let app = router(state);
2483	let resp = app
2484	.oneshot(
2485	Request::builder()
2486	.method("POST")
2487	.uri("/promote/a")
2488	.header("content-type", "application/json")
2489	.body(Body::from(r#"{"version":"9.9.9"}"#))
2490	.unwrap(),
2491	)
2492	.await
2493	.unwrap();
2494	assert_eq!(resp.status(), StatusCode::NOT_FOUND);
2495	}
2496
2497	// ---- GET /logs/{version}/{gate} ----
2498
2499	async fn state_with_logs_root(logs_root: PathBuf) -> AppState {
2500	let mut s = test_state().await;
2501	let mut cfg = (*s.cfg).clone();
2502	cfg.logs_root = logs_root;
2503	s.cfg = Arc::new(cfg);
2504	s
2505	}
2506
2507	#[tokio::test]
2508	async fn get_gate_log_returns_file_contents() {
2509	let tmp = tempfile::tempdir().unwrap();
2510	let dir = tmp.path().join("0.9.5");
2511	tokio::fs::create_dir_all(&dir).await.unwrap();
2512	tokio::fs::write(dir.join("cargo_test.log"), b"hello sandod\n")
2513	.await
2514	.unwrap();
2515
2516	let state = state_with_logs_root(tmp.path().to_path_buf()).await;
2517	let app = router(state);
2518	let resp = app
2519	.oneshot(
2520	Request::builder()
2521	.uri("/logs/0.9.5/cargo_test")
2522	.body(Body::empty())
2523	.unwrap(),
2524	)
2525	.await
2526	.unwrap();
2527	assert_eq!(resp.status(), StatusCode::OK);
2528	assert_eq!(body_string(resp).await, "hello sandod\n");
2529	}
2530
2531	#[tokio::test]
2532	async fn get_gate_log_404s_when_missing() {
2533	let tmp = tempfile::tempdir().unwrap();
2534	let state = state_with_logs_root(tmp.path().to_path_buf()).await;
2535	let app = router(state);
2536	let resp = app
2537	.oneshot(
2538	Request::builder()
2539	.uri("/logs/0.9.5/cargo_test")
2540	.body(Body::empty())
2541	.unwrap(),
2542	)
2543	.await
2544	.unwrap();
2545	assert_eq!(resp.status(), StatusCode::NOT_FOUND);
2546	}
2547
2548	// ---- CF2: bearer-token auth on deploy mutators ----
2549
2550	#[tokio::test]
2551	async fn mutating_route_requires_bearer_when_token_set() {
2552	let mut state = test_state().await;
2553	state.api_token = Some(std::sync::Arc::from("s3cr3t"));
2554	let app = router(state);
2555
2556	// No Authorization header -> 401, before any deploy logic runs.
2557	let resp = app
2558	.clone()
2559	.oneshot(
2560	Request::builder()
2561	.method("POST")
2562	.uri("/promote/a")
2563	.body(Body::empty())
2564	.unwrap(),
2565	)
2566	.await
2567	.unwrap();
2568	assert_eq!(resp.status(), StatusCode::UNAUTHORIZED);
2569
2570	// Wrong token -> 401.
2571	let resp = app
2572	.clone()
2573	.oneshot(
2574	Request::builder()
2575	.method("POST")
2576	.uri("/promote/a")
2577	.header("authorization", "Bearer nope")
2578	.body(Body::empty())
2579	.unwrap(),
2580	)
2581	.await
2582	.unwrap();
2583	assert_eq!(resp.status(), StatusCode::UNAUTHORIZED);
2584
2585	// Correct token -> passes auth (then blocked downstream by gates /
2586	// missing predecessor version, but specifically NOT 401).
2587	let resp = app
2588	.clone()
2589	.oneshot(
2590	Request::builder()
2591	.method("POST")
2592	.uri("/promote/a")
2593	.header("authorization", "Bearer s3cr3t")
2594	.body(Body::empty())
2595	.unwrap(),
2596	)
2597	.await
2598	.unwrap();
2599	assert_ne!(resp.status(), StatusCode::UNAUTHORIZED);
2600	}
2601
2602	#[tokio::test]
2603	async fn read_routes_require_token_when_set() {
2604	// Reads expose prod state (versions, SHAs, gate logs, the event stream),
2605	// so they are bearer-gated too — not just the mutators. A tailnet peer
2606	// without the token gets 401; the TUI presents the token and gets 200.
2607	let mut state = test_state().await;
2608	state.api_token = Some(std::sync::Arc::from("s3cr3t"));
2609	let app = router(state);
2610
2611	// No token -> 401 on a read.
2612	let resp = app
2613	.clone()
2614	.oneshot(
2615	Request::builder()
2616	.uri("/state")
2617	.body(Body::empty())
2618	.unwrap(),
2619	)
2620	.await
2621	.unwrap();
2622	assert_eq!(resp.status(), StatusCode::UNAUTHORIZED);
2623
2624	// Correct token -> 200.
2625	let resp = app
2626	.oneshot(
2627	Request::builder()
2628	.uri("/state")
2629	.header("authorization", "Bearer s3cr3t")
2630	.body(Body::empty())
2631	.unwrap(),
2632	)
2633	.await
2634	.unwrap();
2635	assert_eq!(resp.status(), StatusCode::OK);
2636	}
2637
2638	#[tokio::test]
2639	async fn read_routes_open_without_a_token() {
2640	// The loopback/dev posture: no token configured, reads pass through so a
2641	// local TUI needs no credential.
2642	let state = test_state().await;
2643	let app = router(state);
2644	let resp = app
2645	.oneshot(
2646	Request::builder()
2647	.uri("/state")
2648	.body(Body::empty())
2649	.unwrap(),
2650	)
2651	.await
2652	.unwrap();
2653	assert_eq!(resp.status(), StatusCode::OK);
2654	}
2655
2656	#[tokio::test]
2657	async fn self_update_malformed_body_is_400_not_422() {
2658	// TypedBody funnels a JSON deserialize failure through the Error envelope
2659	// (400), not axum's raw 422 — keeping every mutator on one error contract.
2660	let state = test_state().await; // no token -> auth passes, body is the gate
2661	let app = router(state);
2662	let resp = app
2663	.oneshot(
2664	Request::builder()
2665	.method("POST")
2666	.uri("/self-update")
2667	.header("content-type", "application/json")
2668	.body(Body::from("{ not valid json"))
2669	.unwrap(),
2670	)
2671	.await
2672	.unwrap();
2673	assert_eq!(resp.status(), StatusCode::BAD_REQUEST);
2674	}
2675
2676	#[tokio::test]
2677	async fn gate_log_rejects_unknown_gate_kind() {
2678	// The gate segment is an allowlisted GateKind, not a free-form filename:
2679	// an unknown kind is a 404, so `*.log` basenames can't be probed.
2680	let state = test_state().await;
2681	let app = router(state);
2682	let resp = app
2683	.oneshot(
2684	Request::builder()
2685	.uri("/logs/0.9.6/passwd")
2686	.body(Body::empty())
2687	.unwrap(),
2688	)
2689	.await
2690	.unwrap();
2691	assert_eq!(resp.status(), StatusCode::NOT_FOUND);
2692	}
2693
2694	/// Path-traversal guard: a `..` segment must not escape logs_root.
2695	/// axum's `{name}` param already rejects a literal `/` in the value, but
2696	/// `..` as a whole segment is structurally valid and must be blocked at
2697	/// the handler.
2698	#[tokio::test]
2699	async fn get_gate_log_rejects_dotdot_segments() {
2700	let tmp = tempfile::tempdir().unwrap();
2701	let state = state_with_logs_root(tmp.path().to_path_buf()).await;
2702	let app = router(state);
2703	let resp = app
2704	.oneshot(
2705	Request::builder()
2706	.uri("/logs/../etc/passwd")
2707	.body(Body::empty())
2708	.unwrap(),
2709	)
2710	.await
2711	.unwrap();
2712	assert_eq!(resp.status(), StatusCode::NOT_FOUND);
2713	}
2714
2715	// ---- build-identity path (wiki release-artifact-identity) ----
2716
2717	/// Insert a settled (passed) build_runs row with content identity and return
2718	/// its id. `staged_path` is the content-addressed release dir the bundle was
2719	/// published to (`releases/<digest16>`).
2720	async fn seed_build(pool: &SqlitePool, sha: &str, version: &str, staged_path: &str) -> i64 {
2721	sqlx::query_scalar(
2722	"INSERT INTO build_runs (sha, version, phase, result, started_at, bundle_digest, staged_path)
2723	VALUES (?, ?, 'done', 'passed', datetime('now'), ?, ?) RETURNING id",
2724	)
2725	.bind(sha)
2726	.bind(version)
2727	.bind(format!("{sha}-digest"))
2728	.bind(staged_path)
2729	.fetch_one(pool)
2730	.await
2731	.unwrap()
2732	}
2733
2734	async fn insert_gate_build(
2735	pool: &SqlitePool,
2736	tier: &str,
2737	version: &str,
2738	kind: &str,
2739	passed: i64,
2740	build_id: i64,
2741	) {
2742	let status = if passed == 1 { "passed" } else { "failed" };
2743	sqlx::query(
2744	"INSERT INTO gate_runs (version, tier, gate_kind, started_at, finished_at, status, build_id) \
2745	VALUES (?, ?, ?, datetime('now'), datetime('now'), ?, ?)",
2746	)
2747	.bind(version)
2748	.bind(tier)
2749	.bind(kind)
2750	.bind(status)
2751	.bind(build_id)
2752	.execute(pool)
2753	.await
2754	.unwrap();
2755	}
2756
2757	#[tokio::test]
2758	async fn unsatisfied_gates_keys_build_evidence_on_build_id() {
2759	let pool = fresh_pool().await;
2760	seed(&pool, "a", "3.0.0").await;
2761	// Two builds of the SAME version string. Only b1's gate ran.
2762	let b1 = seed_build(&pool, "sha1", "3.0.0", "/rel/1111111111111111").await;
2763	let b2 = seed_build(&pool, "sha2", "3.0.0", "/rel/2222222222222222").await;
2764	insert_gate_build(&pool, "a", "3.0.0", "cargo_test", 1, b1).await;
2765
2766	// b1's own evidence satisfies.
2767	let ok = unsatisfied_gates(
2768	&pool,
2769	&tid("a"),
2770	&[Gate::CargoTest],
2771	"3.0.0",
2772	Some(b1),
2773	false,
2774	)
2775	.await
2776	.unwrap();
2777	assert!(ok.is_empty(), "the build that passed its gate is satisfied");
2778
2779	// b2 shares the version but has no gate row of its own: fail closed. This
2780	// is the hole — a rebuild reusing the version must not ride b1's evidence.
2781	let bad = unsatisfied_gates(
2782	&pool,
2783	&tid("a"),
2784	&[Gate::CargoTest],
2785	"3.0.0",
2786	Some(b2),
2787	false,
2788	)
2789	.await
2790	.unwrap();
2791	assert_eq!(
2792	bad,
2793	vec!["cargo_test".to_string()],
2794	"a different build of the same version does not inherit the evidence"
2795	);
2796
2797	// Legacy (pre-identity) callers still resolve by version string.
2798	let legacy = unsatisfied_gates(&pool, &tid("a"), &[Gate::CargoTest], "3.0.0", None, false)
2799	.await
2800	.unwrap();
2801	assert!(legacy.is_empty(), "version-keyed legacy path unchanged");
2802	}
2803
2804	#[tokio::test]
2805	async fn promote_rejects_an_explicit_version_that_is_not_the_source_build() {
2806	// The burn-in hole: `promote --version Y` used to check the SOURCE tier's
2807	// clock/evidence (which belong to whatever is current there), letting Y
2808	// inherit another build's 48h. Now promote resolves the source's current
2809	// build and refuses an explicit version that isn't it.
2810	let state = test_state().await;
2811	seed_version(&state.pool, "3.0.0").await;
2812	let b = seed_build(&state.pool, "shaB", "3.0.0", "/rel/deadbeefdeadbeef").await;
2813	sqlx::query(
2814	"UPDATE tier_state SET current_version='3.0.0', current_build_id=? WHERE tier='host'",
2815	)
2816	.bind(b)
2817	.execute(&state.pool)
2818	.await
2819	.unwrap();
2820
2821	let err = promote_inner(
2822	state,
2823	"a".into(),
2824	PromoteBody {
2825	version: Some("2.0.0".into()),
2826	..Default::default()
2827	},
2828	)
2829	.await
2830	.expect_err("promoting a version other than the source build must be refused");
2831	assert!(
2832	matches!(&err, crate::error::Error::GateBlocked(m) if m.contains("vouched for")),
2833	"the refusal must name the mismatch, got: {err:?}",
2834	);
2835	}
2836
2837	#[tokio::test]
2838	async fn promote_identity_path_deploys_the_source_build_and_advances_build_id() {
2839	// Full promote through the identity path: source tier points at a build,
2840	// promote resolves the artifact through it, deploys build_runs.staged_path
2841	// (content-addressed), and advances the target's current_build_id.
2842	let mut state = test_state().await;
2843	let node_root = tempfile::tempdir().unwrap();
2844	// Point the a-local node at a real tempdir so the symlink swap is checkable.
2845	let mut topo = (*state.topo).clone();
2846	topo.tiers[1].nodes[0].release_root = node_root.path().to_string_lossy().into_owned();
2847	topo.tiers[1].gates = vec![]; // isolate the deploy fan-out
2848	state.topo = Arc::new(topo);
2849	state.executors = Arc::new(crate::state::build_executors(&state.topo));
2850
2851	// deploys.node FKs into `nodes`, so the promote path needs the row.
2852	sqlx::query("INSERT INTO nodes (name, tier, ssh_target, release_root) VALUES ('a-local', 'a', 'local', ?)")
2853	.bind(node_root.path().to_string_lossy().into_owned())
2854	.execute(&state.pool)
2855	.await
2856	.unwrap();
2857
2858	seed_version(&state.pool, "3.0.0").await;
2859	let staged = format!(
2860	"{}/releases/abc123abc123abc1",
2861	node_root.path().to_string_lossy()
2862	);
2863	let b = seed_build(&state.pool, "shaB", "3.0.0", &staged).await;
2864	sqlx::query(
2865	"UPDATE tier_state SET current_version='3.0.0', current_build_id=? WHERE tier='host'",
2866	)
2867	.bind(b)
2868	.execute(&state.pool)
2869	.await
2870	.unwrap();
2871
2872	let pool = state.pool.clone();
2873	let Json(body) = promote_inner(state, "a".into(), PromoteBody::default())
2874	.await
2875	.expect("identity-path promote succeeds");
2876	assert_eq!(body["version"], "3.0.0");
2877
2878	// Target tier advanced with the build identity, not just the version.
2879	let (cur_v, cur_b): (Option<String>, Option<i64>) = sqlx::query_as(
2880	"SELECT current_version, current_build_id FROM tier_state WHERE tier = 'a'",
2881	)
2882	.fetch_one(&pool)
2883	.await
2884	.unwrap();
2885	assert_eq!(cur_v.as_deref(), Some("3.0.0"));
2886	assert_eq!(cur_b, Some(b), "target records the promoted build id");
2887
2888	// The deploy row is attributed to the build.
2889	let deploy_b: Option<i64> = sqlx::query_scalar(
2890	"SELECT build_id FROM deploys WHERE tier = 'a' ORDER BY id DESC LIMIT 1",
2891	)
2892	.fetch_one(&pool)
2893	.await
2894	.unwrap();
2895	assert_eq!(deploy_b, Some(b));
2896
2897	// The node's `current` points at the content-addressed release dir, whose
2898	// name is the staged_path's basename — not the version.
2899	let link = tokio::fs::read_link(node_root.path().join("current"))
2900	.await
2901	.unwrap();
2902	assert_eq!(link.to_string_lossy(), "releases/abc123abc123abc1");
2903	}
2904
2905	/// Insert a bare `versions` row (FK target for tier_state.current_version).
2906	async fn seed_version(pool: &SqlitePool, version: &str) {
2907	sqlx::query("INSERT INTO versions (version, git_sha, built_at, artifact_path) VALUES (?, 'sha', datetime('now'), '/tmp/x') ON CONFLICT DO NOTHING")
2908	.bind(version).execute(pool).await.unwrap();
2909	}
2910	}
2911