crates/gunnar-growbench/src/twin_drill.rs
//! **T1 — the twin drill** (`.nornir/twin-transaction-log.md` §drill; PLAN-3VM L4 (e), "the only
//! proof that counts").
//!
//! Two guests under KVM at once: a PRIMARY from the keyless medium (4 small members) and a TWIN
//! from the primary's own medium, byte for byte, with no role on its command line (ONE data volume). The primary takes repositories, an LFS
//! object, users; the fresh twin is joined through `Bootstrap.BecomeTwin` and its key enrolled
//! on the primary in group `twin`; two waves of writes are followed to the watermark
//! (`applied == head`). Then the primary's disks are DESTROYED, a fresh primary is installed on
//! blank disks and answers where the old one did, `gunnar control twin replay` streams the
//! twin's log into its `Bootstrap.Restore`, the box restarts, and every repository is cloned and
//! `git fsck`'d, the LFS object compared, the users compared, and the twin seen following the
//! restored primary. Last, the twin's data volume is read from the HOST with `xfs_db` and every
//! file on it is held to the cache rule — nothing that is cache may exist on the twin.
use std::collections::BTreeSet;
use std::os::unix::fs::FileExt;
use std::os::unix::io::AsRawFd;
use std::path::{Path, PathBuf};
use std::process::Command;
use std::sync::mpsc::{self, Receiver, Sender};
use std::time::{Duration, Instant};
use super::{boot_failed, print_boot, seed_repos, Opts, Session, Status, Verdict};
use crate::data::{self, RepoManifest};
use crate::rig::{Ports, Rig, RigSpec};
const WAIT: Duration = Duration::from_secs(900);
enum ToTwin {
PrimaryAt(Ports),
Sync(u64),
/// The restored primary is up and has written a transaction ONLY it can have: this head.
RestoreDone(u64),
Stop,
}
enum FromTwin {
Joined { key: String, ports: Ports },
Synced(String),
Final(String),
Failed(String),
}
/// The same predicate as `gunnar_core::txlog::is_cache_path` (cache only — a box's own `crash/`
/// and `log/` are records, not cache), spelled here because this bench is a detached crate
/// with no gunnar dependency. Held identical by eye and by the note's table.
pub fn is_cache_path(rel: &str) -> bool {
let rel = rel.trim_start_matches("./").trim_start_matches('/');
// With no /serialdat partition the installer lays that tree at `/data/.serialdat` and links
// `crash`, `log`, `gc`, `quarantine` … to it: judge a path by its place in that tree.
let rel = rel
.strip_prefix(".serialdat/")
.or_else(|| rel.strip_prefix("serialdat/"))
.unwrap_or(rel);
let first = rel.split('/').next().unwrap_or("");
let name = rel.rsplit('/').next().unwrap_or("");
matches!(
first,
".korp-disktest" | "gc" | "quarantine" | "meta" | "serialdat"
) || rel.split('/').any(|c| c.starts_with("objects.exploded"))
|| name == "objects.graph"
|| name.ends_with(".tmp")
|| name.starts_with("incoming-")
|| name == "grow-journal"
}
pub(crate) fn field(out: &str, key: &str) -> Option<String> {
out.lines()
.find(|l| l.split_whitespace().next() == Some(key))
.and_then(|l| l.split_whitespace().nth(1))
.map(str::to_string)
}
pub(crate) fn twin_status(cli: &Path, key: &Path, at: &str) -> Result<String, String> {
data::cli(
cli,
&[
"control",
"twin",
"status",
"--at",
at,
"-i",
&key.display().to_string(),
],
)
}
pub(crate) fn head_of(cli: &Path, key: &Path, at: &str) -> Result<u64, String> {
let out = twin_status(cli, key, at)?;
field(&out, "head_seq")
.and_then(|v| v.parse().ok())
.ok_or_else(|| format!("no head_seq in:\n{out}"))
}
pub(crate) fn one_line(s: &str) -> String {
s.lines()
.map(str::trim)
.filter(|l| !l.is_empty())
.collect::<Vec<_>>()
.join(" · ")
}
/// Poll the twin until `pred` holds on its `LogHead`; the admin key only works there once the
/// identity table has replicated, so early refusals are retried.
pub(crate) fn wait_twin(
cli: &Path,
key: &Path,
at: &str,
what: &str,
pred: impl Fn(&str) -> bool,
) -> Result<String, String> {
let t0 = Instant::now();
let mut last = String::new();
while t0.elapsed() < WAIT {
match twin_status(cli, key, at) {
Ok(out) if pred(&out) => {
return Ok(format!(
"{} in {:.0} s",
one_line(&out),
t0.elapsed().as_secs_f64()
))
}
Ok(out) => last = one_line(&out),
Err(e) => last = e,
}
std::thread::sleep(Duration::from_secs(2));
}
Err(format!(
"the twin never reached `{what}` within {} s; last: {last}",
WAIT.as_secs()
))
}
fn twin_drive(
cli: &Path,
key: &Path,
tp: Ports,
rx: &Receiver<ToTwin>,
tx: &Sender<FromTwin>,
) -> Result<String, String> {
let at = format!("127.0.0.1:{}", tp.grpc_private);
let p = match rx.recv_timeout(WAIT) {
Ok(ToTwin::PrimaryAt(p)) => p,
_ => return Err("the primary never came up".into()),
};
// The twin reaches the primary through its slirp gateway: the host's loopback forward.
let primary = format!("10.0.2.2:{}", p.grpc_private);
let key_line = data::cli(
cli,
&[
"control",
"twin",
"join",
"--at",
&at,
"--primary",
&primary,
"-i",
&key.display().to_string(),
],
)?
.trim()
.to_string();
if !key_line.starts_with("ssh-") {
return Err(format!("twin join answered no key: {key_line}"));
}
let _ = tx.send(FromTwin::Joined {
key: key_line,
ports: tp,
});
let mut said = Vec::new();
loop {
match rx.recv_timeout(WAIT * 3) {
Ok(ToTwin::Sync(head)) => {
let s = wait_twin(
cli,
key,
&at,
&format!("applied >= {head}, caught up"),
|o| {
field(o, "applied")
.and_then(|v| v.parse::<u64>().ok())
.is_some_and(|a| a >= head)
&& field(o, "caught_up").as_deref() == Some("true")
},
)?;
said.push(s.clone());
let _ = tx.send(FromTwin::Synced(s));
}
Ok(ToTwin::RestoreDone(head)) => {
// ★ **A poll of the RESTORED primary, not an old answer.** MEASURED 2026-09-21 (T11
// reading this drill's logs): T1e passed on `head 262`, the last thing the twin heard
// DURING the replay boot, while every poll after the restored box came up failed
// (broken pipe, then connection refused). `head` is a transaction written on the
// restored primary after it came up, so only a twin that really follows it can
// have applied it.
let s = wait_twin(cli, key, &at, &format!("following the restored primary, applied >= {head} (written after the restore)"), |o| {
field(o, "state").as_deref() == Some("following")
&& field(o, "caught_up").as_deref() == Some("true")
&& field(o, "applied").and_then(|v| v.parse::<u64>().ok()).is_some_and(|a| a >= head)
})?;
said.push(s.clone());
let _ = tx.send(FromTwin::Final(s));
}
Ok(ToTwin::Stop) => return Ok(said.join(" | ")),
Ok(ToTwin::PrimaryAt(_)) => return Err("the primary was announced twice".into()),
Err(_) => return Err("the drill went silent".into()),
}
}
}
fn recv(rx: &Receiver<FromTwin>) -> Result<FromTwin, String> {
match rx.recv_timeout(WAIT * 2) {
Ok(FromTwin::Failed(e)) => Err(format!("twin: {e}")),
Ok(m) => Ok(m),
Err(mpsc::RecvTimeoutError::Disconnected) => Err(
"the twin's boot ended before it answered (never ready, or its hold ran out)".into(),
),
Err(mpsc::RecvTimeoutError::Timeout) => Err("the twin went silent".into()),
}
}
/// A repository (in [`data::PUSH_CREATE_FORMAT`]) with one LFS-tracked object.
pub(crate) fn make_lfs_repo(trees: &Path, seed: u64) -> Result<(RepoManifest, String), String> {
let dir = trees.join("lfs-assets");
let _ = std::fs::remove_dir_all(&dir);
std::fs::create_dir_all(&dir).map_err(|e| e.to_string())?;
data::git(
&dir,
"ssh",
&["init", "-q", data::PUSH_CREATE_FORMAT, "-b", "main"],
)?;
data::git(&dir, "ssh", &["lfs", "install", "--local"])?;
data::git(&dir, "ssh", &["lfs", "track", "*.bin"])?;
// A plain first commit: git-lfs uploads in PRE-PUSH, before create-on-push has made the
// repository (measured: `ssh.exec.denied reason=no-such-repository` for git-lfs-transfer), so
// the drill creates the repository with this commit and pushes the LFS object after it.
data::git(&dir, "ssh", &["add", ".gitattributes"])?;
data::git(
&dir,
"ssh",
&[
"-c",
"user.name=growbench",
"-c",
"user.email=growbench@example.invalid",
"commit",
"-qm",
"create",
],
)?;
let mut buf = vec![0u8; 3 << 20];
data::Rng::new(seed).fill(&mut buf);
std::fs::write(dir.join("asset.bin"), &buf).map_err(|e| e.to_string())?;
let sha = data::sha256_hex(&buf);
data::git(&dir, "ssh", &["add", "asset.bin"])?;
data::git(
&dir,
"ssh",
&[
"-c",
"user.name=growbench",
"-c",
"user.email=growbench@example.invalid",
"commit",
"-qm",
"lfs asset",
],
)?;
let head = data::git(&dir, "ssh", &["rev-parse", "HEAD"])?
.trim()
.to_string();
let tree = data::git(&dir, "ssh", &["rev-parse", "HEAD^{tree}"])?
.trim()
.to_string();
Ok((
RepoManifest {
name: "lfs-assets".into(),
tree,
head,
files: Vec::new(),
bytes: buf.len() as u64,
dir,
},
sha,
))
}
pub(crate) fn verify_lfs(into: &Path, ssh: &str, url: &str, sha: &str) -> Result<String, String> {
let _ = std::fs::remove_dir_all(into);
std::fs::create_dir_all(into).map_err(|e| e.to_string())?;
data::clone_retry(into, ssh, url, "lfs-assets", Duration::from_secs(300))?;
let dir = into.join("lfs-assets");
data::git(&dir, ssh, &["fsck", "--strict", "--full", "--no-progress"])?;
data::git(&dir, ssh, &["lfs", "install", "--local"])?;
data::git(&dir, ssh, &["lfs", "pull"])?;
let got = data::sha256_file(&dir.join("asset.bin"))?;
if got != sha {
return Err(format!(
"asset.bin sha256 {}… ≠ pushed {}…",
&got[..12],
&sha[..12]
));
}
Ok("lfs-assets cloned+fsck-clean, asset.bin (3 MiB, LFS) sha256-identical".into())
}
pub(crate) fn users(cli: &Path, key: &Path, at: &str) -> Result<BTreeSet<String>, String> {
let out = data::cli(
cli,
&[
"control",
"user",
"list",
"-i",
&key.display().to_string(),
"--at",
at,
],
)?;
Ok(out
.lines()
.map(str::trim)
.filter(|l| !l.is_empty())
.map(str::to_string)
.collect())
}
pub(crate) fn pubkey(path: &Path) -> Result<String, String> {
std::fs::read_to_string(path.with_extension("pub"))
.map(|s| s.trim().to_string())
.map_err(|e| format!("{}: {e}", path.display()))
}
// ── what makes the box ITSELF, read over its own control plane ──────────────
//
// The drill used to compare the repositories, the LFS object and the user rows, and call that
// "the twin holds everything". It is not: a box is also its SSH host key, its groups, its
// grants, its namespaces, its repository rows, its signed membership log and its outbound
// configuration. Each of those is one CLASS below and one verdict above, so "the twin holds
// everything" is a row per class rather than a sentence.
/// One class of configuration, as the control plane answers it. `Err` is what the server said:
/// a build without the feature answers `Unimplemented`, which is an UNRUN row, not a red one.
type Class = (&'static str, Result<String, String>);
/// The server refused because this build does not contain the capability (a cargo feature the
/// medium was composed without) — never a failure of the twin.
fn unimplemented(e: &str) -> bool {
let e = e.to_ascii_lowercase();
// MEASURED 2026-09-17: the server says "Operation is not implemented or not supported: not
// implemented: outbound webhooks: this server was built without the `webhooks` feature".
// A predicate looking only for the one word "unimplemented" read that as a red row.
e.contains("unimplemented")
|| e.contains("not implemented")
|| e.contains("built without the")
|| e.contains("not compiled")
|| e.contains("no such subcommand")
}
/// **The SSH host key a client would see**, read from outside with `ssh-keyscan` — the same
/// bytes `known_hosts` pins. A restored box with a new one is a box every client refuses with a
/// changed-fingerprint warning, which is a failure no repository comparison can see.
pub(crate) fn host_key(port: u16) -> Result<String, String> {
let t0 = Instant::now();
let mut last = String::new();
while t0.elapsed() < Duration::from_secs(300) {
let out = Command::new("ssh-keyscan")
.env("LC_ALL", "C")
.args(["-T", "10", "-p", &port.to_string(), "127.0.0.1"])
.output()
.map_err(|e| format!("ssh-keyscan: {e}"))?;
let keys: Vec<String> = String::from_utf8_lossy(&out.stdout)
.lines()
.filter(|l| !l.trim_start().starts_with('#') && !l.trim().is_empty())
.map(|l| l.split_whitespace().skip(1).collect::<Vec<_>>().join(" "))
.collect();
if !keys.is_empty() {
let mut keys = keys;
keys.sort();
return Ok(keys.join(" | "));
}
last = String::from_utf8_lossy(&out.stderr).trim().to_string();
std::thread::sleep(Duration::from_secs(3));
}
Err(format!(
"ssh-keyscan read no host key on :{port} within 300 s: {last}"
))
}
/// The `SHA256:` fingerprint of a host-key line, for a verdict a human can compare by eye.
fn host_key_fingerprint(line: &str) -> String {
let dir = std::env::temp_dir().join(format!("growbench-hk-{}", std::process::id()));
let _ = std::fs::create_dir_all(&dir);
let p = dir.join("hostkey.pub");
if std::fs::write(&p, format!("{line}\n")).is_err() {
return "unfingerprinted".into();
}
let out = Command::new("ssh-keygen").args(["-lf"]).arg(&p).output();
let _ = std::fs::remove_file(&p);
match out {
Ok(o) => String::from_utf8_lossy(&o.stdout)
.split_whitespace()
.nth(1)
.unwrap_or("unfingerprinted")
.to_string(),
Err(_) => "unfingerprinted".into(),
}
}
/// **Every class of control-plane configuration this box holds**, in a fixed order, each read
/// through the verb an operator would use. Nothing here mints or creates: a reader that made
/// what it was looking for could not fail for its absence.
pub(crate) fn control_state(cli: &Path, key: &Path, at: &str, repos: &[String]) -> Vec<Class> {
let k = key.display().to_string();
let run = |args: &[&str]| -> Result<String, String> {
let mut v = vec!["control"];
v.extend_from_slice(args);
v.extend_from_slice(&["-i", &k, "--at", at]);
data::cli(cli, &v).map(|s| one_line(&s))
};
let mut out: Vec<Class> = vec![
("users", run(&["user", "list"])),
("keys", run(&["user", "keys"])),
("groups", run(&["group", "list"])),
("group members", run(&["group", "members", TWIN_REVIEWERS])),
("grants", run(&["grant", "list"])),
("namespaces", run(&["namespace", "list"])),
// M7, the signed append-only membership log: `group history` is the one appender a
// keyless medium can be made to write from the outside.
(
"the signed membership log",
run(&["group", "history", TWIN_REVIEWERS]),
),
("webhooks", run(&["webhook", "list"])),
(
"replication config",
run(&["replication", "list"]).map(|s| scrub_process_clock(&s)),
),
];
// One row per repository: hash kind, default branch and visibility, which live in the
// catalog document and nowhere else.
let rows: Result<String, String> = repos
.iter()
.map(|r| run(&["repo", "get", r]).map(|s| format!("{r}: {s}")))
.collect::<Result<Vec<_>, _>>()
.map(|v| v.join(" ; "));
out.push(("repository rows (hash, default branch, visibility)", rows));
out.push(("pull mirrors", run(&["mirror", "status", MIRROR_REPO])));
out
}
/// The group the drill makes, grants to and reads the signed history of.
pub(crate) const TWIN_REVIEWERS: &str = "reviewers";
/// The namespace the group is granted on — a second one, so "namespaces back" can fail.
pub(crate) const TWIN_NAMESPACE: &str = "audit";
/// The branch `bench/mixed` is moved to: NOT the one `repo create` hands out, so a
/// `default_branch` comparison against `main` could not fail.
pub(crate) const TWIN_DEFAULT_BRANCH: &str = "refs/heads/trunk";
/// The repository the pull mirror is registered on: `carol`'s, so the mirror row and the
/// repository rows are not the same document read twice.
pub(crate) const MIRROR_REPO: &str = "bench/after-join";
/// **One volatile number in an otherwise replicated answer.** `replication list` prints
/// "(this record is held in memory and begins at <ms>)" — the moment THIS PROCESS started
/// counting, which is not configuration and is not replicated. It is scrubbed so the row
/// compares the standby roster it is named after; nothing else is touched, and the identity
/// table's own timestamps (which ARE replicated) are still compared digit for digit.
fn scrub_process_clock(s: &str) -> String {
const MARK: &str = "begins at ";
let mut out = String::with_capacity(s.len());
let mut rest = s;
while let Some(i) = rest.find(MARK) {
out.push_str(&rest[..i + MARK.len()]);
rest = &rest[i + MARK.len()..];
let n = rest
.find(|c: char| !c.is_ascii_digit())
.unwrap_or(rest.len());
out.push_str("<when this process started>");
rest = &rest[n..];
}
out.push_str(rest);
out
}
// ── the audit trail (lane T11, 2026-09-21 — the D6 class) ──────────────────
//
// Every other class above is "does the restored box still hold X". This one is different in
// the way that matters: the trail GROWS while you look at it, because reading it is itself an
// audited decision. So it is never compared tip to tip. It is compared at a named sequence —
// the primary's last — and because each entry's hash commits to its `prev`, one matching hash
// at that sequence proves the whole prefix came back byte for byte.
/// The audit chain's shape, as `gunnar control audit` prints it.
#[derive(Debug, Clone)]
pub(crate) struct AuditShape {
/// Is there a durable trail at all? A box serving without one is not a box with an empty
/// one, and this drill must never confuse the two.
pub(crate) present: bool,
pub(crate) entries: u64,
pub(crate) tip: String,
pub(crate) tip_seq: u64,
/// Does the chain still hold, end to end, on the box that answered?
pub(crate) verified: bool,
/// Is every entry also a transaction — i.e. does the twin hold this trail? A `no` on a
/// PRIMARY is the D6 defect itself.
pub(crate) replicated: bool,
/// The hash at the sequence that was asked for, when one was.
pub(crate) hash_at_seq: String,
}
/// Read the trail's shape over the control plane.
///
/// Parsed from the printed lines rather than a second gRPC client: the bench drives the same
/// binary an operator drives, so a verb that prints something other than what it means is a
/// failure this drill sees.
pub(crate) fn audit_shape(
cli: &Path,
key: &Path,
at: &str,
at_seq: u64,
) -> Result<AuditShape, String> {
let k = key.display().to_string();
let seq = at_seq.to_string();
let mut args = vec!["control", "audit", "-i", &k, "--at", at];
if at_seq != 0 {
args.push("--at-seq");
args.push(&seq);
}
let out = data::cli(cli, &args)?;
parse_audit_shape(&out, at_seq)
}
/// The printed lines of `gunnar control audit`, read back. Separate from the call so the reading
/// is provable without a box.
pub(crate) fn parse_audit_shape(out: &str, at_seq: u64) -> Result<AuditShape, String> {
let field = |name: &str| -> String {
out.lines()
.find_map(|l| l.trim().strip_prefix(name).map(|v| v.trim().to_string()))
.unwrap_or_default()
};
if out.contains("audit trail: NONE") {
return Ok(AuditShape {
present: false,
entries: 0,
tip: String::new(),
tip_seq: 0,
verified: false,
replicated: false,
hash_at_seq: String::new(),
});
}
let entries: u64 = out
.lines()
.find_map(|l| l.trim().strip_prefix("audit trail:"))
.and_then(|v| {
v.trim()
.split_whitespace()
.next()
.and_then(|n| n.parse().ok())
})
.ok_or_else(|| format!("`control audit` printed no entry count:\n{out}"))?;
Ok(AuditShape {
present: true,
entries,
tip: field("tip:"),
tip_seq: field("tip seq:").parse().unwrap_or(0),
verified: field("verifies:").starts_with("yes"),
replicated: field("replicated:").starts_with("yes"),
// **`(not held)` is the CLI saying "no entry at that seq", not a hash.** MEASURED
// 2026-09-21 (T3, medium `ff2c7c9`): this parser took the words for a hash, so the row
// read "DIFFERS at seq 40 … the restored box holds: (not held)" — a claim about a hash
// that did not exist — instead of "holds no entry at seq 40 (it holds N)", which is the
// sentence that pointed at the real defect: the restored box had lost the trail.
hash_at_seq: {
let h = field(&format!("hash at seq {at_seq}:"));
if h.starts_with('(') {
String::new()
} else {
h
}
},
})
}
/// **Did the audit trail survive the restore?** — one verdict, in the drill's own idiom.
///
/// UNRUN, never a vacuous PASS, when the primary could not be read for a trail at all: a
/// medium from before `Admin.AuditTrail` existed, or a box serving without one. A row that
/// cannot go red is worse than no row.
fn audit_verdict(
id: &'static str,
before: &Result<AuditShape, String>,
after: &Result<AuditShape, String>,
) -> Verdict {
const NAME: &str = "the control-plane audit trail back: the primary's own decisions, chained, at the same hashes";
let unrun = |detail: String| Verdict {
id,
name: NAME,
status: Status::Unrun,
detail,
};
let fail = |detail: String| Verdict {
id,
name: NAME,
status: Status::Fail,
detail,
};
let b = match before {
Ok(b) => b,
Err(e) if unimplemented(e) => {
return unrun(format!(
"this medium's build answers no `control audit`: {}",
one_line(e)
))
}
Err(e) => {
return fail(format!(
"the PRIMARY's audit trail could not be read: {}",
one_line(e)
))
}
};
if !b.present {
return unrun("the PRIMARY was serving without a durable audit trail, so this row could not fail for losing one".into());
}
if b.entries == 0 {
return unrun("the PRIMARY's audit trail was EMPTY, so this row could not fail for losing it — the drill's own decisions should have filled it".into());
}
// **The defect, seen on the primary.** A trail that was never replicated cannot come back,
// and saying that here is worth more than the red row below.
if !b.replicated {
return fail(format!(
"the PRIMARY's audit trail ({} entries, tip seq {}) is NOT replicated into its transaction log, so the twin never held it and no restore can bring it back. This is D6",
b.entries, b.tip_seq
));
}
if !b.verified {
return fail("the PRIMARY's own audit chain did not verify before its disks were destroyed; nothing after this is a claim about the restore".into());
}
let a = match after {
Ok(a) => a,
Err(e) => {
return fail(format!(
"the RESTORED box's audit trail could not be read: {}",
one_line(e)
))
}
};
if !a.present {
return fail("the restored box came back with NO audit trail at all — the record of what happened is gone on exactly the day it is needed".into());
}
if !a.verified {
return fail(format!(
"the restored box holds {} entries and the chain does NOT verify: {}",
a.entries, a.tip
));
}
// **Which trail answered.** A box that is still mid-restore, or thinks it is a twin, answers
// from its OWN per-box trail and says `replicated: no`. MEASURED 2026-09-21 (T3, `ff2c7c9`):
// exactly that — the `txlog-restoring` unlink was lost in the hard stop and the "restored"
// box never left restore mode — and this row, which did not look, blamed the trail instead.
if !a.replicated {
return fail(format!(
"the restored box answered from its OWN per-box trail ({} entries), not the replicated one: it is still mid-restore or believes it is a twin, so it is not yet the primary it was restored as",
a.entries
));
}
if a.hash_at_seq.is_empty() {
return fail(format!(
"the restored box's trail holds no entry at seq {} — the primary's history did not come back (it holds {} entries)",
b.tip_seq, a.entries
));
}
if a.hash_at_seq != b.tip {
return fail(format!(
"the restored box's audit trail DIFFERS at seq {}:\n the primary wrote: {}\n the restored box holds: {}",
b.tip_seq, b.tip, a.hash_at_seq
));
}
if a.entries < b.entries {
return fail(format!(
"the restored box holds {} entries where the primary held {}",
a.entries, b.entries
));
}
Verdict {
id,
name: NAME,
status: Status::Pass,
detail: format!(
"{} entries came back and the chain verifies; the hash at the primary's last sequence ({}) is identical — {} — so every earlier entry is byte-identical too, and the restored box has continued the SAME chain to {} entries",
b.entries, b.tip_seq, b.tip, a.entries
),
}
}
/// One verdict per class: PASS when before and after agree and the answer is not empty, UNRUN
/// when the build has no such capability, FAIL otherwise. A class that answered on one side and
/// not the other is a FAIL — that asymmetry is exactly the loss this is looking for.
fn class_verdicts(
before: &[Class],
after: &[Class],
ids: &[(&'static str, &'static str)],
seeded: &BTreeSet<&'static str>,
refusals: &std::collections::BTreeMap<&'static str, String>,
) -> Vec<Verdict> {
let mut out = Vec::new();
for (i, (name, b)) in before.iter().enumerate() {
let (id, verdict_name) = ids[i];
let a = after
.get(i)
.map(|(_, a)| a.clone())
.unwrap_or_else(|| Err("not read after the restore".into()));
// **A class the drill never MADE cannot fail for losing it.** MEASURED 2026-09-17:
// `replication list` on a box with no standby answers "no standby configured" — a
// non-empty string, identical before and after, and a green row about nothing. A row
// that cannot go red is worse than no row, so an unseeded class says UNRUN by name.
if !seeded.contains(name) {
out.push(Verdict {
id,
name: verdict_name,
status: Status::Unrun,
detail: match (b, refusals.get(name)) {
(Err(e), _) if unimplemented(e) => format!("this medium's build answers no `{name}`: {}", one_line(e)),
(_, Some(why)) => format!("the drill could not seed {name} on the primary, so this row could not fail for losing one — the server refused: {why}"),
(Err(e), None) => format!("the drill could not seed {name} on the primary, so this row could not fail for losing one: {}", one_line(e)),
(Ok(_), None) => format!("the drill seeded no {name} on the primary, so this row could not fail for losing one"),
},
});
continue;
}
let v = match (b, &a) {
(Err(eb), Err(_)) if unimplemented(eb) => Verdict {
id,
name: verdict_name,
status: Status::Unrun,
detail: format!("this medium's build answers no `{name}`: {}", one_line(eb)),
},
(Err(eb), _) => Verdict {
id,
name: verdict_name,
status: Status::Fail,
detail: format!("the PRIMARY could not be read for {name}: {}", one_line(eb)),
},
(Ok(_), Err(ea)) => Verdict {
id,
name: verdict_name,
status: Status::Fail,
detail: format!("the restored box answers no {name}: {}", one_line(ea)),
},
(Ok(sb), Ok(sa)) if sb == sa && !sb.trim().is_empty() => Verdict {
id,
name: verdict_name,
status: Status::Pass,
detail: format!("{name} identical before and after: {sb}"),
},
(Ok(sb), Ok(_)) if sb.trim().is_empty() => Verdict {
id,
name: verdict_name,
status: Status::Unrun,
detail: format!(
"the primary answered nothing at all for {name}; this row is not claimed"
),
},
(Ok(sb), Ok(sa)) => Verdict {
id,
name: verdict_name,
status: Status::Fail,
detail: format!(
"{name} CHANGED across the restore:\n before: {sb}\n after: {sa}"
),
},
};
out.push(v);
}
out
}
/// The twin's data volume, read from the host: GPT p1 → plain xfs or md v1.2 → xfs; the xfs
/// region copied sparsely (SEEK_DATA/SEEK_HOLE) and walked with `xfs_db -r`. Every file name is
/// held to [`is_cache_path`]. Answers (files, directories, cache hits).
pub fn scan_twin_disk(
img: &Path,
work: &Path,
) -> Result<(usize, usize, Vec<String>, Vec<String>), String> {
let f = std::fs::File::open(img).map_err(|e| format!("{}: {e}", img.display()))?;
let u64_at = |b: &[u8], o: usize| u64::from_le_bytes(b[o..o + 8].try_into().unwrap());
let mut magic = [0u8; 4];
f.read_exact_at(&mut magic, 0).map_err(|e| e.to_string())?;
let disk_len = f.metadata().map_err(|e| e.to_string())?.len();
let (mut start, end, mut shape) = if &magic == b"XFSB" {
(0u64, disk_len, "plain xfs on the whole disk")
} else {
let mut hdr = [0u8; 512];
f.read_exact_at(&mut hdr, 512).map_err(|e| e.to_string())?;
if &hdr[0..8] != b"EFI PART" {
return Err("the twin's data volume carries neither xfs nor a GPT".into());
}
let mut ent = [0u8; 128];
f.read_exact_at(&mut ent, u64_at(&hdr, 72) * 512)
.map_err(|e| e.to_string())?;
let (first, last) = (u64_at(&ent, 32), u64_at(&ent, 40));
(first * 512, (last + 1) * 512, "plain xfs on p1")
};
f.read_exact_at(&mut magic, start)
.map_err(|e| e.to_string())?;
if &magic != b"XFSB" {
let mut sb = [0u8; 256];
f.read_exact_at(&mut sb, start + 4096)
.map_err(|e| e.to_string())?;
if u32::from_le_bytes(sb[0..4].try_into().unwrap()) != 0xa92b_4efc {
return Err("p1 holds neither xfs nor an md v1.2 member".into());
}
start += u64_at(&sb, 128) * 512;
f.read_exact_at(&mut magic, start)
.map_err(|e| e.to_string())?;
if &magic != b"XFSB" {
return Err("the md member's data area holds no xfs".into());
}
shape = "xfs inside md v1.2 on p1";
}
let copy = work.join("twin-data-xfs.img");
let out = std::fs::File::create(©).map_err(|e| e.to_string())?;
out.set_len(end - start).map_err(|e| e.to_string())?;
let fd = f.as_raw_fd();
let mut pos = start;
let mut buf = vec![0u8; 4 << 20];
while pos < end {
let d = unsafe { libc::lseek(fd, pos as libc::off_t, libc::SEEK_DATA) };
if d < 0 || d as u64 >= end {
break;
}
let h = (unsafe { libc::lseek(fd, d, libc::SEEK_HOLE) } as u64).min(end);
let mut o = d as u64;
while o < h {
let n = ((h - o) as usize).min(buf.len());
f.read_exact_at(&mut buf[..n], o)
.map_err(|e| e.to_string())?;
out.write_all_at(&buf[..n], o - start)
.map_err(|e| e.to_string())?;
o += n as u64;
}
pos = h;
}
drop(out);
eprintln!(
" twin disk: {shape}; xfs region copied sparsely to {}",
copy.display()
);
let (mut files, mut dirs, mut hits, mut cache_dirs) = (0usize, 0usize, Vec::new(), Vec::new());
let mut stack = vec![String::from("/")];
while let Some(dir) = stack.pop() {
dirs += 1;
let o = Command::new("/usr/sbin/xfs_db")
.env("LC_ALL", "C")
.args(["-r", "-f"])
.arg(©)
.args(["-c", &format!("ls {dir}")])
.output()
.map_err(|e| format!("xfs_db: {e}"))?;
if !o.status.success() {
return Err(format!(
"xfs_db ls {dir}: {}",
String::from_utf8_lossy(&o.stderr)
));
}
for line in String::from_utf8_lossy(&o.stdout).lines() {
let t: Vec<&str> = line.split_whitespace().collect();
if t.len() < 7 || t[0].ends_with(':') {
continue;
}
let name = t[5..t.len() - 1].join(" ");
if name == "." || name == ".." {
continue;
}
let path = if dir == "/" {
format!("/{name}")
} else {
format!("{dir}/{name}")
};
// A FILE is what the rule is about. A cache-named directory is walked like any other:
// every file under it matches the rule by its prefix. Empty ones (the installer lays
// `gc/` and `quarantine/` on /data when a box has no /serialdat) are reported, not failed.
// Only a REGULAR file is a cache file. The installer's links (`gc -> .serialdat/gc`) and
// empty directories are reported by name, never failed; a link is not followed, the
// directory it names is walked where it really is.
let kind = t[2];
if is_cache_path(&path) {
match kind {
"regular" => hits.push(path.clone()),
_ => cache_dirs.push(format!("{path} ({kind})")),
}
}
match kind {
"directory" => stack.push(path),
"regular" => files += 1,
_ => {}
}
}
}
let _ = std::fs::remove_file(©);
Ok((files, dirs, hits, cache_dirs))
}
fn v(id: &'static str, name: &'static str, ok: Result<String, String>) -> Verdict {
match ok {
Ok(detail) => Verdict {
id,
name,
status: Status::Pass,
detail,
},
Err(detail) => Verdict {
id,
name,
status: Status::Fail,
detail,
},
}
}
/// **The ids of one drill's rows.** Spelled per scenario rather than generated, so a verdict
/// line can be grepped for and a row cannot silently move between scenarios.
pub struct Ids {
follow: &'static str,
restored: &'static str,
users: &'static str,
entitlements: &'static str,
twin_follows: &'static str,
purge: &'static str,
cache: &'static str,
host_key: &'static str,
classes: &'static [(&'static str, &'static str)],
issuer: &'static str,
/// **The audit trail** (lane T11): the box's own record of who did what, which until
/// 2026-09-21 no restore brought back and no row noticed.
audit: &'static str,
geometry: &'static str,
}
/// **The machine the restore lands on.** The original primary is always the appliance's own
/// stripe (`--members`, `--member-gib`); this is what the RE-INSTALLED box has, which in the
/// field is whatever hardware was to hand.
pub struct Shape {
pub id: &'static str,
pub name: &'static str,
/// Members the fresh primary is installed on. `None` = the same as the original.
pub fresh_members: Option<usize>,
/// GiB per member on the fresh primary. `None` = the same as the original.
pub fresh_member_gib: Option<u64>,
pub ids: &'static Ids,
}
const T1_IDS: Ids = Ids {
follow: "T1a",
restored: "T1b",
users: "T1c",
entitlements: "T1d",
twin_follows: "T1e",
purge: "T1g",
cache: "T1f",
host_key: "T1h",
classes: &[
("T1i", "users, read as identity rows"),
("T1j", "public keys back"),
("T1k", "groups back"),
("T1l", "group membership back"),
("T1m", "grants back"),
("T1n", "namespaces back"),
("T1o", "the signed membership log (an Append frame) back"),
("T1p", "webhook configuration back"),
("T1q", "replication configuration back"),
(
"T1r",
"repository rows back: hash kind, default branch, visibility",
),
("T1s", "pull-mirror configuration back"),
],
issuer: "T1t",
audit: "T1v",
geometry: "T1u",
};
const T2_IDS: Ids = Ids {
follow: "T2a",
restored: "T2b",
users: "T2c",
entitlements: "T2d",
twin_follows: "T2e",
purge: "T2g",
cache: "T2f",
host_key: "T2h",
classes: &[
("T2i", "users, read as identity rows"),
("T2j", "public keys back"),
("T2k", "groups back"),
("T2l", "group membership back"),
("T2m", "grants back"),
("T2n", "namespaces back"),
("T2o", "the signed membership log (an Append frame) back"),
("T2p", "webhook configuration back"),
("T2q", "replication configuration back"),
(
"T2r",
"repository rows back: hash kind, default branch, visibility",
),
("T2s", "pull-mirror configuration back"),
],
issuer: "T2t",
audit: "T2v",
geometry: "T2u",
};
/// **T1 — the twin drill onto the SAME shape of machine.**
pub fn t1(o: &Opts) -> Verdict {
drill(
o,
&Shape {
id: "T1",
name: "twin drill (follow, destroy the primary, restore from the twin)",
fresh_members: None,
fresh_member_gib: None,
ids: &T1_IDS,
},
)
}
/// **T2 — the same drill onto MORE devices than the set left with** (the owner's 4 → 8).
///
/// The data set is md RAID0 over its members and a re-installed box cannot assume the geometry
/// it had: `korp_installer`'s `plan_set` computes the stripe from the disks the machine
/// actually has. Nothing in the restore reproduces a member layout — it streams frames into a
/// filesystem — so the only question is whether the set the installer lays for the NEW machine
/// takes every object and document. This asks it with eight members where there were four.
pub fn t2(o: &Opts) -> Verdict {
drill(
o,
&Shape {
id: "T2",
name: "restore onto MORE devices than the set left with (4 → 8): the set is laid for the machine it is on",
fresh_members: Some(o.fresh_members.unwrap_or(8)),
fresh_member_gib: o.fresh_member_gib,
ids: &T2_IDS,
},
)
}
fn drill(o: &Opts, shape: &Shape) -> Verdict {
let (id, name) = (shape.id, shape.name);
let ids = shape.ids;
let fresh_members = shape.fresh_members.unwrap_or(o.members.max(4));
let fresh_member_gib = shape.fresh_member_gib.unwrap_or(o.member_gib);
// **ONE MEDIUM.** The twin installs the PRIMARY's own image (`--iso`), with
// no role parameter at all: the role is conferred afterwards, over the control plane, by
// `twin join` → `Bootstrap.BecomeTwin`, which writes `twin.json`. That file — not a flag and
// not an install parameter — is what `stores.is_twin()` reads in `serve.rs` (the follower
// starts), `serve.rs` (no git listener) and `control_plane.rs` (the role it reports).
// There is no second medium (`--twin-iso` is retired); the T1m row still hashes both paths.
let twin_iso = o.iso.clone();
let work = o.work.join(id);
let _ = std::fs::create_dir_all(&work);
let spec = |sub: &str, iso: PathBuf, boot_gib: u64, members: usize, gib: u64| RigSpec {
iso,
work: work.join(sub),
boot_gib,
member_gib: gib,
members,
mem_mb: o.mem_mb,
smp: o.smp,
settle_secs: o.settle_secs.min(10),
explicit_blockdevs: false,
};
let run = || -> Result<Vec<Verdict>, String> {
let mut primary = Rig::create(spec("primary", o.iso.clone(), o.boot_gib, 4, o.member_gib))?;
primary.hold = Duration::from_secs(4 * 3600);
// **The twin's system disk must be LARGER than its data volume, and that is the
// installer's rule, not a preference** (MEASURED 2026-09-17, lane TWIN-COMPLETENESS,
// against the 2026-09-16 media).
//
// This used to be a 10 GiB system disk beside a 20 GiB volume, with a comment reading
// "with one data disk the installer takes the SMALLER disk as boot, so the volume must
// be the larger". That was true of the 2026-08-29 installer and is the exact inverse of
// today's: `korp_installer::layout` takes the largest equal GROUP as `/data` and boots
// the remainder, and with two unequal disks there is no group —
// `pick_boot_without_a_group` decides by class, then by the LARGEST. So the ESP and the
// rootfs landed on the VOLUME:
//
// INSTALLER: data set: boot vdb (20GiB) · data 1×10GiB plain [vda (blank)]
// boot.img p1 is "korp-data-0", not the ESP
//
// korp-installer names this itself (`pick_boot_declared`): "no size heuristic can serve
// both … where the sizes are silent the layout has to be DECLARED", and the declaration
// is `korp.data.disks=` on the medium's command line. **The `gunnar-twin` medium does
// not carry one**, so the estate's twin shape (estate.toml [machines.twin]: a 10 GB
// system disk and a 20 GB volume) CANNOT be installed by it. That is a defect in the
// twin variant's recipe, not in this bench, and the bench must not hide it.
//
// **The system disk stays at 10 GiB and the volume must be smaller.** Making the system
// disk 50 GiB instead also satisfies the rule, and was tried — but a 50 GiB boot disk
// earns its own P3 (`INSTALLER: p3=korp-data(47.9GiB, mounted at /serialdat)`), which
// moves the slow tree off the volume and left the cache scan reading an EMPTY data
// volume (`T1f: 0 files`). The twin's shape is one volume and no P3; keeping the system
// disk at the estate's 10 GiB keeps it.
//
// **MEASURED AGAIN 2026-09-21 (lane T3).** The rule above is about the ELECTION, and the
// election only falls through to class-then-LARGEST when there is no equal-size GROUP.
// Give the twin the SAME number of equal members the primary has (`--twin-members`) and
// the group exists, so `/data` is elected with no `korp.data.disks=` token at all — which
// is the whole of what the `gunnar-twin` medium was carrying that the primary's does not.
const TWIN_BOOT_GIB: u64 = 10;
let twin_members = o.twin_members.unwrap_or(1);
let twin_boot_gib_asked = o.twin_boot_gib.unwrap_or(TWIN_BOOT_GIB);
if twin_members == 1 && o.twin_boot_gib.is_none() && o.twin_member_gib >= TWIN_BOOT_GIB {
return Err(format!(
"the twin would be laid with a {TWIN_BOOT_GIB} GiB system disk beside a {} GiB data volume. Today's korp-installer boots the LARGER of two unequal disks, so the ESP would land on the volume and the box would not boot. Pass `--twin-member-gib` under {TWIN_BOOT_GIB}, or give the twin as many equal members as the primary (`--twin-members`) so the installer elects `/data` as a group",
o.twin_member_gib
));
}
let twin_boot_gib = twin_boot_gib_asked;
eprintln!(
"── {id}: the twin's medium is {} ({twin_members} × {} GiB data, {twin_boot_gib} GiB system)",
"THE PRIMARY'S OWN — one medium, no role on the command line",
o.twin_member_gib
);
let mut twin = Rig::create(spec(
"twin",
twin_iso.clone(),
twin_boot_gib,
twin_members,
o.twin_member_gib,
))?;
twin.hold = Duration::from_secs(4 * 3600);
twin.grpc_only_ready = true;
eprintln!("── {id}: primary {} ", primary.describe());
eprintln!("── {id}: twin {} ", twin.describe());
primary.install()?;
twin.install()?;
// **The medium, as a hash and not a claim.** With `--twin-same-medium` these two lines
// are the same 64 characters, which is the owner's "EXACTLY like the main" measured.
let medium_id = |p: &Path| -> String {
match std::fs::read(p) {
Ok(b) => {
use sha2::{Digest, Sha256};
format!("{:x}", Sha256::digest(&b))
}
Err(e) => format!("unreadable: {e}"),
}
};
let (primary_medium, twin_medium) = (medium_id(&o.iso), medium_id(&twin_iso));
let one_medium = if primary_medium == twin_medium {
Ok(format!(
"ONE medium, installed twice: sha256 {primary_medium} — primary {} and twin {}",
o.iso.display(),
twin_iso.display()
))
} else {
Err(format!("two DIFFERENT media: primary {} sha256 {primary_medium}; twin {} sha256 {twin_medium}", o.iso.display(), twin_iso.display()))
};
eprintln!(
"── {id}: medium: {}",
match &one_medium {
Ok(s) => s.clone(),
Err(e) => e.clone(),
}
);
// The installer's own words, off the ESP: the UDP sink never hears the installer phase.
for (who, rig) in [("primary", &primary), ("twin", &twin)] {
match rig.esp_bootlog() {
Ok(log) => {
let elect: Vec<&str> = log
.lines()
.filter(|l| {
l.contains("data set")
|| l.contains("ESP")
|| l.contains("p1=")
|| l.contains("elect")
})
.take(8)
.collect();
eprintln!(
"── {id}: {who} ESP BOOTLOG.TXT ({} lines); election:\n {}",
log.lines().count(),
elect.join("\n ")
);
}
Err(e) => eprintln!("── {id}: {who} ESP BOOTLOG.TXT: {e}"),
}
}
// ── the bench's monetize key (T1d, TK1, TK2) ─────────────────────────
//
// The box must TRUST the bench's key, and that is read off what the installer laid, not
// assumed from how the medium was meant to be composed.
let bench_key = o
.monetize_seed
.as_deref()
.map(crate::money::load_or_make)
.transpose()?;
let wrong_iso = o.restore_wrong_key_iso.clone();
if wrong_iso.is_some() && bench_key.is_none() {
return Err("--restore-wrong-key-iso needs --monetize-seed: a box that trusts no bench key cannot be given the wrong one".into());
}
let trusts_bench: Result<String, String> = match &bench_key {
None => Err("no --monetize-seed: the bench holds no key to sign a fact with".into()),
Some(k) => {
let want = crate::money::pub_hex(k);
match primary.esp_cmdline().map(|c| crate::money::pubkey_in(&c)) {
Ok(Some(have)) if have.eq_ignore_ascii_case(&want) => Ok(format!("the primary's installed command line trusts the bench key {}…", &want[..12])),
Ok(Some(have)) => Err(format!("the medium trusts {}…, not the bench key {}…: compose it `--param monetize_pubkey={want}`", &have[..have.len().min(12)], &want[..12])),
Ok(None) => Err("the medium trusts NO monetize key (a community medium, or one composed without `monetize_pubkey`)".into()),
Err(e) => Err(format!("the installed command line is unreadable: {e}")),
}
}
};
eprintln!(
"── {id}: monetize: {}",
match &trusts_bench {
Ok(s) => s.clone(),
Err(e) => e.clone(),
}
);
let mut sess = Session::new(&work, &o.cli)?;
let keys = work.join("keys");
let carol_key = keys.join("carol");
let dave_key = keys.join("dave");
data::keygen(&carol_key)?;
data::keygen(&dave_key)?;
let repos = seed_repos(&sess.trees, 0x7417_0001)?;
let (lfs, lfs_sha) = make_lfs_repo(&sess.trees, 0x7417_0002)?;
let notes = data::make_repo(&sess.trees, "notes", &[("n.bin", 2 << 20)], 0x7417_0003)?;
let later = data::make_repo(
&sess.trees,
"after-join",
&[("a.bin", 8 << 20), ("b.bin", 3 * data::CHUNK_BYTES + 7)],
0x7417_0004,
)?;
// Item 4 of the loose ends: a repository the twin receives in wave 1 and the primary
// purges in wave 2 — gone on the twin's replay, and a restore must not resurrect it.
let doomed = data::make_repo(&sess.trees, "doomed", &[("d.bin", 1 << 20)], 0x7417_0005)?;
let mut doomed_gone = Err::<String, String>("not reached".into());
// The repositories whose catalog rows are read back class by class.
let repo_paths: Vec<String> = repos
.iter()
.map(|m| format!("{}/{}", sess.principal, m.name))
.collect();
let carol_ssh = data::ssh_command(&carol_key);
let cli = o.cli.clone();
let admin = sess.key.clone();
let (to_twin, twin_rx) = mpsc::channel::<ToTwin>();
let (twin_tx, from_twin) = mpsc::channel::<FromTwin>();
std::thread::scope(|scope| -> Result<Vec<Verdict>, String> {
let twin_cli = cli.clone();
let twin_admin = admin.clone();
let twin_thread = scope.spawn(move || {
let b = twin.boot("twin", |tp| {
let r = twin_drive(&twin_cli, &twin_admin, tp, &twin_rx, &twin_tx);
if let Err(e) = &r {
let _ = twin_tx.send(FromTwin::Failed(e.clone()));
}
r
});
(twin, b)
});
// ── phase 1: the primary takes everything, the twin follows ─────────
let mut tports: Option<Ports> = None;
let (mut follow, mut users_before) =
(Err::<String, String>("not reached".into()), BTreeSet::new());
let (mut state_before, mut state_after): (Vec<Class>, Vec<Class>) =
(Vec::new(), Vec::new());
// The classes the drill actually MADE something in. Only these may go green.
let mut seeded: BTreeSet<&'static str> = [
"users",
"keys",
"groups",
"group members",
"grants",
"namespaces",
"the signed membership log",
"repository rows (hash, default branch, visibility)",
]
.into_iter()
.collect();
let mut seed_refusals: std::collections::BTreeMap<&'static str, String> =
std::collections::BTreeMap::new();
let (mut host_key_before, mut host_key_after) = (
Err::<String, String>("not reached".into()),
Err::<String, String>("not reached".into()),
);
// The audit trail, before and after (lane T11, the D6 class).
let (mut audit_before, mut audit_after) = (
Err::<AuditShape, String>("not reached".into()),
Err::<AuditShape, String>("not reached".into()),
);
// The facts the bench pushed, as the box LISTS them — before, after, and after the fix.
let (mut ent_before, mut ent_after, mut ent_fixed) = (
Err::<String, String>("not reached".into()),
Err::<String, String>("not reached".into()),
Err::<String, String>("not reached".into()),
);
let mut pinned: Option<Ports> = None;
let first = primary.boot("primary", |p| {
pinned = Some(p);
let at = format!("127.0.0.1:{}", p.grpc_private);
let k = admin.display().to_string();
let boot = sess.bootstrap(p)?;
let pushed = sess.push_all(p, repos.clone())?;
data::git(
&lfs.dir,
&sess.ssh,
&[
"push",
"-q",
&sess.url(p, "lfs-assets"),
"HEAD~1:refs/heads/main",
],
)?;
data::push(
&lfs,
&sess.ssh,
&sess.url(p, "lfs-assets"),
Duration::from_secs(300),
)?;
data::cli(
&cli,
&[
"control",
"user",
"add",
"carol",
"--key",
&pubkey(&carol_key)?,
"-i",
&k,
"--at",
&at,
],
)?;
data::push(
¬es,
&carol_ssh,
&format!("ssh://carol@127.0.0.1:{}/carol/notes", p.ssh),
Duration::from_secs(300),
)?;
// Pushed outside `push_all`: it is not one of the repositories the restore verifies.
data::push(
&doomed,
&sess.ssh,
&sess.url(p, "doomed"),
Duration::from_secs(300),
)?;
let _ = to_twin.send(ToTwin::PrimaryAt(p));
let (key, tp) = match recv(&from_twin)? {
FromTwin::Joined { key, ports } => (key, ports),
_ => return Err("the twin did not join".into()),
};
tports = Some(tp);
data::cli(
&cli,
&[
"control", "user", "add", "twin-box", "--key", &key, "--kind", "machine",
"-i", &k, "--at", &at,
],
)?;
data::cli(
&cli,
&["control", "group", "add", "twin", "-i", &k, "--at", &at],
)?;
data::cli(
&cli,
&[
"control", "group", "member", "add", "twin", "twin-box", "-i", &k, "--at",
&at,
],
)?;
let h1 = head_of(&cli, &admin, &at)?;
let _ = to_twin.send(ToTwin::Sync(h1));
let s1 = match recv(&from_twin)? {
FromTwin::Synced(s) => s,
_ => return Err("no sync".into()),
};
// The second wave, while the twin follows.
let later_pushed = sess.push_all(p, vec![later.clone()])?;
data::cli(
&cli,
&[
"control",
"user",
"add",
"dave",
"--key",
&pubkey(&dave_key)?,
"-i",
&k,
"--at",
&at,
],
)?;
let doomed_path = format!("{}/doomed", sess.principal);
data::cli(
&cli,
&[
"control",
"repo",
"delete",
&doomed_path,
"--purge",
"--yes",
"-i",
&k,
"--at",
&at,
],
)?;
// ── the CONFIGURATION, so each class has something to lose ──────────
//
// Every one of these is a class the design note says is replicated (M4, M7, M8)
// and that no verdict compared. A drill that never made a group cannot notice a
// restore that dropped one.
let mixed = format!("{}/mixed", sess.principal);
data::cli(
&cli,
&[
"control",
"group",
"add",
TWIN_REVIEWERS,
"-i",
&k,
"--at",
&at,
],
)?;
data::cli(
&cli,
&[
"control",
"group",
"member",
"add",
TWIN_REVIEWERS,
"carol",
"-i",
&k,
"--at",
&at,
],
)?;
data::cli(
&cli,
&[
"control",
"group",
"member",
"add",
TWIN_REVIEWERS,
"dave",
"-i",
&k,
"--at",
&at,
],
)?;
data::cli(
&cli,
&[
"control",
"namespace",
"create",
TWIN_NAMESPACE,
"-i",
&k,
"--at",
&at,
],
)?;
data::cli(
&cli,
&[
"control",
"grant",
TWIN_REVIEWERS,
"write",
"--namespace",
TWIN_NAMESPACE,
"-i",
&k,
"--at",
&at,
],
)?;
data::cli(
&cli,
&[
"control",
"repo",
"set-visibility",
&mixed,
"public",
"-i",
&k,
"--at",
&at,
],
)?;
// **The default branch has to EXIST.** MEASURED 2026-09-17: pointing HEAD at a
// branch with no commits made every clone check out nothing and T1b went red on
// `rev-parse HEAD^{tree}` — the drill's own bug, not the twin's. The branch is
// pushed first, at the same commit as `main`, so a clone of the restored box
// checks out the same tree through a DIFFERENT default branch.
let mixed_repo = repos
.iter()
.find(|m| m.name == "mixed")
.ok_or("no mixed repo")?;
data::git(
&mixed_repo.dir,
&sess.ssh,
&[
"push",
"-q",
&sess.url(p, "mixed"),
&format!("main:{TWIN_DEFAULT_BRANCH}"),
],
)?;
data::cli(
&cli,
&[
"control",
"repo",
"set-default-branch",
&mixed,
TWIN_DEFAULT_BRANCH,
"-i",
&k,
"--at",
&at,
],
)?;
// Outbound configuration, all three behind cargo features or an upstream this
// box will never dial. A verb the build does not carry answers "not
// implemented", and its row says UNRUN rather than red; a verb that WORKS seeds
// the class, and only a seeded class is allowed to go green.
// A refusal is KEPT, not swallowed: an UNRUN row that cannot say why it is
// unrun is the same silence as a green one that measured nothing.
let mut seed = |class: &'static str, args: &[&str]| match data::cli(&cli, args) {
Ok(_) => {
seeded.insert(class);
}
Err(e) => {
seed_refusals.insert(class, one_line(&e));
}
};
seed(
"webhooks",
&[
"control",
"webhook",
"add",
"twin-drill",
"http://10.0.2.2:1/hook",
"-i",
&k,
"--at",
&at,
],
);
seed(
"replication config",
&[
"control",
"replication",
"add",
"twin-drill-standby",
"ssh://git@10.0.2.2:2222",
"-i",
&k,
"--at",
&at,
],
);
seed(
"pull mirrors",
&[
"control",
"mirror",
"add",
MIRROR_REPO, // The mirror door refuses an RFC 1918 upstream by name (a mirror dials FROM the server, so
// a private destination the caller could not reach is an SSRF). TEST-NET-3 is public
// and unroutable, and `add` dials nothing anyway — it is a document write.
"ssh://git@203.0.113.7:2222/upstream/notes.git",
"-i",
&k,
"--at",
&at,
],
);
// ── the MONEY class: two facts the bench's key signed, pushed through the operator's
// door, read back as the box ENFORCES them. Before `h2`, so the twin follows them.
if let (Some(key), Ok(_)) = (&bench_key, &trusts_bench) {
let now = std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.map(|d| d.as_millis() as u64)
.unwrap_or(0);
let mut pushed = Vec::new();
let r: Result<(), String> = (|| {
for (i, (ns, state)) in [
(sess.principal.clone(), monetize_embed::State::Paid),
("carol".to_string(), monetize_embed::State::Grace),
]
.into_iter()
.enumerate()
{
let file = work.join(format!("fact-{ns}.json"));
std::fs::write(
&file,
crate::money::signed_fact_json(
key,
&ns,
state,
&[],
now + i as u64,
),
)
.map_err(|e| format!("{}: {e}", file.display()))?;
pushed.push(one_line(&data::cli(
&cli,
&[
"control",
"entitlement",
"set",
&file.display().to_string(),
"-i",
&k,
"--at",
&at,
],
)?));
}
Ok(())
})();
ent_before = r.and_then(|_| entitlements(&cli, &admin, &at)).map(|l| {
eprintln!(" money: pushed {pushed:?}; the box enforces {l:?}");
l
});
}
let h2 = head_of(&cli, &admin, &at)?;
let _ = to_twin.send(ToTwin::Sync(h2));
let s2 = match recv(&from_twin)? {
FromTwin::Synced(s) => s,
_ => return Err("no sync".into()),
};
users_before = users(&cli, &admin, &at)?;
// What the box IS, read the way an operator would, before its disks die.
state_before = control_state(&cli, &admin, &at, &repo_paths);
host_key_before = host_key(p.ssh);
// **The audit trail, as the primary held it on its last day.** Read LAST of
// everything, so that every decision above — every user, group, grant and
// namespace this drill made — is already in the chain it is about to lose.
audit_before = audit_shape(&cli, &admin, &at, 0);
follow = Ok(format!("wave 1 head {h1}: {s1} | wave 2 head {h2}: {s2}"));
Ok(format!(
"{boot}; pushed {pushed}, lfs-assets, carol/notes, {later_pushed}; users {}",
users_before.len()
))
});
print_boot(&first);
let mut out = Vec::new();
out.push(v("T1m", "ONE medium: the primary and the twin are installed from the same image, byte for byte", one_medium));
if let Some(e) = boot_failed(&first) {
let _ = to_twin.send(ToTwin::Stop);
let _ = twin_thread.join();
return Err(format!("phase 1: {e}"));
}
out.push(v(
ids.follow,
"the twin follows the primary to the watermark, two waves",
follow,
));
let (tp, p) = (
tports.ok_or("no twin ports")?,
pinned.ok_or("no primary ports")?,
);
// ── the primary loses its disks; a fresh one answers where it did ───
primary.destroy_disks();
eprintln!(" primary: every disk image DESTROYED");
let mut fresh = Rig::create(spec(
"primary-fresh",
wrong_iso.clone().unwrap_or_else(|| o.iso.clone()),
o.boot_gib,
fresh_members,
fresh_member_gib,
))?;
eprintln!("── {id}: the set left {} × {} GiB and lands on {fresh_members} × {fresh_member_gib} GiB", o.members, o.member_gib);
fresh.pin = Some(p);
fresh.hold = Duration::from_secs(4 * 3600);
let laid = fresh.install()?;
// ── TK: the restore is composed with the WRONG monetize key ─────────
// Which key the box REALLY runs with is read off its boot (the MERGED command line),
// never assumed from the medium's name: see `money_key_in_effect`.
let planted: Result<String, String> = match (&wrong_iso, &bench_key) {
(Some(w), Some(k)) => fresh
.esp_cmdline()
.map(|c| crate::money::pubkey_in(&c))
.and_then(|key| match key {
Some(key) if !key.eq_ignore_ascii_case(&crate::money::pub_hex(k)) => Ok(format!("the restored primary is installed from {} and trusts {}… — not the key that signed its facts", w.display(), &key[..key.len().min(12)])),
Some(_) => Err(format!("{} trusts the SAME key as the bench: it is not a wrong-key medium", w.display())),
None => Err(format!("{} trusts no monetize key at all", w.display())),
}),
_ => Err("no --restore-wrong-key-iso".into()),
};
if wrong_iso.is_some() {
eprintln!(
"── {id}: wrong key: {}",
match &planted {
Ok(s) => s.clone(),
Err(e) => e.clone(),
}
);
}
// **The geometry, as a number and not an adjective.** What the set left with, what
// it landed on, and what the installer itself said it laid — the row an operator
// needs when it later turns out the device count mattered.
let geometry = if laid.member_names.len() == fresh_members {
Ok(format!(
"the set LEFT {} × {} GiB ({} GiB raw) and LANDED on {fresh_members} × {fresh_member_gib} GiB ({} GiB raw); the installer laid {} for the machine it was on: {}",
o.members,
o.member_gib,
o.members as u64 * o.member_gib,
fresh_members as u64 * fresh_member_gib,
laid.member_names.join(", "),
one_line(&laid.lines.join("\n"))
))
} else {
Err(format!(
"the set left {} × {} GiB and the fresh machine has {fresh_members} × {fresh_member_gib} GiB, but the installer laid {} members: {:?}",
o.members,
o.member_gib,
laid.member_names.len(),
laid.member_names
))
};
// The product's own guard on the restore of a DEAD primary (T1w).
let mut stale_guard: Result<String, String> = Err("not reached".into());
let replay = fresh.boot("fresh-restore", |m| {
let st = sess.status_line(m)?;
if !st.contains("OPEN") {
return Err(format!("the fresh primary is not fresh: {st}"));
}
let (twin_at, into, key) = (
format!("127.0.0.1:{}", tp.grpc_private),
format!("127.0.0.1:{}", m.grpc_private),
admin.display().to_string(),
);
// **The operator's command first, exactly as typed.** Only if the stale-source
// guard refuses it BECAUSE THE PRIMARY IS GONE — the very case a twin exists for —
// is the refusal kept as its own verdict and the restore retried with the
// override, so everything after it is still measured.
match data::cli(
&cli,
&[
"control", "twin", "replay", "--twin", &twin_at, "--into", &into, "-i",
&key,
],
) {
Ok(s) => {
stale_guard = Ok("the operator's plain `twin replay` was accepted".into());
Ok(one_line(&s))
}
Err(e) if e.contains("stale-source") => {
stale_guard = Err(one_line(&e));
data::cli(
&cli,
&[
"control",
"twin",
"replay",
"--twin",
&twin_at,
"--into",
&into,
"-i",
&key,
"--stale-ok",
],
)
.map(|s| format!("with --stale-ok: {}", one_line(&s)))
}
Err(e) => Err(one_line(&e)),
}
});
print_boot(&replay);
let replayed = match (&replay.driven, boot_failed(&replay)) {
(Some(Ok(s)), None) => Ok(s.clone()),
(Some(Err(e)), _) => Err(e.clone()),
(_, Some(e)) => Err(e),
_ => Err("the fresh primary never became ready".into()),
};
let (mut restored, mut users_ok, mut twin_after) = (
Err::<String, String>("not reached".into()),
Err::<String, String>("not reached".into()),
Err::<String, String>("not reached".into()),
);
let mut boot_said_mismatch: Option<String> = None;
let mut key_in_effect: Option<String> = None;
let mut refused_wire: Option<String> = None;
if replayed.is_ok() {
let back = fresh.boot("restored", |m| {
let at = format!("127.0.0.1:{}", m.grpc_private);
sess.expect_closed(m)?;
let verify = sess.verify(m);
let carol = data::verify(&work.join("verify-carol"), &carol_ssh, |_| format!("ssh://carol@127.0.0.1:{}/carol/notes", m.ssh), std::slice::from_ref(¬es));
let lfs_ok = verify_lfs(&work.join("verify-lfs"), &sess.ssh, &sess.url(m, "lfs-assets"), &lfs_sha);
restored = match (&verify, carol.intact(), &lfs_ok) {
(Ok(v), true, Ok(l)) => Ok(format!("{v}; carol/notes: {}; {l}", carol.summary())),
_ => Err(format!("{verify:?}; carol/notes: {}; {lfs_ok:?}", carol.summary())),
};
if ent_before.is_ok() {
ent_after = entitlements(&cli, &admin, &at);
refused_wire = refused_on_wire(&cli, &admin, &at);
}
let after = users(&cli, &admin, &at)?;
users_ok = if after == users_before {
Ok(format!("{} identity rows identical before and after (bench, carol, dave, twin-box …)", after.len()))
} else {
Err(format!("before {users_before:?} after {after:?}"))
};
// The purged repository: not in the restored catalog, and not clonable.
let listed = data::cli(&cli, &["control", "repo", "list", "-i", &admin.display().to_string(), "--at", &at])?;
let doomed_path = format!("{}/doomed", sess.principal);
let clone_dir = work.join("verify-doomed");
let _ = std::fs::remove_dir_all(&clone_dir);
let _ = std::fs::create_dir_all(&clone_dir);
let cloned = data::git(&clone_dir, &sess.ssh, &["clone", "-q", &sess.url(m, "doomed")]);
doomed_gone = match (listed.lines().any(|l| l.contains(&doomed_path)), cloned) {
(false, Err(_)) => Ok(format!("{doomed_path} purged in wave 2: absent from the restored catalog ({} rows listed) and its clone refused", listed.lines().filter(|l| !l.trim().is_empty()).count())),
(true, _) => Err(format!("{doomed_path} is back in the restored catalog:\n{listed}")),
(false, Ok(_)) => Err(format!("{doomed_path} is not listed but still clones")),
};
// The same reads, on the box that came back.
state_after = control_state(&cli, &admin, &at, &repo_paths);
host_key_after = host_key(m.ssh);
// The restored box has been answering RPCs since it came up, so its TIP
// has moved. What must not have moved is the hash at the sequence the dead
// primary reported — ask for exactly that one.
audit_after = audit_shape(&cli, &admin, &at, audit_before.as_ref().map(|b| b.tip_seq).unwrap_or(0));
// A transaction only the RESTORED primary can have (after every class was read,
// so no class compare sees it). Its failure is T1e's verdict, never T1b's: the
// repositories came back whether or not the box takes writes.
let moved = (|| -> Result<u64, String> {
let restored_head = head_of(&cli, &admin, &at)?;
data::cli(&cli, &["control", "group", "add", "after-restore", "-i", &admin.display().to_string(), "--at", &at])
.map_err(|e| format!("the restored primary refused a write after its restart — it never left restore mode: {}", one_line(&e)))?;
let fresh_head = head_of(&cli, &admin, &at)?;
// The head must MOVE past what the restore laid: that, and only that,
// shows the box originates transactions again (T11, 2026-09-21).
if fresh_head <= restored_head {
return Err(format!("the restored primary took the write but its head did not move ({restored_head} -> {fresh_head})"));
}
Ok(fresh_head)
})();
match moved {
Ok(fresh_head) => {
let _ = to_twin.send(ToTwin::RestoreDone(fresh_head));
twin_after = match recv(&from_twin)? { FromTwin::Final(s) => Ok(s), _ => Err("no final".into()) };
}
Err(e) => twin_after = Err(e),
}
Ok("verified".into())
});
print_boot(&back);
if let Some(e) = boot_failed(&back) {
restored = Err(e);
}
// The box's own word on the refusal: the WARN `EntitlementSlot::open` logs when
// stored facts do not verify. Read off the console of THAT boot, with the key the
// boot really merged, so a verdict can never be about a key that did not take.
boot_said_mismatch = back
.serial
.lines()
.find(|l| l.contains("entitlement.boot.mismatch"))
.map(one_line);
key_in_effect = money_key_in_effect(&back.serial);
// ── TK2: the remedy is a re-image with the RIGHT medium, keeping the data ──
if let (Some(_), Ok(_)) = (&bench_key, &planted) {
fresh.spec.iso = o.iso.clone();
ent_fixed = match fresh.install() {
Err(e) => Err(format!(
"the keep re-image from the right medium failed: {e}"
)),
Ok(r) => {
eprintln!(
" primary: keep re-image from {} in {} ms",
o.iso.display(),
r.ms
);
let again = fresh.boot("right-key", |m| {
entitlements(&cli, &admin, &format!("127.0.0.1:{}", m.grpc_private))
});
print_boot(&again);
let failed = boot_failed(&again);
match (again.driven, failed) {
(Some(r), None) => r,
(_, Some(e)) => Err(e),
(None, None) => Err("the right-key boot never became ready".into()),
}
}
};
}
}
let _ = to_twin.send(ToTwin::Stop);
let (twin, tb) = twin_thread
.join()
.map_err(|_| "the twin thread panicked".to_string())?;
print_boot(&tb);
out.push(v(ids.restored, "restore from the twin into a re-imaged primary: every repo clones + fsck-clean, LFS identical", replayed.and(restored)));
out.push(v(ids.users, "users back", users_ok));
// Only T1 carries this row; its id is not in `Ids` because no other shape asks it.
if ids.entitlements == "T1d" {
out.push(v("T1w", "the operator's plain `twin replay` restores a primary that is GONE (the case a twin exists for) without an override", stale_guard.clone()));
}
out.push(money_verdicts(
ids.entitlements,
&trusts_bench,
&ent_before,
&ent_after,
&planted,
boot_said_mismatch.as_deref(),
&ent_fixed,
wrong_iso.is_some(),
));
if wrong_iso.is_some() {
let bench_pub = bench_key
.as_ref()
.map(crate::money::pub_hex)
.unwrap_or_default();
out.extend(wrong_key_verdicts(
&ent_before,
&ent_after,
&planted,
key_in_effect.as_deref(),
&bench_pub,
boot_said_mismatch.as_deref(),
refused_wire.as_deref(),
&ent_fixed,
));
}
out.push(v(
ids.twin_follows,
"the twin follows the restored primary",
twin_after,
));
out.push(v(ids.purge, "a repository purged on the primary is not resurrected by the restore from the twin", doomed_gone));
// T8's second half: the memory bound is set at RUNTIME now, so no `ZNIPPY_*` token may
// ride the twin's command line. The appliance has no shell to read `/proc/<pid>/cmdline`
// from; the kernel's own `Command line:` on the twin's console is where those tokens
// came from (`korp.env.ZNIPPY_*`), so it is read there.
let twin_cmdline = tb
.serial
.lines()
.find(|l| l.contains("Command line:"))
.map(str::to_string);
let znippy_tokens: Vec<String> = twin_cmdline
.as_deref()
.unwrap_or("")
.split_whitespace()
.filter(|t| t.contains("ZNIPPY_"))
.map(str::to_string)
.collect();
// The POSITIVE half: the bound as the engine resolved it after `BecomeTwin`
// (`twin.memory_bound`, read back through znippy's own getters). It must say explode
// Off, a 16 MiB budget, 2 workers and a 16 MiB lookahead — or the setting did not take.
let bound_line = tb
.serial
.lines()
.rev()
.find(|l| l.contains("twin.memory_bound"))
.map(one_line);
let bound_ok = bound_line.as_deref().is_some_and(|l| {
l.contains("explode=Off")
&& l.contains("explode_budget_bytes=Some(16777216)")
&& l.contains("resolve_workers=2")
&& l.contains("resolve_lookahead_bytes=16777216")
});
let scan = scan_twin_disk(&twin.members[0], &work).and_then(|(files, dirs, hits, cache_dirs)| {
let said = match (&bound_line, bound_ok) {
(Some(l), true) => format!("the twin's console: {l}"),
(Some(l), false) => return Err(format!("{files} files, 0 cache files — but the twin resolved a DIFFERENT bound: {l}")),
(None, _) => return Err(format!("{files} files, {} cache files — and the twin's console never logged `twin.memory_bound`, so the runtime bound is not shown to have taken", hits.len())),
};
match (hits.is_empty() && files > 0, &twin_cmdline, znippy_tokens.is_empty()) {
(true, Some(_), true) => Ok(format!("{files} files in {dirs} directories on the twin's one data volume; 0 cache files (objects.exploded: 0; cache-named links and directories with no file under them, laid by the installer: {cache_dirs:?}); the twin's kernel command line carries no ZNIPPY_ token; {said}")),
(true, None, _) => Err(format!("{files} files, 0 cache files — but the twin's console has no `Command line:`, so whether a ZNIPPY_ token rode it is unknown")),
(true, Some(_), false) => Err(format!("{files} files, 0 cache files — but the twin's command line still carries {znippy_tokens:?}")),
(false, _, _) => Err(format!("{files} files, cache files ({}): {hits:?}; ZNIPPY_ tokens on the command line: {znippy_tokens:?}", hits.len())),
}
});
out.push(v(
ids.cache,
"a scan of the twin's disk finds no cache file",
scan,
));
// ── one row per class of configuration ──────────────────────────
out.push(v(
ids.host_key,
"the SSH host key: a restored box no client refuses",
match (&host_key_before, &host_key_after) {
(Ok(b), Ok(a)) if b == a => Ok(format!("{} — identical before and after; no client sees a changed fingerprint", host_key_fingerprint(b))),
(Ok(b), Ok(a)) => Err(format!(
"the restored box serves a DIFFERENT SSH host key: {} became {}. Every client with this box in known_hosts would refuse it",
host_key_fingerprint(b),
host_key_fingerprint(a)
)),
(Err(e), _) => Err(format!("the primary's host key could not be read: {e}")),
(_, Err(e)) => Err(format!("the restored box's host key could not be read: {e}")),
},
));
out.extend(class_verdicts(
&state_before,
&state_after,
ids.classes,
&seeded,
&seed_refusals,
));
out.push(audit_verdict(ids.audit, &audit_before, &audit_after));
out.push(v(ids.geometry, "the data set is laid for the machine the restore lands on, not for the one it left", geometry));
out.push(Verdict {
id: ids.issuer,
name: "the certificate issuer back",
status: Status::Unrun,
detail: "no control-plane read of the PKI document exists: `gunnar issuer show` takes --data-root and the appliance has no shell, and the issuer certificate is returned only by Bootstrap.Enroll, which a restored box can never answer again. THIS IS A GAP IN THE PRODUCT, not in the twin — an operator cannot ask a running appliance which CA it mints with. Proven natively meanwhile: `twin::tests::a_restored_box_comes_back_as_the_same_box_not_just_the_same_repositories` compares the issuer certificate, its fingerprint and its private key after the re-seal, and mints a certificate with it"
.into(),
});
if !o.keep {
fresh.destroy_disks();
twin.destroy_disks();
}
Ok(out)
})
};
match run() {
Ok(lines) => {
for l in &lines {
eprintln!("{}", l.line());
}
let failed: Vec<&str> = lines
.iter()
.filter(|l| matches!(l.status, Status::Fail))
.map(|l| l.id)
.collect();
let detail = lines
.iter()
.map(|l| {
format!(
"{} {}",
l.id,
match l.status {
Status::Pass => "PASS",
Status::Fail => "FAIL",
Status::Unrun => "UNRUN",
}
)
})
.collect::<Vec<_>>()
.join(", ");
Verdict {
id,
name,
status: if failed.is_empty() {
Status::Pass
} else {
Status::Fail
},
detail,
}
}
Err(e) => Verdict {
id,
name,
status: Status::Fail,
detail: e,
},
}
}
// ── T3: the restore onto FEWER, SMALLER devices ─────────────────────────────
/// **T3 — the set left four members and lands on two that cannot hold it.**
///
/// This is the direction that can lose data quietly. Two devices may hold LESS TOTAL SPACE than
/// the set being restored, and the two honest outcomes are: it fits and restores whole, or it
/// **refuses by name, before it writes anything, saying how much is short**. A restore that
/// fills two disks and stops with the tail missing is the failure this scenario exists to catch.
///
/// Written red first: the fresh set is deliberately made too small (`--fresh-members 2
/// --fresh-member-gib 1`) and the twin's log deliberately made bigger than it
/// (`--restore-fill-mib`), and the scenario then proves the refusal came BEFORE the first byte —
/// the box's bootstrap door is still OPEN, so no identity document of the twin's ever landed.
///
/// **If it cannot create the situation it is about, it says UNRUN naming the numbers** — a
/// scenario that passed because the set turned out big enough would be the worst of the three
/// possible answers.
pub fn t3(o: &Opts) -> Verdict {
let (id, name) = ("T3", "restore onto FEWER, SMALLER devices (4 → 2): too small refuses by name before the first byte, never a silent short restore");
// ONE medium: the twin installs the primary's own `--iso` and is made a twin by `twin join`.
let twin_iso = o.iso.clone();
let fresh_members = o.fresh_members.unwrap_or(2);
let fresh_member_gib = o.fresh_member_gib.unwrap_or(1);
let work = o.work.join(id);
let _ = std::fs::create_dir_all(&work);
let spec = |sub: &str, iso: PathBuf, boot_gib: u64, members: usize, gib: u64| RigSpec {
iso,
work: work.join(sub),
boot_gib,
member_gib: gib,
members,
mem_mb: o.mem_mb,
smp: o.smp,
settle_secs: o.settle_secs.min(10),
explicit_blockdevs: false,
};
let run = || -> Result<Vec<Verdict>, String> {
let mut primary = Rig::create(spec(
"primary",
o.iso.clone(),
o.boot_gib,
o.members.max(4),
o.member_gib,
))?;
primary.hold = Duration::from_secs(4 * 3600);
// The system disk is larger than the volume, and stays small enough to earn no P3:
// see the note in `drill`.
if o.twin_member_gib >= 10 {
return Err(format!("the twin's volume ({} GiB) must be under its 10 GiB system disk; see the note in `drill`", o.twin_member_gib));
}
let mut twin = Rig::create(spec("twin", twin_iso.clone(), 10, 1, o.twin_member_gib))?;
twin.hold = Duration::from_secs(4 * 3600);
twin.grpc_only_ready = true;
eprintln!("── T3: primary {} ", primary.describe());
eprintln!("── T3: twin {} ", twin.describe());
eprintln!("── T3: the restore will land on {fresh_members} × {fresh_member_gib} GiB, and the log will be made bigger than that");
primary.install()?;
twin.install()?;
let mut sess = Session::new(&work, &o.cli)?;
let repos = seed_repos(&sess.trees, 0x7317_0001)?;
// The fill: enough log that the small set cannot hold twice it plus the box's reserve.
let fill_each = (o.restore_fill_mib / 4).max(1);
let fill: Vec<RepoManifest> = (0..4)
.map(|i| {
data::make_repo(
&sess.trees,
&format!("fill-{i}"),
&[("f.bin", fill_each << 20)],
0x7317_1000 + i,
)
})
.collect::<Result<_, _>>()?;
let cli = o.cli.clone();
let admin = sess.key.clone();
let (to_twin, twin_rx) = mpsc::channel::<ToTwin>();
let (twin_tx, from_twin) = mpsc::channel::<FromTwin>();
std::thread::scope(|scope| -> Result<Vec<Verdict>, String> {
let twin_cli = cli.clone();
let twin_admin = admin.clone();
let twin_thread = scope.spawn(move || {
let b = twin.boot("twin", |tp| {
let r = twin_drive(&twin_cli, &twin_admin, tp, &twin_rx, &twin_tx);
if let Err(e) = &r {
let _ = twin_tx.send(FromTwin::Failed(e.clone()));
}
r
});
(twin, b)
});
let mut tports: Option<Ports> = None;
let mut pinned: Option<Ports> = None;
let mut log_bytes = 0u64;
let mut follow = Err::<String, String>("not reached".into());
let first = primary.boot("primary", |p| {
pinned = Some(p);
let at = format!("127.0.0.1:{}", p.grpc_private);
let boot = sess.bootstrap(p)?;
let pushed = sess.push_all(p, repos.clone())?;
let filled = sess.push_all(p, fill.clone())?;
let _ = to_twin.send(ToTwin::PrimaryAt(p));
let (key, tp) = match recv(&from_twin)? {
FromTwin::Joined { key, ports } => (key, ports),
_ => return Err("the twin did not join".into()),
};
tports = Some(tp);
let k = admin.display().to_string();
data::cli(
&cli,
&[
"control", "user", "add", "twin-box", "--key", &key, "--kind", "machine",
"-i", &k, "--at", &at,
],
)?;
data::cli(
&cli,
&["control", "group", "add", "twin", "-i", &k, "--at", &at],
)?;
data::cli(
&cli,
&[
"control", "group", "member", "add", "twin", "twin-box", "-i", &k, "--at",
&at,
],
)?;
let h = head_of(&cli, &admin, &at)?;
let _ = to_twin.send(ToTwin::Sync(h));
let s = match recv(&from_twin)? {
FromTwin::Synced(s) => s,
_ => return Err("no sync".into()),
};
log_bytes = field(&twin_status(&cli, &admin, &at)?, "bytes")
.and_then(|v| v.parse().ok())
.unwrap_or(0);
follow = Ok(format!("head {h}: {s}"));
Ok(format!(
"{boot}; pushed {pushed}; fill {filled}; log {} B",
log_bytes
))
});
print_boot(&first);
if let Some(e) = boot_failed(&first) {
let _ = to_twin.send(ToTwin::Stop);
let _ = twin_thread.join();
return Err(format!("phase 1: {e}"));
}
let mut out = vec![v(
"T3a",
"the twin follows the primary to the watermark",
follow,
)];
let (tp, p) = (
tports.ok_or("no twin ports")?,
pinned.ok_or("no primary ports")?,
);
primary.destroy_disks();
eprintln!(" primary: every disk image DESTROYED");
let mut fresh = Rig::create(spec(
"primary-fresh",
o.iso.clone(),
o.boot_gib,
fresh_members,
fresh_member_gib,
))?;
fresh.pin = Some(p);
fresh.hold = Duration::from_secs(4 * 3600);
let laid = fresh.install()?;
// Does this run actually create the situation? A set that HOLDS the restore makes
// every verdict below a claim about nothing, so the scenario says so and stops.
let set_bytes = fresh_members as u64 * fresh_member_gib * (1 << 30);
let need = log_bytes * 2 + (set_bytes / 20).max(256 << 20);
let situation = format!(
"the set LEFT {} × {} GiB and LANDS on {fresh_members} × {fresh_member_gib} GiB ({} raw); \
the twin holds {} of log, which a restore needs about {} for (twice the log plus the box's reserve)",
o.members.max(4),
o.member_gib,
crate::rig::human(set_bytes),
crate::rig::human(log_bytes),
crate::rig::human(need)
);
if log_bytes == 0 || need <= set_bytes {
let _ = to_twin.send(ToTwin::Stop);
let (_twin, tb) = twin_thread
.join()
.map_err(|_| "the twin thread panicked".to_string())?;
print_boot(&tb);
out.push(Verdict {
id: "T3b",
name: "a restore onto a set too small for it refuses by name",
status: Status::Unrun,
detail: format!(
"this run could not create the situation: {situation} — which FITS. Raise the log with `--restore-fill-mib N` or shrink the set with `--fresh-member-gib M`. Never claimed green on a set that was big enough."
),
});
return Ok(out);
}
let mut refused = Err::<String, String>("not reached".into());
let mut untouched = Err::<String, String>("not reached".into());
let replay = fresh.boot("fresh-too-small", |m| {
let st = sess.status_line(m)?;
if !st.contains("OPEN") {
return Err(format!("the fresh primary is not fresh: {st}"));
}
let answer = data::cli(&cli, &["control", "twin", "replay", "--twin", &format!("127.0.0.1:{}", tp.grpc_private), "--into", &format!("127.0.0.1:{}", m.grpc_private), "-i", &admin.display().to_string()]);
refused = match &answer {
Err(e) if e.contains("NOT ENOUGH ROOM") && e.contains("SHORT BY") => Ok(format!("{situation}. The box refused: {}", one_line(e))),
Err(e) => Err(format!(
"the replay failed, but NOT with the room refusal this scenario is about ({situation}). A medium composed BEFORE `LogChunk.restore_bytes` cannot refuse — it has nothing to check against — so it fills the volume and dies mid-apply, which is this row's whole point. Compose the medium from a build that carries the check:\n{}",
one_line(e)
)),
Ok(s) => Err(format!(
"the replay SUCCEEDED onto a set that cannot hold it ({situation}) — the silent short restore this scenario exists to catch. A medium composed BEFORE `LogChunk.restore_bytes` has nothing to check against; compose one from a build that carries the check:\n{}",
one_line(s)
)),
};
// Before the first byte: a single ControlDoc of the twin's would have written the
// identity table and shut the bootstrap door. It is still open.
let after = sess.status_line(m)?;
untouched = if after.contains("OPEN") {
Ok(format!("the bootstrap door is still OPEN after the refusal ({after}): no identity document of the twin's landed, so the refusal came before the first byte"))
} else {
Err(format!("the door is no longer open after a REFUSED restore ({after}): something of the twin's log landed before the box said no"))
};
Ok("refused".into())
});
print_boot(&replay);
if let Some(e) = boot_failed(&replay) {
refused = Err(format!("the fresh primary never came up: {e}"));
}
let _ = to_twin.send(ToTwin::Stop);
let (twin, tb) = twin_thread
.join()
.map_err(|_| "the twin thread panicked".to_string())?;
print_boot(&tb);
out.push(v(
"T3b",
"a restore onto a set too small for it refuses by name, saying how much is short",
refused,
));
out.push(v(
"T3c",
"the refusal comes BEFORE the first byte lands",
untouched,
));
out.push(v(
"T3d",
"the installer laid the small set for the machine it was on",
if laid.member_names.len() == fresh_members {
Ok(format!(
"{fresh_members} members laid ({}): {}",
laid.member_names.join(", "),
one_line(&laid.lines.join("\n"))
))
} else {
Err(format!(
"{fresh_members} devices, {} members laid: {:?}",
laid.member_names.len(),
laid.member_names
))
},
));
if !o.keep {
fresh.destroy_disks();
twin.destroy_disks();
}
Ok(out)
})
};
match run() {
Ok(lines) => {
for l in &lines {
eprintln!("{}", l.line());
}
let failed: Vec<&str> = lines
.iter()
.filter(|l| matches!(l.status, Status::Fail))
.map(|l| l.id)
.collect();
let unrun = lines
.iter()
.all(|l| matches!(l.status, Status::Unrun) || l.id == "T3a");
let detail = lines
.iter()
.map(|l| {
format!(
"{} {}",
l.id,
match l.status {
Status::Pass => "PASS",
Status::Fail => "FAIL",
Status::Unrun => "UNRUN",
}
)
})
.collect::<Vec<_>>()
.join(", ");
let status = if !failed.is_empty() {
Status::Fail
} else if unrun {
Status::Unrun
} else {
Status::Pass
};
Verdict {
id,
name,
status,
detail,
}
}
Err(e) => Verdict {
id,
name,
status: Status::Fail,
detail: e,
},
}
}
#[cfg(test)]
mod audit_cell_tests {
use super::*;
/// What the CLI printed, in the order it prints it (`gunnar-cli`'s `audit_trail`).
fn printed(entries: u64, tip: &str, tip_seq: u64, at_seq: u64, hash_at: &str) -> String {
format!(
"audit trail: {entries} entries\n tip: {tip}\n tip seq: {tip_seq}\n verifies: yes\n replicated: yes — every entry is a transaction\n hash at seq {at_seq}: {hash_at}\n"
)
}
/// **`(not held)` is not a hash** (lane T11, 2026-09-21 — T3's KVM run of `ff2c7c9`).
///
/// The row read "DIFFERS at seq 40 … the restored box holds: (not held)": the parser had
/// taken the CLI's words for a hash, and the verdict then compared two "hashes". The honest
/// sentence — the restored box holds NO entry at the primary's last sequence, and how many
/// it does hold — was never printed, and it is the one that pointed at the real defect.
///
/// **Fails before the fix**: delete the `if h.starts_with('(')` guard in
/// `parse_audit_shape` and the verdict says "DIFFERS" again.
#[test]
fn not_held_is_read_as_no_entry_and_the_verdict_says_so() {
let tip = "e4d6ff67e20ef24b246304d8f006cda86f55c8f2e6eda9389777222e97c1c1e6";
let before = parse_audit_shape(&printed(41, tip, 40, 0, ""), 0).unwrap();
// The measured KVM answer: the restored box held only its own 16 decisions.
let after = parse_audit_shape(&printed(16, "f0", 15, 40, "(not held)"), 40).unwrap();
assert_eq!(
after.hash_at_seq, "",
"the CLI's `(not held)` is the absence of a hash"
);
let v = audit_verdict("T1v", &Ok(before.clone()), &Ok(after));
assert!(matches!(v.status, Status::Fail));
assert!(
v.detail.contains("holds no entry at seq 40") && v.detail.contains("16 entries"),
"{}",
v.detail
);
assert!(!v.detail.contains("DIFFERS"), "{}", v.detail);
// And the PASS it must reach when the trail came back and the box continued it.
let whole = parse_audit_shape(&printed(57, "aa", 56, 40, tip), 40).unwrap();
assert!(matches!(
audit_verdict("T1v", &Ok(before), &Ok(whole)).status,
Status::Pass
));
}
/// **A restored box answering from its own per-box trail is named as such** — the real cause
/// of T3's `ff2c7c9` failure: the box reopened still mid-restore and read its `.local` trail.
///
/// **Fails before the fix**: delete the `if !a.replicated` block in `audit_verdict` and the
/// row blames the trail ("holds no entry at seq 40") instead of the box's state.
#[test]
fn a_restored_box_still_mid_restore_is_named_not_blamed_on_the_trail() {
let tip = "e4d6ff67e20ef24b246304d8f006cda86f55c8f2e6eda9389777222e97c1c1e6";
let before = parse_audit_shape(&printed(41, tip, 40, 0, ""), 0).unwrap();
let own = printed(16, "f0", 15, 40, "(not held)").replace(
"replicated: yes — every entry is a transaction",
"replicated: no — this box's trail is written here only",
);
let after = parse_audit_shape(&own, 40).unwrap();
assert!(!after.replicated);
let v = audit_verdict("T1v", &Ok(before), &Ok(after));
assert!(matches!(v.status, Status::Fail));
assert!(
v.detail.contains("OWN per-box trail") && v.detail.contains("mid-restore"),
"{}",
v.detail
);
}
}
// ── TE: the disk election, on the twin's own shape, from the ONE medium ──────
/// **TE — does the installer lay a BOOTABLE box on the twin's asymmetric shape with no
/// `korp.data.disks=` token?**
///
/// The owner's ruling of 2026-09-21 is that the WEAKEST machine becomes the twin, elected by
/// measurement after install. The twin is therefore permanently the differently-shaped box, and
/// the one medium carries nothing per-box — which puts the whole weight on
/// `korp_installer::layout`'s discovery.
///
/// The failure this reproduces (MEASURED 2026-09-17, commit `81b8b3a`): with one small system
/// disk and one LARGER data volume there is no equal-size group, so the election falls through
/// to class-then-LARGEST, the volume wins `/data`… and the ESP lands on it, leaving the disk the
/// firmware boots with no boot half at all — `No bootable device`.
///
/// The shape is `--twin-boot-gib` × `--twin-members` × `--twin-member-gib`, so the SAME scenario
/// states both halves: a system disk LARGER than the volume boots, a smaller one is the defect.
///
/// The evidence is the ESP's own `BOOTLOG.TXT`: the installer phase brings no network up, so not
/// one line of it ever reaches the `korp.log_udp` sink. And a `BOOTLOG.TXT` that cannot be read
/// at all is the loudest evidence of the defect — GPT slot 0 of the boot disk is then not an ESP.
pub fn te(o: &Opts) -> Verdict {
let id = "TE";
let name = "the disk election from the ONE medium, with no `korp.data.disks=` token";
let work = o.work.join(id);
let _ = std::fs::create_dir_all(&work);
let boot_gib = o.twin_boot_gib.unwrap_or(10);
let members = o.twin_members.unwrap_or(1);
let run = || -> Result<String, String> {
let mut rig = Rig::create(RigSpec {
iso: o.iso.clone(),
work: work.join("box"),
boot_gib,
member_gib: o.twin_member_gib,
members,
mem_mb: o.mem_mb,
smp: o.smp,
settle_secs: o.settle_secs.min(10),
explicit_blockdevs: false,
})?;
rig.hold = Duration::from_secs(600);
eprintln!("── {id}: {}", rig.describe());
let laid = rig.install()?;
eprintln!(
"── {id}: the installer laid: {}",
one_line(&laid.lines.join("\n"))
);
let esp = rig.esp_bootlog();
match &esp {
Ok(log) => {
let says: Vec<&str> = log
.lines()
.filter(|l| l.contains("data set") || l.contains("partition GPT"))
.collect();
eprintln!(
"── {id}: ESP BOOTLOG.TXT ({} lines):\n {}",
log.lines().count(),
says.join("\n ")
);
}
Err(e) => eprintln!("── {id}: ESP BOOTLOG.TXT: {e}"),
}
// The only honest test of "bootable" is a boot.
let b = rig.boot("elected", |_| Ok("the installed box answered".into()));
print_boot(&b);
let shape = format!(
"system {boot_gib} GiB + {members} × {} GiB data",
o.twin_member_gib
);
let words = match &esp {
Ok(log) => one_line(
&log.lines()
.filter(|l| l.contains("data set"))
.collect::<Vec<_>>()
.join(" "),
),
Err(e) => format!("no ESP on the boot disk: {e}"),
};
if !o.keep {
rig.destroy_disks();
}
match boot_failed(&b) {
None => Ok(format!("{shape}: BOOTS from the one medium with no korp.data.disks token — {words}")),
Some(e) => Err(format!(
"{shape}: the installed box does NOT come up — {e}. The election put the boot half somewhere the firmware does not look. Installer said: {words}"
)),
}
};
match run() {
Ok(detail) => Verdict {
id,
name,
status: Status::Pass,
detail,
},
Err(detail) => Verdict {
id,
name,
status: Status::Fail,
detail,
},
}
}
// ── the money class: T1d, TK1, TK2 ───────────────────────────────────────────
/// `gunnar control entitlement list`, one line per ENFORCED fact, sorted. An empty string is a
/// box enforcing nothing — every namespace Free.
pub(crate) fn entitlements(cli: &Path, admin: &Path, at: &str) -> Result<String, String> {
let out = data::cli(
cli,
&[
"control",
"entitlement",
"list",
"-i",
&admin.display().to_string(),
"--at",
at,
],
)?;
// The enforced facts only: the `REFUSED AT BOOT` line (what the box HOLDS and does not
// enforce) is read on its own by `refused_on_wire`.
let mut lines: Vec<&str> = out
.lines()
.map(str::trim)
.filter(|l| !l.is_empty() && !l.starts_with("REFUSED AT BOOT"))
.collect();
lines.sort_unstable();
Ok(lines.join("\n"))
}
/// The `REFUSED AT BOOT` line of `gunnar control entitlement list`, if the box carries one —
/// the refusal ON THE WIRE, readable on a machine with no serial console.
pub(crate) fn refused_on_wire(cli: &Path, admin: &Path, at: &str) -> Option<String> {
data::cli(
cli,
&[
"control",
"entitlement",
"list",
"-i",
&admin.display().to_string(),
"--at",
at,
],
)
.ok()?
.lines()
.find(|l| l.starts_with("REFUSED AT BOOT"))
.map(|l| l.trim().to_string())
}
/// **T1d — entitlements back.** Judged on the box that came back under the key that signed the
/// facts: the restore itself when the key was right, the right-key boot when TK planted a wrong
/// one first.
#[allow(clippy::too_many_arguments)]
fn money_verdicts(
id: &'static str,
trusts: &Result<String, String>,
before: &Result<String, String>,
after: &Result<String, String>,
planted: &Result<String, String>,
_said: Option<&str>,
fixed: &Result<String, String>,
wrong: bool,
) -> Verdict {
let name = "entitlements back";
if let Err(e) = trusts {
return Verdict {
id,
name,
status: Status::Unrun,
detail: e.clone(),
};
}
let before = match before {
Ok(b) if !b.is_empty() => b,
Ok(_) => {
return v(
id,
name,
Err("the box accepted the bench's facts and then listed none of them".into()),
)
}
Err(e) => {
return v(
id,
name,
Err(format!("the facts could not be put on the primary: {e}")),
)
}
};
let (label, got) = if wrong && planted.is_ok() {
("after the right key was given back", fixed)
} else {
("after the restore", after)
};
v(
id,
name,
match got {
Ok(g) if g == before => Ok(format!(
"{} fact(s) identical {label}: {}",
before.lines().count(),
one_line(before)
)),
Ok(g) => Err(format!("before:\n{before}\n{label}:\n{g}")),
Err(e) => Err(format!("{label}: {e}")),
},
)
}
/// **TK1 / TK2 — the restore handed the WRONG monetize key.** TK1: every fact is still on the
/// box, none is enforced, the reads say Free (not the stored plan), and the box's own boot line
/// says how many it refused. TK2: the right key, and nothing else, brings every plan back.
fn wrong_key_verdicts(
before: &Result<String, String>,
after: &Result<String, String>,
planted: &Result<String, String>,
in_effect: Option<&str>,
bench_pub: &str,
said: Option<&str>,
wire: Option<&str>,
fixed: &Result<String, String>,
) -> Vec<Verdict> {
let tk1 = "a restore under the wrong monetize key: every signed namespace served Free, the reads say so, and the boot line counts the refused";
let tk2 = "the remedy is a keep re-image from the right medium: every plan back with no fact re-pushed";
let (before, planted) = match (before, planted) {
(Ok(b), Ok(p)) if !b.is_empty() => (b, p),
(_, Err(e)) => {
return vec![
v(
"TK1",
tk1,
Err(format!("the wrong key was not planted: {e}")),
),
Verdict {
id: "TK2",
name: tk2,
status: Status::Unrun,
detail: "TK1 did not run".into(),
},
]
}
_ => {
return vec![
Verdict {
id: "TK1",
name: tk1,
status: Status::Unrun,
detail: "no facts were on the primary to lose".into(),
},
Verdict {
id: "TK2",
name: tk2,
status: Status::Unrun,
detail: "TK1 did not run".into(),
},
]
}
};
// ★ The guard this cell was missing on its first run: a verdict about a wrong key is only a
// verdict if the boot RAN with it. MEASURED 2026-09-21: a key planted on the ESP was
// overridden by the signed medium, the box booted with the bench key, and TK1 read that as
// "the box reports plans it cannot have verified" — a false red about a key that never took.
match in_effect {
Some(k) if k.eq_ignore_ascii_case(bench_pub) => {
let d = format!("the restored box booted with the BENCH key {}… (its MERGED command line), so the wrong key never took effect and nothing was measured", &k[..k.len().min(12)]);
return vec![
Verdict {
id: "TK1",
name: tk1,
status: Status::Unrun,
detail: d.clone(),
},
Verdict {
id: "TK2",
name: tk2,
status: Status::Unrun,
detail: d,
},
];
}
None => {
let d = "the restored boot's console carries no `KORP-BOOT: cmdline MERGED` monetize key, so which key it ran with is unknown".to_string();
return vec![
Verdict {
id: "TK1",
name: tk1,
status: Status::Unrun,
detail: d.clone(),
},
Verdict {
id: "TK2",
name: tk2,
status: Status::Unrun,
detail: d,
},
];
}
Some(_) => {}
}
let n = before.lines().count();
// The refusal must be READABLE, not only logged: on the UpCloud machine there is no console.
let wire_ok = wire.is_some_and(|w| w.contains(&format!("REFUSED AT BOOT {n} fact(s)")));
let first = match (after, said) {
(Ok(a), _) if a.is_empty() && !wire_ok => Err(format!("{planted}; every namespace Free but `entitlement list` carries no `REFUSED AT BOOT {n} fact(s)` line (got {wire:?}) — the refusal is not on the wire")),
(Ok(a), Some(line)) if a.is_empty() && line.contains(&format!("refused={n}")) => Ok(format!("{planted}; the restored box lists NO plan (all {n} namespace(s) Free); on the wire: {}; and its boot said: {line}", wire.unwrap_or(""))),
(Ok(a), Some(line)) if a.is_empty() => Err(format!("Free, and the boot line counts the wrong number (expected refused={n}): {line}")),
(Ok(a), None) if a.is_empty() => Err(format!("{planted}; every namespace Free and the box SAID NOTHING on its console — the silent downgrade")),
(Ok(a), _) => Err(format!("{planted}; the restored box REPORTS plans it cannot have verified — the stored facts, not the enforced ones:\n{a}")),
(Err(e), _) => Err(format!("{planted}; the restored box could not be asked: {e}")),
};
let second = match fixed {
Ok(f) if f == before => Ok(format!("{n} fact(s) enforced again after a keep re-image from the medium that trusts the key that signed them — nothing re-pushed")),
Ok(f) => Err(format!("before:\n{before}\nwith the right key:\n{f}")),
Err(e) => Err(e.clone()),
};
vec![v("TK1", tk1, first), v("TK2", tk2, second)]
}
/// The monetize key a boot REALLY ran with: the `korp.env.GUNNAR_MONETIZE_PUBKEY=` token on
/// korp-boot's `cmdline MERGED` line (the signed baked half wins over the ESP's), falling back
/// to the kernel's own `Command line:`.
fn money_key_in_effect(serial: &str) -> Option<String> {
let pick = |prefix: &str| {
serial
.lines()
.filter(|l| l.contains(prefix))
.find_map(crate::money::pubkey_in)
};
pick("KORP-BOOT: cmdline MERGED").or_else(|| pick("Command line:"))
}