Skip to main content

max / makenotwork

133.0 KB · 3322 lines History Blame Raw
1 //! Gate execution. Each gate kind has a runner that produces a pass/fail
2 //! outcome plus an optional detail string (typically a stderr tail or a
3 //! human-readable reason). Outcomes are persisted to `gate_runs` so /state
4 //! and the TUI can show them.
5
6 use crate::classify;
7 use crate::config::Config;
8 use crate::domain::{GateKind, GateRunId, TierId, Version};
9 use crate::events::{self, Event, EventTx};
10 use crate::outcome::{GateBlocker, GateFailure, GateOutcome, LogRef, PassNote};
11 use crate::topology::Gate;
12 use anyhow::{Context, Result};
13 use chrono::Utc;
14 use ops_core::live_log::LiveLog;
15 use ops_core::remote::LogSink; // brings `LiveLog::write_chunk` (the sink trait) into scope
16 use sqlx::SqlitePool;
17 use std::path::PathBuf;
18 use std::sync::Arc;
19 use tokio::io::AsyncReadExt;
20 use tokio::process::Command;
21
22 /// The gate live-log callback: emit each chunk as a `GateLogChunk` event so the
23 /// TUI sees the tail stream in real time. `ops_core::live_log::LiveLog` owns the
24 /// disk append and the per-run sequence counter; this closure is the one
25 /// tool-specific bit (Sando previously carried a whole `live_log.rs` copy that
26 /// hardcoded exactly this emit).
27 fn gate_chunk_cb(events: EventTx, run_id: GateRunId) -> ops_core::live_log::ChunkCallback {
28 Box::new(move |seq, text| {
29 events::emit(
30 &events,
31 Event::GateLogChunk {
32 run_id,
33 seq,
34 text: text.to_owned(),
35 },
36 );
37 })
38 }
39
40 pub struct GateCtx {
41 pub pool: SqlitePool,
42 pub cfg: Arc<Config>,
43 pub tier: TierId,
44 pub version: Version,
45 pub worktree: PathBuf,
46 pub events: EventTx,
47 /// Nodes the `node_health` post-deploy gate probes. Empty for build-time
48 /// gate runs on the host (where `node_health` never appears); filled at
49 /// promote time with each freshly-deployed node and its executor.
50 pub nodes: Vec<NodeProbe>,
51 /// The `build_runs.id` this gate run vouches for — the artifact identity
52 /// (wiki [[release-artifact-identity]]). Recorded on every `gate_runs` row so
53 /// promote can resolve the artifact through the evidence for a specific build,
54 /// not through a version string that a later rebuild can silently reuse.
55 /// `None` for legacy/pre-identity runs and gate unit tests.
56 pub build_id: Option<i64>,
57 }
58
59 /// One node the `node_health` gate verifies: its id, the systemd unit to
60 /// confirm active after the restart, an optional HTTP readiness URL, and the
61 /// executor that reaches it (the same transport the deploy used).
62 pub struct NodeProbe {
63 pub node: crate::domain::NodeId,
64 pub service: String,
65 pub health_url: Option<String>,
66 pub executor: Arc<dyn ops_exec::Executor>,
67 }
68
69 /// Run a single gate end-to-end: insert the in-flight row, execute the gate,
70 /// update the row with the outcome. Returns the outcome for the caller.
71 pub async fn run(ctx: &GateCtx, gate: &Gate) -> Result<GateOutcome> {
72 let kind = gate.kind();
73 let started_at = Utc::now().to_rfc3339();
74
75 let id: i64 = sqlx::query_scalar(
76 "INSERT INTO gate_runs (version, tier, gate_kind, started_at, build_id) VALUES (?, ?, ?, ?, ?)
77 RETURNING id",
78 )
79 .bind(&ctx.version)
80 .bind(&ctx.tier)
81 .bind(kind)
82 .bind(&started_at)
83 .bind(ctx.build_id)
84 .fetch_one(&ctx.pool)
85 .await?;
86 let run_id = GateRunId(id);
87
88 tracing::info!(
89 run_id = %run_id, tier = %ctx.tier, version = %ctx.version, gate = %kind,
90 "gate start",
91 );
92 events::emit(
93 &ctx.events,
94 Event::GateStart {
95 run_id,
96 tier: ctx.tier.clone(),
97 version: ctx.version.clone(),
98 gate: kind,
99 },
100 );
101
102 let outcome = match gate {
103 // cargo_test bounds its own run internally (it kills the specific child).
104 Gate::CargoTest => cargo_test(ctx, run_id).await,
105 // hardening_test bounds its own run internally, same as cargo_test.
106 Gate::HardeningTest => hardening_test(ctx, run_id).await,
107 // Each bounds itself the same way cargo_test does: one deadline across
108 // every target, so N crates cannot multiply the ceiling by N.
109 Gate::Clippy => clippy(ctx, run_id).await,
110 Gate::Fmt => fmt_check(ctx, run_id).await,
111 Gate::CargoAudit => supply_chain(ctx, run_id, GateKind::CargoAudit).await,
112 Gate::CargoDeny => supply_chain(ctx, run_id, GateKind::CargoDeny).await,
113 // migration_dry_run's psql restore + sqlx migrate could wedge; bound the
114 // whole gate here. Its bash restore sets kill_on_drop, so a timeout-drop
115 // doesn't orphan it.
116 Gate::MigrationDryRun => {
117 let ceiling = std::time::Duration::from_secs(ctx.cfg.gate_timeout_secs);
118 match tokio::time::timeout(ceiling, migration_dry_run(ctx, run_id)).await {
119 Ok(res) => res,
120 Err(_elapsed) => Ok(GateOutcome::failed(GateFailure::Timeout {
121 gate: GateKind::MigrationDryRun,
122 after_s: ctx.cfg.gate_timeout_secs as u32,
123 })
124 .with_log_ref(LogRef::new(&ctx.version, GateKind::MigrationDryRun))),
125 }
126 }
127 // code_smoke boots the real binary (migrate-from-scratch + seed + serve),
128 // any step of which could wedge; bound the whole gate here. Both child
129 // processes set kill_on_drop, so a timeout-drop can't orphan them. A
130 // timeout leaves the throwaway DB behind; the next run's createdb drops
131 // it first (DROP IF EXISTS), same as migration_dry_run's scratch reset.
132 Gate::CodeSmoke => {
133 let ceiling = std::time::Duration::from_secs(ctx.cfg.gate_timeout_secs);
134 match tokio::time::timeout(ceiling, code_smoke(ctx, run_id)).await {
135 Ok(res) => res,
136 Err(_elapsed) => Ok(GateOutcome::failed(GateFailure::Timeout {
137 gate: GateKind::CodeSmoke,
138 after_s: ctx.cfg.gate_timeout_secs as u32,
139 })
140 .with_log_ref(LogRef::new(&ctx.version, GateKind::CodeSmoke))),
141 }
142 }
143 Gate::BootSmoke => boot_smoke(ctx, run_id).await,
144 Gate::NodeHealth => node_health(ctx).await,
145 Gate::BurnIn { hours } => burn_in(ctx, *hours).await,
146 Gate::ManualConfirm => manual_confirm(ctx).await,
147 };
148
149 let outcome = outcome.unwrap_or_else(|e| {
150 GateOutcome::failed(GateFailure::Unclassified {
151 legacy_detail: Some(format!("gate runner errored: {e}")),
152 })
153 });
154
155 let outcome_json = serde_json::to_string(&outcome)
156 .unwrap_or_else(|e| format!("{{\"_serialize_error\":{e:?}}}"));
157 sqlx::query(
158 "UPDATE gate_runs
159 SET finished_at = ?, status = ?, outcome_json = ?, log_ref = ?
160 WHERE id = ?",
161 )
162 .bind(Utc::now().to_rfc3339())
163 .bind(outcome.status_str())
164 .bind(&outcome_json)
165 .bind(outcome.log_ref.as_ref().map(super::outcome::LogRef::as_str))
166 .bind(id)
167 .execute(&ctx.pool)
168 .await?;
169
170 tracing::info!(
171 tier = %ctx.tier, version = %ctx.version, gate = %kind,
172 status = outcome.status_str(), "gate done",
173 );
174 events::emit(
175 &ctx.events,
176 Event::GateDone {
177 run_id,
178 tier: ctx.tier.clone(),
179 version: ctx.version.clone(),
180 gate: kind,
181 outcome: outcome.clone(),
182 },
183 );
184
185 Ok(outcome)
186 }
187
188 /// Run every gate in order and return the kinds that did not pass (empty means
189 /// green). We deliberately do NOT short-circuit on first failure — every gate's
190 /// outcome is recorded in `gate_runs`, which is the operator's only visibility
191 /// into pipeline health. Hiding later gates because an earlier one failed makes
192 /// diagnosis worse.
193 ///
194 /// Returning the failing kinds rather than a bare bool is what lets the promote
195 /// path name them in the tier's `partial_reason` and in the error it returns to
196 /// the operator, instead of a generic "something was red".
197 pub async fn run_all(ctx: &GateCtx, gates: &[Gate]) -> Result<Vec<GateKind>> {
198 let mut failed = Vec::new();
199 for g in gates {
200 let o = run(ctx, g).await?;
201 if !o.is_passed() {
202 failed.push(g.kind());
203 }
204 }
205 Ok(failed)
206 }
207
208 // ---- individual gate runners ----
209
210 /// Run every configured `test_target`'s suite, in order, under one gate.
211 ///
212 /// This used to be hardcoded to `worktree/server`, which meant every other
213 /// crate in the repo shipped ungated — including `mnw-cli`, which is built as a
214 /// companion and installed onto prod-1 in the same promote. The targets are now
215 /// configured (`[[test_target]]` in the daemon config), defaulting to the
216 /// historical single `server` entry.
217 ///
218 /// The whole set shares one `gate_runs` row and one log file: from the
219 /// pipeline's point of view "the tests" either pass or don't. The first failing
220 /// target ends the gate, since a red suite blocks the promote regardless of what
221 /// the remaining crates would have said, and running them would only delay the
222 /// operator's answer. Its name is carried in the failure so the summary points
223 /// at the crate, not just the test.
224 async fn cargo_test(ctx: &GateCtx, run_id: GateRunId) -> Result<GateOutcome> {
225 let log_path = gate_log_path(ctx, GateKind::CargoTest);
226 let log_ref = LogRef::new(&ctx.version, GateKind::CargoTest);
227
228 // Best-effort: drop our own role's stale `mnw_test_*` databases (the
229 // template + any per-test clones orphaned by a previously-killed run)
230 // before the suite, so they can't accumulate or collide. Foreign-owned
231 // leftovers are left alone — the harness now namespaces its template per
232 // role, so they no longer wedge the gate.
233 if let Some(scratch_url) = ctx.cfg.scratch_db_url.as_deref() {
234 clean_stale_test_dbs(scratch_url).await;
235 }
236
237 let started = std::time::Instant::now();
238 // One ceiling for the whole gate, not per target: the point is to bound how
239 // long a hung suite can block the pipeline, and N targets each allowed the
240 // full timeout would multiply that by N.
241 let deadline = started + std::time::Duration::from_secs(ctx.cfg.gate_timeout_secs);
242 let mut ran = 0usize;
243
244 for target in &ctx.cfg.test_targets {
245 let dir = ctx.worktree.join(&target.dir);
246 // A target absent from this sha is skipped, not fatal: sando has to be
247 // able to build older shas (bisect, rollback rebuild) from a config that
248 // describes the tip. The zero-targets-ran check below is what stops this
249 // from quietly turning the gate into a no-op.
250 if !dir.join("Cargo.toml").is_file() {
251 tracing::warn!(
252 target = %target.dir.display(), version = %ctx.version,
253 "test_target has no Cargo.toml in this worktree; skipping",
254 );
255 continue;
256 }
257 let features: Vec<&str> = target.features.iter().map(String::as_str).collect();
258
259 // Fast pre-gate: compile the test targets WITHOUT running them. This
260 // builds the exact artifacts the full run needs (so the subsequent run
261 // reuses the cache — no wasted work), but fails in ~minutes with the
262 // real `error[Ennnn]: ...` on a test-only-target compile break. That
263 // class (a field missing in a `#[cfg(test)]`-only binary like `load`)
264 // otherwise compiles fine under the build step and only blows up here,
265 // after a full build, as an opaque mass test failure.
266 let banner = format!("\n==== test_target: {} ====\n", target.dir.display());
267 append_to_log(&log_path, banner.as_bytes()).await;
268 let mut pre = match cargo_test_command(ctx, &dir, target, &features, &["--no-run"]).spawn()
269 {
270 Ok(c) => c,
271 Err(e) => {
272 return Ok(GateOutcome::failed(GateFailure::SpawnFailed {
273 message: format!("{}: {e}", target.dir.display()),
274 })
275 .with_log_ref(log_ref));
276 }
277 };
278 let (pre_out, pre_err, pre_status) = match run_to_deadline_for(
279 &mut pre,
280 ctx,
281 run_id,
282 log_path.clone(),
283 deadline,
284 started,
285 GateKind::CargoTest,
286 )
287 .await?
288 {
289 Ok(v) => v,
290 Err(timeout) => return Ok(timeout.with_log_ref(log_ref)),
291 };
292 if !pre_status.success() {
293 let failure = classify::classify_compile_error(&pre_out, &pre_err);
294 return Ok(GateOutcome::failed(name_target(failure, &target.dir)).with_log_ref(log_ref));
295 }
296
297 // Full run: the test binaries are already built above, so cargo's
298 // up-to-date check skips compilation and this just runs the tests.
299 let mut child = match cargo_test_command(ctx, &dir, target, &features, &[]).spawn() {
300 Ok(c) => c,
301 Err(e) => {
302 return Ok(GateOutcome::failed(GateFailure::SpawnFailed {
303 message: format!("{}: {e}", target.dir.display()),
304 })
305 .with_log_ref(log_ref));
306 }
307 };
308 let (stdout_buf, stderr_buf, status) = match run_to_deadline_for(
309 &mut child,
310 ctx,
311 run_id,
312 log_path.clone(),
313 deadline,
314 started,
315 GateKind::CargoTest,
316 )
317 .await?
318 {
319 Ok(v) => v,
320 Err(timeout) => return Ok(timeout.with_log_ref(log_ref)),
321 };
322 if !status.success() {
323 let failure = classify::classify_cargo_test(&stdout_buf, &stderr_buf);
324 return Ok(GateOutcome::failed(name_target(failure, &target.dir)).with_log_ref(log_ref));
325 }
326 ran += 1;
327 }
328
329 // Every configured target was missing from the worktree. Exiting green here
330 // would report "tests passed" having run none of them.
331 if ran == 0 {
332 return Ok(GateOutcome::failed(GateFailure::Unclassified {
333 legacy_detail: Some(format!(
334 "cargo_test ran no targets: none of the {} configured test_target dir(s) \
335 exist in this worktree",
336 ctx.cfg.test_targets.len(),
337 )),
338 })
339 .with_log_ref(log_ref));
340 }
341
342 let duration_s = started.elapsed().as_secs() as u32;
343 Ok(GateOutcome::passed(PassNote::TestsPassed { duration_s }).with_log_ref(log_ref))
344 }
345
346 /// Stream a child to the live log, bounded by the gate-wide `deadline`. `Ok(Err(_))`
347 /// is the timeout outcome (child killed); `Err(_)` is an IO error on the stream.
348 #[allow(clippy::type_complexity)]
349 async fn run_to_deadline_for(
350 child: &mut tokio::process::Child,
351 ctx: &GateCtx,
352 run_id: GateRunId,
353 log_path: PathBuf,
354 deadline: std::time::Instant,
355 started: std::time::Instant,
356 kind: GateKind,
357 ) -> Result<std::result::Result<(Vec<u8>, Vec<u8>, std::process::ExitStatus), GateOutcome>> {
358 let remaining = deadline.saturating_duration_since(std::time::Instant::now());
359 let stream = stream_child_to_live_log(child, ctx.events.clone(), run_id, log_path);
360 match tokio::time::timeout(remaining, stream).await {
361 Ok(res) => Ok(Ok(res?)),
362 Err(_elapsed) => {
363 child.start_kill().ok();
364 let _ = child.wait().await;
365 Ok(Err(GateOutcome::failed(GateFailure::Timeout {
366 gate: kind,
367 after_s: started.elapsed().as_secs() as u32,
368 })))
369 }
370 }
371 }
372
373 /// Prefix a test/compile failure's headline with the crate it came from, so a
374 /// red gate across many targets says *which* crate broke. Other failure kinds
375 /// are single-target by construction and pass through untouched.
376 fn name_target(failure: GateFailure, dir: &std::path::Path) -> GateFailure {
377 let at = dir.display();
378 match failure {
379 GateFailure::CargoTest {
380 failed_count,
381 first_failed,
382 first_panic,
383 } => GateFailure::CargoTest {
384 failed_count,
385 first_failed: Some(match first_failed {
386 Some(name) => format!("{at}: {name}"),
387 None => at.to_string(),
388 }),
389 first_panic,
390 },
391 GateFailure::CompileError {
392 error_count,
393 first_error,
394 } => GateFailure::CompileError {
395 error_count,
396 first_error: Some(match first_error {
397 Some(e) => format!("{at}: {e}"),
398 None => at.to_string(),
399 }),
400 },
401 other => other,
402 }
403 }
404
405 /// Append raw bytes to a gate log outside the child-streaming path (target
406 /// banners). Best-effort, same as `LiveLog`: a broken log dir never turns a
407 /// passing gate red.
408 async fn append_to_log(path: &std::path::Path, bytes: &[u8]) {
409 use tokio::io::AsyncWriteExt;
410 if let Some(parent) = path.parent()
411 && tokio::fs::create_dir_all(parent).await.is_err()
412 {
413 return;
414 }
415 if let Ok(mut f) = tokio::fs::OpenOptions::new()
416 .create(true)
417 .append(true)
418 .open(path)
419 .await
420 {
421 let _ = f.write_all(bytes).await;
422 }
423 }
424
425 /// `cargo clippy --all-targets -- -D warnings` over every configured
426 /// `test_target`.
427 ///
428 /// Before this gate, `-D warnings` was enforced in exactly one place —
429 /// `server/deploy/run-ci.sh`, which died with the astra pipeline — so lint drift
430 /// reached prod ungated. Two crates even carried comments referring to "the
431 /// -D warnings CI gate" for a gate that did not exist.
432 async fn clippy(ctx: &GateCtx, run_id: GateRunId) -> Result<GateOutcome> {
433 lint_over_targets(ctx, run_id, GateKind::Clippy, |target, features| {
434 let mut args = vec!["clippy".to_string(), "--all-targets".to_string()];
435 if target.all_features {
436 args.push("--all-features".to_string());
437 } else if !features.is_empty() {
438 args.push("--features".to_string());
439 args.push(features.join(","));
440 }
441 // Everything after `--` goes to rustc, which is where -D lives.
442 args.push("--".to_string());
443 args.push("-D".to_string());
444 args.push("warnings".to_string());
445 args
446 })
447 .await
448 }
449
450 /// `cargo fmt --check` over every configured `test_target`.
451 ///
452 /// No `rustfmt.toml` exists anywhere in the tree, so this is plain rustfmt
453 /// defaults. Cheap: no compilation, just a parse.
454 async fn fmt_check(ctx: &GateCtx, run_id: GateRunId) -> Result<GateOutcome> {
455 lint_over_targets(ctx, run_id, GateKind::Fmt, |_target, _features| {
456 vec!["fmt".to_string(), "--check".to_string()]
457 })
458 .await
459 }
460
461 /// `cargo audit` / `cargo deny check`, over the crates that carry the matching
462 /// config file.
463 ///
464 /// Config-gated on purpose. Both tools are only meaningful against a triaged
465 /// posture: four crates in this repo fail `cargo audit` today purely because
466 /// they have no `.cargo/audit.toml` recording which transitive advisories have
467 /// been reviewed and accepted. Running them everywhere would make the gate
468 /// permanently and uninformatively red. Dropping the config file into a crate
469 /// is what opts it in.
470 async fn supply_chain(ctx: &GateCtx, run_id: GateRunId, kind: GateKind) -> Result<GateOutcome> {
471 let (config_rel, args): (&str, Vec<String>) = match kind {
472 GateKind::CargoAudit => (".cargo/audit.toml", vec!["audit".into()]),
473 GateKind::CargoDeny => ("deny.toml", vec!["deny".into(), "check".into()]),
474 other => unreachable!("supply_chain called for {other:?}"),
475 };
476 run_over_targets(ctx, run_id, kind, |target_dir| {
477 target_dir.join(config_rel).is_file().then(|| args.clone())
478 })
479 .await
480 }
481
482 /// Shared driver for the lint gates: run `cargo <args>` in every configured
483 /// `test_target` that exists in this worktree.
484 async fn lint_over_targets(
485 ctx: &GateCtx,
486 run_id: GateRunId,
487 kind: GateKind,
488 build_args: impl Fn(&crate::config::TestTarget, &[String]) -> Vec<String>,
489 ) -> Result<GateOutcome> {
490 let targets = ctx.cfg.test_targets.clone();
491 run_over_targets(ctx, run_id, kind, move |dir| {
492 let t = targets.iter().find(|t| ctx.worktree.join(&t.dir) == dir)?;
493 Some(build_args(t, &t.features))
494 })
495 .await
496 }
497
498 /// Run one cargo invocation per configured `test_target`, under a single
499 /// gate-wide deadline. `args_for` returns `None` to skip a target (used by the
500 /// supply-chain gates, which only apply where their config file lives).
501 ///
502 /// Shares `cargo_test`'s conventions: per-target log banners, stop at the first
503 /// failure with the crate named, skip targets absent from the worktree, and fail
504 /// closed if that leaves nothing to run.
505 async fn run_over_targets(
506 ctx: &GateCtx,
507 run_id: GateRunId,
508 kind: GateKind,
509 args_for: impl Fn(&std::path::Path) -> Option<Vec<String>>,
510 ) -> Result<GateOutcome> {
511 let log_path = gate_log_path(ctx, kind);
512 let log_ref = LogRef::new(&ctx.version, kind);
513 let started = std::time::Instant::now();
514 let deadline = started + std::time::Duration::from_secs(ctx.cfg.gate_timeout_secs);
515 let mut ran = 0usize;
516
517 for target in &ctx.cfg.test_targets {
518 let dir = ctx.worktree.join(&target.dir);
519 if !dir.join("Cargo.toml").is_file() {
520 tracing::warn!(
521 gate = kind.as_str(), target = %target.dir.display(),
522 "target has no Cargo.toml in this worktree; skipping",
523 );
524 continue;
525 }
526 let Some(args) = args_for(&dir) else { continue };
527
528 append_to_log(
529 &log_path,
530 format!("\n==== {}: {} ====\n", kind.as_str(), target.dir.display()).as_bytes(),
531 )
532 .await;
533
534 let mut cmd = Command::new("cargo");
535 cmd.args(&args)
536 .current_dir(&dir)
537 .stdout(std::process::Stdio::piped())
538 .stderr(std::process::Stdio::piped())
539 .kill_on_drop(true);
540 if let Some(t) = ctx.cfg.cargo_target_dir.as_deref() {
541 cmd.env("CARGO_TARGET_DIR", t);
542 }
543 // clippy type-checks, so it needs the same sqlx online-mode env the
544 // build and cargo_test steps get. fmt/audit/deny never touch the DB.
545 if kind == GateKind::Clippy
546 && let Some(url) = ctx
547 .cfg
548 .scratch_db_url
549 .as_deref()
550 .filter(|_| target.scratch_db)
551 {
552 cmd.env("DATABASE_URL", url);
553 cmd.env(
554 "TEST_DATABASE_URL",
555 url.split_once('?').map_or(url, |(b, _)| b),
556 );
557 }
558
559 let mut child = match cmd.spawn() {
560 Ok(c) => c,
561 Err(e) => {
562 return Ok(GateOutcome::failed(GateFailure::SpawnFailed {
563 message: format!("{}: {e}", target.dir.display()),
564 })
565 .with_log_ref(log_ref));
566 }
567 };
568 let (stdout_buf, stderr_buf, status) = match run_to_deadline_for(
569 &mut child,
570 ctx,
571 run_id,
572 log_path.clone(),
573 deadline,
574 started,
575 kind,
576 )
577 .await?
578 {
579 Ok(v) => v,
580 Err(timeout) => return Ok(timeout.with_log_ref(log_ref)),
581 };
582 if !status.success() {
583 let failure = match kind {
584 // clippy speaks rustc diagnostics, so the compile-error
585 // classifier extracts the real `error: ...` headline.
586 GateKind::Clippy => classify::classify_compile_error(&stdout_buf, &stderr_buf),
587 _ => GateFailure::Unclassified {
588 legacy_detail: Some(first_meaningful_line(&stdout_buf, &stderr_buf)),
589 },
590 };
591 return Ok(GateOutcome::failed(name_target(failure, &target.dir)).with_log_ref(log_ref));
592 }
593 ran += 1;
594 }
595
596 if ran == 0 {
597 return Ok(GateOutcome::failed(GateFailure::Unclassified {
598 legacy_detail: Some(format!(
599 "{} ran nothing: no configured target in this worktree qualified",
600 kind.as_str(),
601 )),
602 })
603 .with_log_ref(log_ref));
604 }
605 Ok(GateOutcome::passed(PassNote::TestsPassed {
606 duration_s: started.elapsed().as_secs() as u32,
607 })
608 .with_log_ref(log_ref))
609 }
610
611 /// First line that looks like a diagnostic, for gates whose tools have no
612 /// dedicated classifier (`cargo audit`, `cargo deny`). Falls back to a generic
613 /// note rather than an empty string.
614 fn first_meaningful_line(stdout: &[u8], stderr: &[u8]) -> String {
615 for buf in [stderr, stdout] {
616 let text = String::from_utf8_lossy(buf);
617 if let Some(line) = text
618 .lines()
619 .map(str::trim)
620 .find(|l| l.starts_with("error") || l.contains("vulnerabilit") || l.contains("FAILED"))
621 {
622 return line.chars().take(200).collect();
623 }
624 }
625 "tool reported failure; see the gate log".into()
626 }
627
628 /// The tests `cargo_test` cannot reach, run against production constants.
629 ///
630 /// `cargo_test` builds with `--features fast-tests`, which relaxes
631 /// `AUTH_RATE_LIMIT_BURST` 5 → 20, `SANDBOX_RATE_LIMIT_MS` 30s → 10ms, and
632 /// argon2 from 46 MiB/t=2 to 8 MiB/t=1 — and the rate-limiting suite is
633 /// `#[cfg_attr(feature = "fast-tests", ignore)]`d on top of that, because a
634 /// bucket refilling at 100/sec never depletes under parallel test threads. The
635 /// net effect was that Sando's only code gate silently skipped every test of
636 /// the auth hardening it most needs to protect.
637 ///
638 /// So: no features, `--test-threads=1` (these tests key on a shared per-IP
639 /// bucket and must not interleave), and a name filter rather than the whole
640 /// suite — the rest of the suite is tuned for `fast-tests` and would only go
641 /// slow and flaky here. The filter is a substring match, so it catches
642 /// `..._rate_limit_...` and `..._rate_limited` alike.
643 ///
644 /// This costs a second compile of the lib + integration binary (a different
645 /// feature set is a different cfg, so no artifact sharing with `cargo_test`).
646 /// That is the price of the coverage; the filter keeps the *run* to seconds.
647 async fn hardening_test(ctx: &GateCtx, run_id: GateRunId) -> Result<GateOutcome> {
648 let server_dir = ctx.worktree.join("server");
649 // No features is the whole point; the scratch DB is needed because the
650 // server's sqlx macros type-check against it. Unlike cargo_test, this gate
651 // is deliberately not driven by `test_targets`: it targets one specific
652 // suite in one specific crate, not "the repo's tests".
653 let target = crate::config::TestTarget {
654 dir: std::path::PathBuf::from("server"),
655 features: Vec::new(),
656 all_features: false,
657 scratch_db: true,
658 };
659 let log_path = gate_log_path(ctx, GateKind::HardeningTest);
660 let log_ref = LogRef::new(&ctx.version, GateKind::HardeningTest);
661
662 if let Some(scratch_url) = ctx.cfg.scratch_db_url.as_deref() {
663 clean_stale_test_dbs(scratch_url).await;
664 }
665
666 let started = std::time::Instant::now();
667
668 // Same two-step shape as cargo_test: compile first so a test-target break
669 // reports as a compile error rather than an opaque mass test failure.
670 let mut pre = match cargo_test_command(
671 ctx,
672 &server_dir,
673 &target,
674 &[],
675 &["--no-run", "--test", "integration"],
676 )
677 .spawn()
678 {
679 Ok(c) => c,
680 Err(e) => {
681 return Ok(GateOutcome::failed(GateFailure::SpawnFailed {
682 message: e.to_string(),
683 })
684 .with_log_ref(log_ref));
685 }
686 };
687 let (pre_out, pre_err, pre_status) =
688 stream_child_to_live_log(&mut pre, ctx.events.clone(), run_id, log_path.clone()).await?;
689 if !pre_status.success() {
690 let failure = classify::classify_compile_error(&pre_out, &pre_err);
691 return Ok(GateOutcome::failed(failure).with_log_ref(log_ref));
692 }
693
694 let mut child = match cargo_test_command(
695 ctx,
696 &server_dir,
697 &target,
698 &[],
699 &[
700 "--test",
701 "integration",
702 "--",
703 "--test-threads=1",
704 "rate_limit",
705 ],
706 )
707 .spawn()
708 {
709 Ok(c) => c,
710 Err(e) => {
711 return Ok(GateOutcome::failed(GateFailure::SpawnFailed {
712 message: e.to_string(),
713 })
714 .with_log_ref(log_ref));
715 }
716 };
717 let ceiling = std::time::Duration::from_secs(ctx.cfg.gate_timeout_secs);
718 let stream = stream_child_to_live_log(&mut child, ctx.events.clone(), run_id, log_path);
719 let (stdout_buf, stderr_buf, status) = match tokio::time::timeout(ceiling, stream).await {
720 Ok(res) => res?,
721 Err(_elapsed) => {
722 child.start_kill().ok();
723 let _ = child.wait().await;
724 return Ok(GateOutcome::failed(GateFailure::Timeout {
725 gate: GateKind::HardeningTest,
726 after_s: started.elapsed().as_secs() as u32,
727 })
728 .with_log_ref(log_ref));
729 }
730 };
731 let duration_s = started.elapsed().as_secs() as u32;
732 if status.success() {
733 // A filter that matches nothing exits 0, which would make this gate a
734 // green no-op the day someone renames the tests. Fail closed instead.
735 if tests_run(&stdout_buf) == 0 {
736 return Ok(GateOutcome::failed(GateFailure::Unclassified {
737 legacy_detail: Some(
738 "hardening_test ran 0 tests: the `rate_limit` filter matched nothing. \
739 The suite was renamed or moved — this gate is proving nothing."
740 .into(),
741 ),
742 })
743 .with_log_ref(log_ref));
744 }
745 Ok(GateOutcome::passed(PassNote::TestsPassed { duration_s }).with_log_ref(log_ref))
746 } else {
747 let failure = classify::classify_cargo_test(&stdout_buf, &stderr_buf);
748 Ok(GateOutcome::failed(failure).with_log_ref(log_ref))
749 }
750 }
751
752 /// Count the tests libtest reports as run, from its `test result: ok. N passed`
753 /// summary line. Returns 0 when no summary is present, which is itself the
754 /// "nothing ran" case the caller fails on.
755 fn tests_run(stdout: &[u8]) -> u32 {
756 String::from_utf8_lossy(stdout)
757 .lines()
758 .filter_map(|l| l.trim().strip_prefix("test result:"))
759 .filter_map(|rest| rest.split_once(" passed"))
760 .filter_map(|(head, _)| {
761 head.rsplit(' ')
762 .find(|t| !t.is_empty())?
763 .parse::<u32>()
764 .ok()
765 })
766 .sum()
767 }
768
769 /// Configure (but don't spawn) `cargo test --release [--features <features>]
770 /// <extra>` in `dir`, wired to the scratch DB. Shared by the `--no-run`
771 /// pre-gate compile and the full test run so both go through one env setup,
772 /// and by both test gates so they can't drift apart on env.
773 ///
774 /// `cargo_test` passes `fast-tests`, matching what the retired astra CI did:
775 /// it relaxes the auth rate-limit burst (5 → 20) and argon2 cost so
776 /// signup-heavy + lockout workflow tests complete without hitting Governor
777 /// before the hand-rolled lockout check (documented at
778 /// `server/src/constants.rs:87`). `hardening_test` passes no features, which is
779 /// the whole point of that gate — see `GateKind::HardeningTest`.
780 fn cargo_test_command(
781 ctx: &GateCtx,
782 dir: &std::path::Path,
783 target: &crate::config::TestTarget,
784 features: &[&str],
785 extra: &[&str],
786 ) -> Command {
787 let mut cmd = Command::new("cargo");
788 cmd.args(["test", "--release"]);
789 if target.all_features {
790 cmd.arg("--all-features");
791 } else if !features.is_empty() {
792 cmd.args(["--features", &features.join(",")]);
793 }
794 cmd.args(extra)
795 .current_dir(dir)
796 .stdout(std::process::Stdio::piped())
797 .stderr(std::process::Stdio::piped())
798 .kill_on_drop(true);
799 // Share the build step's target dir so the test compile reuses its
800 // artifacts (and the `--no-run` precompile reuses them again). Must match
801 // `build.rs` or the gate would clean-compile the whole tree a second time.
802 if let Some(target) = ctx.cfg.cargo_target_dir.as_deref() {
803 cmd.env("CARGO_TARGET_DIR", target);
804 }
805 // Same online-mode rationale as the build step: sqlx query macros need a
806 // live DB to type-check against. The scratch DB is left in migrated state
807 // by the preceding build, so we can reuse it here.
808 //
809 // Opt-in per target: setting DATABASE_URL switches sqlx OUT of offline mode,
810 // so a crate that ships `.sqlx` query data would stop using it and try to
811 // type-check against a database that has none of its tables.
812 if let Some(scratch_url) = ctx
813 .cfg
814 .scratch_db_url
815 .as_deref()
816 .filter(|_| target.scratch_db)
817 {
818 cmd.env("DATABASE_URL", scratch_url);
819 // The server test harness (tests/harness/db.rs) parses TEST_DATABASE_URL
820 // with rfind('/'), which mangles URLs whose query string contains '/'
821 // (e.g. `?host=/var/run/postgresql`). Strip the query — libpq defaults
822 // to /var/run/postgresql on Debian/Ubuntu when host is unspecified.
823 let test_url = scratch_url
824 .split_once('?')
825 .map_or(scratch_url, |(base, _)| base);
826 cmd.env("TEST_DATABASE_URL", test_url);
827 }
828 cmd
829 }
830
831 async fn migration_dry_run(ctx: &GateCtx, run_id: GateRunId) -> Result<GateOutcome> {
832 let log = GateLog::open(ctx, run_id, GateKind::MigrationDryRun).await;
833 let outcome = migration_dry_run_inner(ctx, &log).await;
834 log.close().await;
835 outcome.map(|o| o.with_log_ref(LogRef::new(&ctx.version, GateKind::MigrationDryRun)))
836 }
837
838 /// The staged interior of [`migration_dry_run`], writing every step through the
839 /// gate's live log. The caller owns the sink so it can flush it on every exit
840 /// path, and attaches the `log_ref` once instead of at each return.
841 async fn migration_dry_run_inner(ctx: &GateCtx, log: &GateLog) -> Result<GateOutcome> {
842 let Some(db_url) = ctx.cfg.scratch_db_url.as_deref() else {
843 log.line("scratch_db_url unset in daemon config\n").await;
844 return Ok(GateOutcome::blocked(GateBlocker::ScratchDbUrlUnset));
845 };
846
847 let backup: Option<(String, String)> =
848 sqlx::query_as("SELECT local_path, fetched_at FROM backups ORDER BY id DESC LIMIT 1")
849 .fetch_optional(&ctx.pool)
850 .await?;
851 let Some((backup_path, fetched_at)) = backup else {
852 log.line("no backup fetched; call /backup/fetch first\n")
853 .await;
854 return Ok(GateOutcome::blocked(GateBlocker::NoBackupAvailable));
855 };
856
857 // Presence is not freshness. A fetch that quietly stopped working leaves this
858 // row in place, and restoring it dry-runs the migrations against a schema prod
859 // has moved past — green, and worthless. Block on age instead. An unparsable
860 // timestamp is treated as stale: this row is daemon-written RFC 3339, so a
861 // value that will not parse means something is wrong, and failing closed on a
862 // freshness check is the whole point.
863 let age_hours = chrono::DateTime::parse_from_rfc3339(&fetched_at).map_or(i64::MAX, |t| {
864 (Utc::now() - t.with_timezone(&Utc)).num_hours()
865 });
866 let max_age_hours = ctx.cfg.backup_max_age_hours;
867 if age_hours > i64::from(max_age_hours) {
868 let msg = format!(
869 "backup {backup_path} was fetched {fetched_at} ({age_hours}h ago, max \
870 {max_age_hours}h); re-run /backup/fetch\n"
871 );
872 log.line(&msg).await;
873 return Ok(GateOutcome::blocked(GateBlocker::BackupStale {
874 age_hours,
875 max_age_hours,
876 }));
877 }
878
879 log.line("---- reset_scratch ----\n").await;
880 if let Err(e) = reset_scratch(db_url, &ctx.cfg.scratch_owner_role).await {
881 let msg = format!("scratch reset: {e}");
882 log.line(&msg).await;
883 return Ok(GateOutcome::failed(GateFailure::RestoreFailed {
884 reason: msg,
885 }));
886 }
887 log.line(&format!("---- restore_dump ({backup_path}) ----\n"))
888 .await;
889 if let Err(e) = restore_dump(db_url, &backup_path, log).await {
890 let msg = format!("restore: {e}");
891 log.line(&msg).await;
892 return Ok(GateOutcome::failed(GateFailure::RestoreFailed {
893 reason: msg,
894 }));
895 }
896
897 let migrations_dir = ctx.worktree.join("server").join("migrations");
898 log.line("---- run_migrator ----\n").await;
899 match run_migrator(db_url, &migrations_dir).await {
900 Ok(()) => {
901 log.line(&format!("restored {backup_path} + migrated"))
902 .await;
903 Ok(GateOutcome::passed(PassNote::Migrated { backup_path }))
904 }
905 Err(e) => {
906 let err_s = e.to_string();
907 log.line(&err_s).await;
908 Ok(GateOutcome::failed(classify::classify_migration_error(
909 &err_s, None,
910 )))
911 }
912 }
913 }
914
915 pub(crate) async fn reset_scratch(db_url: &str, owner_role: &str) -> Result<()> {
916 use sqlx::Executor;
917 use sqlx::postgres::PgPoolOptions;
918 let pool = PgPoolOptions::new()
919 .max_connections(1)
920 .connect(db_url)
921 .await?;
922 // `owner_role` is validated `[A-Za-z0-9_]+` at config load, so interpolating
923 // it into DDL is sound. It still goes through `format('%I')` inside the DO
924 // block for the quoting Postgres expects on an identifier.
925 let sql = format!(
926 r#"
927 DO $$
928 DECLARE s text;
929 BEGIN
930 -- The dump restores objects owned by the prod role and re-grants to
931 -- it (`ALTER ... OWNER TO {owner_role}`), which errors if the role
932 -- is absent — superuser does not imply the role exists. Create it
933 -- NOLOGIN: the scratch DB needs the role as an *owner* only, never
934 -- as a connecting identity. Idempotent, so a re-reset is a no-op.
935 IF NOT EXISTS (SELECT 1 FROM pg_roles WHERE rolname = '{owner_role}') THEN
936 EXECUTE format('CREATE ROLE %I NOLOGIN', '{owner_role}');
937 END IF;
938
939 -- Drop every non-system schema, not just public — migrations create
940 -- custom schemas (e.g. tower_sessions) that survive `DROP SCHEMA
941 -- public CASCADE` and then collide on the next migration run.
942 FOR s IN
943 SELECT nspname FROM pg_namespace
944 WHERE nspname NOT LIKE 'pg_%'
945 AND nspname NOT IN ('information_schema')
946 LOOP
947 EXECUTE format('DROP SCHEMA IF EXISTS %I CASCADE', s);
948 END LOOP;
949 EXECUTE 'CREATE SCHEMA public';
950 -- Restore the pre-PG15 public-schema default on the throwaway
951 -- scratch DB. Without this, the freshly-created public is owned by
952 -- the connecting role (sando) with no grant to anyone else, so a
953 -- migration's FK/trigger check that Postgres runs as a *restored*
954 -- prod-owned table's owner ({owner_role} from the backup dump)
955 -- fails with "permission denied for schema public". Granting to
956 -- PUBLIC is role-agnostic and safe here — this DB is disposable and
957 -- exists only to dry-run migrations.
958 EXECUTE 'GRANT USAGE, CREATE ON SCHEMA public TO PUBLIC';
959 -- PG15+: the new owner needs CREATE on public in its own right, not
960 -- only via PUBLIC, for the restore's owner-scoped DDL.
961 EXECUTE format('GRANT USAGE, CREATE ON SCHEMA public TO %I', '{owner_role}');
962 END $$;
963 "#
964 );
965 pool.execute(sqlx::raw_sql(sqlx::AssertSqlSafe(sql)))
966 .await?;
967 pool.close().await;
968 Ok(())
969 }
970
971 /// Startup assertion for the scratch cluster: the gates reset it, seed an owner
972 /// role into it, and drop leftover test databases in it — none of which a plain
973 /// unprivileged role can do. Historically these preconditions were satisfied by
974 /// hand on fw13 (`ALTER ROLE sando SUPERUSER`, a hand-created `makenotwork`
975 /// role) and nothing recorded that, so a rebuild elsewhere would fail one gate
976 /// at a time with an opaque permissions error. Assert once, at boot, loudly.
977 ///
978 /// Not part of `--check-config`: that path is pure by design (no DB, no
979 /// network), and a green there must mean "this build understands its config",
980 /// not "the cluster is reachable".
981 pub async fn preflight_scratch_privileges(db_url: &str) -> Result<()> {
982 use sqlx::postgres::PgPoolOptions;
983 let pool = PgPoolOptions::new()
984 .max_connections(1)
985 .connect(db_url)
986 .await
987 .context("connecting to scratch_db_url for the startup privilege check")?;
988 let (is_super, can_signal): (bool, bool) = sqlx::query_as(
989 "SELECT rolsuper, pg_catalog.pg_has_role(current_user, 'pg_signal_backend', 'USAGE')
990 FROM pg_roles WHERE rolname = current_user",
991 )
992 .fetch_one(&pool)
993 .await?;
994 pool.close().await;
995 anyhow::ensure!(
996 is_super || can_signal,
997 "the scratch_db_url role has neither SUPERUSER nor pg_signal_backend; migration_dry_run \
998 and cargo_test cannot reset the scratch DB or clear stale test databases. Grant one:\n \
999 ALTER ROLE <role> SUPERUSER; -- what fw13 uses\n \
1000 GRANT pg_signal_backend TO <role>; -- narrower: terminate only, cannot drop \
1001 foreign-owned databases",
1002 );
1003 if !is_super {
1004 tracing::warn!(
1005 "scratch role has pg_signal_backend but not SUPERUSER: stale test databases owned by \
1006 another role cannot be dropped, and the scratch owner role cannot be created if absent"
1007 );
1008 }
1009 Ok(())
1010 }
1011
1012 /// Best-effort cleanup of stale per-test database clones (`mnw_test_<uuid>`)
1013 /// left behind by a previously-killed `cargo_test` run.
1014 ///
1015 /// Drops **foreign-owned leftovers too**, which is why the daemon asserts
1016 /// SUPERUSER at startup (`preflight_scratch_privileges`): `DROP DATABASE`
1017 /// requires ownership or superuser, and the `WITH (FORCE)` terminate requires
1018 /// superuser or `pg_signal_backend`. Without both, orphans from a run under a
1019 /// different role accumulate and degrade the gate — the failure this cleanup
1020 /// exists to prevent.
1021 ///
1022 /// OPERATIONAL HAZARD: fw13 runs one Postgres cluster shared with local `cargo
1023 /// test` as `max`, so a gate firing mid-local-test will force-drop that run's
1024 /// databases out from under it. That collision is known and tracked separately
1025 /// (give the gate its own cluster); until then, do not run local tests on fw13
1026 /// while a Sando gate is live.
1027 ///
1028 /// Deliberately **excludes the template** (`mnw_test_template_*`): the harness
1029 /// reuses it across runs when it's migration-current (skipping a full
1030 /// drop+migrate), so dropping it here would force a needless rebuild every
1031 /// gate run. Templates are bounded (one per role) and never accumulate, so
1032 /// leaving them is free. Never returns an error: a cleanup miss must not turn a
1033 /// deploy red.
1034 async fn clean_stale_test_dbs(db_url: &str) {
1035 use sqlx::Executor;
1036 use sqlx::postgres::PgPoolOptions;
1037 let pool = match PgPoolOptions::new()
1038 .max_connections(1)
1039 .connect(db_url)
1040 .await
1041 {
1042 Ok(p) => p,
1043 Err(e) => {
1044 tracing::warn!(error = %e, "stale test-db cleanup: could not connect; skipping");
1045 return;
1046 }
1047 };
1048 // Every per-test clone, whoever owns it. The ownership filter this used to
1049 // carry is what let foreign-owned orphans pile up; superuser (asserted at
1050 // startup) makes them droppable.
1051 let names: Vec<(String,)> = sqlx::query_as(
1052 "SELECT datname FROM pg_database
1053 WHERE datname LIKE 'mnw_test_%'
1054 AND datname NOT LIKE '%template%'",
1055 )
1056 .fetch_all(&pool)
1057 .await
1058 .unwrap_or_default();
1059 let count = names.len();
1060 for (name,) in names {
1061 // `name` comes straight from pg_database; quoting it is sufficient.
1062 if let Err(e) = pool
1063 .execute(sqlx::raw_sql(sqlx::AssertSqlSafe(format!(
1064 "DROP DATABASE IF EXISTS \"{name}\" WITH (FORCE)"
1065 ))))
1066 .await
1067 {
1068 tracing::warn!(error = %e, db = %name, "stale test-db cleanup: drop failed");
1069 }
1070 }
1071 if count > 0 {
1072 tracing::info!(
1073 count,
1074 "stale test-db cleanup: dropped leftover mnw_test_* databases"
1075 );
1076 }
1077 pool.close().await;
1078 }
1079
1080 /// Build the restore shell line. Two pipelines we accept:
1081 /// *.sql -> psql -v ON_ERROR_STOP=1 $url < dump
1082 /// *.sql.gz -> set -o pipefail; gunzip -c dump | psql -v ON_ERROR_STOP=1 $url
1083 ///
1084 /// Two safety flags are load-bearing (CF4):
1085 /// - `ON_ERROR_STOP=1`: without it, psql exits 0 even when individual statements
1086 /// error, so a partial/corrupt restore would *pass* the gate.
1087 /// - `set -o pipefail`: without it a shell pipeline reports only the last
1088 /// command's status, so a `gunzip` failure on a truncated archive is masked by
1089 /// psql's exit. pipefail is a bash builtin (not POSIX sh), so the runner uses
1090 /// `bash -c`.
1091 fn restore_shell(db_url: &str, dump: &str) -> String {
1092 if std::path::Path::new(dump)
1093 .extension()
1094 .is_some_and(|ext| ext.eq_ignore_ascii_case("gz"))
1095 {
1096 format!(
1097 "set -o pipefail; gunzip -c {q} | psql -v ON_ERROR_STOP=1 {url}",
1098 q = shell_escape(dump),
1099 url = shell_escape(db_url),
1100 )
1101 } else {
1102 format!(
1103 "psql -v ON_ERROR_STOP=1 {url} < {q}",
1104 url = shell_escape(db_url),
1105 q = shell_escape(dump),
1106 )
1107 }
1108 }
1109
1110 async fn restore_dump(db_url: &str, dump: &str, log: &GateLog) -> Result<()> {
1111 // Split the password out of the URL and hand it to psql via PGPASSWORD, so it
1112 // never lands in argv (visible in /proc/<pid>/cmdline to any local user).
1113 // The sanitized URL — user/host/db, no secret — goes on the command line.
1114 let (sanitized, password) = split_pg_password(db_url);
1115 let shell = restore_shell(&sanitized, dump);
1116 // `bash` (not `sh`): `set -o pipefail` is a bash builtin. The restore runs
1117 // locally on the Sando host (fw13), which has bash.
1118 let mut cmd = Command::new("bash");
1119 cmd.arg("-c").arg(&shell);
1120 // kill_on_drop so the gate's wall-clock ceiling (dispatcher-level timeout on
1121 // migration_dry_run) can't orphan a wedged psql restore.
1122 cmd.kill_on_drop(true);
1123 if let Some(pw) = password {
1124 cmd.env("PGPASSWORD", pw);
1125 }
1126 // Streamed, not `.output()`: a prod-sized restore runs for minutes, and
1127 // psql's progress is the only thing an operator has to watch during it.
1128 let (_stdout, stderr, status) = log.run(&mut cmd).await?;
1129 anyhow::ensure!(
1130 status.success(),
1131 "restore failed: {}",
1132 String::from_utf8_lossy(&stderr),
1133 );
1134 Ok(())
1135 }
1136
1137 /// Split a `postgres://user:password@host/db` URL into its password-free form and
1138 /// the (percent-decoded) password. Returns the URL unchanged with `None` when
1139 /// there is no userinfo password. psql reads the password from `PGPASSWORD`, so
1140 /// keeping it off the command line removes the /proc exposure.
1141 fn split_pg_password(db_url: &str) -> (String, Option<String>) {
1142 let Some(after) = db_url.find("://").map(|i| i + 3) else {
1143 return (db_url.to_string(), None);
1144 };
1145 // The authority ends at the first '/', '?' or '#'; the password (if any) is
1146 // between the first ':' and the '@' within the userinfo of that authority.
1147 let authority_end = db_url[after..]
1148 .find(['/', '?', '#'])
1149 .map_or(db_url.len(), |i| after + i);
1150 let Some(at) = db_url[after..authority_end].find('@').map(|i| after + i) else {
1151 return (db_url.to_string(), None);
1152 };
1153 let userinfo = &db_url[after..at];
1154 let Some(colon) = userinfo.find(':') else {
1155 return (db_url.to_string(), None);
1156 };
1157 let password = percent_decode(&userinfo[colon + 1..]);
1158 let sanitized = format!(
1159 "{}{}{}",
1160 &db_url[..after],
1161 &userinfo[..colon],
1162 &db_url[at..]
1163 );
1164 (sanitized, Some(password))
1165 }
1166
1167 /// Minimal `%XX` percent-decode for a URL userinfo component. Non-escape bytes
1168 /// pass through; a malformed escape is left literal.
1169 fn percent_decode(s: &str) -> String {
1170 let b = s.as_bytes();
1171 let mut out = Vec::with_capacity(b.len());
1172 let mut i = 0;
1173 while i < b.len() {
1174 if b[i] == b'%'
1175 && i + 2 < b.len()
1176 && let (Some(h), Some(l)) = (hex_val(b[i + 1]), hex_val(b[i + 2]))
1177 {
1178 out.push((h << 4) | l);
1179 i += 3;
1180 } else {
1181 out.push(b[i]);
1182 i += 1;
1183 }
1184 }
1185 String::from_utf8_lossy(&out).into_owned()
1186 }
1187
1188 fn hex_val(c: u8) -> Option<u8> {
1189 match c {
1190 b'0'..=b'9' => Some(c - b'0'),
1191 b'a'..=b'f' => Some(c - b'a' + 10),
1192 b'A'..=b'F' => Some(c - b'A' + 10),
1193 _ => None,
1194 }
1195 }
1196
1197 pub(crate) async fn run_migrator(db_url: &str, dir: &std::path::Path) -> Result<()> {
1198 use sqlx::postgres::PgPoolOptions;
1199 let pool = PgPoolOptions::new()
1200 .max_connections(1)
1201 .connect(db_url)
1202 .await?;
1203 let migrator = sqlx::migrate::Migrator::new(dir).await?;
1204 migrator.run(&pool).await?;
1205 pool.close().await;
1206 Ok(())
1207 }
1208
1209 fn shell_escape(s: &str) -> String {
1210 format!("'{}'", s.replace('\'', "'\\''"))
1211 }
1212
1213 // ---- code_smoke ----
1214
1215 /// A 32+ char throwaway signing secret for the `code_smoke` boot. The server
1216 /// only enforces length (>= 32) outside production, and code_smoke's loopback
1217 /// `HOST_URL` keeps it in dev mode, so the value is irrelevant beyond that.
1218 const CODE_SMOKE_SIGNING_SECRET: &str = "sando-code-smoke-dummy-signing-secret-0000000000";
1219
1220 /// Seconds to wait for the real server to come up and serve `GET /health`
1221 /// during `code_smoke`. Longer than `boot_smoke`'s 3s: this boots the *full*
1222 /// server (config, pool, session store, webauthn, doc load, app build), not the
1223 /// minimal no-DB smoke server. The whole gate is also bounded by
1224 /// `gate_timeout_secs` at the dispatcher.
1225 const CODE_SMOKE_READY_SECS: u64 = 30;
1226
1227 /// `code_smoke` — the first host gate. Boots the freshly-built binary against a
1228 /// throwaway *empty* DB it migrates from scratch and seeds the example catalog
1229 /// into, then proves the real server serves `GET /health` against that
1230 /// nonempty DB. Fast and infra-light (one local Postgres, no prod-dump restore,
1231 /// no scratch-role reset, no external services), so a green here proves the
1232 /// code is sound and isolates a later `cargo_test`/`migration_dry_run` red as an
1233 /// environment problem rather than a code one.
1234 ///
1235 /// Reuses existing binary entrypoints, so the server needs no smoke-specific
1236 /// mode: `<bin> --seed-examples` loads config, connects, migrates from scratch,
1237 /// seeds, and exits; a plain `<bin>` then serves the real app. Both run with CWD
1238 /// at the server crate root so `docs/business/assumptions.toml` + `site-docs/`
1239 /// resolve (a missing assumptions file panics real startup), and with a loopback
1240 /// `HOST_URL` so config stays in dev mode (no CDN/S3/signing-secret prod
1241 /// enforcement). The seed's host allowlist already admits `127.0.0.1`, and the
1242 /// fresh DB trivially satisfies its no-real-users guard.
1243 async fn code_smoke(ctx: &GateCtx, run_id: GateRunId) -> Result<GateOutcome> {
1244 let log = GateLog::open(ctx, run_id, GateKind::CodeSmoke).await;
1245 let outcome = code_smoke_inner(ctx, &log).await;
1246 log.close().await;
1247 outcome.map(|o| o.with_log_ref(LogRef::new(&ctx.version, GateKind::CodeSmoke)))
1248 }
1249
1250 /// The staged interior of [`code_smoke`], writing every step through the gate's
1251 /// live log. Same split as [`migration_dry_run_inner`]: the caller owns the sink
1252 /// and attaches the `log_ref`.
1253 async fn code_smoke_inner(ctx: &GateCtx, log: &GateLog) -> Result<GateOutcome> {
1254 let Some(scratch_url) = ctx.cfg.scratch_db_url.as_deref() else {
1255 log.line("scratch_db_url unset in daemon config\n").await;
1256 return Ok(GateOutcome::blocked(GateBlocker::ScratchDbUrlUnset));
1257 };
1258
1259 // The staged binary (set by build_and_run_host before gating). code_smoke
1260 // runs first among the host gates, but staging precedes all gating, so the
1261 // artifact path is already recorded.
1262 let bin: Option<(String,)> =
1263 sqlx::query_as("SELECT artifact_path FROM versions WHERE version = ?")
1264 .bind(&ctx.version)
1265 .fetch_optional(&ctx.pool)
1266 .await?;
1267 let Some((bin,)) = bin else {
1268 return Ok(GateOutcome::blocked(GateBlocker::ArtifactMissing {
1269 version: ctx.version.clone(),
1270 }));
1271 };
1272
1273 // Frontend builds, before anything else: they need no DB and no staged
1274 // binary, and a `tsc` error is the one failure the Rust build deliberately
1275 // swallows (both MNW build scripts emit `cargo::warning` and succeed against
1276 // a stale `static/dist/`). Failing here is what stops the deploy rsyncing
1277 // the previous build's bundle.
1278 if let Some(outcome) = code_smoke_frontends(ctx, log).await {
1279 return Ok(outcome);
1280 }
1281
1282 // Docs integrity, first and cheapest: run the staged binary's DB-free
1283 // `MNW_CHECK_DOCS` mode before creating the throwaway DB. A broken internal
1284 // docs link (a `[..](x.md)` resolving to a slug no page serves) fails here
1285 // in well under a second instead of after a full migrate+seed+boot, and a
1286 // rotted link never reaches prod as a live 404. Collisions are reported by
1287 // the check but do not fail it; only broken links do.
1288 if let Some(outcome) = code_smoke_docs_check(ctx, &bin, log).await {
1289 return Ok(outcome);
1290 }
1291
1292 let dbname = code_smoke_db_name(&ctx.version);
1293 let maintenance_url = pg_url_with_dbname(scratch_url, "postgres");
1294 let throwaway_url = pg_url_with_dbname(scratch_url, &dbname);
1295
1296 // Create the throwaway DB (dropping any stale one from a killed prior run).
1297 log.line(&format!("---- createdb {dbname} ----\n")).await;
1298 if let Err(e) = pg_create_db(&maintenance_url, &dbname).await {
1299 let reason = format!("createdb {dbname}: {e}");
1300 log.line(&reason).await;
1301 return Ok(GateOutcome::failed(GateFailure::CodeSmokeSetup { reason }));
1302 }
1303
1304 // Everything past createdb must drop the DB on the way out, pass or fail.
1305 let outcome = code_smoke_body(ctx, &bin, &throwaway_url, log).await;
1306
1307 log.line(&format!("\n---- dropdb {dbname} ----\n")).await;
1308 if let Err(e) = pg_drop_db(&maintenance_url, &dbname).await {
1309 // A teardown miss must not turn a passing gate red — log it and move on.
1310 // The next run's createdb drops it first anyway.
1311 tracing::warn!(error = %e, db = %dbname, "code_smoke: dropdb failed; next run will reclaim it");
1312 log.line(&format!("dropdb warning (non-fatal): {e}")).await;
1313 }
1314
1315 Ok(outcome)
1316 }
1317
1318 /// Compile every configured `frontend_build` in the worktree.
1319 ///
1320 /// Returns `Some(failed)` on the first project that does not build; `None` when
1321 /// all of them do (or none are configured). Output streams to `log` either way.
1322 ///
1323 /// `npm ci` runs only when `node_modules` is absent. Usually it is not: the app
1324 /// build script installed it during the `cargo build` that produced the artifact
1325 /// this gate is about to smoke, so the common path here is just `npm run build`
1326 /// against a warm install — seconds. The install branch covers the gate running
1327 /// against a worktree whose build script was skipped or failed at `npm ci`, and
1328 /// it is as fatal as a compile failure, because the alternative is compiling
1329 /// against whatever some earlier sha installed.
1330 ///
1331 /// Unlike the app build scripts, nothing here is best-effort. That asymmetry is
1332 /// the point of the gate.
1333 async fn code_smoke_frontends(ctx: &GateCtx, log: &GateLog) -> Option<GateOutcome> {
1334 for fe in &ctx.cfg.frontend_builds {
1335 let dir = ctx.worktree.join(&fe.dir);
1336 let label = fe.dir.display().to_string();
1337 log.line(&format!("---- frontend build ({label}) ----\n"))
1338 .await;
1339
1340 if !dir.is_dir() {
1341 // An older sha predating the frontend, mid-bisect. Skipping keeps
1342 // sando able to rebuild history; the log says so out loud.
1343 log.line(&format!("{label} absent from this worktree; skipping\n"))
1344 .await;
1345 continue;
1346 }
1347
1348 if !dir.join("node_modules").is_dir()
1349 && let Some(outcome) = run_npm(&dir, &label, &["ci"], "npm ci", ctx, log).await
1350 {
1351 return Some(outcome);
1352 }
1353
1354 if let Some(outcome) = run_npm(
1355 &dir,
1356 &label,
1357 &["run", &fe.script],
1358 &format!("npm run {}", fe.script),
1359 ctx,
1360 log,
1361 )
1362 .await
1363 {
1364 return Some(outcome);
1365 }
1366 }
1367 None
1368 }
1369
1370 /// One `npm` invocation for [`code_smoke_frontends`], bounded by the gate
1371 /// timeout so a wedged install cannot hold the whole pipeline (the enclosing
1372 /// `code_smoke` ceiling would catch it eventually, but this attributes the
1373 /// failure to the project that hung).
1374 async fn run_npm(
1375 dir: &std::path::Path,
1376 label: &str,
1377 args: &[&str],
1378 what: &str,
1379 ctx: &GateCtx,
1380 log: &GateLog,
1381 ) -> Option<GateOutcome> {
1382 log.line(&format!("$ {what}\n")).await;
1383 let mut cmd = tokio::process::Command::new("npm");
1384 cmd.args(args).current_dir(dir).kill_on_drop(true);
1385 let ceiling = std::time::Duration::from_secs(ctx.cfg.gate_timeout_secs);
1386 // On the timeout branch the whole `run` future is dropped, which drops the
1387 // child; `kill_on_drop` is what turns that into an actual kill.
1388 let status = match tokio::time::timeout(ceiling, log.run(&mut cmd)).await {
1389 Ok(Ok((_stdout, _stderr, status))) => status,
1390 Ok(Err(e)) => {
1391 // A missing `npm` lands here. Fatal, not skipped: a build host
1392 // without Node cannot produce the bundle the release serves, and
1393 // silently passing is how the stale bundle shipped in the first place.
1394 log.line(&format!("{what} could not be spawned: {e}\n"))
1395 .await;
1396 return Some(GateOutcome::failed(GateFailure::SpawnFailed {
1397 message: format!("{what} in {label}: {e}"),
1398 }));
1399 }
1400 Err(_elapsed) => {
1401 log.line(&format!(
1402 "{what} timed out after {}s\n",
1403 ctx.cfg.gate_timeout_secs
1404 ))
1405 .await;
1406 return Some(GateOutcome::failed(GateFailure::CodeSmokeFrontend {
1407 dir: label.to_string(),
1408 exit_code: None,
1409 }));
1410 }
1411 };
1412 if status.success() {
1413 return None;
1414 }
1415 Some(GateOutcome::failed(GateFailure::CodeSmokeFrontend {
1416 dir: label.to_string(),
1417 exit_code: status.code(),
1418 }))
1419 }
1420
1421 /// Run the staged binary's DB-free docs integrity check (`MNW_CHECK_DOCS=1`).
1422 ///
1423 /// Returns `Some(failed)` if the check reports broken links, cannot be spawned,
1424 /// or overruns its ceiling; `None` when the docs are clean. Output streams to
1425 /// `log` either way. The 60s ceiling backstops the case where the staged binary
1426 /// predates the flag and would fall through to a normal (DB-needing) boot and
1427 /// hang.
1428 async fn code_smoke_docs_check(ctx: &GateCtx, bin: &str, log: &GateLog) -> Option<GateOutcome> {
1429 let server_dir = ctx.worktree.join("server");
1430 log.line("---- docs check (MNW_CHECK_DOCS) ----\n").await;
1431 let mut cmd = tokio::process::Command::new(bin);
1432 cmd.env("MNW_CHECK_DOCS", "1")
1433 .current_dir(&server_dir)
1434 .kill_on_drop(true);
1435 let (stdout, _stderr, status) =
1436 match tokio::time::timeout(std::time::Duration::from_mins(1), log.run(&mut cmd)).await {
1437 Ok(Ok(out)) => out,
1438 Ok(Err(e)) => {
1439 log.line(&format!("docs check spawn failed: {e}\n")).await;
1440 return Some(GateOutcome::failed(GateFailure::SpawnFailed {
1441 message: e.to_string(),
1442 }));
1443 }
1444 Err(_elapsed) => {
1445 let reason =
1446 "docs check timed out after 60s (staged binary may predate MNW_CHECK_DOCS)"
1447 .to_string();
1448 log.line(&format!("{reason}\n")).await;
1449 return Some(GateOutcome::failed(GateFailure::CodeSmokeSetup { reason }));
1450 }
1451 };
1452 if status.success() {
1453 return None;
1454 }
1455 Some(GateOutcome::failed(GateFailure::CodeSmokeDocs {
1456 broken: parse_check_docs_broken_count(&stdout),
1457 }))
1458 }
1459
1460 /// Best-effort parse of the broken-link count from the `MNW_CHECK_DOCS` sentinel
1461 /// line (`MNW_CHECK_DOCS: N broken link(s)`). Returns 0 if absent — the failure
1462 /// still stands, only the summary count is unknown.
1463 fn parse_check_docs_broken_count(stdout: &[u8]) -> u32 {
1464 let text = String::from_utf8_lossy(stdout);
1465 for line in text.lines() {
1466 if let Some(rest) = line.strip_prefix("MNW_CHECK_DOCS:") {
1467 for tok in rest.split_whitespace() {
1468 if let Ok(n) = tok.parse::<u32>() {
1469 return n;
1470 }
1471 }
1472 }
1473 }
1474 0
1475 }
1476
1477 /// The createdb-to-dropdb interior of `code_smoke`: migrate+seed, then boot and
1478 /// probe. Returns the outcome without a `log_ref` (the caller attaches it after
1479 /// teardown). Never returns `Err` — spawn/child failures map to typed outcomes.
1480 async fn code_smoke_body(ctx: &GateCtx, bin: &str, db_url: &str, log: &GateLog) -> GateOutcome {
1481 let server_dir = ctx.worktree.join("server");
1482
1483 // Phase 1: migrate-from-scratch + seed. `--seed-examples` loads config,
1484 // connects, runs migrations against the empty DB, seeds the catalog, exits.
1485 // A non-zero exit here is the "code is unsound" signal (broken migration,
1486 // seed error, or config-load failure).
1487 log.line("---- migrate + seed (--seed-examples) ----\n")
1488 .await;
1489 let mut seed_cmd = tokio::process::Command::new(bin);
1490 seed_cmd.arg("--seed-examples").current_dir(&server_dir);
1491 code_smoke_env(&mut seed_cmd, ctx, db_url);
1492 seed_cmd.env("ALLOW_EXAMPLE_SEED", "1").kill_on_drop(true);
1493 let seed_status = match log.run(&mut seed_cmd).await {
1494 Ok((_stdout, _stderr, status)) => status,
1495 Err(e) => {
1496 return GateOutcome::failed(GateFailure::SpawnFailed {
1497 message: e.to_string(),
1498 });
1499 }
1500 };
1501 if !seed_status.success() {
1502 return GateOutcome::failed(GateFailure::CodeSmokeSeed {
1503 exit_code: seed_status.code(),
1504 });
1505 }
1506
1507 // Phase 2: boot the real server against the now-migrated + seeded DB and
1508 // assert both startup signals: it logs `listening` (emitted just before the
1509 // socket bind) AND serves GET /health with a 200. The full stdout/stderr is
1510 // persisted for the operator either way.
1511 log.line("\n---- boot + probe /health ----\n").await;
1512 let mut serve_cmd = tokio::process::Command::new(bin);
1513 serve_cmd.current_dir(&server_dir);
1514 code_smoke_env(&mut serve_cmd, ctx, db_url);
1515 serve_cmd
1516 .stdout(std::process::Stdio::piped())
1517 .stderr(std::process::Stdio::piped())
1518 .kill_on_drop(true);
1519 let mut child = match serve_cmd.spawn() {
1520 Ok(c) => c,
1521 Err(e) => {
1522 return GateOutcome::failed(GateFailure::SpawnFailed {
1523 message: e.to_string(),
1524 });
1525 }
1526 };
1527
1528 // Stream stdout/stderr into the gate log (and out as chunk events) while
1529 // the probe loop runs below; the tasks finish when the pipes close (child
1530 // exits or is killed). The buffers they return are what the `listening`
1531 // assertion reads.
1532 let (stdout_task, stderr_task) = log.drain_pipes(&mut child);
1533
1534 let probe_timeout = std::time::Duration::from_millis(500);
1535 let started = std::time::Instant::now();
1536 let window = std::time::Duration::from_secs(CODE_SMOKE_READY_SECS);
1537 let mut probe_ok_after: Option<u32> = None;
1538 let mut last_probe_err = "never responded".to_string();
1539 let mut early_exit = None;
1540 while started.elapsed() < window {
1541 if let Ok(Some(status)) = child.try_wait() {
1542 early_exit = Some(status);
1543 break;
1544 }
1545 match tokio::time::timeout(probe_timeout, probe_health(ctx.cfg.code_smoke_port)).await {
1546 Ok(Ok(())) => {
1547 probe_ok_after = Some(started.elapsed().as_millis() as u32);
1548 break;
1549 }
1550 Ok(Err(e)) => last_probe_err = e,
1551 Err(_) => last_probe_err = "probe timed out".to_string(),
1552 }
1553 tokio::time::sleep(std::time::Duration::from_millis(250)).await;
1554 }
1555
1556 let exit = match early_exit {
1557 Some(status) => Some(status),
1558 None => {
1559 let e = child.try_wait().ok().flatten();
1560 if e.is_none() {
1561 let _ = child.kill().await;
1562 }
1563 e
1564 }
1565 };
1566 // Keep the serve run's own output so the `listening`-log assertion checks it
1567 // (not the seed run's, which exits before binding). The bytes are already in
1568 // the gate log; these buffers exist only for the assertion.
1569 let serve_stdout = stdout_task.await.unwrap_or_default();
1570 let serve_stderr = stderr_task.await.unwrap_or_default();
1571 let logged_listening =
1572 bytes_contain(&serve_stdout, b"listening") || bytes_contain(&serve_stderr, b"listening");
1573
1574 match (exit, probe_ok_after) {
1575 // Exited on its own within the window — panic / config error / bind fail.
1576 (Some(status), _) => GateOutcome::failed(classify::classify_boot_smoke(status.code())),
1577 // Stayed up and served /health. Assert both required startup signals:
1578 // the `listening` bind log AND the /health 200.
1579 (None, Some(after_ms)) if logged_listening => {
1580 GateOutcome::passed(PassNote::HealthyProbe { after_ms })
1581 }
1582 // Served /health but the `listening` log never appeared.
1583 (None, Some(_)) => GateOutcome::failed(GateFailure::CodeSmokeNoListeningLog),
1584 // Stayed up but never served /health — started, not ready.
1585 (None, None) => GateOutcome::failed(GateFailure::BootHealthProbeFailed {
1586 last_error: last_probe_err,
1587 }),
1588 }
1589 }
1590
1591 /// Substring search over raw bytes (the server's log output), for the
1592 /// `code_smoke` startup-log assertion. Avoids a lossy UTF-8 conversion of the
1593 /// whole buffer just to run `str::contains`.
1594 fn bytes_contain(haystack: &[u8], needle: &[u8]) -> bool {
1595 if needle.is_empty() || haystack.len() < needle.len() {
1596 return needle.is_empty();
1597 }
1598 haystack.windows(needle.len()).any(|w| w == needle)
1599 }
1600
1601 /// Apply the minimal env every `code_smoke` invocation shares: point the binary
1602 /// at the throwaway DB, force loopback (dev-mode config, no prod enforcement),
1603 /// hand it a dummy signing secret, and disable file scanning (no AV/YARA on the
1604 /// build host). The worktree is a clean git checkout, so no stray `.env` shadows
1605 /// these (and dotenvy never overrides already-set vars).
1606 fn code_smoke_env(cmd: &mut tokio::process::Command, ctx: &GateCtx, db_url: &str) {
1607 cmd.env("DATABASE_URL", db_url)
1608 .env("HOST", "127.0.0.1")
1609 .env("PORT", ctx.cfg.code_smoke_port.to_string())
1610 .env(
1611 "HOST_URL",
1612 format!("http://127.0.0.1:{}", ctx.cfg.code_smoke_port),
1613 )
1614 .env("SIGNING_SECRET", CODE_SMOKE_SIGNING_SECRET)
1615 .env("SCAN_ENABLED", "false")
1616 .env("INSECURE_COOKIES", "1");
1617 }
1618
1619 /// The throwaway smoke DB name for a version: `sando_code_smoke_<version>` with
1620 /// every non-alphanumeric char folded to `_` and lowercased, capped at Postgres'
1621 /// 63-byte identifier limit. Sanitized to `[a-z0-9_]` so it's safe to quote into
1622 /// DDL. Deterministic per version, so a stale DB from a killed run is reclaimed
1623 /// by the next run's `DROP DATABASE IF EXISTS` rather than accumulating.
1624 fn code_smoke_db_name(version: &Version) -> String {
1625 let mut name = String::from("sando_code_smoke_");
1626 for c in version.to_string().chars() {
1627 name.push(if c.is_ascii_alphanumeric() {
1628 c.to_ascii_lowercase()
1629 } else {
1630 '_'
1631 });
1632 }
1633 name.truncate(63);
1634 name
1635 }
1636
1637 /// Rewrite a `postgres://` URL to point at database `dbname`, preserving scheme,
1638 /// userinfo, host/port, and any query (e.g. the socket `?host=/var/run/postgresql`
1639 /// form) + fragment. Used to derive the maintenance connection (`postgres`) and
1640 /// the throwaway smoke DB URL from the configured `scratch_db_url`.
1641 fn pg_url_with_dbname(url: &str, dbname: &str) -> String {
1642 let Some(after_scheme) = url.find("://").map(|i| i + 3) else {
1643 return url.to_string();
1644 };
1645 let rest = &url[after_scheme..];
1646 // Authority ends at the first '/', '?' or '#'; whatever follows is the
1647 // path (the old dbname) plus an optional query/fragment we must keep.
1648 let auth_end = rest.find(['/', '?', '#']).unwrap_or(rest.len());
1649 let authority = &rest[..auth_end];
1650 let tail = &rest[auth_end..];
1651 let query_and_frag = match tail.find(['?', '#']) {
1652 Some(i) => &tail[i..],
1653 None => "",
1654 };
1655 format!(
1656 "{}{}/{}{}",
1657 &url[..after_scheme],
1658 authority,
1659 dbname,
1660 query_and_frag
1661 )
1662 }
1663
1664 /// Create the throwaway smoke DB on the cluster `maintenance_url` points at,
1665 /// dropping any stale one first. `dbname` is sanitized to `[a-z0-9_]` by
1666 /// `code_smoke_db_name`, so quoting it is sufficient. `CREATE DATABASE` cannot
1667 /// run inside a transaction, so these go through the simple-query protocol (a
1668 /// raw `&str` execute), matching `reset_scratch`.
1669 async fn pg_create_db(maintenance_url: &str, dbname: &str) -> Result<()> {
1670 use sqlx::Executor;
1671 use sqlx::postgres::PgPoolOptions;
1672 let pool = PgPoolOptions::new()
1673 .max_connections(1)
1674 .connect(maintenance_url)
1675 .await?;
1676 pool.execute(sqlx::raw_sql(sqlx::AssertSqlSafe(format!(
1677 "DROP DATABASE IF EXISTS \"{dbname}\" WITH (FORCE)"
1678 ))))
1679 .await?;
1680 pool.execute(sqlx::raw_sql(sqlx::AssertSqlSafe(format!(
1681 "CREATE DATABASE \"{dbname}\""
1682 ))))
1683 .await?;
1684 pool.close().await;
1685 Ok(())
1686 }
1687
1688 /// Drop the throwaway smoke DB, forcing off any lingering connection (the killed
1689 /// server's pool). Best-effort at the call site — a failure is logged, not fatal.
1690 async fn pg_drop_db(maintenance_url: &str, dbname: &str) -> Result<()> {
1691 use sqlx::Executor;
1692 use sqlx::postgres::PgPoolOptions;
1693 let pool = PgPoolOptions::new()
1694 .max_connections(1)
1695 .connect(maintenance_url)
1696 .await?;
1697 pool.execute(sqlx::raw_sql(sqlx::AssertSqlSafe(format!(
1698 "DROP DATABASE IF EXISTS \"{dbname}\" WITH (FORCE)"
1699 ))))
1700 .await?;
1701 pool.close().await;
1702 Ok(())
1703 }
1704
1705 async fn boot_smoke(ctx: &GateCtx, run_id: GateRunId) -> Result<GateOutcome> {
1706 let bin: Option<(String,)> =
1707 sqlx::query_as("SELECT artifact_path FROM versions WHERE version = ?")
1708 .bind(&ctx.version)
1709 .fetch_optional(&ctx.pool)
1710 .await?;
1711 let Some((bin,)) = bin else {
1712 return Ok(GateOutcome::blocked(GateBlocker::ArtifactMissing {
1713 version: ctx.version.clone(),
1714 }));
1715 };
1716
1717 // Readiness smoke: start the binary and confirm it serves `GET /health`
1718 // within the window, not merely that the process stays up. Panics in main,
1719 // missing config, and port-bind failures still surface as an early exit; a
1720 // process that comes up but never serves /health is now its own failure.
1721 //
1722 // The server requires DATABASE_URL or it panics on config load before
1723 // we can observe anything. We point it at the scratch DB (already
1724 // migrated by the build step and refreshed by migration_dry_run if
1725 // that gate ran first). SCAN_ENABLED=false skips loading YARA rules
1726 // from /opt/makenotwork/yara-rules which doesn't exist on the build
1727 // host. SANDO_BOOT_SMOKE_PORT tells the smoke server which loopback port
1728 // to bind so we know where to probe. Other config has sane optional defaults.
1729 let mut cmd = tokio::process::Command::new(&bin);
1730 cmd.env("SANDO_BOOT_SMOKE", "1")
1731 .env("SANDO_BOOT_SMOKE_PORT", ctx.cfg.boot_smoke_port.to_string())
1732 .env("SCAN_ENABLED", "false")
1733 .stdout(std::process::Stdio::piped())
1734 .stderr(std::process::Stdio::piped())
1735 .kill_on_drop(true);
1736 if let Some(scratch_url) = ctx.cfg.scratch_db_url.as_deref() {
1737 cmd.env("DATABASE_URL", scratch_url);
1738 }
1739 let log_path = gate_log_path(ctx, GateKind::BootSmoke);
1740 let log_ref = LogRef::new(&ctx.version, GateKind::BootSmoke);
1741 let mut child = match cmd.spawn() {
1742 Ok(c) => c,
1743 Err(e) => {
1744 // Spawn failures get a one-off log line via LiveLog so the
1745 // on-disk file still exists for `GET /logs/...`.
1746 let mut log = LiveLog::open(log_path, gate_chunk_cb(ctx.events.clone(), run_id)).await;
1747 log.write_chunk(format!("spawn: {e}\n").as_bytes()).await;
1748 log.close().await;
1749 return Ok(GateOutcome::failed(GateFailure::SpawnFailed {
1750 message: e.to_string(),
1751 })
1752 .with_log_ref(log_ref));
1753 }
1754 };
1755
1756 // The boot smoke window is 3s. Drain stdout/stderr concurrently through
1757 // a shared LiveLog sink so the operator sees panics/log lines stream in
1758 // real time before the kill, AND the on-disk log gets the full byte
1759 // stream for post-mortem reads. The drainers exit when their pipe
1760 // closes — which happens when the child exits naturally or after kill.
1761 let log = std::sync::Arc::new(tokio::sync::Mutex::new(
1762 LiveLog::open(log_path, gate_chunk_cb(ctx.events.clone(), run_id)).await,
1763 ));
1764 let stdout_task = tokio::spawn(stream_into_log(child.stdout.take(), log.clone()));
1765 let stderr_task = tokio::spawn(stream_into_log(child.stderr.take(), log.clone()));
1766
1767 // Poll readiness across the 3s window instead of a flat sleep: GET /health
1768 // must return 2xx. A crash mid-window short-circuits to the exit-code
1769 // failure path (try_wait below); a process that stays up but never serves
1770 // /health is a distinct readiness failure.
1771 let probe_timeout = std::time::Duration::from_millis(500);
1772 let started = std::time::Instant::now();
1773 let window = std::time::Duration::from_secs(3);
1774 let mut probe_ok_after: Option<u32> = None;
1775 let mut last_probe_err = "never responded".to_string();
1776 let mut early_exit = None;
1777 while started.elapsed() < window {
1778 if let Some(status) = child.try_wait()? {
1779 early_exit = Some(status);
1780 break;
1781 }
1782 match tokio::time::timeout(probe_timeout, probe_health(ctx.cfg.boot_smoke_port)).await {
1783 Ok(Ok(())) => {
1784 probe_ok_after = Some(started.elapsed().as_millis() as u32);
1785 break;
1786 }
1787 Ok(Err(e)) => last_probe_err = e,
1788 Err(_) => last_probe_err = "probe timed out".to_string(),
1789 }
1790 tokio::time::sleep(std::time::Duration::from_millis(150)).await;
1791 }
1792
1793 // Stop the child unless it already exited, then drain the log tasks.
1794 let exit = match early_exit {
1795 Some(status) => Some(status),
1796 None => {
1797 let e = child.try_wait()?;
1798 if e.is_none() {
1799 let _ = child.kill().await;
1800 }
1801 e
1802 }
1803 };
1804 // The streamed bytes already landed in the live log and the on-disk file for
1805 // the post-mortem reader. Drain the join handles to avoid hangs.
1806 let _ = stdout_task.await;
1807 let _ = stderr_task.await;
1808 // Unique owner of the Arc at this point (both tasks dropped their clones).
1809 if let Ok(mutex) = std::sync::Arc::try_unwrap(log) {
1810 mutex.into_inner().close().await;
1811 }
1812
1813 match (exit, probe_ok_after) {
1814 // Exited on its own within the window — a crash/panic/bind failure.
1815 (Some(status), _) => {
1816 let failure = classify::classify_boot_smoke(status.code());
1817 Ok(GateOutcome::failed(failure).with_log_ref(log_ref))
1818 }
1819 // Stayed up and served /health — readiness proven.
1820 (None, Some(after_ms)) => {
1821 Ok(GateOutcome::passed(PassNote::HealthyProbe { after_ms }).with_log_ref(log_ref))
1822 }
1823 // Stayed up but never served /health — started, not ready.
1824 (None, None) => Ok(GateOutcome::failed(GateFailure::BootHealthProbeFailed {
1825 last_error: last_probe_err,
1826 })
1827 .with_log_ref(log_ref)),
1828 }
1829 }
1830
1831 /// One readiness probe of the boot-smoke server: connect to `127.0.0.1:port`
1832 /// and `GET /health`, returning `Ok(())` only on a `200`. A hand-rolled HTTP/1.0
1833 /// request over a raw `TcpStream` keeps the outbound probe dependency-free
1834 /// (reqwest is dev-only); the smoke server serves the one route over axum, which
1835 /// speaks 1.0. `Err` carries a short reason for the operator's failure note. The
1836 /// caller wraps each call in a timeout.
1837 async fn probe_health(port: u16) -> std::result::Result<(), String> {
1838 use tokio::io::{AsyncReadExt, AsyncWriteExt};
1839 let mut stream = tokio::net::TcpStream::connect((std::net::Ipv4Addr::LOCALHOST, port))
1840 .await
1841 .map_err(|e| format!("connect: {e}"))?;
1842 stream
1843 .write_all(b"GET /health HTTP/1.0\r\nHost: localhost\r\nConnection: close\r\n\r\n")
1844 .await
1845 .map_err(|e| format!("write: {e}"))?;
1846 let mut buf = Vec::new();
1847 stream
1848 .read_to_end(&mut buf)
1849 .await
1850 .map_err(|e| format!("read: {e}"))?;
1851 let text = String::from_utf8_lossy(&buf);
1852 let status_line = text.lines().next().unwrap_or("");
1853 if status_line.contains(" 200 ") {
1854 Ok(())
1855 } else {
1856 Err(format!("unexpected status line: {status_line:?}"))
1857 }
1858 }
1859
1860 /// Sink that drops streamed output. `node_health` keeps the probe's stderr from
1861 /// its `RunOutput` for the failure note, so the live byte stream isn't needed.
1862 struct DiscardSink;
1863 #[async_trait::async_trait]
1864 impl ops_exec::LogSink for DiscardSink {
1865 async fn write_chunk(&mut self, _bytes: &[u8]) {}
1866 }
1867
1868 /// `node_health` — the post-deploy gate that proves the *deployed nodes* are
1869 /// serving, recording one outcome per (tier, version) that the next promote
1870 /// checks. Distinct from `boot_smoke`, which boots the staged artifact on the
1871 /// build host: this probes each node over the same executor the deploy used, so
1872 /// a node that took a corrupt artifact, wrong-arch binary, or failed restart is
1873 /// caught here rather than waved through (Run-2 SERIOUS-3). Fails closed: any
1874 /// unhealthy node fails the gate, and an empty node set is `Blocked`.
1875 async fn node_health(ctx: &GateCtx) -> Result<GateOutcome> {
1876 if ctx.nodes.is_empty() {
1877 return Ok(GateOutcome::blocked(GateBlocker::NoNodesToProbe));
1878 }
1879 for probe in &ctx.nodes {
1880 if let Err(detail) = probe_node(probe).await {
1881 return Ok(GateOutcome::failed(GateFailure::NodeUnhealthy {
1882 node: probe.node.to_string(),
1883 detail,
1884 }));
1885 }
1886 }
1887 Ok(GateOutcome::passed(PassNote::NodesHealthy {
1888 nodes: ctx.nodes.len() as u32,
1889 }))
1890 }
1891
1892 /// Probe one node over its executor: confirm the unit is active post-restart
1893 /// and, when a `health_url` is configured, that it serves a 2xx. Retries across
1894 /// ~10s because the service may still be restarting / warming. Runs under the
1895 /// read-only `Observe(Health)` capability (every Sando node grants it), so the
1896 /// probe needs no deploy authority. `Ok(())` = healthy; `Err(detail)` carries a
1897 /// short reason for the gate's failure note.
1898 async fn probe_node(probe: &NodeProbe) -> std::result::Result<(), String> {
1899 use ops_exec::{Action, ObserveKind, Step, sh_quote};
1900 let svc = sh_quote(&probe.service);
1901 let url = probe
1902 .health_url
1903 .as_deref()
1904 .map_or_else(|| "''".to_string(), sh_quote);
1905 // One executor round-trip with the retry loop on the node: is-active, then
1906 // (if a url is set) curl it for a 2xx. Exit 0 only when both hold.
1907 let script = format!(
1908 "svc={svc}; url={url}; \
1909 for _ in $(seq 1 10); do \
1910 if systemctl is-active --quiet \"$svc\"; then \
1911 if [ -z \"$url\" ] || curl -fsS --max-time 5 \"$url\" >/dev/null 2>&1; then exit 0; fi; \
1912 fi; \
1913 sleep 1; \
1914 done; \
1915 echo 'service not active or health url not 2xx after retries' >&2; exit 1"
1916 );
1917 let step = Step::shell(Action::Observe(ObserveKind::Health), script);
1918 let mut sink = DiscardSink;
1919 let out = probe
1920 .executor
1921 .run_streaming(&step, &mut sink)
1922 .await
1923 .map_err(|e| format!("probe spawn: {e}"))?;
1924 if out.status.success() {
1925 Ok(())
1926 } else {
1927 let code = out
1928 .status
1929 .code()
1930 .map_or_else(|| "signal".to_string(), |c| c.to_string());
1931 let stderr: String = String::from_utf8_lossy(&out.stderr)
1932 .chars()
1933 .take(200)
1934 .collect();
1935 Err(format!("exit {code}: {stderr}"))
1936 }
1937 }
1938
1939 /// Drain `stream` into the shared `LiveLog` (which forwards each chunk to
1940 /// the on-disk log file AND broadcasts a `GateLogChunk` event), and return
1941 /// the concatenated bytes so the classifier can still operate on the full
1942 /// output post-hoc.
1943 async fn stream_into_log<R>(
1944 stream: Option<R>,
1945 log: std::sync::Arc<tokio::sync::Mutex<LiveLog>>,
1946 ) -> Vec<u8>
1947 where
1948 R: tokio::io::AsyncRead + Unpin + Send + 'static,
1949 {
1950 let mut total = Vec::new();
1951 let Some(mut s) = stream else { return total };
1952 let mut buf = [0u8; 4096];
1953 loop {
1954 match s.read(&mut buf).await {
1955 Ok(0) => break,
1956 Err(_) => break,
1957 Ok(n) => {
1958 total.extend_from_slice(&buf[..n]);
1959 log.lock().await.write_chunk(&buf[..n]).await;
1960 }
1961 }
1962 }
1963 total
1964 }
1965
1966 /// Spawn a child, drain its stdout/stderr through a `LiveLog`, return the
1967 /// combined buffers and exit status. Shared by `cargo_test` (no deadline)
1968 /// and ad-hoc callers — `boot_smoke` rolls its own variant because of its
1969 /// 3s kill window.
1970 async fn stream_child_to_live_log(
1971 child: &mut tokio::process::Child,
1972 events: EventTx,
1973 run_id: GateRunId,
1974 log_path: PathBuf,
1975 ) -> Result<(Vec<u8>, Vec<u8>, std::process::ExitStatus)> {
1976 let log = GateLog::new(LiveLog::open(log_path, gate_chunk_cb(events, run_id)).await);
1977 let out = log.stream_child(child).await;
1978 log.close().await;
1979 out
1980 }
1981
1982 /// One gate's live log, held across every step of a multi-step gate.
1983 ///
1984 /// Single-child gates call [`stream_child_to_live_log`] and are done. The
1985 /// staged gates (`code_smoke`, `migration_dry_run`) instead run several
1986 /// children plus banner lines between them, and they hold one `GateLog` across
1987 /// the lot: a single sink means chunk sequence numbers stay monotonic for the
1988 /// whole gate, and the on-disk log reads in the order things actually happened
1989 /// rather than as stdout-then-stderr assembled at the end.
1990 ///
1991 /// Every write is best-effort in the same way `LiveLog` is: a log directory
1992 /// that cannot be written degrades to callback-only and never turns a passing
1993 /// gate red.
1994 struct GateLog {
1995 sink: Arc<tokio::sync::Mutex<LiveLog>>,
1996 }
1997
1998 impl GateLog {
1999 fn new(sink: LiveLog) -> Self {
2000 Self {
2001 sink: Arc::new(tokio::sync::Mutex::new(sink)),
2002 }
2003 }
2004
2005 /// Open the sink for `gate`'s log file, streaming to the TUI under `run_id`.
2006 async fn open(ctx: &GateCtx, run_id: GateRunId, gate: GateKind) -> Self {
2007 Self::new(
2008 LiveLog::open(
2009 gate_log_path(ctx, gate),
2010 gate_chunk_cb(ctx.events.clone(), run_id),
2011 )
2012 .await,
2013 )
2014 }
2015
2016 /// Emit a banner (or any line the gate itself produces) through the same
2017 /// sink the children stream to, so it lands in sequence with their output.
2018 async fn write(&self, bytes: &[u8]) {
2019 self.sink.lock().await.write_chunk(bytes).await;
2020 }
2021
2022 /// Same, for the common `format!`-a-line case.
2023 async fn line(&self, s: &str) {
2024 self.write(s.as_bytes()).await;
2025 }
2026
2027 /// Spawn `cmd` with both pipes captured, stream them into the sink as they
2028 /// arrive, and return the buffers plus the exit status. The buffers are
2029 /// what the classifiers still operate on post-hoc.
2030 async fn run(
2031 &self,
2032 cmd: &mut Command,
2033 ) -> std::io::Result<(Vec<u8>, Vec<u8>, std::process::ExitStatus)> {
2034 cmd.stdout(std::process::Stdio::piped())
2035 .stderr(std::process::Stdio::piped());
2036 let mut child = cmd.spawn()?;
2037 self.stream_child(&mut child)
2038 .await
2039 .map_err(std::io::Error::other)
2040 }
2041
2042 /// Drain an already-spawned child's pipes into the sink and wait for it.
2043 async fn stream_child(
2044 &self,
2045 child: &mut tokio::process::Child,
2046 ) -> Result<(Vec<u8>, Vec<u8>, std::process::ExitStatus)> {
2047 let (stdout_task, stderr_task) = self.drain_pipes(child);
2048 let status = child.wait().await?;
2049 let stdout_buf = stdout_task.await.unwrap_or_default();
2050 let stderr_buf = stderr_task.await.unwrap_or_default();
2051 Ok((stdout_buf, stderr_buf, status))
2052 }
2053
2054 /// Start draining a child's pipes into the sink *without* waiting on the
2055 /// child. For `code_smoke`'s serve phase, which probes `/health` while the
2056 /// server is still up. The caller must await both handles for the buffers
2057 /// (and before [`Self::close`], so the flush isn't skipped).
2058 #[allow(clippy::type_complexity)]
2059 fn drain_pipes(
2060 &self,
2061 child: &mut tokio::process::Child,
2062 ) -> (
2063 tokio::task::JoinHandle<Vec<u8>>,
2064 tokio::task::JoinHandle<Vec<u8>>,
2065 ) {
2066 (
2067 tokio::spawn(stream_into_log(child.stdout.take(), self.sink.clone())),
2068 tokio::spawn(stream_into_log(child.stderr.take(), self.sink.clone())),
2069 )
2070 }
2071
2072 /// Flush the file. A still-outstanding streaming task (only possible if the
2073 /// gate returned without awaiting it) leaves the `Arc` shared, in which case
2074 /// the flush is skipped — `LiveLog` writes unbuffered to the OS either way,
2075 /// so nothing already written is lost.
2076 async fn close(self) {
2077 if let Ok(mutex) = Arc::try_unwrap(self.sink) {
2078 mutex.into_inner().close().await;
2079 }
2080 }
2081 }
2082
2083 fn gate_log_path(ctx: &GateCtx, gate: GateKind) -> PathBuf {
2084 ctx.cfg
2085 .logs_root
2086 .join(ctx.version.to_string())
2087 .join(format!("{}.log", gate.as_str()))
2088 }
2089
2090 /// Live check: has `tier`'s burn-in window of `hours` elapsed since its clock
2091 /// (`tier_state.burn_in_started_at`, started by a promote onto the tier)? Used
2092 /// by the promote-time gate check (`unsatisfied_gates`) so a stale `blocked`
2093 /// row never masks an elapsed — or not-yet-elapsed — window. The `burn_in` gate
2094 /// runner below wraps the same state with a richer outcome for `/state`.
2095 pub async fn burn_in_satisfied(pool: &SqlitePool, tier: &TierId, hours: u32) -> Result<bool> {
2096 let started: Option<String> =
2097 sqlx::query_scalar("SELECT burn_in_started_at FROM tier_state WHERE tier = ?")
2098 .bind(tier)
2099 .fetch_optional(pool)
2100 .await?
2101 .flatten();
2102 let Some(started) = started else {
2103 return Ok(false);
2104 };
2105 let started = chrono::DateTime::parse_from_rfc3339(&started)?.with_timezone(&Utc);
2106 Ok(Utc::now() - started >= chrono::Duration::hours(hours as i64))
2107 }
2108
2109 async fn burn_in(ctx: &GateCtx, hours: u32) -> Result<GateOutcome> {
2110 // Check tier_state.burn_in_started_at on this tier; pass if enough time
2111 // has elapsed. The clock is started by /promote when a version lands on
2112 // the burn-in tier.
2113 let started: Option<String> =
2114 sqlx::query_scalar("SELECT burn_in_started_at FROM tier_state WHERE tier = ?")
2115 .bind(&ctx.tier)
2116 .fetch_optional(&ctx.pool)
2117 .await?
2118 .flatten();
2119 let Some(started) = started else {
2120 return Ok(GateOutcome::blocked(GateBlocker::BurnInClockNotStarted));
2121 };
2122 let started = chrono::DateTime::parse_from_rfc3339(&started)?.with_timezone(&Utc);
2123 let elapsed = Utc::now() - started;
2124 let needed = chrono::Duration::hours(hours as i64);
2125 if elapsed >= needed {
2126 Ok(GateOutcome::passed(PassNote::BurnInElapsed {
2127 hours: elapsed.num_hours() as u32,
2128 }))
2129 } else {
2130 let remaining = (needed - elapsed).num_hours().max(0) as u32;
2131 Ok(GateOutcome::blocked(GateBlocker::BurnInRemaining {
2132 hours_remaining: remaining,
2133 hours_total: hours,
2134 }))
2135 }
2136 }
2137
2138 async fn manual_confirm(ctx: &GateCtx) -> Result<GateOutcome> {
2139 // Pass iff a row in gate_runs exists with status='passed' for this
2140 // (tier, version, manual_confirm) that was inserted out-of-band by an
2141 // operator action. Since the harness inserts the in-flight row itself,
2142 // look for a prior confirmation row.
2143 let prior_at: Option<String> = sqlx::query_scalar(
2144 "SELECT finished_at FROM gate_runs
2145 WHERE tier = ? AND version = ? AND gate_kind = 'manual_confirm' AND status = 'passed'
2146 ORDER BY id DESC LIMIT 1",
2147 )
2148 .bind(&ctx.tier)
2149 .bind(&ctx.version)
2150 .fetch_optional(&ctx.pool)
2151 .await?;
2152 match prior_at {
2153 Some(at_str) => {
2154 let at = chrono::DateTime::parse_from_rfc3339(&at_str)
2155 .map_or_else(|_| Utc::now(), |d| d.with_timezone(&Utc));
2156 Ok(GateOutcome::passed(PassNote::OperatorConfirmed { at }))
2157 }
2158 None => Ok(GateOutcome::blocked(
2159 GateBlocker::AwaitingOperatorConfirmation,
2160 )),
2161 }
2162 }
2163
2164 #[cfg(test)]
2165 mod tests {
2166 use super::*;
2167 use crate::events;
2168 use sqlx::sqlite::SqlitePoolOptions;
2169
2170 fn target(dir: &str) -> crate::config::TestTarget {
2171 crate::config::TestTarget {
2172 dir: std::path::PathBuf::from(dir),
2173 features: Vec::new(),
2174 all_features: false,
2175 scratch_db: false,
2176 }
2177 }
2178
2179 #[test]
2180 fn parse_check_docs_broken_count_reads_the_sentinel() {
2181 // The failing sentinel carries the count.
2182 assert_eq!(
2183 parse_check_docs_broken_count(b"some log\nMNW_CHECK_DOCS: 3 broken link(s)\n"),
2184 3
2185 );
2186 // Count with a preceding log line still parses (first bare int wins).
2187 assert_eq!(
2188 parse_check_docs_broken_count(
2189 b" broken link: a -> b\nMNW_CHECK_DOCS: 1 broken link(s)\n"
2190 ),
2191 1
2192 );
2193 // The parser only runs on failure; the "ok" sentinel is never fed to it,
2194 // and its "(2" token is not a bare int, so it yields 0 harmlessly.
2195 assert_eq!(
2196 parse_check_docs_broken_count(b"MNW_CHECK_DOCS: ok (2 collision(s) reported)\n"),
2197 0
2198 );
2199 // Absent sentinel -> 0; the failure still stands, only the count is lost.
2200 assert_eq!(parse_check_docs_broken_count(b"unrelated output"), 0);
2201 }
2202
2203 #[test]
2204 fn name_target_points_a_test_failure_at_its_crate() {
2205 // With one target the crate was implicit; with fifteen the operator
2206 // needs the summary to say which one broke.
2207 let f = name_target(
2208 GateFailure::CargoTest {
2209 failed_count: 3,
2210 first_failed: Some("workflows::sync::round_trip".into()),
2211 first_panic: None,
2212 },
2213 std::path::Path::new("shared/synckit-client"),
2214 );
2215 assert_eq!(
2216 f.summary(),
2217 "3 test(s) failed; first: shared/synckit-client: workflows::sync::round_trip",
2218 );
2219 }
2220
2221 #[test]
2222 fn name_target_points_a_compile_failure_at_its_crate() {
2223 let f = name_target(
2224 GateFailure::CompileError {
2225 error_count: 1,
2226 first_error: Some("error[E0063]".into()),
2227 },
2228 std::path::Path::new("mnw-cli"),
2229 );
2230 assert_eq!(
2231 f.summary(),
2232 "compile failed (1 error(s)); first: mnw-cli: error[E0063]"
2233 );
2234 }
2235
2236 #[test]
2237 fn name_target_names_the_crate_even_without_a_test_name() {
2238 let f = name_target(
2239 GateFailure::CargoTest {
2240 failed_count: 2,
2241 first_failed: None,
2242 first_panic: None,
2243 },
2244 std::path::Path::new("pom"),
2245 );
2246 assert_eq!(f.summary(), "2 test(s) failed; first: pom");
2247 }
2248
2249 #[test]
2250 fn name_target_leaves_unrelated_failures_alone() {
2251 let f = name_target(
2252 GateFailure::SpawnFailed {
2253 message: "no cargo".into(),
2254 },
2255 std::path::Path::new("pom"),
2256 );
2257 assert!(matches!(f, GateFailure::SpawnFailed { .. }));
2258 }
2259
2260 /// A `GateCtx` over `worktree` with the given frontend projects configured.
2261 /// No DB, no artifact — `code_smoke_frontends` touches neither.
2262 async fn frontend_ctx(worktree: &std::path::Path, dirs: &[&str]) -> GateCtx {
2263 let mut cfg = crate::config::Config::for_tests();
2264 cfg.frontend_builds = dirs
2265 .iter()
2266 .map(|d| crate::config::FrontendBuild {
2267 dir: PathBuf::from(d),
2268 script: "build".into(),
2269 })
2270 .collect();
2271 cfg.logs_root = worktree.join("logs");
2272 GateCtx {
2273 pool: SqlitePoolOptions::new()
2274 .max_connections(1)
2275 .connect("sqlite::memory:")
2276 .await
2277 .unwrap(),
2278 cfg: std::sync::Arc::new(cfg),
2279 tier: TierId::new("host"),
2280 version: "0.1.0".parse().unwrap(),
2281 worktree: worktree.to_path_buf(),
2282 events: events::channel(),
2283 nodes: Vec::new(),
2284 build_id: None,
2285 }
2286 }
2287
2288 /// A `GateCtx` for the `migration_dry_run` freshness checks: a migrated
2289 /// in-memory pool (so `backups` exists) and a scratch URL set, so the gate
2290 /// reaches the backup lookup instead of bailing on config. Nothing here
2291 /// touches postgres — every assertion below blocks before `reset_scratch`.
2292 async fn dry_run_ctx(worktree: &std::path::Path, max_age_hours: u32) -> GateCtx {
2293 let mut cfg = crate::config::Config::for_tests();
2294 cfg.scratch_db_url = Some("postgres:///sando_scratch".into());
2295 cfg.backup_max_age_hours = max_age_hours;
2296 cfg.logs_root = worktree.join("logs");
2297 let pool = SqlitePoolOptions::new()
2298 .max_connections(1)
2299 .connect("sqlite::memory:")
2300 .await
2301 .unwrap();
2302 crate::db::migrate(&pool).await.unwrap();
2303 GateCtx {
2304 pool,
2305 cfg: std::sync::Arc::new(cfg),
2306 tier: TierId::new("host"),
2307 version: "0.1.0".parse().unwrap(),
2308 worktree: worktree.to_path_buf(),
2309 events: events::channel(),
2310 nodes: Vec::new(),
2311 build_id: None,
2312 }
2313 }
2314
2315 /// Record a backup row fetched `hours_ago`, as `/backup/fetch` would.
2316 async fn seed_backup(ctx: &GateCtx, hours_ago: i64) {
2317 let at = (Utc::now() - chrono::Duration::hours(hours_ago)).to_rfc3339();
2318 sqlx::query(
2319 "INSERT INTO backups (fetched_at, source, local_path, byte_size)
2320 VALUES (?, 'file:///x.sql.gz', '/tmp/sando-test-backup.sql.gz', 1000000)",
2321 )
2322 .bind(at)
2323 .execute(&ctx.pool)
2324 .await
2325 .unwrap();
2326 }
2327
2328 #[tokio::test]
2329 async fn migration_dry_run_blocks_on_a_stale_backup() {
2330 // The failure this closes: the gate used to check only that a backups row
2331 // existed, so a fetch that silently stopped working left it green against
2332 // an ever-older schema. Sando ran 45 days that way.
2333 let tmp = tempfile::tempdir().unwrap();
2334 let ctx = dry_run_ctx(tmp.path(), 48).await;
2335 seed_backup(&ctx, 24 * 45).await;
2336 let log = GateLog::open(&ctx, GateRunId(0), GateKind::MigrationDryRun).await;
2337
2338 let outcome = migration_dry_run_inner(&ctx, &log).await.unwrap();
2339 log.close().await;
2340
2341 let crate::outcome::GateStatus::Blocked { blocker } = outcome.status else {
2342 panic!("a 45-day-old backup must block");
2343 };
2344 let GateBlocker::BackupStale {
2345 age_hours,
2346 max_age_hours,
2347 } = blocker
2348 else {
2349 panic!("expected BackupStale, got {blocker:?}");
2350 };
2351 assert_eq!(max_age_hours, 48);
2352 assert!(
2353 age_hours >= 24 * 45,
2354 "reports the real age, got {age_hours}"
2355 );
2356 }
2357
2358 #[tokio::test]
2359 async fn migration_dry_run_accepts_a_fresh_backup() {
2360 // The other side of the boundary: a backup inside the window must not be
2361 // blocked on freshness. It fails later (there is no such dump on disk),
2362 // which is exactly the proof the age check let it through.
2363 let tmp = tempfile::tempdir().unwrap();
2364 let ctx = dry_run_ctx(tmp.path(), 48).await;
2365 seed_backup(&ctx, 6).await;
2366 let log = GateLog::open(&ctx, GateRunId(0), GateKind::MigrationDryRun).await;
2367
2368 let outcome = migration_dry_run_inner(&ctx, &log).await.unwrap();
2369 log.close().await;
2370
2371 assert!(
2372 !matches!(
2373 outcome.status,
2374 crate::outcome::GateStatus::Blocked {
2375 blocker: GateBlocker::BackupStale { .. }
2376 }
2377 ),
2378 "a 6h-old backup is fresh, got {:?}",
2379 outcome.status,
2380 );
2381 }
2382
2383 #[tokio::test]
2384 async fn migration_dry_run_treats_an_unparsable_fetched_at_as_stale() {
2385 // Fail closed: `fetched_at` is daemon-written RFC 3339, so a value that
2386 // will not parse means the row is untrustworthy — and a freshness check
2387 // that shrugs at a timestamp it cannot read is not a freshness check.
2388 let tmp = tempfile::tempdir().unwrap();
2389 let ctx = dry_run_ctx(tmp.path(), 48).await;
2390 sqlx::query(
2391 "INSERT INTO backups (fetched_at, source, local_path, byte_size)
2392 VALUES ('not-a-timestamp', 'file:///x.sql.gz', '/tmp/x.sql.gz', 1000000)",
2393 )
2394 .execute(&ctx.pool)
2395 .await
2396 .unwrap();
2397 let log = GateLog::open(&ctx, GateRunId(0), GateKind::MigrationDryRun).await;
2398
2399 let outcome = migration_dry_run_inner(&ctx, &log).await.unwrap();
2400 log.close().await;
2401
2402 assert!(
2403 matches!(
2404 outcome.status,
2405 crate::outcome::GateStatus::Blocked {
2406 blocker: GateBlocker::BackupStale { .. }
2407 }
2408 ),
2409 "an unreadable fetched_at must block, got {:?}",
2410 outcome.status,
2411 );
2412 }
2413
2414 /// A `code_smoke` live log over `ctx.cfg.logs_root`, for the helpers that
2415 /// take one. `GateRunId(0)` never matches a real row; nothing reads the
2416 /// chunk events in these tests.
2417 async fn test_gate_log(ctx: &GateCtx) -> GateLog {
2418 GateLog::open(ctx, GateRunId(0), GateKind::CodeSmoke).await
2419 }
2420
2421 /// Close `log` (flushing it) and read back what it wrote on disk.
2422 async fn read_gate_log(ctx: &GateCtx, log: GateLog) -> String {
2423 log.close().await;
2424 tokio::fs::read_to_string(gate_log_path(ctx, GateKind::CodeSmoke))
2425 .await
2426 .expect("the gate log must exist on disk")
2427 }
2428
2429 /// Write a minimal npm project at `worktree/<dir>` whose `build` script
2430 /// exits with `exit_code`. Pre-creates `node_modules` so the gate skips
2431 /// `npm ci` — these tests are about the build step, not the network.
2432 fn fake_npm_project(worktree: &std::path::Path, dir: &str, exit_code: u8) {
2433 let root = worktree.join(dir);
2434 std::fs::create_dir_all(root.join("node_modules")).unwrap();
2435 std::fs::write(
2436 root.join("package.json"),
2437 format!(
2438 r#"{{"name":"fake","version":"0.0.0","private":true,
2439 "scripts":{{"build":"exit {exit_code}"}}}}"#
2440 ),
2441 )
2442 .unwrap();
2443 }
2444
2445 #[tokio::test]
2446 async fn frontend_gate_fails_on_a_build_error_and_names_the_project() {
2447 // The whole point of the gate: the app build scripts downgrade this to a
2448 // cargo::warning, so if it passes here nothing stops a stale bundle.
2449 let tmp = tempfile::tempdir().unwrap();
2450 fake_npm_project(tmp.path(), "server/frontend", 0);
2451 fake_npm_project(tmp.path(), "multithreaded/frontend", 2);
2452 let ctx = frontend_ctx(tmp.path(), &["server/frontend", "multithreaded/frontend"]).await;
2453
2454 let log = test_gate_log(&ctx).await;
2455 let outcome = code_smoke_frontends(&ctx, &log)
2456 .await
2457 .expect("a failing tsc must fail the gate");
2458 let crate::outcome::GateStatus::Failed { failure } = &outcome.status else {
2459 panic!("expected a failure, got {:?}", outcome.status)
2460 };
2461 assert!(
2462 matches!(
2463 failure,
2464 GateFailure::CodeSmokeFrontend { dir, exit_code: Some(2) }
2465 if dir == "multithreaded/frontend"
2466 ),
2467 "got: {failure:?}"
2468 );
2469 // The passing project ran first; its output belongs in the log too.
2470 let text = read_gate_log(&ctx, log).await;
2471 assert!(text.contains("server/frontend"), "log: {text}");
2472 }
2473
2474 #[tokio::test]
2475 async fn frontend_gate_passes_when_every_project_builds() {
2476 let tmp = tempfile::tempdir().unwrap();
2477 fake_npm_project(tmp.path(), "server/frontend", 0);
2478 let ctx = frontend_ctx(tmp.path(), &["server/frontend"]).await;
2479 let log = test_gate_log(&ctx).await;
2480 assert!(
2481 code_smoke_frontends(&ctx, &log).await.is_none(),
2482 "a clean build must not fail the gate"
2483 );
2484 }
2485
2486 /// The point of routing `code_smoke` through `LiveLog`: its output reaches
2487 /// the operator *while* the gate runs, as `GateLogChunk` events, instead of
2488 /// appearing all at once when the gate finishes.
2489 #[tokio::test]
2490 async fn code_smoke_streams_chunks_as_it_runs() {
2491 let tmp = tempfile::tempdir().unwrap();
2492 fake_npm_project(tmp.path(), "server/frontend", 0);
2493 let ctx = frontend_ctx(tmp.path(), &["server/frontend"]).await;
2494 let mut rx = ctx.events.subscribe_logs();
2495
2496 let log = GateLog::open(&ctx, GateRunId(7), GateKind::CodeSmoke).await;
2497 assert!(code_smoke_frontends(&ctx, &log).await.is_none());
2498 log.close().await;
2499
2500 let mut chunks = Vec::new();
2501 while let Ok(envelope) = rx.try_recv() {
2502 if let Event::GateLogChunk { run_id, seq, text } = envelope.event {
2503 assert_eq!(run_id, GateRunId(7));
2504 chunks.push((seq, text));
2505 }
2506 }
2507 assert!(!chunks.is_empty(), "no chunk ever reached the bus");
2508 // Sequence numbers are per-run and monotonic across every step of the
2509 // gate, which is why the whole gate shares one sink.
2510 let seqs: Vec<u32> = chunks.iter().map(|(seq, _)| *seq).collect();
2511 assert!(
2512 seqs.windows(2).all(|w| w[0] < w[1]),
2513 "chunk seq must be monotonic, got {seqs:?}"
2514 );
2515 let joined: String = chunks.into_iter().map(|(_, text)| text).collect();
2516 assert!(joined.contains("server/frontend"), "chunks: {joined}");
2517 }
2518
2519 #[tokio::test]
2520 async fn frontend_gate_skips_a_project_absent_from_the_worktree() {
2521 // Rebuilding an older sha that predates the frontend must stay possible.
2522 let tmp = tempfile::tempdir().unwrap();
2523 let ctx = frontend_ctx(tmp.path(), &["multithreaded/frontend"]).await;
2524 let log = test_gate_log(&ctx).await;
2525 assert!(code_smoke_frontends(&ctx, &log).await.is_none());
2526 assert!(
2527 read_gate_log(&ctx, log).await.contains("skipping"),
2528 "the skip must be visible in the log"
2529 );
2530 }
2531
2532 #[tokio::test]
2533 async fn cargo_test_fails_closed_when_no_target_exists_in_the_worktree() {
2534 // A worktree missing every configured crate must not report "tests
2535 // passed" having run none. Uses an empty tempdir as the worktree, so
2536 // no cargo process is ever spawned.
2537 let tmp = tempfile::tempdir().unwrap();
2538 let mut cfg = crate::config::Config::for_tests();
2539 cfg.test_targets = vec![target("server"), target("mnw-cli")];
2540 cfg.logs_root = tmp.path().join("logs");
2541 let pool = SqlitePoolOptions::new()
2542 .max_connections(1)
2543 .connect("sqlite::memory:")
2544 .await
2545 .unwrap();
2546 let ctx = GateCtx {
2547 pool,
2548 cfg: std::sync::Arc::new(cfg),
2549 tier: TierId::new("host"),
2550 version: "0.1.0".parse().unwrap(),
2551 worktree: tmp.path().to_path_buf(),
2552 events: events::channel(),
2553 nodes: Vec::new(),
2554 build_id: None,
2555 };
2556 let out = cargo_test(&ctx, GateRunId(1)).await.unwrap();
2557 assert_eq!(
2558 out.status_str(),
2559 "failed",
2560 "green here would be a silent no-op gate"
2561 );
2562 let crate::outcome::GateStatus::Failed { failure } = &out.status else {
2563 panic!("expected a failure")
2564 };
2565 assert!(
2566 failure.summary().contains("ran no targets"),
2567 "got: {}",
2568 failure.summary()
2569 );
2570 }
2571
2572 #[tokio::test]
2573 async fn scratch_db_env_is_opt_in_per_target() {
2574 // Exporting DATABASE_URL knocks sqlx out of offline mode, so a crate
2575 // shipping .sqlx data must not see it.
2576 let mut cfg = crate::config::Config::for_tests();
2577 cfg.scratch_db_url = Some("postgres://sando@127.0.0.1/sando_scratch".into());
2578 let ctx = GateCtx {
2579 pool: SqlitePoolOptions::new()
2580 .max_connections(1)
2581 .connect_lazy("sqlite::memory:")
2582 .unwrap(),
2583 cfg: std::sync::Arc::new(cfg),
2584 tier: TierId::new("host"),
2585 version: "0.1.0".parse().unwrap(),
2586 worktree: std::path::PathBuf::from("/tmp/wt"),
2587 events: events::channel(),
2588 nodes: Vec::new(),
2589 build_id: None,
2590 };
2591 let dir = std::path::Path::new("/tmp/wt/x");
2592
2593 let off = cargo_test_command(&ctx, dir, &target("x"), &[], &[]);
2594 let has_db = |c: &Command| {
2595 c.as_std()
2596 .get_envs()
2597 .any(|(k, v)| k == "DATABASE_URL" && v.is_some())
2598 };
2599 assert!(!has_db(&off), "scratch_db defaults off");
2600
2601 let mut on_target = target("x");
2602 on_target.scratch_db = true;
2603 assert!(
2604 has_db(&cargo_test_command(&ctx, dir, &on_target, &[], &[])),
2605 "opt-in exports it"
2606 );
2607 }
2608
2609 #[tokio::test]
2610 async fn all_features_replaces_the_feature_list() {
2611 let ctx = GateCtx {
2612 pool: SqlitePoolOptions::new()
2613 .max_connections(1)
2614 .connect_lazy("sqlite::memory:")
2615 .unwrap(),
2616 cfg: std::sync::Arc::new(crate::config::Config::for_tests()),
2617 tier: TierId::new("host"),
2618 version: "0.1.0".parse().unwrap(),
2619 worktree: std::path::PathBuf::from("/tmp/wt"),
2620 events: events::channel(),
2621 nodes: Vec::new(),
2622 build_id: None,
2623 };
2624 let mut t = target("shared/ops-exec");
2625 t.all_features = true;
2626 let cmd = cargo_test_command(&ctx, std::path::Path::new("/tmp/wt"), &t, &[], &[]);
2627 let args: Vec<_> = cmd
2628 .as_std()
2629 .get_args()
2630 .map(|a| a.to_string_lossy().into_owned())
2631 .collect();
2632 assert!(args.iter().any(|a| a == "--all-features"), "got: {args:?}");
2633 assert!(!args.iter().any(|a| a == "--features"), "got: {args:?}");
2634 }
2635
2636 #[test]
2637 fn every_gate_kind_round_trips_through_its_wire_string() {
2638 // gate_kind is a TEXT column and a WS event field; as_str and FromStr
2639 // disagreeing would make a gate's evidence unreadable by
2640 // unsatisfied_gates, which fails the promote closed with no explanation.
2641 for k in [
2642 GateKind::CargoTest,
2643 GateKind::HardeningTest,
2644 GateKind::Clippy,
2645 GateKind::Fmt,
2646 GateKind::CargoAudit,
2647 GateKind::CargoDeny,
2648 GateKind::MigrationDryRun,
2649 GateKind::CodeSmoke,
2650 GateKind::BootSmoke,
2651 GateKind::NodeHealth,
2652 GateKind::BurnIn,
2653 GateKind::ManualConfirm,
2654 ] {
2655 assert_eq!(
2656 k.as_str().parse::<GateKind>().unwrap(),
2657 k,
2658 "round trip for {k:?}"
2659 );
2660 }
2661 }
2662
2663 #[test]
2664 fn first_meaningful_line_prefers_the_diagnostic() {
2665 let stderr = b" Updating crates.io index\nerror: 1 vulnerability found!\n";
2666 assert_eq!(
2667 first_meaningful_line(b"", stderr),
2668 "error: 1 vulnerability found!"
2669 );
2670 }
2671
2672 #[test]
2673 fn first_meaningful_line_finds_a_deny_verdict() {
2674 let out = b"advisories FAILED, bans ok, licenses FAILED, sources ok\n";
2675 assert!(first_meaningful_line(out, b"").contains("FAILED"));
2676 }
2677
2678 #[test]
2679 fn first_meaningful_line_falls_back_rather_than_returning_empty() {
2680 assert!(first_meaningful_line(b"", b"").contains("see the gate log"));
2681 }
2682
2683 #[tokio::test]
2684 async fn supply_chain_gates_skip_crates_without_their_config() {
2685 // Four crates in this repo fail `cargo audit` purely for want of a
2686 // triaged .cargo/audit.toml. Running it there would make the gate
2687 // permanently red, so a target only qualifies once it carries the file.
2688 // The worktree here has a Cargo.toml but no audit config, so nothing
2689 // qualifies and the gate fails closed rather than passing over zero work.
2690 let tmp = tempfile::tempdir().unwrap();
2691 let crate_dir = tmp.path().join("server");
2692 std::fs::create_dir_all(&crate_dir).unwrap();
2693 std::fs::write(crate_dir.join("Cargo.toml"), "[package]\nname = \"x\"\n").unwrap();
2694
2695 let mut cfg = crate::config::Config::for_tests();
2696 cfg.test_targets = vec![target("server")];
2697 cfg.logs_root = tmp.path().join("logs");
2698 let ctx = GateCtx {
2699 pool: SqlitePoolOptions::new()
2700 .max_connections(1)
2701 .connect("sqlite::memory:")
2702 .await
2703 .unwrap(),
2704 cfg: std::sync::Arc::new(cfg),
2705 tier: TierId::new("host"),
2706 version: "0.1.0".parse().unwrap(),
2707 worktree: tmp.path().to_path_buf(),
2708 events: events::channel(),
2709 nodes: Vec::new(),
2710 build_id: None,
2711 };
2712 let out = supply_chain(&ctx, GateRunId(1), GateKind::CargoAudit)
2713 .await
2714 .unwrap();
2715 assert_eq!(out.status_str(), "failed");
2716 let crate::outcome::GateStatus::Failed { failure } = &out.status else {
2717 panic!("expected a failure")
2718 };
2719 assert!(
2720 failure.summary().contains("ran nothing"),
2721 "got: {}",
2722 failure.summary()
2723 );
2724
2725 // Drop the config in and the same target now qualifies.
2726 std::fs::create_dir_all(crate_dir.join(".cargo")).unwrap();
2727 std::fs::write(crate_dir.join(".cargo/audit.toml"), "[advisories]\n").unwrap();
2728 // The fixture crate is not a real cargo project, so the tool itself
2729 // still errors — but on its own terms, not with "ran nothing". That
2730 // distinction is the thing under test: the target was attempted.
2731 let out = supply_chain(&ctx, GateRunId(2), GateKind::CargoAudit)
2732 .await
2733 .unwrap();
2734 if let crate::outcome::GateStatus::Failed { failure } = &out.status {
2735 assert!(
2736 !failure.summary().contains("ran nothing"),
2737 "a target carrying the config must be attempted, not skipped; got: {}",
2738 failure.summary(),
2739 );
2740 }
2741 }
2742
2743 #[test]
2744 fn tests_run_reads_the_libtest_summary() {
2745 let out = b"running 6 tests\ntest auth_rate_limit_triggers_on_burst ... ok\n\n\
2746 test result: ok. 6 passed; 0 failed; 0 ignored; 0 measured; 118 filtered out\n";
2747 assert_eq!(tests_run(out), 6);
2748 }
2749
2750 #[test]
2751 fn tests_run_sums_across_test_binaries() {
2752 let out = b"test result: ok. 6 passed; 0 failed\ntest result: ok. 2 passed; 0 failed\n";
2753 assert_eq!(tests_run(out), 8);
2754 }
2755
2756 #[test]
2757 fn tests_run_is_zero_when_the_filter_matched_nothing() {
2758 // The case hardening_test fails closed on: a filter matching no tests
2759 // exits 0, so a rename would otherwise make the gate a green no-op.
2760 let out = b"running 0 tests\n\n\
2761 test result: ok. 0 passed; 0 failed; 0 ignored; 0 measured; 124 filtered out\n";
2762 assert_eq!(tests_run(out), 0);
2763 }
2764
2765 #[test]
2766 fn tests_run_is_zero_without_a_summary_line() {
2767 assert_eq!(tests_run(b"error: could not compile `makenotwork`\n"), 0);
2768 }
2769
2770 #[tokio::test]
2771 async fn hardening_test_command_carries_no_features() {
2772 // The entire point of the gate: production constants, which means no
2773 // `fast-tests`. A stray feature here silently restores the blind spot.
2774 let ctx = GateCtx {
2775 pool: SqlitePoolOptions::new()
2776 .max_connections(1)
2777 .connect_lazy("sqlite::memory:")
2778 .unwrap(),
2779 cfg: std::sync::Arc::new(crate::config::Config::for_tests()),
2780 tier: TierId::new("host"),
2781 version: "0.1.0".parse().unwrap(),
2782 worktree: std::path::PathBuf::from("/tmp/wt"),
2783 events: events::channel(),
2784 nodes: Vec::new(),
2785 build_id: None,
2786 };
2787 let plain = crate::config::TestTarget {
2788 dir: std::path::PathBuf::from("server"),
2789 features: Vec::new(),
2790 all_features: false,
2791 scratch_db: true,
2792 };
2793 let cmd = cargo_test_command(
2794 &ctx,
2795 std::path::Path::new("/tmp/wt/server"),
2796 &plain,
2797 &[],
2798 &["--test", "integration"],
2799 );
2800 let args: Vec<_> = cmd
2801 .as_std()
2802 .get_args()
2803 .map(|a| a.to_string_lossy().into_owned())
2804 .collect();
2805 assert!(
2806 !args.iter().any(|a| a == "--features"),
2807 "hardening_test must pass no features: {args:?}"
2808 );
2809 assert!(
2810 !args.iter().any(|a| a.contains("fast-tests")),
2811 "got: {args:?}"
2812 );
2813
2814 let fast_target = crate::config::TestTarget {
2815 dir: std::path::PathBuf::from("server"),
2816 features: vec!["fast-tests".into()],
2817 all_features: false,
2818 scratch_db: true,
2819 };
2820 let fast = cargo_test_command(
2821 &ctx,
2822 std::path::Path::new("/tmp/wt/server"),
2823 &fast_target,
2824 &["fast-tests"],
2825 &[],
2826 );
2827 let fast_args: Vec<_> = fast
2828 .as_std()
2829 .get_args()
2830 .map(|a| a.to_string_lossy().into_owned())
2831 .collect();
2832 assert!(
2833 fast_args
2834 .windows(2)
2835 .any(|w| w == ["--features", "fast-tests"]),
2836 "got: {fast_args:?}"
2837 );
2838 }
2839
2840 /// Spawn a one-shot loopback server that answers the first connection with
2841 /// `status_line` + a tiny body, then closes. Returns the bound port.
2842 async fn oneshot_http(status_line: &'static str) -> u16 {
2843 use tokio::io::{AsyncReadExt, AsyncWriteExt};
2844 let listener = tokio::net::TcpListener::bind((std::net::Ipv4Addr::LOCALHOST, 0))
2845 .await
2846 .unwrap();
2847 let port = listener.local_addr().unwrap().port();
2848 tokio::spawn(async move {
2849 if let Ok((mut sock, _)) = listener.accept().await {
2850 let mut scratch = [0u8; 1024];
2851 let _ = sock.read(&mut scratch).await; // drain the request line
2852 let resp =
2853 format!("{status_line}\r\nContent-Length: 2\r\nConnection: close\r\n\r\nok");
2854 let _ = sock.write_all(resp.as_bytes()).await;
2855 }
2856 });
2857 port
2858 }
2859
2860 #[tokio::test]
2861 async fn probe_health_ok_on_200() {
2862 let port = oneshot_http("HTTP/1.1 200 OK").await;
2863 assert!(probe_health(port).await.is_ok());
2864 }
2865
2866 #[tokio::test]
2867 async fn probe_health_err_on_non_200() {
2868 let port = oneshot_http("HTTP/1.1 503 Service Unavailable").await;
2869 let err = probe_health(port).await.unwrap_err();
2870 assert!(err.contains("status line"), "{err}");
2871 }
2872
2873 #[tokio::test]
2874 async fn probe_health_err_on_connection_refused() {
2875 // Bind then drop to get an almost-certainly-free port nothing listens on.
2876 let port = {
2877 let l = tokio::net::TcpListener::bind((std::net::Ipv4Addr::LOCALHOST, 0))
2878 .await
2879 .unwrap();
2880 l.local_addr().unwrap().port()
2881 };
2882 let err = probe_health(port).await.unwrap_err();
2883 assert!(err.contains("connect"), "{err}");
2884 }
2885
2886 /// burn_in returns a typed Blocked when the clock isn't started; the
2887 /// runner persists status='blocked' + outcome_json (the json carries
2888 /// blocker.kind = 'burn_in_clock_not_started').
2889 #[tokio::test]
2890 async fn burn_in_blocked_persists_typed_outcome() {
2891 let pool = SqlitePoolOptions::new()
2892 .max_connections(1)
2893 .connect("sqlite::memory:")
2894 .await
2895 .unwrap();
2896 crate::db::migrate(&pool).await.unwrap();
2897 // Topology sync expects a tier row before gate_runs can reference it.
2898 sqlx::query("INSERT INTO tiers (name, ord, provisioned, canary) VALUES ('host', 0, 0, 'sequential')")
2899 .execute(&pool).await.unwrap();
2900 sqlx::query("INSERT INTO tier_state (tier) VALUES ('host')")
2901 .execute(&pool)
2902 .await
2903 .unwrap();
2904 // versions FK target.
2905 sqlx::query("INSERT INTO versions (version, git_sha, built_at, artifact_path) VALUES ('0.1.0', 'abc1234', '2026-01-01T00:00:00Z', '/tmp/x')")
2906 .execute(&pool).await.unwrap();
2907
2908 let cfg = std::sync::Arc::new(crate::config::Config::for_tests());
2909 let ctx = GateCtx {
2910 pool: pool.clone(),
2911 cfg,
2912 tier: TierId::new("host"),
2913 version: "0.1.0".parse().unwrap(),
2914 worktree: std::path::PathBuf::from("/tmp/unused"),
2915 events: events::channel(),
2916 nodes: Vec::new(),
2917 build_id: None,
2918 };
2919 let out = run(&ctx, &Gate::BurnIn { hours: 24 }).await.unwrap();
2920 assert_eq!(out.status_str(), "blocked");
2921 assert!(!out.is_passed());
2922
2923 // Read the persisted row.
2924 let row: (Option<String>, Option<String>) =
2925 sqlx::query_as("SELECT status, outcome_json FROM gate_runs ORDER BY id DESC LIMIT 1")
2926 .fetch_one(&pool)
2927 .await
2928 .unwrap();
2929 assert_eq!(row.0.as_deref(), Some("blocked"), "typed status");
2930 let json: serde_json::Value = serde_json::from_str(row.1.as_deref().unwrap()).unwrap();
2931 assert_eq!(json["status"]["kind"], "blocked");
2932 assert_eq!(
2933 json["status"]["blocker"]["kind"],
2934 "burn_in_clock_not_started"
2935 );
2936 }
2937
2938 /// node_health fails closed when there are no nodes to probe: a serving tier
2939 /// should always carry nodes, so an empty set is a misconfiguration that must
2940 /// block promotion, not pass it.
2941 #[tokio::test]
2942 async fn node_health_blocks_with_no_nodes() {
2943 let pool = SqlitePoolOptions::new()
2944 .max_connections(1)
2945 .connect("sqlite::memory:")
2946 .await
2947 .unwrap();
2948 crate::db::migrate(&pool).await.unwrap();
2949 sqlx::query(
2950 "INSERT INTO tiers (name, ord, provisioned, canary) VALUES ('b', 2, 1, 'sequential')",
2951 )
2952 .execute(&pool)
2953 .await
2954 .unwrap();
2955 sqlx::query("INSERT INTO tier_state (tier) VALUES ('b')")
2956 .execute(&pool)
2957 .await
2958 .unwrap();
2959 sqlx::query("INSERT INTO versions (version, git_sha, built_at, artifact_path) VALUES ('0.1.0', 'abc1234', '2026-01-01T00:00:00Z', '/tmp/x')")
2960 .execute(&pool).await.unwrap();
2961
2962 let cfg = std::sync::Arc::new(crate::config::Config::for_tests());
2963 let ctx = GateCtx {
2964 pool: pool.clone(),
2965 cfg,
2966 tier: TierId::new("b"),
2967 version: "0.1.0".parse().unwrap(),
2968 worktree: std::path::PathBuf::new(),
2969 events: events::channel(),
2970 nodes: Vec::new(), // no nodes -> fail closed
2971 build_id: None,
2972 };
2973 let out = run(&ctx, &Gate::NodeHealth).await.unwrap();
2974 assert_eq!(out.status_str(), "blocked");
2975 assert!(!out.is_passed());
2976 let row: (Option<String>, Option<String>) =
2977 sqlx::query_as("SELECT status, outcome_json FROM gate_runs ORDER BY id DESC LIMIT 1")
2978 .fetch_one(&pool)
2979 .await
2980 .unwrap();
2981 let json: serde_json::Value = serde_json::from_str(row.1.as_deref().unwrap()).unwrap();
2982 assert_eq!(json["status"]["blocker"]["kind"], "no_nodes_to_probe");
2983 }
2984
2985 /// reset_scratch must drop every non-system schema, not just `public` —
2986 /// otherwise migrations that create custom schemas (e.g. tower_sessions)
2987 /// collide on the next run. This regressed once (Phase 0) and the fix is
2988 /// load-bearing for migration_dry_run.
2989 ///
2990 /// Gated on `SANDO_TEST_PG_URL` so it only runs where postgres is
2991 /// available. Set `SANDO_TEST_PG_URL=postgres:///sando_scratch?host=/var/run/postgresql`
2992 /// (or similar) before `cargo test`.
2993 #[tokio::test]
2994 async fn reset_scratch_drops_all_non_system_schemas() {
2995 let Ok(url) = std::env::var("SANDO_TEST_PG_URL") else {
2996 eprintln!("skipping: SANDO_TEST_PG_URL not set");
2997 return;
2998 };
2999 use sqlx::Executor;
3000 use sqlx::postgres::PgPoolOptions;
3001
3002 let pool = PgPoolOptions::new()
3003 .max_connections(1)
3004 .connect(&url)
3005 .await
3006 .unwrap();
3007 // Plant two non-system schemas + a table in each.
3008 pool.execute(
3009 "DROP SCHEMA IF EXISTS foo CASCADE; CREATE SCHEMA foo; CREATE TABLE foo.t (i int);",
3010 )
3011 .await
3012 .unwrap();
3013 pool.execute("DROP SCHEMA IF EXISTS tower_sessions CASCADE; CREATE SCHEMA tower_sessions; CREATE TABLE tower_sessions.session (id text);")
3014 .await.unwrap();
3015 pool.close().await;
3016
3017 reset_scratch(&url, "makenotwork")
3018 .await
3019 .expect("reset_scratch");
3020
3021 let pool = PgPoolOptions::new()
3022 .max_connections(1)
3023 .connect(&url)
3024 .await
3025 .unwrap();
3026 let rows: Vec<(String,)> = sqlx::query_as(
3027 "SELECT nspname FROM pg_namespace WHERE nspname NOT LIKE 'pg_%' AND nspname <> 'information_schema'",
3028 )
3029 .fetch_all(&pool)
3030 .await
3031 .unwrap();
3032 let names: Vec<String> = rows.into_iter().map(|(s,)| s).collect();
3033 // After reset, only `public` should remain among non-system schemas.
3034 assert_eq!(names, vec!["public".to_string()], "got: {names:?}");
3035 pool.close().await;
3036 }
3037
3038 /// reset_scratch must leave the dump's owner role existing and able to
3039 /// create in `public`, because a prod `pg_dump` carries `ALTER ... OWNER TO
3040 /// <role>` for every object. This was satisfied by a hand-created NOLOGIN
3041 /// role on fw13; nothing recorded it, so any other box failed
3042 /// migration_dry_run at the restore with "role does not exist".
3043 ///
3044 /// Uses a throwaway role name so it can prove the *creation* path rather
3045 /// than passing on fw13's pre-existing `makenotwork`. Same
3046 /// `SANDO_TEST_PG_URL` gate as above; needs a superuser connection.
3047 #[tokio::test]
3048 async fn reset_scratch_seeds_the_dump_owner_role_when_absent() {
3049 let Ok(url) = std::env::var("SANDO_TEST_PG_URL") else {
3050 eprintln!("skipping: SANDO_TEST_PG_URL not set");
3051 return;
3052 };
3053 use sqlx::Executor;
3054 use sqlx::postgres::PgPoolOptions;
3055
3056 let role = "sando_test_owner_probe";
3057 // `DROP ROLE` refuses while the role still holds the grants reset_scratch
3058 // gave it, so drop what it owns first. Idempotent, and a no-op when the
3059 // role is absent (the usual case on a first run).
3060 let drop_role = format!(
3061 "DO $$ BEGIN
3062 IF EXISTS (SELECT 1 FROM pg_roles WHERE rolname = '{role}') THEN
3063 EXECUTE 'DROP OWNED BY {role}';
3064 EXECUTE 'DROP ROLE {role}';
3065 END IF;
3066 END $$;"
3067 );
3068
3069 let pool = PgPoolOptions::new()
3070 .max_connections(1)
3071 .connect(&url)
3072 .await
3073 .unwrap();
3074 pool.execute(sqlx::raw_sql(sqlx::AssertSqlSafe(drop_role.clone())))
3075 .await
3076 .unwrap();
3077 pool.close().await;
3078
3079 reset_scratch(&url, role)
3080 .await
3081 .expect("reset_scratch creates the owner role");
3082
3083 let pool = PgPoolOptions::new()
3084 .max_connections(1)
3085 .connect(&url)
3086 .await
3087 .unwrap();
3088 let (exists, can_login): (bool, bool) =
3089 sqlx::query_as("SELECT true, rolcanlogin FROM pg_roles WHERE rolname = $1")
3090 .bind(role)
3091 .fetch_one(&pool)
3092 .await
3093 .expect("owner role exists after reset");
3094 assert!(exists);
3095 assert!(
3096 !can_login,
3097 "the owner role is an owner only, never a login identity"
3098 );
3099
3100 // The restore's owner-scoped DDL needs CREATE on public in the role's
3101 // own right (PG15+ dropped the implicit grant).
3102 let (has_create,): (bool,) =
3103 sqlx::query_as("SELECT pg_catalog.has_schema_privilege($1, 'public', 'CREATE')")
3104 .bind(role)
3105 .fetch_one(&pool)
3106 .await
3107 .unwrap();
3108 assert!(has_create, "owner role must be able to create in public");
3109
3110 // Idempotent: a second reset must not error on the now-existing role.
3111 pool.close().await;
3112 reset_scratch(&url, role)
3113 .await
3114 .expect("reset_scratch is idempotent");
3115
3116 let pool = PgPoolOptions::new()
3117 .max_connections(1)
3118 .connect(&url)
3119 .await
3120 .unwrap();
3121 pool.execute(sqlx::raw_sql(sqlx::AssertSqlSafe(drop_role)))
3122 .await
3123 .unwrap();
3124 pool.close().await;
3125 }
3126
3127 /// The preflight must pass against a privileged scratch connection. Guards
3128 /// the catalog query itself: a wrong column or a `current_user` that matches
3129 /// no `pg_roles` row would make `fetch_one` error (or, worse, a silently
3130 /// swapped pair would invert the check) and brick startup for everyone.
3131 /// `SANDO_TEST_PG_URL` is expected to be a superuser connection, as the
3132 /// gates require.
3133 #[tokio::test]
3134 async fn preflight_passes_on_a_privileged_scratch_connection() {
3135 let Ok(url) = std::env::var("SANDO_TEST_PG_URL") else {
3136 eprintln!("skipping: SANDO_TEST_PG_URL not set");
3137 return;
3138 };
3139 preflight_scratch_privileges(&url)
3140 .await
3141 .expect("a superuser scratch connection must satisfy the preflight");
3142 }
3143
3144 /// CF4: the restore pipeline must carry `ON_ERROR_STOP=1` (so psql fails on
3145 /// a bad statement instead of exiting 0 on a partial restore) and, for a
3146 /// gzip source, `set -o pipefail` (so a `gunzip` failure on a truncated
3147 /// archive isn't masked by psql's exit). Pure string check — no postgres.
3148 #[test]
3149 fn restore_shell_has_error_stop_and_pipefail() {
3150 let gz = restore_shell("postgres:///scratch", "/srv/sando/backups/latest.sql.gz");
3151 assert!(gz.contains("ON_ERROR_STOP=1"), "gz: {gz}");
3152 assert!(gz.contains("set -o pipefail"), "gz: {gz}");
3153 assert!(gz.contains("gunzip -c"), "gz: {gz}");
3154
3155 let plain = restore_shell("postgres:///scratch", "/srv/sando/backups/dump.sql");
3156 assert!(plain.contains("ON_ERROR_STOP=1"), "plain: {plain}");
3157 // No pipeline for a plain .sql, so pipefail is unnecessary there.
3158 assert!(!plain.contains("gunzip"), "plain: {plain}");
3159 // The db url is single-quote escaped in both forms.
3160 assert!(plain.contains("'postgres:///scratch'"), "plain: {plain}");
3161 }
3162
3163 #[test]
3164 fn split_pg_password_extracts_and_sanitizes() {
3165 // Password lifted out of the URL; the sanitized form keeps user/host/db.
3166 let (url, pw) = split_pg_password("postgres://sando:s3cret@db.host:5432/scratch");
3167 assert_eq!(url, "postgres://sando@db.host:5432/scratch");
3168 assert_eq!(pw.as_deref(), Some("s3cret"));
3169 // Percent-encoded password is decoded for PGPASSWORD.
3170 let (url, pw) = split_pg_password("postgresql://u:p%40ss%2Fword@h/d");
3171 assert_eq!(url, "postgresql://u@h/d");
3172 assert_eq!(pw.as_deref(), Some("p@ss/word"));
3173 }
3174
3175 #[test]
3176 fn split_pg_password_noop_without_password() {
3177 // No userinfo password -> unchanged, None. (A ':' after the '@', e.g. a
3178 // port, must not be mistaken for the password delimiter.)
3179 assert_eq!(
3180 split_pg_password("postgres:///scratch"),
3181 ("postgres:///scratch".to_string(), None),
3182 );
3183 assert_eq!(
3184 split_pg_password("postgres://sando@db.host:5432/scratch"),
3185 ("postgres://sando@db.host:5432/scratch".to_string(), None),
3186 );
3187 }
3188
3189 #[test]
3190 fn percent_decode_handles_escapes_and_malformed() {
3191 assert_eq!(percent_decode("plain"), "plain");
3192 assert_eq!(percent_decode("a%2Fb"), "a/b");
3193 // A malformed trailing escape is left literal, not dropped.
3194 assert_eq!(percent_decode("ab%2"), "ab%2");
3195 assert_eq!(percent_decode("ab%zz"), "ab%zz");
3196 }
3197
3198 #[test]
3199 fn pg_url_with_dbname_rewrites_the_database() {
3200 // user:pass@host:port/db?query — swap db, keep everything else.
3201 assert_eq!(
3202 pg_url_with_dbname(
3203 "postgres://sando:pw@db.host:5432/sando_scratch?sslmode=require",
3204 "postgres"
3205 ),
3206 "postgres://sando:pw@db.host:5432/postgres?sslmode=require",
3207 );
3208 // Socket form: the query carries `host=/var/run/postgresql` and must survive.
3209 assert_eq!(
3210 pg_url_with_dbname(
3211 "postgres:///sando_scratch?host=/var/run/postgresql",
3212 "sando_code_smoke_0_9_6"
3213 ),
3214 "postgres:///sando_code_smoke_0_9_6?host=/var/run/postgresql",
3215 );
3216 // Plain host/db, no query.
3217 assert_eq!(
3218 pg_url_with_dbname("postgres://localhost/scratch", "postgres"),
3219 "postgres://localhost/scratch".replace("scratch", "postgres"),
3220 );
3221 // No authority, no query (loopback socket, default db path).
3222 assert_eq!(
3223 pg_url_with_dbname("postgres:///scratch", "postgres"),
3224 "postgres:///postgres",
3225 );
3226 }
3227
3228 #[test]
3229 fn bytes_contain_matches_listening_in_log_output() {
3230 // JSON release log carries the message field verbatim.
3231 assert!(bytes_contain(
3232 br#"{"timestamp":"...","level":"INFO","fields":{"message":"listening","addr":"127.0.0.1:18182"}}"#,
3233 b"listening",
3234 ));
3235 // Human-format dev log.
3236 assert!(bytes_contain(
3237 b"2026-07-17 INFO makenotwork: listening addr=127.0.0.1:18182",
3238 b"listening"
3239 ));
3240 assert!(!bytes_contain(
3241 b"migrations complete; seeding catalog",
3242 b"listening"
3243 ));
3244 assert!(!bytes_contain(b"", b"listening"));
3245 }
3246
3247 #[test]
3248 fn code_smoke_db_name_sanitizes_and_caps() {
3249 assert_eq!(
3250 code_smoke_db_name(&"0.9.6".parse().unwrap()),
3251 "sando_code_smoke_0_9_6"
3252 );
3253 // Pre-release/build metadata folds to underscores; result stays [a-z0-9_].
3254 let n = code_smoke_db_name(&"1.0.0-rc.1+build".parse().unwrap());
3255 assert_eq!(n, "sando_code_smoke_1_0_0_rc_1_build");
3256 assert!(
3257 n.bytes()
3258 .all(|b| b.is_ascii_lowercase() || b.is_ascii_digit() || b == b'_')
3259 );
3260 assert!(n.len() <= 63);
3261 }
3262
3263 /// code_smoke is Blocked (not Failed) when the daemon has no scratch_db_url:
3264 /// there's no cluster to create the throwaway DB in, and that's an operator
3265 /// precondition, rendered yellow — the same shape as migration_dry_run.
3266 #[tokio::test]
3267 async fn code_smoke_blocks_without_scratch_db_url() {
3268 let pool = SqlitePoolOptions::new()
3269 .max_connections(1)
3270 .connect("sqlite::memory:")
3271 .await
3272 .unwrap();
3273 crate::db::migrate(&pool).await.unwrap();
3274 sqlx::query("INSERT INTO tiers (name, ord, provisioned, canary) VALUES ('host', 0, 1, 'sequential')")
3275 .execute(&pool).await.unwrap();
3276 sqlx::query("INSERT INTO tier_state (tier) VALUES ('host')")
3277 .execute(&pool)
3278 .await
3279 .unwrap();
3280 sqlx::query("INSERT INTO versions (version, git_sha, built_at, artifact_path) VALUES ('0.1.0', 'abc1234', '2026-01-01T00:00:00Z', '/tmp/x')")
3281 .execute(&pool).await.unwrap();
3282
3283 let cfg = std::sync::Arc::new(crate::config::Config::for_tests()); // scratch_db_url: None
3284 let ctx = GateCtx {
3285 pool: pool.clone(),
3286 cfg,
3287 tier: TierId::new("host"),
3288 version: "0.1.0".parse().unwrap(),
3289 worktree: std::path::PathBuf::from("/tmp/unused"),
3290 events: events::channel(),
3291 nodes: Vec::new(),
3292 build_id: None,
3293 };
3294 let out = run(&ctx, &Gate::CodeSmoke).await.unwrap();
3295 assert_eq!(out.status_str(), "blocked");
3296 assert!(!out.is_passed());
3297 let row: (Option<String>, Option<String>) =
3298 sqlx::query_as("SELECT status, outcome_json FROM gate_runs ORDER BY id DESC LIMIT 1")
3299 .fetch_one(&pool)
3300 .await
3301 .unwrap();
3302 assert_eq!(row.0.as_deref(), Some("blocked"));
3303 let json: serde_json::Value = serde_json::from_str(row.1.as_deref().unwrap()).unwrap();
3304 assert_eq!(json["status"]["blocker"]["kind"], "scratch_db_url_unset");
3305 }
3306
3307 /// Sanity: applying MNW migrations from a *non-existent* dir errors,
3308 /// rather than silently no-op'ing. Cheap pure check, no postgres needed
3309 /// (the sqlx::Migrator::new constructor itself reads the dir).
3310 #[tokio::test]
3311 async fn run_migrator_errors_on_missing_dir() {
3312 // The first thing run_migrator does is `Migrator::new(dir)`, which
3313 // needs a real dir to read migration files from.
3314 let res = run_migrator(
3315 "postgres:///does-not-matter",
3316 std::path::Path::new("/nonexistent/sando-test-migrations"),
3317 )
3318 .await;
3319 assert!(res.is_err());
3320 }
3321 }
3322