src/korpprov.rs
//! **KORP-PROVISION-SUITE** — a *fresh* `korp-server --install --system` in a real KVM
//! guest, and then **nothing else**: does the service cut a region, tile it, upload it
//! and become visible to a client, unattended?
//!
//! ## Why this exists
//!
//! On oden, 2026-08-03, the whole chain ran for real: a restart → provisioning started
//! by itself → a 42.7-minute cut → tiling → **28 974 tile objects uploaded by the
//! service itself** → `region provisioning finished — 1 ready, 0 failed`.
//!
//! And it still failed on the last step. `korp::control::Control` resolves its map
//! roster **once**, at construction, and replays that answer for the life of the
//! process. The server had produced a map it could not itself see: `Assets → 0 region
//! asset(s)` with `error: ""` for minutes, against a store that already held every
//! tile. Only a **manual restart** made it visible — exactly the hand-touch this suite
//! exists to forbid.
//!
//! Every check that looked at S3, at `~/.korp/work`, at the region registry or at
//! `$KORP_STATE` was green throughout. The defect is visible **only** from the client
//! side of the wire, on a machine where systemd — not a test — started the service.
//! That is two properties no host-side unit test has, which is why this is a VM suite
//! and not another `#[test]`.
//!
//! ## What it reuses (LAW 5)
//!
//! Everything structural: the plan wire, the in-guest interpreter, the [`Facts`]
//! format, the KVM driver and the four systemd judges all come from
//! [`crate::vmsuite`]. This module adds only what korp-specific *applied output* needs
//! — a parquet footer, a per-zoom object count in the store, the roster a real client
//! got back — as more pure `fn(&KorpCheck, &Facts) -> Verdict`.
//!
//! It does **not** depend on korp. The guest binaries (`korp-server`, `korp-probe`),
//! the map source and the region polygon are handed in as paths, so Skidbladnir grades
//! a product it does not link.
//!
//! ## The scale model — same code, same formats, smaller input
//!
//! A 173 GB planet and a 42.7-minute cut cannot live in a VM loop, so the suite runs
//! The owner's two-level design at 1:31 000 scale:
//!
//! | role | production | scale model |
//! |---|---|---|
//! | "planet" | `planet-260427.osm.bz2`, 173.1 GB | `liechtenstein.osm.bz2`, 5.5 MB |
//! | cached step 1 | Sweden / Skåne (`admin_level` 2 / 4) | Oberland (`admin_level` 6), 3.8 MB `.osm.bz2`, 5.1 s |
//! | benched step 2 | Skåne → Lund | Oberland → Vaduz (`admin_level` 8), 0.27 s |
//!
//! Both polygons come out of the OSM data itself via `osm-katana admin-poly`, never a
//! download — the same block-parallel bz2 sweep that finds Sweden's `admin_level=2`
//! relation, at levels 6 and 8. The input keeps the `.osm.bz2` extension **on purpose**:
//! that is what routes the cut through the identical gatling bz2 reader →
//! `PolygonMask` → GeoParquet writer the planet goes through.
//!
//! What the model therefore does **not** prove: anything about 173 GB — sustained
//! decompression, the external node-coord sort spilling to disk, or a cut long enough
//! for the interruption seam to be observable. [`KorpCase::host_only_blind_spot`] and
//! the suite's own notes say so per case.
//!
//! ## Memory — a VM suite with no ceiling can take the box
//!
//! oden lost its whole tmux server to a global OOM on 2026-08-03. [`KorpSuiteConfig`]
//! therefore carries an explicit [`KorpSuiteConfig::memory_mb`] and
//! [`run_korp_suite`] **refuses to boot** unless the host has
//! [`FREE_MEMORY_FLOOR_MB`] free beyond it. A guest that cannot get its memory is a
//! failed run, never a smaller silent one.
use std::path::PathBuf;
#[cfg(feature = "matrix")]
use std::path::Path;
use std::time::Duration;
#[cfg(feature = "matrix")]
use std::time::Instant;
use anyhow::Result;
#[cfg(feature = "matrix")]
use anyhow::Context;
use serde::{Deserialize, Serialize};
#[cfg(feature = "matrix")]
use crate::machine::kvm::KvmController;
#[cfg(feature = "matrix")]
use crate::machine::{ImageRef, MachineController, MachineSpec, NetMode, Provision, Readiness};
use crate::vmsuite::{Check, Facts, PlanStep, Verdict};
#[cfg(feature = "matrix")]
use crate::vmsuite::parse_report;
/// The unit `korp-server --install --system` creates.
pub const UNIT: &str = "korp-server.service";
/// The dedicated system user that install provisions, and the user the unit declares.
pub const SVC_USER: &str = "korp";
/// The home `--install --system` gives [`SVC_USER`]; `~/.korp` hangs off it.
pub const SVC_HOME: &str = "/home/korp";
/// Where the guest's korp-server binary lands.
pub const KORP_SERVER: &str = "/usr/local/bin/korp-server";
/// Where the guest's thin client probe lands.
pub const KORP_PROBE: &str = "/usr/local/bin/korp-probe";
/// The in-guest scenario interpreter (shared with [`crate::vmsuite`]).
pub const SCENARIO: &str = "/usr/local/bin/skidbladnir-scenario";
/// The host-readable virtio-serial report port.
pub const REPORT_PORT: &str = "/dev/virtio-ports/org.skidbladnir.report";
/// **Where the map source is dropped so korp needs NO configuration to find it.**
///
/// `korp::clip::planet_candidates()` probes this path first among the `.bz2` entries.
/// Putting the source here is what makes the acceptance literally true: the guest is
/// given a binary and a map, and the only command anyone runs is
/// `korp-server --install --system`. No `$KORP_PLANET`, no drop-in, no verb.
pub const PLANET_DROP: &str = "/mnt/work4t/facett-osm-cache/planet.osm.bz2";
/// Where the region's clip polygon is staged in the guest.
pub const POLY_DIR: &str = "/var/lib/korp-fixtures";
/// The S3 bucket korp tiles into (`korp::assets::MAPS_BUCKET`).
pub const MAPS_BUCKET: &str = "nordisk-maps";
/// The embedded rustfs data root, relative to [`SVC_HOME`].
pub const STORE_ROOT: &str = "/home/korp/.korp/rustfs";
/// The region extract root, relative to [`SVC_HOME`] (`~/.korp/work/<id>`).
pub const WORK_ROOT: &str = "/home/korp/.korp/work";
/// Free host memory required **beyond** the guest's own allocation before a boot is
/// allowed. Not a round number for its own sake: qemu's own overhead plus the host
/// page cache the suite's parquet writes will dirty, with room left for the rest of
/// the fleet. See the module docs on the 2026-08-03 OOM.
pub const FREE_MEMORY_FLOOR_MB: u64 = 8_192;
// ─────────────────────────────── the korp judges ─────────────────────────────
/// What to assert about korp's **applied output** — files it wrote, objects it
/// uploaded, and the roster a real client got back over gRPC.
///
/// Every variant is graded by a pure function over [`Facts`], so each one is driven RED
/// from a host unit test with synthetic captures (see this module's `tests`).
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub enum KorpCheck {
/// systemd's `UnitFileState` is `enabled` — the install armed the unit for the
/// NEXT boot, not just this one.
UnitEnabled,
/// The region's extract exists with **both** parquet halves present, non-trivially
/// sized, and each carrying its `PAR1` footer.
///
/// The footer is the point: a cut killed part-way leaves a file that exists, has a
/// plausible size and a correct `PAR1` *header* — and no footer. That is the silent
/// form, and a mere existence check calls it success.
ExtractComplete { region: String },
/// The store holds exactly the tile objects the tiler said it uploaded, and the
/// same per-zoom breakdown the local pyramid has.
///
/// Compares two independent counts — the tiler's own sentence in the journal, and a
/// `find` over the store's on-disk objects. A tiler that reported a number it never
/// achieved, and a store that quietly took fewer, are different repairs.
TilesMatchTiler { region: String },
/// A **real client**, over gRPC, on the running server, sees `region` in the map
/// roster. This is the acceptance criterion, and today's defect.
ClientSeesRegion { region: String },
/// The service's main process is the SAME one the install started: same `MainPID`,
/// and systemd counted no restarts.
///
/// Without this, "the client saw the region" proves nothing — a service that
/// crash-looped its way to a fresh roster would pass, and that is precisely the
/// hand-restart the acceptance forbids.
NoRestartSince { pid_key: String },
/// The disk-margin guard REFUSED before the heavy step, and named both numbers.
ProvisionRefusedForDisk,
/// The region's work dir holds no half-written extract: either nothing at all, or
/// complete halves. A `nodes.parquet` with no `PAR1` footer is the failure.
NoHalfWrittenExtract { region: String },
/// A `WaitFor` whose argv mentions `needle` succeeded within `ms`.
SettledWithin { needle: String, ms: u64 },
}
/// Grade `check` against `facts`. **Pure** — no IO, no clock, no environment.
///
/// The governing rule is [`crate::vmsuite::judge`]'s: a fact that was never captured is
/// a **Fail**, never a Pass. Every judge here therefore separates `None` (the guest
/// never ran that capture) from `Some("")` (it ran and the answer was empty), and both
/// are red.
pub fn judge_korp(check: &KorpCheck, f: &Facts) -> Verdict {
match check {
KorpCheck::UnitEnabled => judge_unit_enabled(f),
KorpCheck::ExtractComplete { region } => judge_extract_complete(region, f, true),
KorpCheck::TilesMatchTiler { region } => judge_tiles_match(region, f),
KorpCheck::ClientSeesRegion { region } => judge_client_sees(region, f),
KorpCheck::NoRestartSince { pid_key } => judge_no_restart(pid_key, f),
KorpCheck::ProvisionRefusedForDisk => judge_refused_for_disk(f),
KorpCheck::NoHalfWrittenExtract { region } => judge_extract_complete(region, f, false),
KorpCheck::SettledWithin { needle, ms } => judge_settled_within(needle, *ms, f),
}
}
fn judge_unit_enabled(f: &Facts) -> Verdict {
match f.show_value("UnitFileState") {
Some("enabled") => Verdict::Pass(format!(
"UnitFileState=enabled (LoadState={})",
f.show_value("LoadState").unwrap_or("-")
)),
Some(other) => Verdict::Fail(format!(
"UnitFileState={other:?} — the service is running but NOT armed for the next boot, so \
a reboot silently un-provisions the box"
)),
None => Verdict::Fail(
"systemd reported no `UnitFileState` at all — the unit is not loaded, so nothing was \
enabled and there is nothing to pass"
.into(),
),
}
}
/// The **key** a case captures a parquet half's last four bytes under.
pub fn footer_key(region: &str, half: &str) -> String {
format!("footer_{region}_{half}")
}
/// The key a case captures a parquet half's byte size under.
pub fn size_key(region: &str, half: &str) -> String {
format!("size_{region}_{half}")
}
/// The two halves an extract must have. `relations.parquet` is written too, but these
/// two are what every read path needs, and what a killed cut truncates first.
pub const HALVES: &[&str] = &["nodes", "ways"];
/// A parquet file's minimum credible size: the 4-byte magic at each end, the 4-byte
/// footer length, and at least a schema in between. Anything at or below this is a
/// stub, not data.
const MIN_PARQUET_BYTES: u64 = 64;
/// Grade the extract. `must_exist` distinguishes the two questions that share this
/// evidence: *"is the extract complete?"* (a finished cut must have produced it) from
/// *"is there no HALF-written extract?"* (an interrupted cut may legitimately have
/// produced nothing at all — but never a footerless file).
fn judge_extract_complete(region: &str, f: &Facts, must_exist: bool) -> Verdict {
let mut seen: Vec<String> = Vec::new();
let mut problems: Vec<String> = Vec::new();
let mut absent = 0usize;
for half in HALVES {
let (fk, sk) = (footer_key(region, half), size_key(region, half));
let (Some(footer), Some(size_raw)) = (f.capture(&fk), f.capture(&sk)) else {
return Verdict::Fail(format!(
"the guest never captured `{fk}`/`{sk}` — nothing was read about \
{region}/{half}.parquet, so nothing can pass"
));
};
let size: u64 = size_raw.trim().parse().unwrap_or(0);
if size_raw.trim().is_empty() || size == 0 {
absent += 1;
seen.push(format!("{half}: ABSENT"));
continue;
}
seen.push(format!(
"{half}: {size} B, tail={:?}",
footer.trim_end_matches('\n')
));
if size < MIN_PARQUET_BYTES {
problems.push(format!(
"{half}.parquet is {size} B — below the {MIN_PARQUET_BYTES} B floor a parquet \
with a schema and both magics needs; it is a stub, not an extract"
));
}
// The silent form: a file that exists, is plausibly sized, and was never closed.
if footer.trim_end_matches('\n') != "PAR1" {
problems.push(format!(
"{half}.parquet has NO `PAR1` footer (last 4 bytes = {:?}) — the writer never \
closed it, which is exactly what a cut killed part-way leaves behind",
footer.trim_end_matches('\n')
));
}
}
if !problems.is_empty() {
return Verdict::Fail(format!("{region}: {}", problems.join("; ")));
}
if absent == HALVES.len() {
return if must_exist {
Verdict::Fail(format!(
"{region}: NEITHER parquet half exists — the cut produced nothing ({})",
seen.join(", ")
))
} else {
Verdict::Pass(format!(
"{region}: no extract at all — an interrupted cut left nothing half-written \
({})",
seen.join(", ")
))
};
}
if absent > 0 {
return Verdict::Fail(format!(
"{region}: {absent} of {} halves missing while the other(s) exist — a partial \
extract ({})",
HALVES.len(),
seen.join(", ")
));
}
Verdict::Pass(format!(
"{region}: both halves complete + PAR1-footered ({})",
seen.join(", ")
))
}
/// The key a case captures the tiler's own journal sentence under.
pub const TILER_JOURNAL_KEY: &str = "tiler_journal";
/// The key a case captures the store's on-disk object listing under.
pub fn store_key(region: &str) -> String {
format!("store_{region}")
}
/// How many objects the tiler SAID it uploaded, from its own sentence
/// (`korp/tile-maps: uploading 28974 tile objects to s3://…`). Pure over the text so
/// the parse is testable without a journal.
pub fn tiler_claimed_uploads(journal: &str) -> Option<usize> {
journal.lines().rev().find_map(|l| {
let rest = l.split("korp/tile-maps: uploading ").nth(1)?;
rest.split_whitespace().next()?.parse::<usize>().ok()
})
}
/// Per-zoom object counts from a `find` listing of store keys
/// (`…/<region>/tiles/z12/1130/590.parquet`), sorted by zoom.
pub fn per_zoom(listing: &str) -> Vec<(String, usize)> {
let mut counts: std::collections::BTreeMap<String, usize> = std::collections::BTreeMap::new();
for line in listing.lines() {
let line = line.trim();
if line.is_empty() {
continue;
}
// The zoom segment is the one directly after `tiles/`.
if let Some(z) = line
.split("/tiles/")
.nth(1)
.and_then(|r| r.split('/').next())
{
if z.starts_with('z') {
*counts.entry(z.to_string()).or_default() += 1;
}
}
}
counts.into_iter().collect()
}
fn judge_tiles_match(region: &str, f: &Facts) -> Verdict {
let sk = store_key(region);
let (Some(journal), Some(listing)) = (f.capture(TILER_JOURNAL_KEY), f.capture(&sk)) else {
return Verdict::Fail(format!(
"the guest never captured `{TILER_JOURNAL_KEY}`/`{sk}` — neither the tiler's own \
count nor the store's contents were read, so there is nothing to compare"
));
};
let zooms = per_zoom(listing);
let in_store: usize = zooms.iter().map(|(_, n)| n).sum();
let breakdown = || {
zooms
.iter()
.map(|(z, n)| format!("{z}:{n}"))
.collect::<Vec<_>>()
.join(" ")
};
let Some(claimed) = tiler_claimed_uploads(journal) else {
return Verdict::Fail(format!(
"the journal carries no `korp/tile-maps: uploading N tile objects` line — the tiler \
never reported a count, so the {in_store} object(s) in the store ({}) match nothing. \
A store populated by something other than this run's tiler is not evidence.",
breakdown()
));
};
if claimed == 0 {
return Verdict::Fail(
"the tiler reported uploading 0 tile objects — an empty pyramid is not a tiled region"
.into(),
);
}
if in_store != claimed {
return Verdict::Fail(format!(
"the tiler said it uploaded {claimed} object(s) to s3://{MAPS_BUCKET}/{region}/tiles/, \
the store holds {in_store} ({}) — {} went missing between the PUT and the disk",
breakdown(),
claimed.abs_diff(in_store),
));
}
if zooms.is_empty() {
return Verdict::Fail(format!(
"the store holds {in_store} object(s) for '{region}' but none under a `z<N>/` prefix — \
the key layout is not `<region>/tiles/z{{Z}}/{{X}}/{{Y}}.parquet`, so no viewport read \
will ever find them"
));
}
Verdict::Pass(format!(
"{claimed} object(s) claimed, {in_store} in s3://{MAPS_BUCKET}/{region}/tiles/ across {} \
zoom(s) — {}",
zooms.len(),
breakdown(),
))
}
/// The key a case captures `korp-probe`'s JSON reply under, per region asked for.
pub fn probe_key(region: &str) -> String {
format!("probe_{region}")
}
fn judge_client_sees(region: &str, f: &Facts) -> Verdict {
let pk = probe_key(region);
let Some(raw) = f.capture(&pk) else {
return Verdict::Fail(format!(
"the guest never captured `{pk}` — no client ever asked this server anything, so \
nothing can pass"
));
};
if raw.trim().is_empty() {
return Verdict::Fail(format!(
"`korp-probe` printed NOTHING when asked for '{region}'. It prints one JSON line on \
every path including failure, so an empty capture means it did not run at all — \
which is not the same as a server serving no regions."
));
}
let v: serde_json::Value = match serde_json::from_str(raw.trim()) {
Ok(v) => v,
Err(e) => {
return Verdict::Fail(format!(
"`korp-probe` output is not the JSON line it promises ({e}): {:?}",
raw.chars().take(200).collect::<String>()
));
}
};
let dial = v["dial_error"].as_str().unwrap_or("");
if !dial.is_empty() {
return Verdict::Fail(format!(
"a client could not reach the server at {} — {dial}",
v["endpoint"].as_str().unwrap_or("?")
));
}
let names: Vec<&str> = v["assets"]
.as_array()
.map(|a| a.iter().filter_map(|x| x["name"].as_str()).collect())
.unwrap_or_default();
let err = v["error"].as_str().unwrap_or("");
if names.contains(®ion) {
let node_count = v["assets"]
.as_array()
.and_then(|a| a.iter().find(|x| x["name"].as_str() == Some(region)))
.and_then(|x| x["node_count"].as_u64())
.unwrap_or(0);
return Verdict::Pass(format!(
"a gRPC client got {} region asset(s) {names:?} from the LIVE server; '{region}' is \
among them ({node_count} nodes)",
names.len()
));
}
Verdict::Fail(format!(
"a gRPC client asked the RUNNING server for its map roster and got {} region asset(s) \
{names:?} — '{region}' is NOT among them, listing error {err:?}. The service cut and \
uploaded this region itself; it cannot see its own output.",
names.len()
))
}
/// The key a case captures a `MainPID` under, for [`KorpCheck::NoRestartSince`].
pub fn pid_key(tag: &str) -> String {
format!("mainpid_{tag}")
}
fn judge_no_restart(pid_key: &str, f: &Facts) -> Verdict {
let Some(then) = f.capture(pid_key) else {
return Verdict::Fail(format!(
"the guest never captured `{pid_key}` — with no earlier PID there is nothing to \
compare, and 'no restart happened' cannot be asserted from one sample"
));
};
let then = then.trim();
let Some(now) = f.show_value("MainPID") else {
return Verdict::Fail(
"systemd reported no `MainPID` now — the unit is not loaded, so it certainly is not \
the same process"
.into(),
);
};
let now = now.trim();
if then.is_empty() || then == "0" {
return Verdict::Fail(format!(
"the earlier `{pid_key}` was {then:?} — the service was not running when the \
comparison point was taken, so a later PID proves nothing"
));
}
if now.is_empty() || now == "0" {
return Verdict::Fail(format!(
"MainPID is {now:?} now — the service is not running at all (Result={}, \
ExecMainStatus={})",
f.show_value("Result").unwrap_or("-"),
f.show_value("ExecMainStatus").unwrap_or("-"),
));
}
if then != now {
return Verdict::Fail(format!(
"the service RESTARTED: MainPID was {then}, is {now}. Whatever the client can see now \
was made visible by a new process — which is the manual restart this suite exists to \
forbid."
));
}
Verdict::Pass(format!(
"same process throughout — MainPID {now}, no restart"
))
}
/// The key a case captures korp-server's journal under.
pub const JOURNAL_KEY: &str = "journal";
fn judge_refused_for_disk(f: &Facts) -> Verdict {
let Some(journal) = f.capture(JOURNAL_KEY) else {
return Verdict::Fail(format!(
"the guest never captured `{JOURNAL_KEY}` — the service's own words were not read, so \
no refusal can be asserted"
));
};
let line = journal
.lines()
.rev()
.find(|l| l.contains("no-headroom") || l.contains("region provisioning REFUSED"));
match line {
None => Verdict::Fail(format!(
"the journal carries NO disk refusal — the service was started against a disk the \
override declared too small and started cutting anyway ({} journal lines read)",
journal.lines().count()
)),
Some(l) => {
// A refusal with no numbers is indistinguishable from a check that never ran.
let digits = l.chars().filter(|c| c.is_ascii_digit()).count();
if digits < 4 {
return Verdict::Fail(format!(
"the refusal names no measurement — a green (or red) with no numbers cannot be \
told from a check that did not run: {l:?}"
));
}
Verdict::Pass(format!(
"refused before the heavy step, with numbers: {}",
l.trim()
))
}
}
}
fn judge_settled_within(needle: &str, ms: u64, f: &Facts) -> Verdict {
let Some((code, waited)) = f.waited(needle) else {
return Verdict::Fail(format!(
"no WAITFOR mentioning {needle:?} ran in this case — nothing was timed, so no latency \
bound can pass"
));
};
if code != 0 {
return Verdict::Fail(format!(
"waiting on {needle:?} TIMED OUT after {waited} ms — the outcome never arrived on the \
service's own schedule"
));
}
if waited > ms {
return Verdict::Fail(format!(
"{needle:?} settled after {waited} ms, the bound is {ms} ms"
));
}
Verdict::Pass(format!("{needle:?} settled in {waited} ms (bound {ms} ms)"))
}
// ────────────────────────────── the case model ───────────────────────────────
/// One expectation: either a systemd judge reused verbatim from [`crate::vmsuite`], or
/// a korp applied-output judge from this module.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub enum AnyCheck {
Systemd(Check),
Korp(KorpCheck),
}
/// Grade either kind.
pub fn judge_any(check: &AnyCheck, f: &Facts) -> Verdict {
match check {
AnyCheck::Systemd(c) => crate::vmsuite::judge(c, f),
AnyCheck::Korp(c) => judge_korp(c, f),
}
}
/// One expectation on a case's facts.
#[derive(Debug, Clone)]
pub struct Expect {
pub name: String,
pub check: AnyCheck,
/// `false` marks a **deliberate red**: the case engineers a state the guard MUST
/// notice. A judge stuck on Pass fails those; a judge stuck on Fail fails the
/// positive ones. Neither hollow direction survives.
pub expect_pass: bool,
}
impl Expect {
pub fn sys(name: &str, c: Check) -> Self {
Expect {
name: name.into(),
check: AnyCheck::Systemd(c),
expect_pass: true,
}
}
pub fn korp(name: &str, c: KorpCheck) -> Self {
Expect {
name: name.into(),
check: AnyCheck::Korp(c),
expect_pass: true,
}
}
/// A guard that MUST go red here.
pub fn red(mut self) -> Self {
self.expect_pass = false;
self
}
}
/// One case of the korp suite.
pub struct KorpCase {
pub id: String,
pub steps: Vec<PlanStep>,
pub checks: Vec<Expect>,
/// Why a host-only (no-VM) test could not have caught what this case catches, or
/// `""` for a setup case that asserts nothing. Data, not a comment, so the test can
/// print it and a reviewer can audit the claim per case.
pub host_only_blind_spot: &'static str,
}
impl KorpCase {
fn setup(id: &str, steps: Vec<PlanStep>) -> Self {
KorpCase {
id: id.into(),
steps,
checks: Vec::new(),
host_only_blind_spot: "",
}
}
}
// ────────────────────────── step-building conveniences ───────────────────────
fn run(argv: &[&str]) -> PlanStep {
PlanStep::Run(argv.iter().map(|s| s.to_string()).collect())
}
fn cap(key: &str, argv: &[&str]) -> PlanStep {
PlanStep::Capture {
key: key.into(),
argv: argv.iter().map(|s| s.to_string()).collect(),
}
}
fn wait(timeout_ms: u64, poll_ms: u64, argv: &[&str]) -> PlanStep {
PlanStep::WaitFor {
timeout_ms,
poll_ms,
argv: argv.iter().map(|s| s.to_string()).collect(),
}
}
fn collect_korp() -> PlanStep {
PlanStep::Collect {
unit: UNIT.into(),
exec: KORP_SERVER.into(),
}
}
/// Capture both parquet halves of `region`'s extract: last four bytes and byte size.
///
/// ONE builder for both, so a case can never capture a size without its footer — which
/// would silently reduce [`KorpCheck::ExtractComplete`] to an existence check.
fn capture_extract(region: &str) -> Vec<PlanStep> {
let mut steps = Vec::new();
for half in HALVES {
let path = format!("{WORK_ROOT}/{region}/{half}.parquet");
steps.push(PlanStep::Capture {
key: footer_key(region, half),
argv: vec!["tail".into(), "-c".into(), "4".into(), path.clone()],
});
// `stat -c %s` prints nothing on a missing file, which the judge reads as
// ABSENT — distinct from a zero-byte file, which prints `0`.
steps.push(PlanStep::Capture {
key: size_key(region, half),
argv: vec!["stat".into(), "-c".into(), "%s".into(), path],
});
}
steps
}
/// Capture every object the store holds under `<region>/tiles/`.
fn capture_store(region: &str) -> PlanStep {
PlanStep::Capture {
key: store_key(region),
argv: vec![
"find".into(),
format!("{STORE_ROOT}/{MAPS_BUCKET}/{region}/tiles"),
"-type".into(),
"f".into(),
"-name".into(),
"*.parquet".into(),
],
}
}
/// The guest-side journal read. `-o cat` so the tiler's own sentence arrives verbatim.
fn capture_journal(key: &str) -> PlanStep {
PlanStep::Capture {
key: key.into(),
argv: vec![
"journalctl".into(),
"-u".into(),
UNIT.into(),
"--no-pager".into(),
"-o".into(),
"cat".into(),
"-n".into(),
"4000".into(),
],
}
}
// ─────────────────────────────────── the plan ────────────────────────────────
/// The region the suite provisions (the "Lund" of the scale model).
pub const REGION: &str = "vaduz";
/// A SECOND configured region, used only by the disk-refusal case: it must never be
/// cut, so its absence is the evidence.
pub const REGION_2: &str = "schaan";
/// A region name that exists nowhere — the live negative control for the client judge.
pub const REGION_ABSENT: &str = "atlantis";
/// korp's config file for the `korp` system user: which regions to provision, and
/// where each one's polygon is.
///
/// This is the deployment's declaration, staged with the binary and the map — NOT a
/// post-install verb. After it, the only command anyone runs is
/// `korp-server --install --system`.
pub fn region_config_json() -> String {
format!(
"{{\n \"regions\": [\n \
{{ \"id\": \"{REGION}\", \"name\": \"Vaduz\", \"poly\": \"{POLY_DIR}/{REGION}.poly\", \"enabled\": true }},\n \
{{ \"id\": \"{REGION_2}\", \"name\": \"Schaan\", \"poly\": \"{POLY_DIR}/{REGION_2}.poly\", \"enabled\": false }}\n \
]\n}}\n"
)
}
/// The drop-in that fakes a full disk, through korp-collectors' OWN override
/// (`KORP_TILE_FREE_BYTES`) — the one way to fake a full disk in this constellation,
/// not a second one (LAW 5).
pub const DISK_DROP_IN: &str = "/etc/systemd/system/korp-server.service.d/90-vmsuite-nodisk.conf";
fn disk_drop_in_body() -> String {
"# vm-suite: make the headroom guard see a disk that cannot hold a convert.\n\
[Service]\n\
Environment=\"KORP_TILE_FREE_BYTES=1024\"\n"
.to_string()
}
/// **Boot A** — the acceptance run, and the engineered reds around it.
///
/// Ordered as a story: install and nothing else; wait for the service to finish its own
/// provisioning; grade what it produced and what a client can see; then two negative
/// controls that must go red against the same live machine; then the disk refusal.
///
/// `cut_timeout_ms` bounds the unattended provisioning wait — the one number that has
/// to grow with the map source (0.27 s for Oberland→Vaduz on 32 cores, tens of minutes
/// for a planet).
pub fn plan_boot_a(cut_timeout_ms: u64) -> Vec<KorpCase> {
let mut cases: Vec<KorpCase> = Vec::new();
// 0 ── THE ONLY COMMAND. Everything after this case is observation.
cases.push(KorpCase {
id: "install-and-nothing-else".into(),
steps: vec![
run(&[KORP_SERVER, "--install", "--system"]),
// Give systemd a moment to fork the service, then take the PID the install
// produced. Every later "no restart" claim is measured against THIS.
wait(60_000, 500, &["systemctl", "is-active", UNIT]),
cap(&pid_key("install"), &["systemctl", "show", "-P", "MainPID", UNIT]),
collect_korp(),
],
checks: vec![
Expect::sys("resolved-user", Check::ResolvedUser(SVC_USER.into())),
Expect::sys("active", Check::Active(true)),
Expect::sys("no-foreign-dropins", Check::NoForeignDropIns(Vec::new())),
Expect::sys("proc-env-home", Check::ProcEnv("HOME".into(), SVC_HOME.into())),
Expect::korp("enabled", KorpCheck::UnitEnabled),
],
host_only_blind_spot:
"`--install --system` provisions a system user and hands a unit to a real init. \
What `User=` RESOLVES to after systemd merges every drop-in, and what the forked \
process actually got in /proc/<MainPID>/environ, exist only on a machine with a \
live systemd. On oden the identical install reported success and ran as `alice`.",
});
// 1 ── the acceptance: the service does the whole job by itself.
let mut steps = vec![
// The applied outcome, not a call: wait until BOTH parquet halves carry their
// PAR1 footer. `tail -c4 | grep -q PAR1` would need a shell, so the wait is on
// the file existing and the FOOTER is graded from the capture below.
wait(
cut_timeout_ms,
1_000,
&["test", "-s", &format!("{WORK_ROOT}/{REGION}/ways.parquet")],
),
// …then until the tiler has put its manifest in the store. That object is
// written LAST (after every tile PUT), so its presence is the tiling's own
// completion marker rather than a sleep long enough to hide a regression.
wait(
cut_timeout_ms,
1_000,
&[
"test",
"-f",
&format!("{STORE_ROOT}/{MAPS_BUCKET}/{REGION}/tiles/manifest.json"),
],
),
// …and finally until a CLIENT can see it. This is the acceptance criterion and
// the 2026-08-03 defect: on today's korp the roster is resolved once at
// startup, so this wait times out against a store that already holds every tile.
wait(120_000, 2_000, &[KORP_PROBE, "--expect", REGION]),
];
steps.extend(capture_extract(REGION));
steps.push(capture_store(REGION));
steps.push(capture_journal(TILER_JOURNAL_KEY));
steps.push(cap(&probe_key(REGION), &[KORP_PROBE, "--expect", REGION]));
steps.push(cap(
&pid_key("install"),
&["systemctl", "show", "-P", "MainPID", UNIT],
));
steps.push(collect_korp());
cases.push(KorpCase {
id: "self-provisions-unattended".into(),
steps,
checks: vec![
Expect::korp("extract-complete", KorpCheck::ExtractComplete { region: REGION.into() }),
Expect::korp("tiles-match-tiler", KorpCheck::TilesMatchTiler { region: REGION.into() }),
Expect::korp("client-sees-region", KorpCheck::ClientSeesRegion { region: REGION.into() }),
Expect::korp("no-restart", KorpCheck::NoRestartSince { pid_key: pid_key("install") }),
Expect::sys("still-active", Check::Active(true)),
],
host_only_blind_spot:
"THE ACCEPTANCE. Nothing here was triggered by the test: systemd started the \
service, the service decided to cut, tile and upload, and a gRPC client asked the \
SAME still-running process what it serves. A host test can call every one of those \
functions and still not answer whether a running server ever notices its own output \
— which is the 2026-08-03 defect exactly.",
});
// 2 ── the LIVE negative control for the client judge: same client, same running
// server, a region that exists nowhere.
cases.push(KorpCase {
id: "client-refuses-a-region-that-is-not-there".into(),
steps: vec![
cap(&probe_key(REGION_ABSENT), &[KORP_PROBE, "--expect", REGION_ABSENT]),
collect_korp(),
],
checks: vec![Expect::korp(
"client-sees-region",
KorpCheck::ClientSeesRegion { region: REGION_ABSENT.into() },
)
.red()],
host_only_blind_spot:
"Proves the client judge is not stuck on Pass, against the SAME live server that \
answered the case above — not against synthetic JSON.",
});
// 3 ── the LIVE negative control for the extract judge: a region nobody cut.
let mut steps = capture_extract(REGION_2);
steps.push(collect_korp());
cases.push(KorpCase {
id: "an-uncut-region-has-no-extract".into(),
steps,
checks: vec![
Expect::korp("extract-complete", KorpCheck::ExtractComplete { region: REGION_2.into() })
.red(),
// …and the SAME evidence read the other way is a PASS: nothing was left
// half-written. One capture, two opposite verdicts — which is what proves
// the footer judge is reading, not guessing.
Expect::korp(
"no-half-written-extract",
KorpCheck::NoHalfWrittenExtract { region: REGION_2.into() },
),
],
host_only_blind_spot:
"The disabled region is real config on a real machine: it proves `enabled: false` is \
honoured by the provisioner rather than by the test's imagination.",
});
// 4 ── the LIVE negative control for the tile judge: a store prefix with nothing in
// it must not read as agreement.
cases.push(KorpCase {
id: "an-untiled-region-does-not-match-a-tiler".into(),
steps: vec![capture_store(REGION_2), capture_journal(TILER_JOURNAL_KEY), collect_korp()],
checks: vec![Expect::korp(
"tiles-match-tiler",
KorpCheck::TilesMatchTiler { region: REGION_2.into() },
)
.red()],
host_only_blind_spot:
"An empty `find` and a journal that never mentions this region: the shape a tile \
check most easily passes on. It must be red.",
});
// 5 ── DISK: the guard refuses before the heavy step. This is the one case that
// restarts the service — deliberately, because a precondition checked at
// startup can only be re-checked at a startup.
cases.push(KorpCase {
id: "disk-margin-refuses-before-it-starts".into(),
steps: vec![
PlanStep::Write {
path: DISK_DROP_IN.into(),
mode: 0o644,
contents: disk_drop_in_body(),
},
// A fresh region for the refusal to be ABOUT: re-enable schaan so the
// provisioner has real work it must decline.
PlanStep::Write {
path: format!("{SVC_HOME}/.config/korp/config.json"),
mode: 0o644,
contents: region_config_json().replace(
&format!("\"{REGION_2}\", \"name\": \"Schaan\", \"poly\": \"{POLY_DIR}/{REGION_2}.poly\", \"enabled\": false"),
&format!("\"{REGION_2}\", \"name\": \"Schaan\", \"poly\": \"{POLY_DIR}/{REGION_2}.poly\", \"enabled\": true"),
),
},
run(&["chown", "-R", "korp:korp", &format!("{SVC_HOME}/.config")]),
run(&["systemctl", "daemon-reload"]),
run(&["systemctl", "restart", UNIT]),
wait(60_000, 500, &["systemctl", "is-active", UNIT]),
PlanStep::Sleep(8_000),
capture_journal(JOURNAL_KEY),
]
.into_iter()
.chain(capture_extract(REGION_2))
.chain([collect_korp()])
.collect(),
checks: vec![
Expect::korp("refused-for-disk", KorpCheck::ProvisionRefusedForDisk),
// The refusal must be a refusal: no extract may appear for the region it
// declined. A guard that logged "no headroom" and cut anyway is worse than
// no guard.
Expect::korp(
"no-half-written-extract",
KorpCheck::NoHalfWrittenExtract { region: REGION_2.into() },
),
Expect::sys("still-active", Check::Active(true)),
],
host_only_blind_spot:
"The decision is made once, at process start, by a service systemd launched. \
Calling `headroom()` in-process proves the arithmetic; only a real start proves \
the service consults it BEFORE it starts cutting, and stays up when it declines.",
});
// 6 ── clear the landmine and arm the reboot.
cases.push(KorpCase::setup(
"arm-boot-b",
vec![
PlanStep::Rm(DISK_DROP_IN.into()),
run(&["systemctl", "daemon-reload"]),
PlanStep::Write {
path: PLAN_B_PATH.into(),
mode: 0o644,
contents: crate::vmsuite::render_plan(&to_vmsuite_cases(&plan_boot_b())),
},
PlanStep::Write {
path: "/etc/systemd/system/korp-bootreport.service".into(),
mode: 0o644,
contents: boot_report_unit_body(PLAN_B_PATH),
},
run(&["systemctl", "daemon-reload"]),
run(&["systemctl", "enable", "korp-bootreport.service"]),
PlanStep::PowerOff,
],
));
cases
}
/// Where the boot-B plan is staged inside the guest.
pub const PLAN_B_PATH: &str = "/var/lib/skid/korp-plan-b.txt";
fn boot_report_unit_body(plan_path: &str) -> String {
format!(
"[Unit]\n\
Description=korp provision suite boot-B reporter\n\
After=multi-user.target network-online.target {UNIT}\n\
[Service]\n\
Type=oneshot\n\
ExecStart={SCENARIO} --plan {plan_path} --report {REPORT_PORT} --also /var/log/korp-report-b.jsonl\n\
[Install]\n\
WantedBy=multi-user.target\n"
)
}
/// **Boot B** — what only a fresh boot can show.
///
/// Two facts live here and nowhere else:
///
/// 1. the install SURVIVED a reboot — systemd brought korp-server up from its
/// `[Install]` wants, as `korp`, ordered after the network;
/// 2. a client sees the region **immediately after a cold boot**. That is the direction
/// that works on today's korp (the roster is resolved once — at startup, when the
/// store is already full), and it is the partner of the boot-A acceptance: together
/// they localise the defect to *"resolved once, never re-resolved"* rather than
/// *"never resolved"*.
pub fn plan_boot_b() -> Vec<KorpCase> {
vec![
KorpCase {
id: "survives-a-reboot".into(),
steps: vec![
wait(120_000, 1_000, &[KORP_PROBE, "--expect", REGION]),
cap(&probe_key(REGION), &[KORP_PROBE, "--expect", REGION]),
capture_store(REGION),
capture_journal(TILER_JOURNAL_KEY),
collect_korp(),
],
checks: vec![
Expect::sys(
"ordered-after-network-online",
Check::OrderedAfterNetworkOnline,
),
Expect::sys("active", Check::Active(true)),
Expect::sys("resolved-user", Check::ResolvedUser(SVC_USER.into())),
Expect::korp(
"client-sees-region",
KorpCheck::ClientSeesRegion {
region: REGION.into(),
},
),
Expect::korp(
"client-settles-fast",
KorpCheck::SettledWithin {
needle: KORP_PROBE.into(),
ms: 120_000,
},
),
],
host_only_blind_spot: "ONLY a fresh boot produces this: systemd, not a test, decided when to start the \
service relative to network-online.target, and the roster it resolved at that \
start is the one a client reads. The tiles were made in the PREVIOUS boot, so \
this is the only case where the store is already full when the process begins.",
},
KorpCase {
id: "the-reboot-did-not-re-cut".into(),
steps: capture_extract(REGION)
.into_iter()
.chain([collect_korp()])
.collect(),
checks: vec![Expect::korp(
"extract-complete",
KorpCheck::ExtractComplete {
region: REGION.into(),
},
)],
host_only_blind_spot: "Idempotence across a real reboot: the extract from boot A is still whole, not \
re-cut and not truncated by a second provisioning pass racing the first.",
},
KorpCase::setup("boot-b-done", vec![PlanStep::PowerOff]),
]
}
/// Project korp cases onto the shape [`crate::vmsuite::render_plan`] renders. The wire
/// is the SAME wire — only the checks differ — so the plan text is produced by the ONE
/// renderer rather than a second one that could drift from the guest's parser.
fn to_vmsuite_cases(cases: &[KorpCase]) -> Vec<crate::vmsuite::SuiteCase> {
cases
.iter()
.map(|c| crate::vmsuite::SuiteCase {
id: c.id.clone(),
steps: c.steps.clone(),
checks: Vec::new(),
host_only_blind_spot: c.host_only_blind_spot,
})
.collect()
}
/// Render the plan wire for `cases`.
pub fn render_korp_plan(cases: &[KorpCase]) -> String {
crate::vmsuite::render_plan(&to_vmsuite_cases(cases))
}
// ──────────────────────────────── the driver ─────────────────────────────────
/// Where the korp suite finds its inputs and puts its scratch.
#[derive(Debug, Clone)]
pub struct KorpSuiteConfig {
/// A cached cloud qcow2 whose glibc can run `korp_server`. Never booted directly —
/// a per-run copy is made.
pub base_image: PathBuf,
/// Grow the per-run disk to this many GiB before boot. korp's headroom guard wants
/// 12 GB free before it will convert, so a 3.5 GB cloud image would make the
/// acceptance case refuse for a reason that has nothing to do with korp.
pub disk_gib: u64,
pub work_root: PathBuf,
pub scenario_bin: PathBuf,
pub probe_bin: PathBuf,
/// The product under test.
pub korp_server_bin: PathBuf,
/// The thin gRPC client that answers "what does a client see?".
pub korp_probe_bin: PathBuf,
/// The map source (`.osm.bz2`), dropped at [`PLANET_DROP`] so korp finds it with no
/// configuration at all.
pub map_source: PathBuf,
/// `(region-id, .poly path)` for [`REGION`] and [`REGION_2`].
pub polys: Vec<(String, PathBuf)>,
pub boot_timeout: Duration,
pub shutdown_grace: Duration,
/// How long the unattended provisioning may take before the wait is a timeout.
pub cut_timeout: Duration,
/// **The guest's memory ceiling.** Explicit and reported; see the module docs.
pub memory_mb: u32,
pub cpus: u32,
/// Runnable-per-core ceiling for [`load_gate`]. See [`LOAD_PER_CORE_CEILING`].
pub load_ceiling: f64,
}
/// One graded expectation.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct Outcome {
pub case: String,
pub check: String,
pub verdict: Verdict,
pub expect_pass: bool,
/// The guard behaved as declared. **This** is the suite's pass/fail.
pub as_declared: bool,
pub elapsed_ms: u128,
}
/// The result of a whole korp suite run.
#[derive(Debug, Default)]
pub struct KorpRun {
pub outcomes: Vec<Outcome>,
pub facts: Vec<Facts>,
pub case_timings: Vec<(String, u128)>,
pub boot_a_ms: u128,
pub boot_b_ms: u128,
pub boot_b_cases: Vec<String>,
/// Anything the run could not determine, verbatim.
pub notes: Vec<String>,
}
impl KorpRun {
pub fn failures(&self) -> Vec<&Outcome> {
self.outcomes.iter().filter(|o| !o.as_declared).collect()
}
}
/// Grade every case's facts. A case with no facts line is a **hard fail per
/// expectation**, never a skip: silence must not read as success.
pub fn grade(cases: &[KorpCase], facts: &[Facts], arrival_ms: &[(String, u128)]) -> Vec<Outcome> {
let mut out = Vec::new();
for c in cases {
if c.checks.is_empty() {
continue;
}
let elapsed = arrival_ms
.iter()
.find(|(id, _)| *id == c.id)
.map(|(_, ms)| *ms)
.unwrap_or(0);
match facts.iter().find(|f| f.case == c.id) {
Some(f) => {
for e in &c.checks {
let v = judge_any(&e.check, f);
let as_declared = v.is_pass() == e.expect_pass;
out.push(Outcome {
case: c.id.clone(),
check: e.name.clone(),
verdict: v,
expect_pass: e.expect_pass,
as_declared,
elapsed_ms: elapsed,
});
}
}
None => {
for e in &c.checks {
out.push(Outcome {
case: c.id.clone(),
check: e.name.clone(),
verdict: Verdict::Fail(
"no facts line arrived for this case — it asserted NOTHING".into(),
),
expect_pass: e.expect_pass,
as_declared: false,
elapsed_ms: 0,
});
}
}
}
}
out
}
/// The cloud-init `user-data` for boot A: install the payload, drop the map source
/// where korp probes for it, stage the region config, run the plan.
///
/// Pure (strings in → string out) so its shape is unit-testable without booting.
///
/// Note what is NOT here: no `korp-server` invocation. The install is a plan step, so
/// it is graded like everything else and its exit code lands in the facts line.
pub fn render_user_data(plan_written_to: &str, polys: &[String]) -> String {
let poly_installs: String = polys
.iter()
.map(|p| format!("install -m0644 /mnt/skid/{p} {POLY_DIR}/{p}; "))
.collect();
format!(
"#cloud-config\n\
# korp provision suite — boot A. The guest is given a binary, a client, a map\n\
# source and a region config. It runs no korp verb: the one install command is\n\
# a graded plan step, so its exit code lands in a facts line like everything else.\n\
growpart:\n\
\x20 mode: auto\n\
\x20 devices: ['/']\n\
resize_rootfs: true\n\
runcmd:\n\
\x20 - [ sh, -c, \"mkdir -p /mnt/skid; for d in /dev/vd* /dev/sd*; do mount -t vfat $d /mnt/skid 2>/dev/null && [ -f /mnt/skid/skidbladnir-scenario ] && break; umount /mnt/skid 2>/dev/null; done\" ]\n\
\x20 - [ sh, -c, \"install -m0755 /mnt/skid/skidbladnir-scenario {SCENARIO}; install -m0755 /mnt/skid/korp-server {KORP_SERVER}; install -m0755 /mnt/skid/korp-probe {KORP_PROBE}\" ]\n\
\x20 - [ sh, -c, \"mkdir -p {POLY_DIR} $(dirname {PLANET_DROP}) {SVC_HOME}/.config/korp /var/lib/skid; {poly_installs}true\" ]\n\
\x20 - [ sh, -c, \"install -m0644 /mnt/skid/map-source.osm.bz2 {PLANET_DROP}\" ]\n\
\x20 - [ sh, -c, \"install -m0644 /mnt/skid/korp-config.json {SVC_HOME}/.config/korp/config.json\" ]\n\
\x20 - [ sh, -c, \"cp /mnt/skid/plan.txt {plan_written_to}\" ]\n\
\x20 - [ sh, -c, \"{SCENARIO} --plan {plan_written_to} --report {REPORT_PORT} --also /var/log/korp-report-a.jsonl\" ]\n"
)
}
/// The FAT payload image — authored by the ONE seed builder
/// (`draupnir::seed::build_fat_image`), never a twinned `fatfs` call.
#[cfg(feature = "matrix")]
fn build_payload(out: &Path, cfg: &KorpSuiteConfig, plan: &str) -> Result<()> {
let mut files: Vec<(String, Vec<u8>)> = vec![
(
"skidbladnir-scenario".into(),
std::fs::read(&cfg.scenario_bin)
.with_context(|| format!("read {}", cfg.scenario_bin.display()))?,
),
(
"korp-server".into(),
std::fs::read(&cfg.korp_server_bin)
.with_context(|| format!("read {}", cfg.korp_server_bin.display()))?,
),
(
"korp-probe".into(),
std::fs::read(&cfg.korp_probe_bin)
.with_context(|| format!("read {}", cfg.korp_probe_bin.display()))?,
),
(
"map-source.osm.bz2".into(),
std::fs::read(&cfg.map_source)
.with_context(|| format!("read {}", cfg.map_source.display()))?,
),
("korp-config.json".into(), region_config_json().into_bytes()),
("plan.txt".into(), plan.as_bytes().to_vec()),
];
for (id, path) in &cfg.polys {
files.push((
format!("{id}.poly"),
std::fs::read(path).with_context(|| format!("read {}", path.display()))?,
));
}
draupnir::seed::build_fat_image(out, "korpdata", &files)?;
Ok(())
}
/// **Runnable-per-core ceiling.** Above this the guest's vCPUs are so oversubscribed
/// that a `WaitFor` measures the host's queue, not the service's schedule.
///
/// Not a taste: MEASURED on oden 2026-08-03 with the SAME suite, image and binary.
/// At load average ≈40 (1.3 runnable/core) the guest reported its install case in
/// ~60 s. At load average 173–221 (5.4–6.9 runnable/core — a 173 GB bz2 planet sweep
/// on 7.5 cores plus a large cargo build) boot A produced **zero** facts in 420 s. The
/// memory gate passed happily through that with 469 GB free, because it measures the
/// wrong resource. 3.0 sits between the two observations, nearer the bad one, so the
/// gate refuses a box that has already been shown to starve a guest.
pub const LOAD_PER_CORE_CEILING: f64 = 3.0;
/// The host's 1-minute load average, or `None` when `/proc/loadavg` cannot be read.
pub fn host_load1() -> Option<f64> {
std::fs::read_to_string("/proc/loadavg")
.ok()?
.split_whitespace()
.next()?
.parse()
.ok()
}
/// **Is this box quiet enough to time a guest on?** `Err` names both numbers.
///
/// Pure over its inputs so both verdicts are asserted without a VM. A gate that cannot
/// measure must not grant — the same rule the memory gate follows, for the same reason.
pub fn load_gate(load1: Option<f64>, cores: usize, ceiling: f64) -> Result<String, String> {
if cores == 0 {
return Err(
"could not determine the host's core count — refusing to time a guest blind".into(),
);
}
match load1 {
None => Err(
"could not read /proc/loadavg — refusing to boot blind. A suite whose latency \
assertions depend on host quiet must be able to measure that quiet."
.into(),
),
Some(l) => {
let per_core = l / cores as f64;
if per_core > ceiling {
Err(format!(
"load average {l:.1} over {cores} core(s) = {per_core:.2} runnable/core, \
ceiling {ceiling:.1} — refusing to boot. A guest starved this badly measures \
the host's run queue, not the service's own schedule (oden 2026-08-03: \
420 s, zero facts)"
))
} else {
Ok(format!(
"load average {l:.1} over {cores} core(s) = {per_core:.2} runnable/core \
≤ {ceiling:.1}"
))
}
}
}
}
/// Free host memory in MiB, from `/proc/meminfo`'s `MemAvailable`. `None` when it
/// cannot be read — which is treated as a refusal to boot, not as permission.
pub fn host_available_mb() -> Option<u64> {
let s = std::fs::read_to_string("/proc/meminfo").ok()?;
let kb: u64 = s
.lines()
.find_map(|l| l.strip_prefix("MemAvailable:"))?
.split_whitespace()
.next()?
.parse()
.ok()?;
Some(kb / 1024)
}
/// **May a guest of `memory_mb` be booted right now?** `Err` names both numbers.
///
/// Pure over its inputs so both verdicts are asserted without a VM — and so the
/// refusal itself cannot be the thing that rots. oden lost its tmux server to a global
/// OOM on 2026-08-03; a suite that boots regardless is a suite that can take the box.
pub fn memory_gate(
available_mb: Option<u64>,
memory_mb: u32,
floor_mb: u64,
) -> Result<String, String> {
let need = memory_mb as u64 + floor_mb;
match available_mb {
None => Err(format!(
"could not read MemAvailable from /proc/meminfo — refusing to boot a {memory_mb} MiB \
guest blind. A memory gate that cannot measure must not grant."
)),
Some(have) if have < need => Err(format!(
"{have} MiB available, the guest needs {memory_mb} MiB plus a {floor_mb} MiB host \
floor = {need} MiB — refusing to boot"
)),
Some(have) => Ok(format!(
"{have} MiB available ≥ {memory_mb} MiB guest + {floor_mb} MiB floor = {need} MiB"
)),
}
}
#[cfg(feature = "matrix")]
const END_MARKER: &str = "\"scenario_end\"";
#[cfg(feature = "matrix")]
fn tail_report(
path: &Path,
want: &[String],
t0: Instant,
timeout: Duration,
) -> (Vec<Facts>, Vec<(String, u128)>, bool) {
let deadline = t0 + timeout;
let mut seen: Vec<Facts> = Vec::new();
let mut timings: Vec<(String, u128)> = Vec::new();
let mut ended = false;
loop {
if let Ok(text) = std::fs::read_to_string(path) {
for f in parse_report(&text) {
if !seen.iter().any(|s| s.case == f.case) {
timings.push((f.case.clone(), t0.elapsed().as_millis()));
seen.push(f);
}
}
ended = text.contains(END_MARKER);
}
if ended && want.iter().all(|w| seen.iter().any(|f| f.case == *w)) {
return (seen, timings, true);
}
if Instant::now() >= deadline {
return (seen, timings, ended);
}
std::thread::sleep(Duration::from_millis(1_000));
}
}
#[cfg(feature = "matrix")]
fn boot_once(
cfg: &KorpSuiteConfig,
name: &str,
disk: &Path,
payload: &Path,
user_data: String,
report_path: &Path,
) -> Result<(KvmController, crate::machine::Handle)> {
let _ = std::fs::write(report_path, b"");
let spec = MachineSpec {
name: name.to_string(),
image: ImageRef::Path(disk.to_path_buf()),
cpus: cfg.cpus,
memory_mb: cfg.memory_mb,
net: NetMode::Nat,
provision: Provision::CloudInit {
user_data,
meta_data: None,
},
mounts: Vec::new(),
readiness: Readiness::Immediate,
iso: None,
data_iso: Some(payload.to_path_buf()),
serial_out: Some(report_path.to_path_buf()),
qga_sock: None,
// The suite handed over its OWN per-run copy and boots it TWICE: once to
// install and provision, once to watch systemd bring the installed system up.
// Without this the second boot gets a pristine image and re-runs cloud-init.
persist_disk: true,
};
let ctrl = KvmController::new();
let handle = ctrl
.start(&spec)
.with_context(|| format!("boot `{name}`"))?;
Ok((ctrl, handle))
}
/// Run the whole korp suite: boot A (install → unattended provisioning → client), then
/// boot B (fresh boot) off the SAME disk.
#[cfg(feature = "matrix")]
pub fn run_korp_suite(cfg: &KorpSuiteConfig) -> Result<KorpRun> {
let mut run = KorpRun::default();
// ── the gates, BEFORE anything is copied or booted ───────────────────────
// Memory first (the OOM lesson), then LOAD — because 469 GB free says nothing
// about whether the box can actually schedule the guest, and the second gate is
// the one that has already been seen to matter here.
match memory_gate(host_available_mb(), cfg.memory_mb, FREE_MEMORY_FLOOR_MB) {
Ok(detail) => run.notes.push(format!("memory gate OK — {detail}")),
Err(detail) => anyhow::bail!("memory gate REFUSED — {detail}"),
}
let cores = std::thread::available_parallelism()
.map(|n| n.get())
.unwrap_or(0);
match load_gate(host_load1(), cores, cfg.load_ceiling) {
Ok(detail) => run.notes.push(format!("load gate OK — {detail}")),
Err(detail) => anyhow::bail!("load gate REFUSED — {detail}"),
}
std::fs::create_dir_all(&cfg.work_root)?;
let disk = cfg.work_root.join("korp-suite-disk.qcow2");
let t_copy = Instant::now();
std::fs::copy(&cfg.base_image, &disk)
.with_context(|| format!("copy base image {}", cfg.base_image.display()))?;
// korp refuses to convert with less than 12 GB free, and it is right to. Grow the
// per-run copy so the acceptance case is declined for korp's reasons or none.
let resize = std::process::Command::new("qemu-img")
.args([
"resize",
&disk.to_string_lossy(),
&format!("{}G", cfg.disk_gib),
])
.output()
.context("qemu-img resize (is qemu-utils installed?)")?;
if !resize.status.success() {
anyhow::bail!(
"qemu-img resize to {} GiB failed: {}",
cfg.disk_gib,
String::from_utf8_lossy(&resize.stderr)
);
}
run.notes.push(format!(
"per-run disk: copied in {} ms, grown to {} GiB (korp's headroom guard wants 12 GB free)",
t_copy.elapsed().as_millis(),
cfg.disk_gib,
));
// ── boot A ───────────────────────────────────────────────────────────────
let cases_a = plan_boot_a(cfg.cut_timeout.as_millis() as u64);
let plan_a = render_korp_plan(&cases_a);
let payload = cfg.work_root.join("korpdata.img");
build_payload(&payload, cfg, &plan_a)?;
run.notes.push(format!(
"payload image {} MiB (korp-server + korp-probe + the {} KiB map source + {} poly/ies)",
std::fs::metadata(&payload)
.map(|m| m.len() / 1_048_576)
.unwrap_or(0),
std::fs::metadata(&cfg.map_source)
.map(|m| m.len() / 1024)
.unwrap_or(0),
cfg.polys.len(),
));
let report_a = cfg.work_root.join("korp-report-a.jsonl");
let poly_names: Vec<String> = cfg
.polys
.iter()
.map(|(id, _)| format!("{id}.poly"))
.collect();
let user_data = render_user_data("/var/lib/skid/korp-plan.txt", &poly_names);
let t0 = Instant::now();
let (ctrl_a, h_a) = boot_once(
cfg,
"korp-provsuite-a",
&disk,
&payload,
user_data.clone(),
&report_a,
)?;
let want_a: Vec<String> = cases_a
.iter()
.filter(|c| !c.checks.is_empty())
.map(|c| c.id.clone())
.collect();
let (facts_a, timings_a, ended_a) = tail_report(&report_a, &want_a, t0, cfg.boot_timeout);
if !ended_a {
run.notes.push(
"boot A never reported `scenario_end` — the arming case that stages boot B may not \
have run"
.into(),
);
}
// A guest that died before the scenario ran produces an EMPTY report, which looks
// exactly like a suite whose cases all failed — and says nothing about why. The
// console is the only place the answer exists, and tunnr keeps it in memory, so it
// must be asked for BEFORE the machine is stopped.
if facts_a.is_empty() {
match ctrl_a.console("korp-provsuite-a") {
Some(log) if !log.trim().is_empty() => {
let tail: Vec<&str> = log.lines().rev().take(60).collect();
run.notes.push(format!(
"boot A produced NO facts at all. Guest console, last {} line(s), newest \
first:\n{}",
tail.len(),
tail.iter()
.map(|l| format!(" | {l}"))
.collect::<Vec<_>>()
.join("\n"),
));
}
_ => run.notes.push(
"boot A produced no facts AND no console output — qemu did not get far enough to \
print anything (check /dev/kvm, the base image and the payload disk)"
.into(),
),
}
}
std::thread::sleep(cfg.shutdown_grace);
run.boot_a_ms = t0.elapsed().as_millis();
let _ = ctrl_a.stop(&h_a);
run.outcomes.extend(grade(&cases_a, &facts_a, &timings_a));
run.case_timings.extend(timings_a);
run.facts.extend(facts_a);
// ── boot B: the SAME disk, booted again ──────────────────────────────────
let cases_b = plan_boot_b();
let report_b = cfg.work_root.join("korp-report-b.jsonl");
let t1 = Instant::now();
let (ctrl_b, h_b) = boot_once(
cfg,
"korp-provsuite-b",
&disk,
&payload,
user_data,
&report_b,
)?;
let want_b: Vec<String> = cases_b
.iter()
.filter(|c| !c.checks.is_empty())
.map(|c| c.id.clone())
.collect();
let (facts_b, timings_b, _) = tail_report(&report_b, &want_b, t1, cfg.boot_timeout);
run.boot_b_ms = t1.elapsed().as_millis();
let _ = ctrl_b.stop(&h_b);
run.boot_b_cases = facts_b.iter().map(|f| f.case.clone()).collect();
run.outcomes.extend(grade(&cases_b, &facts_b, &timings_b));
run.case_timings.extend(timings_b);
run.facts.extend(facts_b);
Ok(run)
}
/// Every case of the suite (both boots), for reporting and for the arms declaration.
pub fn all_cases(cut_timeout_ms: u64) -> Vec<KorpCase> {
let mut v = plan_boot_a(cut_timeout_ms);
v.extend(plan_boot_b());
v
}
#[cfg(test)]
mod tests {
use super::*;
fn f_with(captures: &[(&str, &str)]) -> Facts {
Facts {
case: "t".into(),
captures: captures.iter().map(|(k, v)| format!("{k}={v}")).collect(),
..Default::default()
}
}
// ─────────────────── the extract judge, seen RED four ways ─────────────────
/// The silent form: the file exists, is plausibly sized, and was never closed.
#[test]
fn a_footerless_parquet_is_red_however_big_it_is() {
let f = f_with(&[
("footer_vaduz_nodes", "PAR1"),
("size_vaduz_nodes", "420828"),
// 250 MB of a killed cut: no footer.
("footer_vaduz_ways", "\u{0}\u{0}\u{0}\u{0}"),
("size_vaduz_ways", "251912000"),
]);
let v = judge_korp(
&KorpCheck::ExtractComplete {
region: "vaduz".into(),
},
&f,
);
assert!(!v.is_pass(), "must be red");
assert!(
v.detail().contains("PAR1"),
"names what it looked for: {}",
v.detail()
);
assert!(
v.detail().contains("ways"),
"names the half: {}",
v.detail()
);
}
#[test]
fn one_half_present_and_one_absent_is_a_partial_extract_not_a_pass() {
let f = f_with(&[
("footer_vaduz_nodes", "PAR1"),
("size_vaduz_nodes", "420828"),
("footer_vaduz_ways", ""),
("size_vaduz_ways", ""),
]);
let v = judge_korp(
&KorpCheck::ExtractComplete {
region: "vaduz".into(),
},
&f,
);
assert!(
!v.is_pass(),
"half an extract is not an extract: {}",
v.detail()
);
// …and the SAME facts are red for "nothing half-written" too, because one half
// IS half-written.
assert!(
!judge_korp(
&KorpCheck::NoHalfWrittenExtract {
region: "vaduz".into()
},
&f
)
.is_pass()
);
}
#[test]
fn a_stub_sized_parquet_is_red_even_with_a_footer() {
let f = f_with(&[
("footer_vaduz_nodes", "PAR1"),
("size_vaduz_nodes", "8"),
("footer_vaduz_ways", "PAR1"),
("size_vaduz_ways", "551912"),
]);
let v = judge_korp(
&KorpCheck::ExtractComplete {
region: "vaduz".into(),
},
&f,
);
assert!(!v.is_pass(), "an 8-byte parquet is a stub: {}", v.detail());
}
/// Absence of evidence must not read as evidence of correctness.
#[test]
fn an_uncaptured_extract_is_a_fail_in_both_directions() {
let empty = Facts::default();
assert!(
!judge_korp(
&KorpCheck::ExtractComplete {
region: "vaduz".into()
},
&empty
)
.is_pass()
);
assert!(
!judge_korp(
&KorpCheck::NoHalfWrittenExtract {
region: "vaduz".into()
},
&empty
)
.is_pass(),
"even the negative form — never asked is not the same as nothing there"
);
}
/// …and the judge is not stuck on red: complete halves pass, and an interrupted cut
/// that left NOTHING passes the negative form.
#[test]
fn the_extract_judge_passes_in_both_of_its_directions() {
let good = f_with(&[
("footer_vaduz_nodes", "PAR1"),
("size_vaduz_nodes", "420828"),
("footer_vaduz_ways", "PAR1\n"),
("size_vaduz_ways", "551912"),
]);
assert!(
judge_korp(
&KorpCheck::ExtractComplete {
region: "vaduz".into()
},
&good
)
.is_pass()
);
let nothing = f_with(&[
("footer_vaduz_nodes", ""),
("size_vaduz_nodes", ""),
("footer_vaduz_ways", ""),
("size_vaduz_ways", ""),
]);
assert!(
judge_korp(
&KorpCheck::NoHalfWrittenExtract {
region: "vaduz".into()
},
¬hing
)
.is_pass(),
"a cancelled cut that deleted what it opened is the CORRECT outcome"
);
assert!(
!judge_korp(
&KorpCheck::ExtractComplete {
region: "vaduz".into()
},
¬hing
)
.is_pass(),
"…and the same facts must still be red for 'the cut produced an extract'"
);
}
// ─────────────────────── the tile judge, seen RED ──────────────────────────
#[test]
fn the_tiler_count_is_parsed_from_its_own_sentence_and_only_from_that() {
let j = "korp-server: 'vaduz' cut — tiling it …\n\
korp/tile-maps: uploading 41 tile objects to s3://nordisk-maps/vaduz/tiles/ …\n\
korp/tile-maps: 'vaduz' tiles built + uploaded\n";
assert_eq!(tiler_claimed_uploads(j), Some(41));
assert_eq!(tiler_claimed_uploads("nothing of the sort\n"), None);
// The LAST claim wins — a second tiling in the same journal must not be read
// through the first one's number.
let two = format!("{j}korp/tile-maps: uploading 7 tile objects to s3://x/y/tiles/ …\n");
assert_eq!(tiler_claimed_uploads(&two), Some(7));
}
#[test]
fn per_zoom_counts_come_from_the_key_layout_not_from_a_guess() {
let listing = "/home/korp/.korp/rustfs/nordisk-maps/vaduz/tiles/z12/2168/1424.parquet\n\
/home/korp/.korp/rustfs/nordisk-maps/vaduz/tiles/z12/2168/1425.parquet\n\
/home/korp/.korp/rustfs/nordisk-maps/vaduz/tiles/z9/271/178.parquet\n";
assert_eq!(
per_zoom(listing),
vec![("z12".to_string(), 2), ("z9".to_string(), 1)]
);
assert!(per_zoom("").is_empty());
// A path with no `tiles/` segment contributes nothing rather than a wrong zoom.
assert!(per_zoom("/somewhere/else/z12/1/2.parquet\n").is_empty());
}
#[test]
fn a_store_that_took_fewer_objects_than_the_tiler_claimed_is_red() {
let f = f_with(&[
(
TILER_JOURNAL_KEY,
"korp/tile-maps: uploading 41 tile objects to s3://nordisk-maps/vaduz/tiles/ …\n",
),
(
"store_vaduz",
"/x/nordisk-maps/vaduz/tiles/z12/1/1.parquet\n/x/nordisk-maps/vaduz/tiles/z12/1/2.parquet\n",
),
]);
let v = judge_korp(
&KorpCheck::TilesMatchTiler {
region: "vaduz".into(),
},
&f,
);
assert!(!v.is_pass());
assert!(
v.detail().contains("41") && v.detail().contains('2'),
"both numbers: {}",
v.detail()
);
}
/// The shape a tile check most easily passes on: nothing said, nothing found.
#[test]
fn an_empty_store_and_a_silent_journal_do_not_agree() {
let f = f_with(&[
(TILER_JOURNAL_KEY, "korp-server: started\n"),
("store_schaan", ""),
]);
let v = judge_korp(
&KorpCheck::TilesMatchTiler {
region: "schaan".into(),
},
&f,
);
assert!(!v.is_pass(), "0 == 0 must NOT be a match: {}", v.detail());
assert!(
!judge_korp(
&KorpCheck::TilesMatchTiler {
region: "schaan".into()
},
&Facts::default()
)
.is_pass()
);
}
#[test]
fn the_tile_judge_passes_when_the_two_counts_really_agree() {
let listing: String = (0..41)
.map(|i| format!("/x/nordisk-maps/vaduz/tiles/z12/2168/{i}.parquet\n"))
.collect();
let f = f_with(&[
(
TILER_JOURNAL_KEY,
"korp/tile-maps: uploading 41 tile objects to s3://nordisk-maps/vaduz/tiles/ …\n",
),
("store_vaduz", &listing),
]);
let v = judge_korp(
&KorpCheck::TilesMatchTiler {
region: "vaduz".into(),
},
&f,
);
assert!(v.is_pass(), "{}", v.detail());
assert!(
v.detail().contains("z12:41"),
"carries the breakdown: {}",
v.detail()
);
}
// ───────────────────── the client judge, seen RED ──────────────────────────
/// **The 2026-08-03 defect, as a unit test.** The store is full; the server answers
/// zero with an empty error.
#[test]
fn the_roster_cache_defect_is_red_and_the_message_says_what_the_client_got() {
let f = f_with(&[(
"probe_vaduz",
r#"{"endpoint":"http://127.0.0.1:50051","ok":true,"count":0,"assets":[],"error":"","dial_error":""}"#,
)]);
let v = judge_korp(
&KorpCheck::ClientSeesRegion {
region: "vaduz".into(),
},
&f,
);
assert!(!v.is_pass(), "must be red");
assert!(
v.detail().contains("0 region asset"),
"names the count: {}",
v.detail()
);
assert!(
v.detail().contains("cannot see its own output"),
"{}",
v.detail()
);
}
/// A server that is up and serving something ELSE must not pass for the region
/// asked about — the identity-value trap in roster form.
#[test]
fn a_non_empty_roster_without_the_region_is_still_red() {
let f = f_with(&[(
"probe_vaduz",
r#"{"ok":true,"count":1,"assets":[{"name":"sweden","node_count":9}],"error":"","dial_error":""}"#,
)]);
let v = judge_korp(
&KorpCheck::ClientSeesRegion {
region: "vaduz".into(),
},
&f,
);
assert!(!v.is_pass(), "{}", v.detail());
assert!(
v.detail().contains("sweden"),
"names what it DID see: {}",
v.detail()
);
}
#[test]
fn a_client_that_could_not_dial_is_red_and_says_so_distinctly() {
let f = f_with(&[(
"probe_vaduz",
r#"{"endpoint":"http://127.0.0.1:50051","ok":false,"count":0,"assets":[],"error":"","dial_error":"transport error"}"#,
)]);
let v = judge_korp(
&KorpCheck::ClientSeesRegion {
region: "vaduz".into(),
},
&f,
);
assert!(!v.is_pass());
assert!(
v.detail().contains("could not reach"),
"a dial failure is not an empty roster: {}",
v.detail()
);
}
/// An empty capture is the trap: `korp-probe` prints a line on EVERY path, so
/// nothing at all means it never ran — which must not read as "serves nothing".
#[test]
fn an_empty_or_absent_probe_capture_is_red_not_an_empty_roster() {
assert!(
!judge_korp(
&KorpCheck::ClientSeesRegion {
region: "vaduz".into()
},
&Facts::default()
)
.is_pass()
);
let f = f_with(&[("probe_vaduz", "")]);
let v = judge_korp(
&KorpCheck::ClientSeesRegion {
region: "vaduz".into(),
},
&f,
);
assert!(!v.is_pass());
assert!(v.detail().contains("did not run"), "{}", v.detail());
}
#[test]
fn the_client_judge_passes_when_the_region_really_is_in_the_roster() {
let f = f_with(&[(
"probe_vaduz",
r#"{"ok":true,"count":1,"assets":[{"name":"vaduz","node_count":38594}],"error":"","dial_error":""}"#,
)]);
let v = judge_korp(
&KorpCheck::ClientSeesRegion {
region: "vaduz".into(),
},
&f,
);
assert!(v.is_pass(), "{}", v.detail());
assert!(
v.detail().contains("38594"),
"carries the applied number: {}",
v.detail()
);
}
// ────────────────── restart / disk / latency judges, seen RED ──────────────
#[test]
fn a_changed_mainpid_is_the_manual_restart_the_suite_forbids() {
let mut f = f_with(&[("mainpid_install", "812")]);
f.show = vec!["MainPID=1907".into()];
let v = judge_korp(
&KorpCheck::NoRestartSince {
pid_key: "mainpid_install".into(),
},
&f,
);
assert!(!v.is_pass());
assert!(
v.detail().contains("812") && v.detail().contains("1907"),
"{}",
v.detail()
);
// …and it passes when it really is the same process.
let mut same = f_with(&[("mainpid_install", "812")]);
same.show = vec!["MainPID=812".into()];
assert!(
judge_korp(
&KorpCheck::NoRestartSince {
pid_key: "mainpid_install".into()
},
&same
)
.is_pass()
);
}
/// A comparison point of 0 is no comparison point at all.
#[test]
fn a_zero_baseline_pid_cannot_prove_anything() {
let mut f = f_with(&[("mainpid_install", "0")]);
f.show = vec!["MainPID=1907".into()];
assert!(
!judge_korp(
&KorpCheck::NoRestartSince {
pid_key: "mainpid_install".into()
},
&f
)
.is_pass()
);
assert!(
!judge_korp(
&KorpCheck::NoRestartSince {
pid_key: "nope".into()
},
&f
)
.is_pass()
);
}
#[test]
fn a_disk_refusal_with_no_numbers_is_not_a_refusal() {
let numberless = f_with(&[(JOURNAL_KEY, "korp-server: region provisioning REFUSED\n")]);
assert!(!judge_korp(&KorpCheck::ProvisionRefusedForDisk, &numberless).is_pass());
let silent = f_with(&[(JOURNAL_KEY, "korp-server: started, serving :50051\n")]);
let v = judge_korp(&KorpCheck::ProvisionRefusedForDisk, &silent);
assert!(!v.is_pass());
assert!(v.detail().contains("NO disk refusal"), "{}", v.detail());
assert!(!judge_korp(&KorpCheck::ProvisionRefusedForDisk, &Facts::default()).is_pass());
let real = f_with(&[(
JOURNAL_KEY,
"korp-server: region provisioning REFUSED — free 1024 B < 12884901888 B needed \
(no-headroom)\n",
)]);
assert!(judge_korp(&KorpCheck::ProvisionRefusedForDisk, &real).is_pass());
}
#[test]
fn a_timed_out_wait_is_red_and_a_slow_one_is_too() {
let mut f = Facts {
exits: vec![
"1 WAITFOR 120000ms/120000ms /usr/local/bin/korp-probe --expect vaduz :: TIMED OUT ::"
.into(),
],
..Default::default()
};
let v = judge_korp(
&KorpCheck::SettledWithin {
needle: "korp-probe".into(),
ms: 120_000,
},
&f,
);
assert!(!v.is_pass());
assert!(v.detail().contains("TIMED OUT"), "{}", v.detail());
f.exits =
vec!["0 WAITFOR 90000ms/120000ms /usr/local/bin/korp-probe --expect vaduz".into()];
assert!(
!judge_korp(
&KorpCheck::SettledWithin {
needle: "korp-probe".into(),
ms: 30_000
},
&f
)
.is_pass()
);
assert!(
judge_korp(
&KorpCheck::SettledWithin {
needle: "korp-probe".into(),
ms: 120_000
},
&f
)
.is_pass()
);
// A wait that never happened is not a fast one.
assert!(
!judge_korp(
&KorpCheck::SettledWithin {
needle: "nothing".into(),
ms: 1
},
&f
)
.is_pass()
);
}
#[test]
fn unit_enabled_is_red_for_every_way_it_can_be_wrong() {
let mut f = Facts::default();
assert!(
!judge_korp(&KorpCheck::UnitEnabled, &f).is_pass(),
"not loaded"
);
f.show = vec!["UnitFileState=disabled".into()];
let v = judge_korp(&KorpCheck::UnitEnabled, &f);
assert!(!v.is_pass());
assert!(
v.detail().contains("next boot"),
"says why it matters: {}",
v.detail()
);
f.show = vec!["UnitFileState=enabled".into()];
assert!(judge_korp(&KorpCheck::UnitEnabled, &f).is_pass());
}
// ───────────────────────── the memory gate, both ways ──────────────────────
#[test]
fn the_memory_gate_refuses_a_box_that_cannot_spare_the_guest() {
// The 2026-08-03 shape: a box with almost nothing left.
let v = memory_gate(Some(4_000), 4_096, FREE_MEMORY_FLOOR_MB);
assert!(v.is_err(), "must refuse");
assert!(v.unwrap_err().contains("4000"), "names what it measured");
// A gate that cannot measure must not grant.
assert!(memory_gate(None, 4_096, FREE_MEMORY_FLOOR_MB).is_err());
// …and it is not stuck on refuse.
assert!(memory_gate(Some(400_000), 4_096, FREE_MEMORY_FLOOR_MB).is_ok());
// Exactly at the boundary is enough; one MiB below is not.
assert!(
memory_gate(
Some(4_096 + FREE_MEMORY_FLOOR_MB),
4_096,
FREE_MEMORY_FLOOR_MB
)
.is_ok()
);
assert!(
memory_gate(
Some(4_096 + FREE_MEMORY_FLOOR_MB - 1),
4_096,
FREE_MEMORY_FLOOR_MB
)
.is_err()
);
}
/// The gate the memory gate could not give: 469 GB free and the guest still
/// starved. Both observed loads are asserted against the ceiling, so a later edit
/// that moved it away from the measurement is caught here.
#[test]
fn the_load_gate_refuses_the_box_that_actually_starved_a_guest() {
// MEASURED, oden 2026-08-03, 32 cores: 221 → boot A produced zero facts in
// 420 s; ~40 → the install case reported in ~60 s.
let starved = load_gate(Some(221.0), 32, LOAD_PER_CORE_CEILING);
assert!(
starved.is_err(),
"the load that really starved a guest must be refused"
);
let msg = starved.unwrap_err();
assert!(
msg.contains("221") && msg.contains("6.9"),
"names both numbers: {msg}"
);
assert!(
load_gate(Some(40.0), 32, LOAD_PER_CORE_CEILING).is_ok(),
"…and the load that really worked must be allowed — a gate stuck on refuse is as \
useless as one stuck on grant"
);
// A gate that cannot measure must not grant.
assert!(load_gate(None, 32, LOAD_PER_CORE_CEILING).is_err());
assert!(
load_gate(Some(1.0), 0, LOAD_PER_CORE_CEILING).is_err(),
"zero cores is unknowable"
);
// The boundary, both sides.
assert!(
load_gate(Some(96.0), 32, 3.0).is_ok(),
"exactly at the ceiling is allowed"
);
assert!(load_gate(Some(96.1), 32, 3.0).is_err());
}
// ───────────────────────── the suite's own shape ───────────────────────────
/// A case that reports nothing must fail every one of its expectations — including
/// the deliberate reds. Silence is the easiest way for a VM suite to pass.
#[test]
fn a_case_with_no_facts_line_fails_every_check_including_the_negative_ones() {
let cases = plan_boot_a(60_000);
let out = grade(&cases, &[], &[]);
assert!(!out.is_empty(), "there are expectations to fail");
assert!(out.iter().all(|o| !o.as_declared), "every one is a failure");
assert!(
out.iter().any(|o| !o.expect_pass),
"including at least one declared RED"
);
}
/// A judge stuck on Pass fails the negative cases; a judge stuck on Fail fails the
/// positive ones. Neither hollow direction survives — asserted structurally, so a
/// later edit that removed every red is caught here.
#[test]
fn both_boots_carry_deliberate_reds_and_deliberate_passes() {
for (label, cases) in [("boot A", plan_boot_a(60_000)), ("boot B", plan_boot_b())] {
let reds: usize = cases
.iter()
.flat_map(|c| &c.checks)
.filter(|e| !e.expect_pass)
.count();
let passes: usize = cases
.iter()
.flat_map(|c| &c.checks)
.filter(|e| e.expect_pass)
.count();
assert!(passes > 0, "{label} declares no positive expectation");
if label == "boot A" {
assert!(reds >= 3, "{label} declares only {reds} engineered red(s)");
}
}
}
#[test]
fn every_asserting_case_declares_its_host_only_blind_spot() {
for c in all_cases(60_000) {
if c.checks.is_empty() {
continue;
}
assert!(
c.host_only_blind_spot.len() > 40,
"case `{}` asserts {} thing(s) but does not say what a host-only test would \
miss — the claim that this needs a VM must be auditable per case",
c.id,
c.checks.len()
);
}
}
/// The acceptance case must contain the ONE install command and no other korp verb.
/// If a later edit slipped a `systemctl restart` or a `--tile-maps` in, the suite
/// would still be green while proving something weaker than the acceptance.
#[test]
fn the_acceptance_path_runs_exactly_one_korp_command_and_it_is_the_install() {
let cases = plan_boot_a(60_000);
let acceptance: Vec<&KorpCase> = cases
.iter()
.filter(|c| c.id == "install-and-nothing-else" || c.id == "self-provisions-unattended")
.collect();
assert_eq!(
acceptance.len(),
2,
"both halves of the acceptance are present"
);
let mut korp_cmds: Vec<String> = Vec::new();
for c in &acceptance {
for s in &c.steps {
let argv = match s {
PlanStep::Run(a) => a.clone(),
PlanStep::WaitFor { argv, .. } => argv.clone(),
PlanStep::Capture { argv, .. } => argv.clone(),
_ => continue,
};
if argv[0].contains("korp-server") {
korp_cmds.push(argv.join(" "));
}
assert!(
!argv.iter().any(|a| a == "restart"),
"the acceptance path must contain no restart: {argv:?}"
);
}
}
assert_eq!(
korp_cmds,
vec![format!("{KORP_SERVER} --install --system")],
"exactly one korp-server invocation, and it is the install"
);
}
/// The install case must not be graded on evidence the provisioning case gathered:
/// each case's captures have to be taken in that case, because the guest resets
/// them at every COLLECT.
#[test]
fn every_korp_check_has_its_evidence_captured_in_its_own_case() {
for c in all_cases(60_000) {
let keys: Vec<String> = c
.steps
.iter()
.filter_map(|s| match s {
PlanStep::Capture { key, .. } => Some(key.clone()),
_ => None,
})
.collect();
for e in &c.checks {
let need: Option<String> = match &e.check {
AnyCheck::Korp(KorpCheck::ClientSeesRegion { region }) => {
Some(probe_key(region))
}
AnyCheck::Korp(KorpCheck::TilesMatchTiler { region }) => {
Some(store_key(region))
}
AnyCheck::Korp(KorpCheck::NoRestartSince { pid_key }) => Some(pid_key.clone()),
AnyCheck::Korp(KorpCheck::ProvisionRefusedForDisk) => Some(JOURNAL_KEY.into()),
AnyCheck::Korp(KorpCheck::ExtractComplete { region })
| AnyCheck::Korp(KorpCheck::NoHalfWrittenExtract { region }) => {
Some(footer_key(region, "nodes"))
}
_ => None,
};
if let Some(k) = need {
assert!(
keys.contains(&k),
"case `{}` asserts `{}` but never captures `{k}` — it would be graded \
on evidence it did not gather",
c.id,
e.name
);
}
}
}
}
/// The config the guest gets must name the regions the plan asserts on, and must
/// parse — a typo here would make the provisioner find nothing and the suite would
/// blame the server.
#[test]
fn the_region_config_is_valid_json_naming_exactly_the_regions_the_plan_uses() {
let v: serde_json::Value = serde_json::from_str(®ion_config_json()).expect("valid JSON");
let regions = v["regions"].as_array().expect("a regions array");
let names: Vec<&str> = regions.iter().filter_map(|r| r["id"].as_str()).collect();
assert_eq!(names, vec![REGION, REGION_2]);
assert_eq!(regions[0]["enabled"], serde_json::Value::Bool(true));
assert_eq!(
regions[1]["enabled"],
serde_json::Value::Bool(false),
"the control region is OFF"
);
for r in regions {
assert!(
r["poly"].as_str().unwrap().starts_with(POLY_DIR),
"every polygon is staged where cloud-init puts it"
);
}
}
/// The disk case re-enables the control region; if that string surgery ever stopped
/// matching, the case would restart the service with nothing to refuse ABOUT and
/// pass for the wrong reason.
#[test]
fn the_disk_case_really_re_enables_the_control_region() {
let case = plan_boot_a(60_000)
.into_iter()
.find(|c| c.id == "disk-margin-refuses-before-it-starts")
.expect("the disk case exists");
let written = case
.steps
.iter()
.find_map(|s| match s {
PlanStep::Write { path, contents, .. } if path.ends_with("config.json") => {
Some(contents.clone())
}
_ => None,
})
.expect("the case rewrites the config");
let v: serde_json::Value = serde_json::from_str(&written).expect("still valid JSON");
let r2 = v["regions"]
.as_array()
.unwrap()
.iter()
.find(|r| r["id"] == REGION_2)
.unwrap();
assert_eq!(
r2["enabled"],
serde_json::Value::Bool(true),
"the control region must be ENABLED, or the refusal has nothing to refuse"
);
}
#[test]
fn the_cloud_init_stages_the_map_where_korp_probes_for_it_with_no_config() {
let ud = render_user_data("/var/lib/skid/korp-plan.txt", &["vaduz.poly".into()]);
assert!(ud.starts_with("#cloud-config\n"));
assert!(
ud.contains(PLANET_DROP),
"the map lands on korp's own probe path: {ud}"
);
assert!(
ud.contains("growpart"),
"the root fs is grown, or korp's 12 GB floor refuses"
);
assert!(ud.contains("vaduz.poly"), "the polygon is staged");
assert!(
!ud.contains("KORP_PLANET") && !ud.contains("--install"),
"no env override and no korp verb in cloud-init — the install is a graded plan step"
);
}
/// The plan wire must be exactly what the guest parser accepts — the korp cases
/// ride the SAME renderer, so this pins that they survive it.
#[test]
fn the_korp_plan_renders_the_new_verbs_the_guest_understands() {
let text = render_korp_plan(&plan_boot_a(60_000));
assert!(text.contains("RUN /usr/local/bin/korp-server --install --system\n"));
assert!(
text.contains("WAITFOR "),
"the unattended waits ride the wire"
);
assert!(text.contains("CAPTURE footer_vaduz_nodes tail -c 4 "));
assert!(text.contains("COLLECT korp-server.service /usr/local/bin/korp-server\n"));
// Every CASE is terminated; an unterminated one makes the guest abort the
// whole plan and report nothing.
assert_eq!(
text.matches("\nCASE ").count() + usize::from(text.starts_with("CASE ")),
text.matches("\nEND\n").count(),
"every CASE has its END"
);
}
}