max / makenotwork

135.3 KB · 3570 lines History Blame Raw

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