crates/gunnar-growbench/src/todo_drill.rs
//! **TODO — one cell per open item** (the owner's order, 2026-09-21: "every open item gets a test
//! in the functional test bench"). Written RED first: each cell is run against today's code and
//! its FAIL is the measurement of the defect; it turns green when the fix lands, and only then.
//!
//! ```text
//! gunnar-growbench --scenarios TODO --iso M [--work W] [--cli GUNNAR]
//! [--mock-kvm PATH] [--mock PATH] [--grow-probe PATH]
//! [--tf-decl DIR[,DIR]] [--tf-decl-unfixed DIR[,DIR]] [--tf-providers DIR]
//! [--todo-silence-flag NAME] [--todo-silence-secs N] [--todo-tunnel-enrol ARGS]
//! ```
//!
//! | cell | the open item | where it runs |
//! |------|----------------------------------------------------------------------|--------------------------|
//! | TD1 | the twin follows the RESTORED primary — a FRESH poll, not a stale one | OVMF estate (shared) |
//! | TD2 | the audit trail after a crash: its lost tail rebuilt from the log | OVMF estate (shared) |
//! | TD3 | a long-silent twin: plain replay refuses naming the age; --stale-ok | OVMF estate (shared) |
//! | TD4 | install from the CD alone (no virtio copy of the medium) | OVMF rig (TD4) + mock (TD4M) |
//! | TD5 | two boxes from ONE medium get different WireGuard inner addresses | OVMF, two rigs |
//! | TD6 | a data-set growth meets 511 HOTPLUG_FAILED and falls back | mock-upcloud-kvm |
//! | TD7 | terraform after a growth: `No changes`, never 400 SIZE_INVALID | mock-upcloud + provider |
//! | TD8 | the whole twin install rehearsal | mock-upcloud-kvm |
//!
//! TD1–TD3 share ONE installed estate (a primary and a twin from the same medium), in the order
//! the situations happen: the primary crashes and loses its trail's tail (TD2), comes back, dies
//! for good and goes silent (TD3), is restored from the twin, and the twin must be seen following
//! the box that came back (TD1). Asking for any of the three runs the estate once.
//!
//! **No cell may pass because nothing was seeded.** Every PASS names what it measured; a cell that
//! could not create its situation is UNRUN and says which part is missing.
use std::collections::BTreeMap;
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;
use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH};
use korp_installer::isoproof::{self, INSTALL_OK};
use korp_installer::kvmproof::{self, Machine};
use super::twin_drill::{field, head_of, one_line, pubkey, twin_status, wait_twin};
use super::{boot_failed, print_boot, seed_repos, Opts, Session, Status, Verdict};
use crate::data;
use crate::rig::{tail, Ports, Rig, RigSpec, GIB};
// ── the cells' names ────────────────────────────────────────────────────────
pub const TD1_NAME: &str = "the twin follows the RESTORED primary: a SUCCESSFUL poll after that box booted, at a head it did not have before";
pub const TD2_NAME: &str = "the audit trail after a crash: the primary loses its trail's tail on /serialdat, the start rebuilds it from the log, and the twin's copy still verifies";
pub const TD3_NAME: &str = "a long-silent twin: plain `twin replay` refuses and names the age once the silence passes the threshold, `--stale-ok` accepts, and under the threshold plain replay accepts";
pub const TD4_NAME: &str = "install from the CD alone: the medium ONLY as a cdrom (AHCI on q35), no virtio copy — the installer must read it";
pub const TD4M_NAME: &str = "install from the CD alone on the UpCloud shape (mock-upcloud-kvm, SeaBIOS, -serial none): the medium only in the tray";
pub const TD5_NAME: &str = "WireGuard address from bootstrap: two boxes from ONE medium come up on the tunnel with DIFFERENT inner addresses, and their traffic goes over wg";
pub const TD6_NAME: &str = "hot-plug 511: a data-set growth on a guest without PCI hot-plug gets 511 HOTPLUG_FAILED and REFUSES untouched, naming it — never Unavailable, and never a restart (growth never restarts the appliance)";
pub const TD7_NAME: &str = "the terraform size lock: after the grow engine grew the data volumes, `terraform plan` with the variables unchanged says No changes (never 400 SIZE_INVALID)";
pub const TD8_NAME: &str = "the whole twin install rehearsal on the UpCloud shape: one medium, main + twin, weakest-is-twin, BecomeTwin, follow, the primary dies, replay into a new primary, every class checked";
// ── small shared helpers ────────────────────────────────────────────────────
/// A flag of this family, read off the process's own argv (the family keeps its knobs out of
/// `Opts` so the drills other people edit never have to learn them).
pub fn flag(name: &str) -> Option<String> {
let args: Vec<String> = std::env::args().collect();
args.iter()
.position(|a| a == name)
.and_then(|i| args.get(i + 1))
.cloned()
}
/// [`super::twin_drill::one_line`], re-exported for the sibling TODO modules.
pub fn one_line_pub(s: &str) -> String {
one_line(s)
}
pub fn now_ms() -> u64 {
SystemTime::now()
.duration_since(UNIX_EPOCH)
.map(|d| d.as_millis() as u64)
.unwrap_or(0)
}
pub fn verdict(id: &'static str, name: &'static str, r: Result<String, String>) -> Verdict {
match r {
Ok(detail) => Verdict {
id,
name,
status: Status::Pass,
detail,
},
Err(detail) => Verdict {
id,
name,
status: Status::Fail,
detail,
},
}
}
pub fn unrun(id: &'static str, name: &'static str, why: impl Into<String>) -> Verdict {
Verdict {
id,
name,
status: Status::Unrun,
detail: why.into(),
}
}
/// The CLI with BOTH streams kept: the replay guard speaks on stderr, and a verdict about what it
/// said must read it whether the verb succeeded or not.
pub fn cli_all(cli: &Path, args: &[&str]) -> (bool, String, String) {
match Command::new(cli)
.args(args)
.env("GUNNAR_LOG", "warn")
.output()
{
Ok(o) => (
o.status.success(),
String::from_utf8_lossy(&o.stdout).into_owned(),
String::from_utf8_lossy(&o.stderr).into_owned(),
),
Err(e) => (
false,
String::new(),
format!("{} {}: {e}", cli.display(), args.join(" ")),
),
}
}
/// `its last answer, N s ago` out of the `as_of` line the CLI prints for a twin.
pub fn answer_age_s(status: &str) -> Option<u64> {
let l = status
.lines()
.find(|l| l.trim_start().starts_with("as_of"))?;
// Two wordings are in the field: T4's (`… at its last answer, N s ago; …`) and the older
// (`the primary last answered N s ago (unix ms …)`). Both are read; neither is guessed at.
["its last answer, ", "the primary last answered "]
.iter()
.find_map(|w| {
let i = l.find(w)? + w.len();
l[i..].split_whitespace().next()?.parse().ok()
})
}
/// A log file of this family's own, next to the work dir: every cell's detail and the lines it
/// quoted, so a verdict can be read back without the console that printed it.
pub fn log_line(work: &Path, cell: &str, line: &str) {
use std::io::Write;
let p = work.join("TODO").join(format!("{cell}.log"));
let _ = std::fs::create_dir_all(p.parent().unwrap());
if let Ok(mut f) = std::fs::OpenOptions::new()
.create(true)
.append(true)
.open(&p)
{
let _ = writeln!(f, "{} {line}", now_ms());
}
}
// ── the audit trail as `gunnar control audit` prints it ─────────────────────
#[derive(Debug, Clone, Default)]
pub struct Trail {
pub present: bool,
pub entries: u64,
pub tip: String,
pub tip_seq: u64,
pub verified: bool,
/// Parsed for the log line's sake (`raw`); no verdict reads it yet.
#[allow(dead_code)]
pub replicated: bool,
pub hash_at_seq: String,
pub raw: String,
}
pub fn trail(cli: &Path, key: &Path, at: &str, at_seq: u64) -> Result<Trail, 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)?;
let pick = |name: &str| {
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(Trail {
raw: one_line(&out),
..Trail::default()
});
}
let entries = out
.lines()
.find_map(|l| l.trim().strip_prefix("audit trail:"))
.and_then(|v| v.split_whitespace().next().and_then(|n| n.parse().ok()))
.ok_or_else(|| format!("`control audit` printed no entry count: {}", one_line(&out)))?;
Ok(Trail {
present: true,
entries,
tip: pick("tip:"),
tip_seq: pick("tip seq:").parse().unwrap_or(0),
verified: pick("verifies:").starts_with("yes"),
replicated: pick("replicated:").starts_with("yes"),
hash_at_seq: pick(&format!("hash at seq {at_seq}:")),
raw: one_line(&out),
})
}
// ── the boot disk's P3, read and cut from the HOST ──────────────────────────
/// Byte range of GPT entry `index` on a raw disk image.
pub fn gpt_range(img: &Path, index: u64) -> Result<(u64, u64), String> {
let f = std::fs::File::open(img).map_err(|e| format!("{}: {e}", img.display()))?;
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(format!("{}: no GPT", img.display()));
}
let u64_at = |b: &[u8], o: usize| u64::from_le_bytes(b[o..o + 8].try_into().unwrap());
let esz = u32::from_le_bytes(hdr[84..88].try_into().unwrap()) as u64;
let mut ent = vec![0u8; esz as usize];
f.read_exact_at(&mut ent, u64_at(&hdr, 72) * 512 + index * esz)
.map_err(|e| e.to_string())?;
let (first, last) = (u64_at(&ent, 32), u64_at(&ent, 40));
if first == 0 {
return Err(format!(
"{}: GPT entry {} is empty",
img.display(),
index + 1
));
}
Ok((first * 512, (last + 1) * 512))
}
/// Copy `[start, end)` of `src` into `dst` at `dst_off + (pos - start)`, data regions only.
pub fn copy_sparse(
src: &Path,
start: u64,
end: u64,
dst: &Path,
dst_off: u64,
create: bool,
) -> Result<u64, String> {
let f = std::fs::File::open(src).map_err(|e| format!("{}: {e}", src.display()))?;
let out = std::fs::OpenOptions::new()
.write(true)
.create(create)
.truncate(create)
.open(dst)
.map_err(|e| format!("{}: {e}", dst.display()))?;
if create {
out.set_len(end - start).map_err(|e| e.to_string())?;
}
let fd = f.as_raw_fd();
let (mut pos, mut copied) = (start, 0u64);
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], dst_off + (o - start))
.map_err(|e| e.to_string())?;
o += n as u64;
copied += n as u64;
}
pos = h;
}
out.sync_all().map_err(|e| e.to_string())?;
Ok(copied)
}
fn xfs_db(img: &Path, expert: bool, cmds: &[&str]) -> Result<String, String> {
let mut c = Command::new("/usr/sbin/xfs_db");
c.env("LC_ALL", "C")
.arg(if expert { "-x" } else { "-r" })
.arg("-f")
.arg(img);
for x in cmds {
c.args(["-c", x]);
}
let o = c.output().map_err(|e| format!("xfs_db: {e}"))?;
let (so, se) = (
String::from_utf8_lossy(&o.stdout).into_owned(),
String::from_utf8_lossy(&o.stderr).into_owned(),
);
if !o.status.success() || se.to_ascii_lowercase().contains("error") {
return Err(format!(
"xfs_db {cmds:?}: {} {}",
one_line(&so),
one_line(&se)
));
}
Ok(so)
}
fn xfs_num(out: &str, key: &str) -> Option<u64> {
out.lines().find_map(|l| {
l.trim()
.strip_prefix(key)
.and_then(|r| r.trim().strip_prefix('='))
.and_then(|v| v.trim().parse().ok())
})
}
/// Read one regular file's bytes out of an xfs image by its extent map.
fn xfs_read(img: &Path, path: &str) -> Result<Vec<u8>, String> {
let sb = xfs_db(img, false, &["sb 0", "print blocksize", "print agblocks"])?;
let (bs, agb) = (
xfs_num(&sb, "blocksize").ok_or("no blocksize")?,
xfs_num(&sb, "agblocks").ok_or("no agblocks")?,
);
let size = xfs_num(
&xfs_db(img, false, &[&format!("path {path}"), "print core.size"])?,
"core.size",
)
.ok_or("no core.size")?;
let bmap = xfs_db(img, false, &[&format!("path {path}"), "bmap"])?;
let f = std::fs::File::open(img).map_err(|e| e.to_string())?;
let mut out = vec![0u8; size as usize];
for l in bmap.lines() {
// `data offset 0 startblock 24 (0/24) count 3 flag 0`
let t: Vec<&str> = l.split_whitespace().collect();
let (Some(off), Some(ag), Some(cnt)) = (
t.iter()
.position(|x| *x == "offset")
.and_then(|i| t.get(i + 1))
.and_then(|v| v.parse::<u64>().ok()),
t.iter()
.find(|x| x.starts_with('('))
.map(|x| x.trim_matches(|c| c == '(' || c == ')').to_string()),
t.iter()
.position(|x| *x == "count")
.and_then(|i| t.get(i + 1))
.and_then(|v| v.parse::<u64>().ok()),
) else {
continue;
};
let (agno, agbno) = ag
.split_once('/')
.and_then(|(a, b)| Some((a.parse::<u64>().ok()?, b.parse::<u64>().ok()?)))
.ok_or_else(|| format!("bmap line: {l}"))?;
let phys = (agno * agb + agbno) * bs;
for b in 0..cnt {
let fo = (off + b) * bs;
if fo >= size {
break;
}
let n = bs.min(size - fo) as usize;
f.read_exact_at(&mut out[fo as usize..fo as usize + n], phys + b * bs)
.map_err(|e| e.to_string())?;
}
}
Ok(out)
}
/// **The crash's loss, made on purpose.** On the STOPPED primary's boot disk: the P3 (`/serialdat`,
/// xfs `KORP_DATA`) is copied out sparsely, `log/audit-chain.jsonl` is cut in the middle of its
/// third-from-last line (the last whole entries gone and a torn one left, which is what a lost
/// page-cache tail looks like), the xfs log is formatted clean so no replay can put the bytes back,
/// and the region is written back. Returns what was cut, as numbers.
pub fn cut_trail_tail(boot_img: &Path, work: &Path) -> Result<String, String> {
const TRAIL: &str = "/log/audit-chain.jsonl";
let (start, end) = gpt_range(boot_img, 2)?;
let p3 = work.join("p3-serialdat.img");
let copied = copy_sparse(boot_img, start, end, &p3, 0, true)?;
let mut magic = [0u8; 4];
std::fs::File::open(&p3)
.and_then(|f| f.read_exact_at(&mut magic, 0))
.map_err(|e| e.to_string())?;
if &magic != b"XFSB" {
return Err("the boot disk's P3 is not xfs: there is no /serialdat partition on this primary to lose a tail from".into());
}
// **The log first.** A mounted xfs killed hard has a dirty log, and replaying it at the next
// mount writes the logged inode core back — the size this cut shortens included. MEASURED
// 2026-09-21 on loki (run2): with the log left dirty the trail came back whole on start and
// the cell read green on nothing. `xfs_db logformat` refuses a dirty log ("Please mount to
// replay"), so the log is ZEROED with `xfs_repair -L` — which is itself what a crash that lost
// its log looks like — and only then is the size cut.
let lf = {
let o = Command::new("/usr/sbin/xfs_repair")
.env("LC_ALL", "C")
.args(["-L", "-f"])
.arg(&p3)
.output()
.map_err(|e| format!("xfs_repair: {e}"))?;
let said = format!(
"{}{}",
String::from_utf8_lossy(&o.stdout),
String::from_utf8_lossy(&o.stderr)
);
if !o.status.success() {
return Err(format!(
"xfs_repair -L on the P3 copy failed: {}",
one_line(&said)
));
}
let alert: Vec<&str> = said
.lines()
.filter(|l| l.contains("ALERT") || l.contains("zeroing") || l.contains("log"))
.take(3)
.collect();
format!(
"log zeroed by xfs_repair -L: {}",
if alert.is_empty() {
"the log was already clean".to_string()
} else {
alert.join(" / ")
}
)
};
let before = xfs_read(&p3, TRAIL)?;
let lines: Vec<usize> = before
.iter()
.enumerate()
.filter(|(_, b)| **b == b'\n')
.map(|(i, _)| i)
.collect();
if lines.len() < 6 {
return Err(format!(
"the trail holds {} lines — too few to lose a tail from and still name what was lost",
lines.len()
));
}
// Keep everything up to the end of line n-4; then half of line n-3.
let keep_whole = lines[lines.len() - 4] + 1;
let torn = (lines[lines.len() - 3] - keep_whole) / 2;
let new_size = (keep_whole + torn) as u64;
xfs_db(
&p3,
true,
&[
&format!("path {TRAIL}"),
&format!("write core.size {new_size}"),
],
)?;
let after = xfs_read(&p3, TRAIL)?;
if after.len() as u64 != new_size {
return Err(format!(
"the cut did not take: the trail reads {} bytes after writing {new_size}",
after.len()
));
}
let wrote = copy_sparse(&p3, 0, end - start, boot_img, start, false)?;
let _ = std::fs::remove_file(&p3);
Ok(format!(
"trail {} B / {} lines → {} B: the last 3 entries lost and a torn half-line of {torn} B left (P3 at {start}..{end}, {copied} B copied out, {wrote} B written back; {})",
before.len(),
lines.len(),
new_size,
one_line(&lf)
))
}
// ── TD1–TD3: the shared OVMF estate ─────────────────────────────────────────
enum ToTwin {
Stop,
}
/// Runs the estate once and answers TD1, TD2 and TD3.
pub fn estate(o: &Opts) -> Vec<Verdict> {
let work = o.work.join("TODO-estate");
let _ = std::fs::create_dir_all(&work);
let mut out = BTreeMap::<&'static str, Verdict>::new();
let r = estate_run(o, &work, &mut out);
for (id, name) in [("TD1", TD1_NAME), ("TD2", TD2_NAME), ("TD3", TD3_NAME)] {
if !out.contains_key(id) {
let why = match &r {
Err(e) => format!("the estate stopped before this cell was reached: {e}"),
Ok(()) => "the estate ended without answering this cell (a bench defect)".into(),
};
out.insert(
id,
Verdict {
id,
name,
status: Status::Fail,
detail: why,
},
);
}
}
let mut v: Vec<Verdict> = ["TD1", "TD2", "TD3"]
.iter()
.filter_map(|k| out.remove(k))
.collect();
for x in &mut v {
x.detail = format!("OVMF/q35 (virtio disks) · {}", x.detail);
}
for x in &v {
log_line(&o.work, x.id, &x.line());
}
v
}
fn estate_run(
o: &Opts,
work: &Path,
out: &mut BTreeMap<&'static str, Verdict>,
) -> Result<(), String> {
let spec = |sub: &str, boot_gib: u64, members: usize, gib: u64| RigSpec {
iso: o.iso.clone(),
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,
};
// The primary's boot disk is big enough to earn a P3, so its trail lives on /serialdat — the
// partition TD2 cuts. The twin is the drill's shape: one medium, one volume.
let mut primary = Rig::create(spec("primary", 24, 4, 6))?;
primary.hold = Duration::from_secs(4 * 3600);
// A medium with T9's pairing (`twin certify` in the CLI) serves the twin over mTLS on :50052.
let pairing = cli_all(&o.cli, &["control", "twin", "certify", "--help"]).0;
primary.mtls_forward = pairing;
// Only the PRIMARY's door is dialled: the twin must not take host 50052 from it.
let mut twin = Rig::create(spec("twin", 12, 1, 8))?;
twin.hold = Duration::from_secs(4 * 3600);
twin.grpc_only_ready = true;
eprintln!("── TODO estate: primary {}", primary.describe());
eprintln!("── TODO estate: twin {}", twin.describe());
let laid = primary.install()?;
if laid.boot_p3 != "korp-data" {
return Err(format!("the primary's boot disk carries no P3 (`{}`): there is no /serialdat to lose a tail from", laid.boot_p3));
}
twin.install()?;
let mut sess = Session::new(work, &o.cli)?;
let cli = o.cli.clone();
let admin = sess.key.clone();
let k = admin.display().to_string();
let repos = seed_repos(&sess.trees, 0x7d0d_0001)?;
let (twin_ports_tx, twin_ports_rx) = mpsc::channel::<Ports>();
let (stop_tx, stop_rx) = mpsc::channel::<ToTwin>();
std::thread::scope(|scope| -> Result<(), String> {
let tw = scope.spawn(move || {
let b = twin.boot("twin", |tp| {
let _ = twin_ports_tx.send(tp);
match stop_rx.recv_timeout(Duration::from_secs(4 * 3600)) {
Ok(ToTwin::Stop) | Err(_) => Ok("stopped".into()),
}
});
(twin, b)
});
let stop_twin = |why: &str| {
eprintln!(" twin: stopping ({why})");
let _ = stop_tx.send(ToTwin::Stop);
};
let tp = match twin_ports_rx.recv_timeout(Duration::from_secs(1800)) {
Ok(p) => p,
Err(_) => {
stop_twin("never ready");
let (_, b) = tw.join().map_err(|_| "twin thread panicked")?;
print_boot(&b);
return Err(format!(
"the twin never became ready: {}",
boot_failed(&b).unwrap_or_default()
));
}
};
let twin_at = format!("127.0.0.1:{}", tp.grpc_private);
// ── phase 1: the primary takes data, the twin joins and follows ──────
let mut before: Result<Trail, String> = Err("not reached".into());
let mut pinned: Option<Ports> = None;
primary.crash_after_drive = true;
let first = primary.boot("primary", |p| {
pinned = Some(p);
let at = format!("127.0.0.1:{}", p.grpc_private);
sess.bootstrap(p)?;
sess.push_all(p, repos.clone())?;
let key_line = data::cli(
&cli,
&[
"control",
"twin",
"join",
"--at",
&twin_at,
"--primary",
&format!("10.0.2.2:{}", p.grpc_private),
],
)?
.trim()
.to_string();
if !key_line.starts_with("ssh-") {
return Err(format!("twin join answered no key: {key_line}"));
}
data::cli(
&cli,
&[
"control", "user", "add", "twin-box", "--key", &key_line, "--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,
],
)?;
if pairing {
// The twin makes its key on the box, the primary signs it, the twin pins the issuer.
let c = data::cli(
&cli,
&[
"control",
"twin",
"certify",
"--at",
&twin_at,
"--primary-at",
&at,
"-i",
&k,
],
)
.map_err(|e| format!("twin certify: {}", one_line(&e)))?;
eprintln!(" twin certify: {}", one_line(&c));
log_line(&o.work, "TD1", &format!("twin certify: {}", one_line(&c)));
}
// Decisions for the trail to hold, each an audit entry.
for g in ["td-a", "td-b", "td-c"] {
data::cli(&cli, &["control", "group", "add", g, "-i", &k, "--at", &at])?;
}
data::cli(
&cli,
&[
"control",
"namespace",
"create",
"td-ns",
"-i",
&k,
"--at",
&at,
],
)?;
let h = head_of(&cli, &admin, &at)?;
wait_twin(
&cli,
&admin,
&twin_at,
&format!("applied >= {h}, caught up"),
|s| {
field(s, "applied")
.and_then(|v| v.parse::<u64>().ok())
.is_some_and(|a| a >= h)
&& field(s, "caught_up").as_deref() == Some("true")
},
)?;
before = trail(&cli, &admin, &at, 0);
// Let the guest's writeback and the xfs log settle, so the cut below is the ONLY loss.
eprintln!(
" primary: level with the twin at head {h}; settling 60 s, then a HARD STOP"
);
std::thread::sleep(Duration::from_secs(60));
Ok(format!("head {h}"))
});
print_boot(&first);
if let Some(e) = boot_failed(&first).or_else(|| match &first.driven {
Some(Err(e)) => Some(e.clone()),
None => Some("the primary never became ready".into()),
_ => None,
}) {
stop_twin("phase 1 failed");
let _ = tw.join();
return Err(format!("phase 1: {e}"));
}
let p = pinned.ok_or("no primary ports")?;
// The twin follows a fixed address: the box that comes back answers where it did.
primary.pin = Some(p);
let at = format!("127.0.0.1:{}", p.grpc_private);
// ── TD2: the crash lost the tail ────────────────────────────────────
let cut = match &before {
Ok(t) if t.present && t.verified && t.entries >= 6 => {
cut_trail_tail(&primary.boot, work)
}
Ok(t) => Err(format!(
"the primary had no trail worth cutting before the crash: {}",
t.raw
)),
Err(e) => Err(format!(
"the primary's trail could not be read before the crash: {e}"
)),
};
eprintln!(
" TD2 cut: {}",
match &cut {
Ok(s) | Err(s) => s,
}
);
let mut after_crash: Result<Trail, String> = Err("not reached".into());
let mut twin_after_crash: Result<String, String> = Err("not reached".into());
let mut last_primary_head = 0u64;
let tip_before = before.as_ref().map(|t| t.tip_seq).unwrap_or(0);
let second = primary.boot("after-crash", |_| {
after_crash = trail(&cli, &admin, &at, tip_before);
// Something new, so the twin must follow the box that came back.
data::cli(
&cli,
&[
"control",
"group",
"add",
"td-after-crash",
"-i",
&k,
"--at",
&at,
],
)?;
let h = head_of(&cli, &admin, &at)?;
twin_after_crash = wait_twin(
&cli,
&admin,
&twin_at,
&format!("applied >= {h}, caught up"),
|s| {
field(s, "applied")
.and_then(|v| v.parse::<u64>().ok())
.is_some_and(|a| a >= h)
&& field(s, "caught_up").as_deref() == Some("true")
},
);
last_primary_head = head_of(&cli, &admin, &at).unwrap_or(h);
Ok("ok".into())
});
print_boot(&second);
let audit_console: Vec<String> = second
.serial
.lines()
.filter(|l| l.to_ascii_lowercase().contains("audit"))
.take(8)
.map(|l| l.trim().to_string())
.collect();
let td2_start: Result<String, String> = (|| {
let cut = cut.clone()?;
if let Some(e) = boot_failed(&second) {
return Err(format!(
"after the cut ({cut}) the primary did not come back: {e}"
));
}
let b = before.clone()?;
let a = after_crash
.clone()
.map_err(|e| format!("after the cut ({cut}) the trail could not be read: {e}"))?;
if !a.present {
return Err(format!("after the cut ({cut}) the primary serves WITHOUT a durable trail: {} · console: {audit_console:?}", a.raw));
}
if !a.verified {
return Err(format!("after the cut ({cut}) the trail does NOT verify: {} · console: {audit_console:?}", a.raw));
}
if a.entries < b.entries || a.hash_at_seq != b.tip {
return Err(format!(
"after the cut ({cut}) the trail is not the one the primary had: {} entries (had {}), hash at seq {} is `{}` (was `{}`) · console: {audit_console:?}",
a.entries, b.entries, b.tip_seq, a.hash_at_seq, b.tip
));
}
Ok(format!("{cut}; on start the trail holds {} entries (had {}), verifies, and seq {} hashes to {} as before · console: {audit_console:?}", a.entries, b.entries, b.tip_seq, b.tip))
})();
// ── TD3 + TD1: the primary dies for good and a fresh one is restored ─
primary.destroy_disks();
let silent_since = Instant::now();
eprintln!(" primary: every disk image DESTROYED; the twin is on its own");
let mut fresh = Rig::create(spec("primary-fresh", 24, 4, 6))?;
fresh.pin = Some(p);
fresh.mtls_forward = pairing;
fresh.hold = Duration::from_secs(4 * 3600);
fresh.install()?;
let short_flag = flag("--todo-silence-flag").unwrap_or_else(|| "--max-silence".into());
let short_secs: u64 = flag("--todo-silence-secs")
.and_then(|v| v.parse().ok())
.unwrap_or(60);
let (mut refused, mut accepted_plain, mut accepted_override) = (
Err::<String, String>("not reached".into()),
Err::<String, String>("not reached".into()),
Err::<String, String>("not reached".into()),
);
let mut restore: Result<String, String> = Err("not reached".into());
let twin_before_restore = twin_status(&cli, &admin, &twin_at).unwrap_or_default();
let head_before = field(&twin_before_restore, "primary")
.and_then(|v| v.parse::<u64>().ok())
.unwrap_or(last_primary_head);
let replay = fresh.boot("fresh-restore", |m| {
let into = format!("127.0.0.1:{}", m.grpc_private);
if !sess.status_line(m)?.contains("OPEN") {
return Err("the fresh primary is not fresh".into());
}
// Past the short threshold, by the twin's own clock.
let t0 = Instant::now();
let mut st = twin_status(&cli, &admin, &twin_at).unwrap_or_default();
while answer_age_s(&st).unwrap_or(0) <= short_secs + 5 && t0.elapsed() < Duration::from_secs(short_secs + 300) {
std::thread::sleep(Duration::from_secs(5));
st = twin_status(&cli, &admin, &twin_at).unwrap_or_default();
}
let age = answer_age_s(&st);
// `--max-silence` takes a duration in words (`60s`, `24h`), the `gc --prune-grace` parser.
let secs = flag("--todo-silence-value").unwrap_or_else(|| format!("{short_secs}s"));
// (1) over the threshold, plain: must refuse and NAME the age — and write nothing.
let (ok, so, se) = cli_all(&cli, &["control", "twin", "replay", "--twin", &twin_at, "--into", &into, "-i", &k, &short_flag, &secs]);
refused = if se.contains("unexpected argument") || se.contains("unrecognized") || se.contains("wasn't expected") {
Err(format!("the CLI has no threshold knob `{short_flag}`: {} — the configurable long-silence threshold is not built", one_line(&se)))
} else if ok {
Err(format!("a twin silent for {age:?} s was ACCEPTED under a {short_secs} s threshold: {}", one_line(&format!("{se}\n{so}"))))
} else if !se.contains("stale-source") {
Err(format!("refused, but not by the stale-source guard: {}", one_line(&se)))
} else if !age.is_some_and(|a| (a.saturating_sub(3)..=a + 3).any(|x| se.contains(&format!("{x} s")))) || !se.contains("longer than the") {
Err(format!("refused, but the refusal does not name both the age ({age:?} s) and the threshold: {}", one_line(&se)))
} else if !sess.status_line(m)?.contains("OPEN") {
Err(format!("refused — but the fresh box is no longer OPEN, so the refusal came AFTER a write: {}", one_line(&se)))
} else {
Ok(format!("silent {age:?} s by the twin's clock, threshold {short_secs} s: {}", one_line(&se)))
};
// (2) over the threshold, --stale-ok: the guard must let it through. Pointed at the
// twin itself so nothing is restored by it — only the guard's word is measured.
let (_, _, se2) = cli_all(&cli, &["control", "twin", "replay", "--twin", &twin_at, "--into", &twin_at, "-i", &k, &short_flag, &secs, "--stale-ok"]);
accepted_override = if se2.contains("--stale-ok: restoring what this twin holds") {
Ok(one_line(se2.lines().filter(|l| l.starts_with("--stale-ok")).collect::<Vec<_>>().join("\n").as_str()))
} else {
Err(format!("--stale-ok under a {short_secs} s threshold did not pass the guard: {}", one_line(&se2)))
};
// (3) under the DEFAULT threshold (24 h), plain: accepted — and this IS the restore.
let (ok3, so3, se3) = cli_all(&cli, &["control", "twin", "replay", "--twin", &twin_at, "--into", &into, "-i", &k]);
accepted_plain = if ok3 { Ok(format!("silent {:?} s, default threshold: {}", answer_age_s(&twin_status(&cli, &admin, &twin_at).unwrap_or_default()), one_line(&se3))) } else { Err(one_line(&format!("{se3}\n{so3}"))) };
restore = if ok3 {
Ok(one_line(&so3))
} else {
// The restore is what TD1 stands on: take it with the override if the plain one refused.
let (ok4, so4, se4) = cli_all(&cli, &["control", "twin", "replay", "--twin", &twin_at, "--into", &into, "-i", &k, "--stale-ok"]);
if ok4 { Ok(format!("(with --stale-ok) {}", one_line(&so4))) } else { Err(one_line(&format!("{se4}\n{so4}"))) }
};
Ok(format!("silent {:.0} s", silent_since.elapsed().as_secs_f64()))
});
print_boot(&replay);
// OBSERVATION (not a verdict): the restore window's one host-key WARN — measured on loki
// 2026-09-21, once, ~0.4 s after "Restore finished; the box must restart", gone after the
// restart. Harmless by the coordinator's reading; T17 owns it. Logged when it recurs.
if let Some(l) = replay
.serial
.lines()
.find(|l| l.contains("wrong key-encrypting key"))
{
log_line(
&o.work,
"TD1",
&format!("OBSERVATION restore-window host-key WARN: {}", l.trim()),
);
}
out.insert(
"TD3",
verdict(
"TD3",
TD3_NAME,
match (&refused, &accepted_override, &accepted_plain) {
(Ok(a), Ok(b), Ok(c)) => {
Ok(format!("over: {a} | --stale-ok: {b} | under: {c}"))
}
_ => Err(format!(
"over the threshold, plain: {} | over, --stale-ok: {} | under, plain: {}",
show(&refused),
show(&accepted_override),
show(&accepted_plain)
)),
},
),
);
// ── TD1: the restored box boots; the twin must be SEEN polling it ────
let mut follows: Result<String, String> =
Err(format!("the restore did not happen: {}", show(&restore)));
let mut td2_twin: Result<String, String> = Err("not reached".into());
if restore.is_ok() {
let booted_ms = now_ms();
let back = fresh.boot("restored", |m| {
let at = format!("127.0.0.1:{}", m.grpc_private);
sess.expect_closed(m)?;
td2_twin = match (&before, trail(&cli, &admin, &at, tip_before)) {
(Ok(b), Ok(a)) if a.present && a.verified && a.hash_at_seq == b.tip => {
Ok(format!("the box restored from the twin holds the trail: {} entries, verifies, seq {} hashes to {} as the primary had it", a.entries, b.tip_seq, b.tip))
}
(Ok(b), Ok(a)) => Err(format!("the box restored from the twin: {} (the primary had seq {} = {})", a.raw, b.tip_seq, b.tip)),
(_, Err(e)) => Err(format!("the restored box's trail could not be read: {e}")),
(Err(e), _) => Err(format!("no trail before: {e}")),
};
// **A head the twin cannot have heard of before this boot.** The restored box makes
// one decision; its sequence exists only here, so a twin that reports it has polled
// THIS box, after it booted, successfully. (MEASURED 2026-09-21 on loki: without it
// the restored head equals the twin's own, and "follows" can be a stale answer.)
let k = admin.display().to_string();
data::cli(&cli, &["control", "group", "add", "td-after-restore", "-i", &k, "--at", &at]).map_err(|e| format!("the restored primary refuses a write, so it has no head a fresh poll could show: {}", one_line(&e)))?;
let h_new = head_of(&cli, &admin, &at)?;
// T3's alarm (159bdf3): a box that opened with its restore marker still present says
// `twin.restoring_at_open` on Admin.Health. The restored box must NOT say it.
let health = data::cli(&cli, &["control", "health", "-i", &k, "--at", &at]).map_err(|e| format!("Admin.Health unreadable on the restored box: {}", one_line(&e)))?;
if health.contains("twin.restoring_at_open") {
return Err(format!("the restored box raises twin.restoring_at_open on Admin.Health: {}", one_line(&health)));
}
let t0 = Instant::now();
let mut last = String::new();
while t0.elapsed() < Duration::from_secs(600) {
let st = twin_status(&cli, &admin, &twin_at).unwrap_or_else(|e| e);
let since_boot_s = (now_ms().saturating_sub(booted_ms)) / 1000;
let age = answer_age_s(&st);
let state = field(&st, "state").unwrap_or_default();
let head = field(&st, "primary").and_then(|v| v.parse::<u64>().ok());
let fresh_poll = age.is_some_and(|a| a + 2 < since_boot_s);
if state == "following" && field(&st, "caught_up").as_deref() == Some("true") && fresh_poll && head.is_some_and(|h| h >= h_new && h > head_before) {
return Ok(format!(
"Admin.Health carries no twin.restoring_at_open; the restored box wrote head {h_new}; the twin reports primary head {} (it had {head_before} before the restore), last answer {} s old, the box booted {since_boot_s} s ago: {}",
head.unwrap_or(0),
age.unwrap_or(0),
one_line(&st)
));
}
last = format!("state {state:?}, answer age {age:?} s vs boot {since_boot_s} s ago, primary head {head:?} (restored box at {h_new}, twin had {head_before}): {}", one_line(&st));
std::thread::sleep(Duration::from_secs(5));
}
Err(format!("no fresh successful poll of the restored primary within 600 s — {last}"))
});
print_boot(&back);
follows = match (&back.driven, boot_failed(&back)) {
(Some(r), None) => r.clone(),
(_, Some(e)) => Err(format!("the restored box: {e}")),
(None, None) => Err("the restored box never became ready".into()),
};
}
out.insert("TD1", verdict("TD1", TD1_NAME, follows));
out.insert(
"TD2",
verdict(
"TD2",
TD2_NAME,
match (&td2_start, &twin_after_crash, &td2_twin) {
(Ok(a), Ok(b), Ok(c)) => Ok(format!(
"{a} | the twin followed the box that came back: {b} | {c}"
)),
_ => Err(format!(
"start: {} | twin after the crash: {} | the twin's copy: {}",
show(&td2_start),
show(&twin_after_crash),
show(&td2_twin)
)),
},
),
);
stop_twin("done");
let (twin, tb) = tw.join().map_err(|_| "twin thread panicked")?;
print_boot(&tb);
if !o.keep {
fresh.destroy_disks();
twin.destroy_disks();
}
Ok(())
})
}
fn show(r: &Result<String, String>) -> String {
match r {
Ok(s) => format!("OK {s}"),
Err(e) => format!("NO {e}"),
}
}
// ── TD4: install from the CD alone, OVMF ────────────────────────────────────
/// The rig's own install, minus the read-only virtio copy of the medium: the medium is ONLY the
/// `-cdrom` (q35 puts it on the ICH9 AHCI controller), which is how UpCloud's tray reaches a guest.
pub fn td4(o: &Opts) -> Verdict {
let work = o.work.join("TODO-TD4");
let r = (|| -> Result<String, String> {
let spec = RigSpec {
iso: o.iso.clone(),
work: work.clone(),
boot_gib: 24,
member_gib: 6,
members: 1,
mem_mb: o.mem_mb,
smp: o.smp,
settle_secs: 5,
explicit_blockdevs: false,
};
let mut rig = Rig::create(spec)?;
let qemu = kvmproof::which("qemu-system-x86_64").ok_or("no qemu-system-x86_64")?;
let (code, vars) = kvmproof::locate_ovmf().ok_or("no OVMF")?;
let mut machine = Machine::uefi(&code, &vars);
machine.mem_mb = o.mem_mb;
machine.smp = o.smp;
let mut args = vec![
"-drive".to_string(),
format!("file={},if=virtio,format=raw,id=boot", rig.boot.display()),
];
args.push("-drive".into());
args.push(format!(
"file={},if=virtio,format=raw,id=data0",
rig.members[0].display()
));
isoproof::attach_cdrom(&mut args, &o.iso);
isoproof::attach_slirp_nics(&mut args, &[], 2);
let wd = work.join("b1-install-cd-only");
let run = kvmproof::boot_hold(
&qemu,
&machine,
&args,
&wd,
Duration::from_secs(600),
|_| false,
);
let cd001 = run
.serial
.lines()
.find(|l| l.contains("CD001"))
.map(|l| l.trim().to_string());
if !run.serial.contains(INSTALL_OK) {
return Err(format!(
"OVMF/q35, medium ONLY as -cdrom (AHCI): no {INSTALL_OK} after {} ms — {} · serial {}",
run.elapsed.as_millis(),
cd001.unwrap_or_else(|| format!("tail: {}", one_line(&tail(&run.serial, 6)))),
wd.join("serial.log").display()
));
}
let b = rig.boot("cd-only-installed", |p| {
Session::new(&work, &o.cli).and_then(|s| s.status_line(p))
});
print_boot(&b);
match (&b.driven, boot_failed(&b)) {
(Some(Ok(st)), None) => Ok(format!("OVMF/q35, medium ONLY as -cdrom: {INSTALL_OK} in {} ms, and the installed box serves: {st}", run.elapsed.as_millis())),
(_, Some(e)) => Err(format!("installed from the CD alone, but the box did not come up: {e}")),
(Some(Err(e)), _) => Err(format!("installed from the CD alone, but: {e}")),
_ => Err("installed from the CD alone, but the box never became ready".into()),
}
})();
let v = verdict("TD4", TD4_NAME, r);
log_line(&o.work, "TD4", &v.line());
v
}
// ── TD5: the WireGuard inner address comes from bootstrap ───────────────────
pub fn td5(o: &Opts) -> Verdict {
let v = (|| -> Verdict {
let (ok, so, se) = cli_all(&o.cli, &["control", "tunnel", "--help"]);
if !ok {
return verdict(
"TD5",
TD5_NAME,
Err(format!(
"the CLI at {} has no `control tunnel` verb ({}): no box can be given a tunnel address at bootstrap, so two boxes from one medium carry the one `[Interface] Address` baked into it — or none",
o.cli.display(),
one_line(&format!("{se}{so}")).chars().take(200).collect::<String>()
)),
);
}
let work = o.work.join("TODO-TD5");
let _ = std::fs::create_dir_all(&work);
// ONE site: one pool, one ledger, one hub config, shared by both boxes — exactly what makes
// two boxes from one medium distinguishable. `--todo-tunnel-enrol` overrides the lot.
let enrol = flag("--todo-tunnel-enrol").unwrap_or_else(|| {
format!(
"--pool 10.99.0.0/24 --ledger {} --hub-peers {}",
work.join("tunnel-ledger").display(),
work.join("hub-peers.conf").display()
)
});
let mut said = Vec::new();
let mut addrs = Vec::new();
for who in ["box-a", "box-b"] {
let spec = RigSpec {
iso: o.iso.clone(),
work: work.join(who),
boot_gib: 24,
member_gib: 6,
members: 4,
mem_mb: o.mem_mb,
smp: o.smp,
settle_secs: 5,
explicit_blockdevs: false,
};
let mut rig = match Rig::create(spec) {
Ok(r) => r,
Err(e) => return verdict("TD5", TD5_NAME, Err(e)),
};
if let Err(e) = rig.install() {
return verdict("TD5", TD5_NAME, Err(format!("{who}: {e}")));
}
let mut sess = match Session::new(&work.join(who), &o.cli) {
Ok(s) => s,
Err(e) => return verdict("TD5", TD5_NAME, Err(e)),
};
let cli = o.cli.clone();
let b = rig.boot(who, |p| {
let at = format!("127.0.0.1:{}", p.grpc_private);
// `enrol` takes a FRESH box (its door is Bootstrap.AssignTunnelAddress), so it
// comes before the bootstrap; the box is then bootstrapped as any other.
let mut a = vec!["control", "tunnel", "enrol", "--at", &at, "--label", who];
a.extend(enrol.split_whitespace());
let e = data::cli(&cli, &a)?;
sess.bootstrap(p)?;
let s = data::cli(&cli, &["control", "tunnel", "show", "--at", &at])?;
Ok(format!("{}\n{s}", one_line(&e)))
});
print_boot(&b);
match (&b.driven, boot_failed(&b)) {
(Some(Ok(show)), None) => {
addrs.push(field(show, "address").unwrap_or_default());
said.push(format!("{who}: {}", one_line(show)));
}
(Some(Err(e)), _) => return verdict("TD5", TD5_NAME, Err(format!("{who}: {e}"))),
(_, Some(e)) => return verdict("TD5", TD5_NAME, Err(format!("{who}: {e}"))),
_ => return verdict("TD5", TD5_NAME, Err(format!("{who}: never ready"))),
}
}
if addrs.iter().any(String::is_empty) || addrs[0] == addrs[1] {
return verdict(
"TD5",
TD5_NAME,
Err(format!(
"the inner addresses are {addrs:?}: {}",
said.join(" | ")
)),
);
}
unrun(
"TD5",
TD5_NAME,
format!(
"addresses DIFFER ({} vs {}) — {}; the traffic half is not measured: the OVMF rig forwards TCP only (slirp hostfwd), so no UDP path for wg exists between two rigs yet",
addrs[0],
addrs[1],
said.join(" | ")
),
)
})();
log_line(&o.work, "TD5", &v.line());
v
}
/// Unused-import guard for helpers other cells in this family share.
#[allow(dead_code)]
fn _uses(_: PathBuf, _: u64) -> u64 {
GIB
}
#[allow(dead_code)]
fn _pubkey(p: &Path) -> Result<String, String> {
pubkey(p)
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn the_answer_age_is_read_out_of_as_of() {
let s = "state unreachable\nas_of it has applied 12 of the head 12 its primary reported at its last answer, 93 s ago; it was LEVEL";
assert_eq!(answer_age_s(s), Some(93));
assert_eq!(
answer_age_s("as_of the primary last answered 7 s ago (unix ms 1790016926610)"),
Some(7)
);
assert_eq!(answer_age_s("as_of it has applied 3 and has had no answer from its primary since it started"), None);
}
#[test]
fn xfs_db_numbers_are_read_by_key() {
assert_eq!(
xfs_num("blocksize = 4096\nagblocks = 65536\n", "agblocks"),
Some(65536)
);
assert_eq!(xfs_num("core.size = 12345", "core.size"), Some(12345));
}
}