crates/gunnar-growbench/src/restore_drill.rs
//! **H1 — the dual heartbeat, and R6 — the one-step restore** (private-gunnar-ops
//! `DATA-SET-GROWTH-DESIGN.md` §18), under KVM in ONE estate so both media are installed once.
//!
//! A PRIMARY (4 members) and a TWIN (one data volume) come up; the primary takes repositories and
//! users, the twin is joined and followed to the watermark. Then:
//!
//! * **H1a** the twin is killed (SIGKILL, no flush): the primary's `Admin.Health` twin row must read
//! `SILENT` within the bound (`SILENT_AFTER_MS` + one poll + the bench's tick). A repository is
//! pushed while it is down; the twin is booted again and must be `ok` and caught up.
//! * **H1b** the primary is killed: the twin's `Admin.Health` primary row must read `SILENT`
//! within the same bound.
//! * **R8** (only when `R8` is in `--scenarios`) the RETENTION HOLE, `DATA-SET-GROWTH-DESIGN.md`
//! §20 B8 — the one twin path the design names as having no driver at all. While the twin is
//! dead, `Mirror.LogRetain` trims the primary's log past the entry the twin wants next; when
//! the twin comes back, no `LogFetch` can close that hole and it must take a `SnapshotFetch`
//! instead. Proven, not assumed: the twin's OWN `first_seq` must have moved off 1, which only
//! `Log::restart_after` (the snapshot install) does. R6 then restores from a SNAPSHOTTED twin.
//! * **R10** (only when `R10` is in `--scenarios`) the primary dies MID-PUSH: a 96 MiB push is
//! still in flight when QEMU is killed under it. After the restore the half-push must be
//! either absent from the catalog (the transaction never reached the twin) or present and
//! `git fsck`-clean — never a repository that is listed and will not clone.
//! * **R9** (only when `R9` is in `--scenarios`) the INTERRUPTED RESCUE: the `twin replay` that
//! restores the re-imaged primary is KILLED in flight, and the same command is run again. It
//! must resume — `applied` strictly below `head_seq` — because a box that threw the broken
//! restore away and started over would apply every entry a second time.
//!
//! **What R9 does NOT prove, and why.** The other half of the same contract is that a stream
//! which ends cleanly BELOW the head the sender announced (`LogChunk.restore_head`, added by
//! the `replay-abort` work) is not a restore: the box must stay `restoring`. That is a
//! property of the SERVER in the medium, and it cannot be read off the wire from here — an
//! older server ignores the unknown field and finishes the short stream, which looks from the
//! client exactly like a stream that was long enough. So R9 measures the resume, which is real
//! on any medium, and the announced-head refusal is proven where it can be: the unit tests in
//! `gunnar-cli/src/control/twin.rs` (`pump_into` with a twin whose fetch breaks). To assert it
//! under KVM the bench would need a door that reports the box's restore state after a cut —
//! `Bootstrap.Status` saying `restoring` — and no such field is on the wire today. **Compose
//! the media from a commit that carries `restore_head` or this phase is half a proof.**
//! * **R6** the steps `cargo xtask restore main` takes, on the killed primary: (a) the twin holds
//! the watermark; (b) the primary's DATA disks are destroyed and the medium re-installed onto
//! its existing system disk (a re-image), and the box must read FRESH (bootstrap OPEN, the
//! catalog refuses the old admin); (c) `gunnar control twin replay` streams the twin into its
//! `Bootstrap.Restore`; (d) the box restarts only when the restore said `restart true`;
//! (e) every repository clones `fsck`-clean with every file sha256-identical, and the user
//! list and the repository list equal what the primary held before it died; the twin follows
//! the restored primary.
use std::collections::BTreeSet;
use std::path::{Path, PathBuf};
use std::sync::mpsc::{self, Receiver, Sender};
use std::time::{Duration, Instant};
use super::twin_drill::{
self, field, head_of, make_lfs_repo, one_line, pubkey, users, verify_lfs, wait_twin,
};
use super::{boot_failed, print_boot, Opts, Session, Status, Verdict};
use crate::data::{self, RepoManifest};
use crate::rig::{self, Ports, Rig, RigSpec, GIB};
const WAIT: Duration = Duration::from_secs(900);
/// `gunnar_grpc::heartbeat::SILENT_AFTER_MS`, spelled here because this bench is a detached crate.
pub const SILENT_AFTER_MS: u64 = 60_000;
/// One twin poll (2 s), the bench's own tick (1 s) and a CLI round trip on a loaded laptop.
pub const SLACK_MS: u64 = 10_000;
/// How long a box that has just restarted out of a restore may take to answer its control plane.
const CONTROL_WAIT: Duration = Duration::from_secs(180);
enum ToTwin {
PrimaryAt(Ports),
Sync(u64),
Kill,
Boot,
RestoreDone,
Stop,
}
enum FromTwin {
Joined { key: String, ports: Ports },
Up,
Synced(String),
Killed(Instant),
Final(String),
Failed(String),
}
/// The state word of a heartbeat row printed by `gunnar control health`:
/// `twin <principal> <STATE> — …` or `primary <address> <STATE> — …`.
pub fn row_state<'a>(health: &'a str, kind: &str) -> Option<(&'a str, &'a str)> {
health
.lines()
.find(|l| l.split_whitespace().next() == Some(kind))
.and_then(|l| l.split_whitespace().nth(2).map(|s| (s, l.trim())))
}
/// `gunnar control repo list` → the set of `namespace/repo` paths. The CLI prints one bare
/// `namespace/repo` per line (or `no repositories`); `repo get` style `repo ns/name` lines are
/// read too. MEASURED 2026-09-15 run 1: a reader of only the `repo ` form counted 0 repositories
/// on a primary holding four, and a catalog check of empty against empty would have passed.
pub fn repo_names(list: &str) -> BTreeSet<String> {
list.lines()
.map(|l| l.trim())
.map(|l| l.strip_prefix("repo ").map(str::trim).unwrap_or(l))
.filter(|l| {
!l.contains(char::is_whitespace)
&& l.matches('/').count() == 1
&& !l.starts_with('/')
&& !l.ends_with('/')
})
.map(str::to_string)
.collect()
}
/// Did a heartbeat go silent within the bound, measured from the kill?
pub fn within_bound(elapsed: Duration) -> bool {
elapsed.as_millis() as u64 <= SILENT_AFTER_MS + SLACK_MS
}
fn health(cli: &Path, key: &Path, at: &str) -> Result<String, String> {
data::cli(
cli,
&[
"control",
"health",
"-i",
&key.display().to_string(),
"--at",
at,
],
)
}
/// Poll `Admin.Health` at `at` until the `kind` row reads `want` (case-insensitive); answers the
/// row and how long after `since` it was first seen.
fn wait_row(
cli: &Path,
key: &Path,
at: &str,
kind: &str,
want: &str,
since: Instant,
budget: Duration,
) -> Result<(Duration, String), String> {
let t0 = Instant::now();
let mut last = String::new();
while t0.elapsed() < budget {
match health(cli, key, at) {
Ok(out) => match row_state(&out, kind) {
Some((state, line)) if state.eq_ignore_ascii_case(want) => {
return Ok((since.elapsed(), line.to_string()))
}
Some((_, line)) => last = line.to_string(),
None => last = format!("no `{kind}` row in Admin.Health"),
},
Err(e) => last = one_line(&e),
}
std::thread::sleep(Duration::from_secs(1));
}
Err(format!(
"the {kind} row at {at} never read {want} within {} s; last: {last}",
budget.as_secs()
))
}
/// Poll `control status` until the restored box's control plane answers. Answers how long that
/// took, so a box that is slow to reopen shows up as a number rather than as a flake.
fn wait_control(cli: &Path, at: &str) -> Result<Duration, String> {
let t0 = Instant::now();
let mut last = String::new();
while t0.elapsed() < CONTROL_WAIT {
match data::cli(cli, &["control", "status", "--at", at]) {
Ok(_) => return Ok(t0.elapsed()),
Err(e) => last = one_line(&e),
}
std::thread::sleep(Duration::from_secs(2));
}
Err(format!(
"the restored box's control plane never answered within {} s: {last}",
CONTROL_WAIT.as_secs()
))
}
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 thread ended before it answered".into())
}
Err(mpsc::RecvTimeoutError::Timeout) => Err("the twin went silent".into()),
}
}
/// One boot of the twin: join on the first, then obey until killed or stopped.
fn twin_drive(
cli: &Path,
key: &Path,
tp: Ports,
joined: &mut bool,
killed_at: &mut Option<Instant>,
end: &mut bool,
rx: &Receiver<ToTwin>,
tx: &Sender<FromTwin>,
) -> Result<String, String> {
let at = format!("127.0.0.1:{}", tp.grpc_private);
if !*joined {
let p = match rx.recv_timeout(WAIT) {
Ok(ToTwin::PrimaryAt(p)) => p,
_ => return Err("the primary never came up".into()),
};
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}"));
}
*joined = true;
let _ = tx.send(FromTwin::Joined {
key: key_line,
ports: tp,
});
} else {
wait_twin(cli, key, &at, "answering LogHead as a twin", |o| {
field(o, "twin").as_deref() == Some("yes")
})?;
let _ = tx.send(FromTwin::Up);
}
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) => {
let s = wait_twin(
cli,
key,
&at,
"following the restored primary, caught up",
|o| {
field(o, "state").as_deref() == Some("following")
&& field(o, "caught_up").as_deref() == Some("true")
},
)?;
said.push(s.clone());
let _ = tx.send(FromTwin::Final(s));
}
Ok(ToTwin::Kill) => {
*killed_at = Some(Instant::now());
return Ok(said.join(" | "));
}
Ok(ToTwin::Stop) => {
*end = true;
return Ok(said.join(" | "));
}
Ok(ToTwin::PrimaryAt(_)) | Ok(ToTwin::Boot) => {
return Err("an order the running twin cannot take".into())
}
Err(_) => return Err("the drill went silent".into()),
}
}
}
/// The twin's whole life: boots until stopped; every boot ends in a SIGKILL (a kill is the point,
/// and the last boot's disk is not read afterwards).
fn twin_life(twin: &mut Rig, cli: &Path, key: &Path, rx: &Receiver<ToTwin>, tx: &Sender<FromTwin>) {
let mut joined = false;
let mut n = 0;
loop {
n += 1;
let (mut killed_at, mut end, mut ports) = (None, false, None);
twin.crash_after_drive = true;
let b = twin.boot(&format!("twin-{n}"), |tp| {
ports = Some(tp);
let r = twin_drive(cli, key, tp, &mut joined, &mut killed_at, &mut end, rx, tx);
if let Err(e) = &r {
let _ = tx.send(FromTwin::Failed(e.clone()));
end = true;
}
r
});
print_boot(&b);
if twin.pin.is_none() {
twin.pin = ports;
}
if b.driven.is_none() {
let _ = tx.send(FromTwin::Failed(
boot_failed(&b).unwrap_or_else(|| "the twin never became ready".into()),
));
return;
}
if end {
return;
}
let _ = tx.send(FromTwin::Killed(killed_at.unwrap_or_else(Instant::now)));
match rx.recv_timeout(WAIT * 3) {
Ok(ToTwin::Boot) => continue,
_ => return,
}
}
}
struct Restored<'a> {
cli: &'a Path,
admin: &'a Path,
carol_ssh: &'a str,
notes: &'a RepoManifest,
work: &'a Path,
users_before: &'a BTreeSet<String>,
repos_before: &'a BTreeSet<String>,
/// The pushed LFS object's sha256: a restore that brings the pointer back but not the object is not a restore.
lfs_sha: &'a str,
/// R10's half-push: the one repository the catalog may have GAINED, because the primary
/// died inside its push. `None` when R10 did not run, and then the catalog must match
/// exactly, as it always did.
half: Option<&'a str>,
}
impl Restored<'_> {
/// (e): repositories clone fsck-clean and byte-identical, users and catalog equal.
fn verify(&self, sess: &mut Session, m: Ports) -> Result<String, String> {
let at = format!("127.0.0.1:{}", m.grpc_private);
// MEASURED 2026-09-16: a box that has just restarted out of a restore answers SSH
// (which is what `Rig` calls ready) before its control plane is listening — the run
// failed at this exact line with `transport error` while the twin, moments later, found
// the same box serving. The re-seal of every restored secret happens at open, and it
// takes longer the more the box was given back. So the check waits for the control
// plane, in a bounded and reported way, instead of racing it.
let waited = wait_control(self.cli, &at)?;
sess.expect_closed(m)?;
let repos = sess.verify(m);
let carol = data::verify(
&self.work.join("verify-carol"),
self.carol_ssh,
|_| format!("ssh://carol@127.0.0.1:{}/carol/notes", m.ssh),
std::slice::from_ref(self.notes),
);
let users_after = users(self.cli, self.admin, &at)?;
let lfs = verify_lfs(
&self.work.join("verify-lfs"),
&sess.ssh,
&sess.url(m, "lfs-assets"),
self.lfs_sha,
);
let listed = repo_names(&data::cli(
self.cli,
&[
"control",
"repo",
"list",
"-i",
&self.admin.display().to_string(),
"--at",
&at,
],
)?);
let mut bad = Vec::new();
if let Err(e) = &lfs {
bad.push(format!("lfs-assets: {e}"));
}
if repos.is_err() {
bad.push(format!("{repos:?}"));
}
if !carol.intact() {
bad.push(format!("carol/notes: {}", carol.summary()));
}
if users_after != *self.users_before {
bad.push(format!(
"users before {:?} after {users_after:?}",
self.users_before
));
}
// R10's half-push is the one repository the catalog may have gained: the primary died
// inside its create-on-push, so the twin's log may or may not carry that transaction.
// Either is right; a HALF of it is not, and `verify_half` below is what says so.
let mut catalog = listed.clone();
if let Some(h) = self.half {
catalog.retain(|r| !r.ends_with(&format!("/{h}")));
}
if catalog != *self.repos_before {
bad.push(format!(
"repos before {:?} after {catalog:?} (of {listed:?})",
self.repos_before
));
}
if bad.is_empty() {
Ok(format!("control plane answered {:.1} s after the box was ready; {}; carol/notes {}; {}; users {} identical; repo list {} identical {:?}", waited.as_secs_f64(), repos.unwrap_or_default(), carol.summary(), lfs.unwrap_or_default(), users_after.len(), listed.len(), listed))
} else {
Err(bad.join(" · "))
}
}
/// **R10's answer**: what became of the repository the primary died inside. Two outcomes are
/// right and one is not. Absent means the transaction never reached the twin — nothing was
/// promised and nothing was lost. Present means it did, and then it must clone and `git fsck`
/// clean like any other. A repository that is in the catalog but will not clone, or clones
/// with a broken object, is the half nobody may ship.
fn verify_half(&self, sess: &Session, m: Ports) -> Result<String, String> {
let at = format!("127.0.0.1:{}", m.grpc_private);
let listed = repo_names(&data::cli(
self.cli,
&[
"control",
"repo",
"list",
"-i",
&self.admin.display().to_string(),
"--at",
&at,
],
)?);
let name = format!("/{HALF}");
let Some(path) = listed.iter().find(|r| r.ends_with(&name)) else {
return Ok(format!("`{HALF}` is absent from the restored catalog ({} repositories): the push was never a transaction on the twin, so nothing of it came back — the right half of the rule", listed.len()));
};
let into = self.work.join("verify-half");
let _ = std::fs::remove_dir_all(&into);
std::fs::create_dir_all(&into).map_err(|e| e.to_string())?;
data::clone_retry(&into, &sess.ssh, &sess.url(m, HALF), HALF, Duration::from_secs(300)).map_err(|e| format!("`{path}` is in the restored catalog but will not clone — that is the half this rule forbids: {}", one_line(&e)))?;
let out = data::git(
&into.join(HALF),
&sess.ssh,
&["fsck", "--strict", "--no-progress"],
)
.map_err(|e| format!("`{path}` cloned but does not fsck: {}", one_line(&e)))?;
let _ = std::fs::remove_dir_all(&into);
Ok(format!("`{path}` came back WHOLE and git fsck-clean ({}): the transaction reached the twin before the primary died", if out.trim().is_empty() { "no complaint".to_string() } else { one_line(&out) }))
}
}
fn verdict(
id: &'static str,
name: &'static str,
parts: &[(&str, &Result<String, String>)],
) -> Verdict {
let status = if parts.iter().all(|(_, r)| r.is_ok()) {
Status::Pass
} else {
Status::Fail
};
let detail = parts
.iter()
.map(|(k, r)| match r {
Ok(s) => format!("{k} ok: {s}"),
Err(e) => format!("{k} FAILED: {e}"),
})
.collect::<Vec<_>>()
.join(" | ");
Verdict {
id,
name,
status,
detail,
}
}
pub const H1_NAME: &str = "dual heartbeat under KVM: twin killed → primary reads it silent, primary killed → twin reads it silent, within the bound";
/// The prefix of an outcome that is neither a pass nor a fail: the bench could not put the
/// box in the state the scenario is about, so the run MEASURED NOTHING. A scenario that
/// reported PASS from here would be reporting on a situation that never happened.
pub const UNMEASURED: &str = "UNMEASURED: ";
/// R10's repository: pushed, never acknowledged, because the primary dies under it.
pub const HALF: &str = "half-push";
/// How long R10 lets the push run before the primary is killed. Long enough that the pack is
/// moving, short enough that a push of `R10_BYTES` over slirp into a KVM guest cannot have
/// finished — and if it has, the run says UNMEASURED rather than PASS.
pub const R10_IN_FLIGHT_MS: u64 = 1_200;
pub const R10_BYTES: u64 = 96 << 20;
/// When R9 cuts the replay — a window in the middle of a restore made long enough on purpose
/// (see [`R9_BALLAST_BYTES`]). A cut that lands before the first entry is applied, or after the
/// last one, says UNMEASURED; only a resume that applies SOME of the head is a proof.
///
/// MEASURED, three runs, and the misses came from BOTH ends before this number was earned:
/// * 1500 ms, no ballast — the whole 29 MB log was already replayed. Too late.
/// * 200 ms, no ballast, twin snapshotted by R8 — same, in under 200 ms. Too late.
/// * 700 ms, 320 MiB of ballast — the resume applied ALL 169 of head 169: at 700 ms the replay
/// was still streaming and had not applied a single entry yet. **Too early.**
/// * 8000 ms, the same 320 MiB — "the whole replay was over within 8000 ms". **Too late**: a
/// restore streams the log raw over gRPC and is far faster than the git pushes that filled it
/// (those measured ~25 MB/s; the replay of ~364 MB finished inside 8 s, so better than 45).
///
/// **The window is therefore bracketed by measurement, not by guesswork: later than 0.7 s and
/// earlier than 8 s.** 3 s is its middle, and each further miss narrows this bracket rather than
/// starting the search again — a miss from the early end raises the floor, one from the late end
/// lowers the ceiling, and both are written here.
///
/// A clock is the wrong instrument for this and the bench knows it. The right one is a door that
/// reports how far a restore has got, which the wire does not have; until it does, R9 brackets.
pub const R9_CUT_MS: u64 = 3_000;
/// Acknowledged, replicated repository weight pushed before the primary dies, so the restore has
/// to stream it all back and takes long enough to interrupt. MEASURED 2026-09-16, twice: without
/// it the ordinary log replays in under 1.5 s and, once R8 has trimmed it to a snapshot, in under
/// 200 ms — the bench lost that race both times and said UNMEASURED both times. Pushed only when
/// R9 is asked for; every other phase of this estate is faster without it.
pub const R9_BALLAST_BYTES: u64 = 320 << 20;
pub const R10_NAME: &str = "the primary dies MID-PUSH: the twin's log ends on a whole transaction, and the restore gives back every acknowledged repository plus, of the half-push, either nothing or something git fsck calls clean — never a half";
pub const R9_NAME: &str = "the interrupted rescue: the replay stream is cut mid-flight and the same command RESUMES from where it stopped, applying only what the box was still missing";
/// R9's two steps: kill a `twin replay` in flight, then run it again and read whether it
/// resumed. `applied < head_seq` on the second run is the proof — the entries already on the
/// box are skipped (`bootstrap.rs`: `if done.seq < next { continue }`, before `applied += 1`),
/// so a box that had thrown the broken restore away and started over would report
/// `applied == head_seq` instead.
fn cut_and_resume(
cli: &Path,
twin_at: &str,
at: &str,
key: &str,
) -> (Result<String, String>, Option<String>) {
let replay = |args_key: &str| {
data::cli(
cli,
&[
"control", "twin", "replay", "--twin", twin_at, "--into", at, "-i", args_key,
],
)
};
// **R8 and R9 cannot be measured in the same estate.** MEASURED 2026-09-16: after R8 the twin
// has no log left to send — it answers `first_seq 155, bytes 0` — so `twin replay` streams a
// SNAPSHOT, which `Bootstrap.Restore` stages and installs WHOLESALE (`restore_snapshot_chunk`
// sets `next = at + 1`, and every following entry is then skipped). A snapshot restore's
// `applied` is therefore 0 whether it resumed or started over, and R9's only proof is gone.
// The bench says so instead of reporting a pass it cannot stand behind.
match twin_drill::twin_status(cli, key.as_ref(), twin_at)
.ok()
.and_then(|o| field(&o, "first_seq").and_then(|v| v.parse::<u64>().ok()))
{
Some(first) if first > 1 => {
return (
Err(format!(
"{UNMEASURED}the twin's log begins at {first}, so it is SNAPSHOTTED (R8 ran in this estate) and the restore is a snapshot, installed whole — `applied` is 0 whether the replay resumed or started over, so a cut proves nothing. Run R9 in an estate WITHOUT R8: `--scenarios H1,R10,R9,R6`."
)),
None,
);
}
_ => {}
}
let spawned = std::process::Command::new(cli)
.args([
"control", "twin", "replay", "--twin", twin_at, "--into", at, "-i", key,
])
.stdout(std::process::Stdio::piped())
.stderr(std::process::Stdio::null())
.spawn();
let mut child = match spawned {
Ok(c) => c,
Err(e) => return (Err(format!("spawning the replay to cut: {e}")), None),
};
std::thread::sleep(Duration::from_millis(R9_CUT_MS));
match child.try_wait() {
Err(e) => return (Err(format!("try_wait: {e}")), None),
// The knife missed: the whole replay was over. It also RESTORED the box, so its own
// answer is handed back — R6 must not send a second Restore to a box that finished one.
Ok(Some(st)) => {
let out = child
.wait_with_output()
.map(|o| String::from_utf8_lossy(&o.stdout).into_owned())
.unwrap_or_default();
return (
Err(format!("{UNMEASURED}the whole replay was over within {R9_CUT_MS} ms (exit {st}), ballast and all — raise R9_BALLAST_BYTES (now {}) or cut sooner", rig::human(R9_BALLAST_BYTES))),
Some(out),
);
}
Ok(None) => {}
}
if let Err(e) = child.kill() {
return (Err(format!("killing the replay: {e}")), None);
}
let st = match child.wait() {
Ok(st) => st,
Err(e) => return (Err(format!("waiting for the killed replay: {e}")), None),
};
// The resume: the same command, no flag, no hint of what happened. Its answer is also R6's.
let out = match replay(key) {
Ok(out) => out,
Err(e) => {
return (
Err(format!(
"the resume was refused after the cut — a broken rescue must be resumable: {}",
one_line(&e)
)),
None,
)
}
};
let verdict = (|| -> Result<String, String> {
if st.success() {
return Err(format!(
"{UNMEASURED}the replay finished between the check and the kill (exit {st})"
));
}
let head = field(&out, "head_seq")
.and_then(|v| v.parse::<u64>().ok())
.ok_or_else(|| format!("no head_seq in the resume:\n{out}"))?;
let applied = field(&out, "applied")
.and_then(|v| v.parse::<u64>().ok())
.ok_or_else(|| format!("no applied in the resume:\n{out}"))?;
let lost = field(&out, "lost").unwrap_or_default();
if lost != "0" {
return Err(format!(
"the resume lost {lost} entries: {}",
one_line(&out)
));
}
// The two ways the knife can miss. Neither is a defect and neither is a proof.
if applied == 0 {
return Err(format!("{UNMEASURED}the resume applied nothing: the cut landed after the first replay had already finished, so there was no interrupted restore to resume — {}", one_line(&out)));
}
if applied >= head {
return Err(format!("{UNMEASURED}the resume applied all {applied} of head {head}: the cut landed before the first replay had applied a single entry, so nothing was resumed — cut later — {}", one_line(&out)));
}
Ok(format!("the replay was killed in flight after {R9_CUT_MS} ms; the same command resumed and applied {applied} of head {head} (the {} already on the box were skipped), lost 0 — {}", head - applied, one_line(&out)))
})();
(verdict, Some(out))
}
pub const R8_NAME: &str = "the retention hole: the primary trims its log behind a dead twin, and the twin closes the hole with a SnapshotFetch, not a LogFetch (DESIGN §20 B8)";
pub const R6_NAME: &str = "one-step restore: data disks destroyed, primary re-imaged FRESH, replay from the twin, restart if asked, repos fsck-clean, users and catalog identical";
/// Every verdict this one estate answers, in the order its phases run: `[H1, R8, R10, R9, R6]`.
/// The list is fixed whatever `--scenarios` asked for, so `main`'s lookup by id always finds
/// its row; a phase nobody asked for answers UNRUN with that as the reason.
const IDS: [(&str, &str); 5] = [
("H1", H1_NAME),
("R8", R8_NAME),
("R10", R10_NAME),
("R9", R9_NAME),
("R6", R6_NAME),
];
/// All four verdicts of the estate.
pub fn h1_r6(o: &Opts) -> Vec<Verdict> {
let all = |status: Status, detail: String| {
IDS.iter()
.map(|(id, name)| Verdict {
id,
name,
status,
detail: detail.clone(),
})
.collect::<Vec<_>>()
};
// ONE medium: the twin installs the primary's own `--iso` and is made a twin over the wire
// (`twin join` → `Bootstrap.BecomeTwin`) — owner, 2026-09-21.
match run(o, o.iso.clone()) {
Ok(v) => v,
Err(e) => all(Status::Fail, e),
}
}
fn run(o: &Opts, twin_iso: PathBuf) -> Result<Vec<Verdict>, String> {
let work = o.work.join("HR");
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 mut primary = Rig::create(spec("primary", o.iso.clone(), o.boot_gib, 4, o.member_gib))?;
primary.hold = Duration::from_secs(4 * 3600);
let mut twin = Rig::create(spec("twin", twin_iso, 10, 1, o.twin_member_gib))?;
twin.hold = Duration::from_secs(4 * 3600);
twin.grpc_only_ready = true;
eprintln!("── H1/R6: primary {}", primary.describe());
eprintln!("── H1/R6: twin {}", twin.describe());
primary.install()?;
twin.install()?;
let mut sess = Session::new(&work, &o.cli)?;
let keys = work.join("keys");
let (carol_key, dave_key) = (keys.join("carol"), keys.join("dave"));
data::keygen(&carol_key)?;
data::keygen(&dave_key)?;
let alpha = data::make_repo(
&sess.trees,
"alpha",
&[
("a.bin", 4 << 20),
("edge.bin", data::CHUNK_BYTES + 1),
("deep/t.bin", 1024),
],
0x4861_0001,
)?;
let beta = data::make_repo(&sess.trees, "beta", &[("b.bin", 12 << 20)], 0x4861_0002)?;
let gamma = data::make_repo(
&sess.trees,
"while-twin-down",
&[("g.bin", 6 << 20)],
0x4861_0003,
)?;
let notes = data::make_repo(&sess.trees, "notes", &[("n.bin", 2 << 20)], 0x4861_0004)?;
let (lfs, lfs_sha) = make_lfs_repo(&sess.trees, 0x4861_0005)?;
let half = data::make_repo(&sess.trees, HALF, &[("h.bin", R10_BYTES)], 0x4861_0006)?;
let carol_ssh = data::ssh_command(&carol_key);
let cli = o.cli.clone();
let admin = sess.key.clone();
let k = admin.display().to_string();
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 (tcli, tadmin) = (cli.clone(), admin.clone());
let twin_thread = scope.spawn(move || {
twin_life(&mut twin, &tcli, &tadmin, &twin_rx, &twin_tx);
twin
});
// ── the estate takes data; H1a on the live primary; the primary is then killed ──
let (mut pinned, mut tports) = (None::<Ports>, None::<Ports>);
let (mut h1a, mut h1a_back) = (
Err::<String, String>("not reached".into()),
Err::<String, String>("not reached".into()),
);
let (mut users_before, mut repos_before, mut head_final) =
(BTreeSet::new(), BTreeSet::new(), 0u64);
let mut primary_killed_at = None::<Instant>;
// R8 is a phase of THIS estate, not a second install: asking for it makes the
// twin's catch-up go through `SnapshotFetch`, and R6 then restores from a
// SNAPSHOTTED twin. Not asking for it leaves H1 and R6 exactly as they were.
let want_r8 = o.scenarios.iter().any(|s| s == "R8");
let (mut r8_hole, mut r8_closed) = (
Err::<String, String>("R8 was not asked for".into()),
Err::<String, String>("not reached".into()),
);
let want_r9 = o.scenarios.iter().any(|s| s == "R9");
let mut r9 = Err::<String, String>("R9 was not asked for".into());
let want_r10 = o.scenarios.iter().any(|s| s == "R10");
let mut r10 = Err::<String, String>("R10 was not asked for".into());
let mut r10_after = Err::<String, String>("not reached".into());
primary.crash_after_drive = true;
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, vec![alpha.clone(), beta.clone()])?;
// LFS: create the repository with the plain first commit, then push the object (pre-push runs before create-on-push).
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))?;
data::cli(&cli, &["control", "user", "add", "dave", "--key", &pubkey(&dave_key)?, "-i", &k, "--at", &at])?;
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 h = head_of(&cli, &admin, &at)?;
let _ = to_twin.send(ToTwin::Sync(h));
let s1 = match recv(&from_twin)? { FromTwin::Synced(s) => s, _ => return Err("no sync".into()) };
// What the twin holds when it dies: R8's hole is measured against this.
let twin_applied = h;
// H1a: the twin row is ok, the twin dies, the row goes silent within the bound.
let now = Instant::now();
let (_, ok_row) = wait_row(&cli, &admin, &at, "twin", "ok", now, Duration::from_secs(90))?;
let _ = to_twin.send(ToTwin::Kill);
let t0 = match recv(&from_twin)? { FromTwin::Killed(t) => t, _ => return Err("the twin was not killed".into()) };
h1a = wait_row(&cli, &admin, &at, "twin", "silent", t0, Duration::from_millis(SILENT_AFTER_MS + SLACK_MS + 60_000)).and_then(|(took, row)| {
if within_bound(took) {
Ok(format!("before the kill: `{ok_row}`; SILENT {:.1} s after the SIGKILL (bound {} s + {} s slack): `{row}`", took.as_secs_f64(), SILENT_AFTER_MS / 1000, SLACK_MS / 1000))
} else {
Err(format!("SILENT only after {:.1} s (bound {} s + {} s slack): `{row}`", took.as_secs_f64(), SILENT_AFTER_MS / 1000, SLACK_MS / 1000))
}
});
// A write while the twin is down, then the twin comes back and catches up.
let gamma_pushed = sess.push_all(p, vec![gamma.clone()])?;
// ── R8, the RETENTION HOLE (only when asked for) ─────────────────
// `DATA-SET-GROWTH-DESIGN.md` §20 names this the one unproven twin path:
// "nothing in gunnar-growbench trims the log (`Mirror.LogRetain`) behind a
// stopped twin and catches it up through `SnapshotFetch` … that drive
// belongs in growbench beside T1". Here it is: while the twin is dead, the
// primary is told to retain only from its own head, which puts the twin's
// next wanted sequence BELOW the primary's `first_seq` — a hole no
// `LogFetch` can close. The twin must notice and take a snapshot instead.
if want_r8 {
r8_hole = (|| -> Result<String, String> {
let primary_log = twin_drill::twin_status(&cli, &admin, &at)?;
let head_now = head_of(&cli, &admin, &at)?;
// `--from head + 1`, not `--from head`. MEASURED 2026-09-16: `retain_from`
// rolls a fresh segment and then drops every segment whose LAST sequence is
// below `from` (`txlog::truncate_before`: `last < seq`), so `--from head`
// leaves the segment that ends at `head` in place and first_seq never moves.
// The first run of R8 said UNMEASURED for exactly that reason.
let from = head_now + 1;
let out = data::cli(&cli, &["control", "twin", "retain", "--at", &at, "--from", &from.to_string(), "--allow-past-twin", "-i", &k])?;
let first = field(&out, "first_seq").and_then(|v| v.parse::<u64>().ok()).ok_or_else(|| format!("no first_seq in twin retain:\n{out}"))?;
// The threshold: the hole must be REAL. The twin last applied
// `twin_applied` and wants `twin_applied + 1`; a trim that left that
// entry on the primary proves nothing, because a plain LogFetch
// would still close it.
if first <= twin_applied + 1 {
return Err(format!("{UNMEASURED}the retain from {from} left first_seq at {first} and the twin wants {} — no hole opened, so nothing was measured (the primary's log: {})", twin_applied + 1, one_line(&primary_log)));
}
Ok(format!("trimmed the primary's log to first_seq {first} while the twin was dead holding {twin_applied}: the twin's next ({}) is gone", twin_applied + 1))
})();
eprintln!(" R8: {r8_hole:?}");
}
let _ = to_twin.send(ToTwin::Boot);
match recv(&from_twin)? { FromTwin::Up => {}, _ => return Err("the twin did not come back".into()) }
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 after the twin's reboot".into()) };
h1a_back = wait_row(&cli, &admin, &at, "twin", "ok", Instant::now(), Duration::from_secs(120)).map(|(took, row)| format!("twin back and caught up (head {h2}: {s2}), row ok in {:.0} s: `{row}`", took.as_secs_f64()));
// R8's second half: the twin is caught up again — but HOW? A twin that
// closed the hole by `SnapshotFetch` called `Log::restart_after(at)`, so its
// OWN log begins after the snapshot: `first_seq > 1`. A twin that merely
// fetched entries still begins at 1. That is the difference between a
// snapshot and a log fetch, read off the wire rather than off a log line
// the next poll would have overwritten.
if want_r8 && r8_hole.is_ok() {
let tw_at = format!("127.0.0.1:{}", tports.ok_or("no twin ports")?.grpc_private);
r8_closed = (|| -> Result<String, String> {
let out = twin_drill::twin_status(&cli, &admin, &tw_at)?;
let first = field(&out, "first_seq").and_then(|v| v.parse::<u64>().ok()).ok_or_else(|| format!("no first_seq on the twin:\n{out}"))?;
let applied = field(&out, "applied").and_then(|v| v.parse::<u64>().ok()).unwrap_or(0);
if first <= 1 {
return Err(format!("the twin is caught up (applied {applied}) but its log still begins at {first}: it did not install a snapshot, so this run proves nothing about the retention hole — {}", one_line(&out)));
}
if applied < h2 {
return Err(format!("the twin's log begins at {first} but it applied only {applied} of {h2}: {}", one_line(&out)));
}
Ok(format!("the twin closed the hole with a snapshot: its own log now begins at {first} (was 1) and it applied {applied} of {h2} — {}", one_line(&out)))
})();
eprintln!(" R8: {r8_closed:?}");
}
// R9's ballast: acknowledged repository weight, so the restore takes long enough
// that a knife can be thrown into it. Pushed before the final watermark so the twin
// carries it and the replay has to stream it back.
let ballast = if want_r9 {
let m = data::make_repo(&sess.trees, "ballast", &[("w.bin", R9_BALLAST_BYTES)], 0x4861_0007)?;
// Pushed with `data::push`, NOT `sess.push_all`: the ballast is weight for the
// WIRE, and `push_all` would also enrol it in the manifest set that every later
// verify clones and re-hashes. MEASURED 2026-09-16: that made R6's verification
// of the restored box take longer than every other phase of the estate put
// together, for a repository whose content nothing is asking about. The catalog
// comparison sees it on both sides, so it still has to come back.
let took = data::push(&m, &sess.ssh, &sess.url(p, "ballast"), Duration::from_secs(900))?;
format!("; ballast {} in {:.1} s (wire weight for R9, not verified)", rig::human(m.bytes), took.as_secs_f64())
} else {
String::new()
};
users_before = users(&cli, &admin, &at)?;
repos_before = repo_names(&data::cli(&cli, &["control", "repo", "list", "-i", &k, "--at", &at])?);
head_final = head_of(&cli, &admin, &at)?;
let _ = to_twin.send(ToTwin::Sync(head_final));
let s3 = match recv(&from_twin)? { FromTwin::Synced(s) => s, _ => return Err("no final sync".into()) };
// ── R10: the primary dies MID-PUSH (only when asked for) ─────
// Every other kill in this bench lands on a quiet box. A real
// primary dies while someone is pushing, and the question is what
// the twin's log ends on. The push runs on its own thread; this
// closure returns while it is still in flight, and returning is
// what kills QEMU (`crash_after_drive`).
if want_r10 {
let (u, ssh, dir) = (sess.url(p, HALF), sess.ssh.clone(), half.dir.clone());
let pushing = std::thread::spawn(move || data::git(&dir, &ssh, &["push", "-q", &u, "main"]));
std::thread::sleep(Duration::from_millis(R10_IN_FLIGHT_MS));
r10 = if pushing.is_finished() {
Err(format!("{UNMEASURED}the {HALF} push was over within {R10_IN_FLIGHT_MS} ms, so the primary would not have died mid-push; make that repository bigger"))
} else {
Ok(format!("the primary is killed with the {HALF} push ({}) still in flight after {R10_IN_FLIGHT_MS} ms", rig::human(half.bytes)))
};
eprintln!(" R10: {r10:?}");
// The thread is left to die with the process it is pushing to.
}
primary_killed_at = Some(Instant::now());
Ok(format!("{boot}; pushed {pushed}, carol/notes, {gamma_pushed}{ballast}; wave 1: {s1}; final head {head_final}: {s3}; users {} repos {}", users_before.len(), repos_before.len()))
});
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 (p, tp) = (
pinned.ok_or("no primary ports")?,
tports.ok_or("no twin ports")?,
);
let twin_at = format!("127.0.0.1:{}", tp.grpc_private);
let killed = primary_killed_at.ok_or("the primary was never killed")?;
// H1b: the twin reads the primary silent.
let h1b = wait_row(
&cli,
&admin,
&twin_at,
"primary",
"silent",
killed,
Duration::from_millis(SILENT_AFTER_MS + SLACK_MS + 60_000),
)
.and_then(|(took, row)| {
if within_bound(took) {
Ok(format!(
"SILENT on the twin {:.1} s after the primary's SIGKILL: `{row}`",
took.as_secs_f64()
))
} else {
Err(format!(
"SILENT only after {:.1} s: `{row}`",
took.as_secs_f64()
))
}
});
// R6 (a): the twin holds the watermark the primary died at.
let r6a = wait_twin(
&cli,
&admin,
&twin_at,
&format!("applied >= {head_final}"),
|o| {
field(o, "applied")
.and_then(|v| v.parse::<u64>().ok())
.is_some_and(|a| a >= head_final)
},
);
// (b) data disks destroyed, the medium re-installed onto the same system disk.
// The restore wall time: from the re-image's first step to (e) verified on the serving box.
let restore_t0 = Instant::now();
let mut restore_wall = Err::<String, String>("not reached".into());
let r6b_disks = (|| -> Result<String, String> {
for m in &primary.members {
std::fs::remove_file(m).map_err(|e| format!("{}: {e}", m.display()))?;
let f = std::fs::File::create(m).map_err(|e| format!("{}: {e}", m.display()))?;
f.set_len(o.member_gib * GIB).map_err(|e| e.to_string())?;
}
let system = if o.blank_system {
let b = &primary.boot;
let len = std::fs::metadata(b)
.map_err(|e| format!("{}: {e}", b.display()))?
.len();
std::fs::remove_file(b).map_err(|e| format!("{}: {e}", b.display()))?;
std::fs::File::create(b)
.and_then(|f| f.set_len(len))
.map_err(|e| format!("{}: {e}", b.display()))?;
"onto a BLANK system disk (new system storage)"
} else {
"onto the old system disk"
};
let medium = match &o.reimage_iso {
Some(p) => {
primary.spec.iso = p.clone();
format!(
"{} (reinstall=wipe)",
p.file_name()
.map(|n| n.to_string_lossy().into_owned())
.unwrap_or_default()
)
}
None => "the first install's medium (keep: an upgrade)".to_string(),
};
let r = primary.install()?;
Ok(format!("4 data disks destroyed and blank again; re-installed {system} from {medium} in {} ms, data set laid {:?}", r.ms, r.member_names))
})();
eprintln!(" primary: {r6b_disks:?}");
let (mut r6b, mut r6c, mut r6d, mut r6e, mut r6f) = (
Err::<String, String>("not reached".into()),
Err::<String, String>("not reached".into()),
Err::<String, String>("not reached".into()),
Err::<String, String>("not reached".into()),
Err::<String, String>("not reached".into()),
);
let check = Restored {
cli: &cli,
admin: &admin,
carol_ssh: &carol_ssh,
notes: ¬es,
work: &work,
users_before: &users_before,
repos_before: &repos_before,
lfs_sha: &lfs_sha,
half: want_r10.then_some(HALF),
};
if r6b_disks.is_ok() {
primary.pin = Some(p);
let mut restart = None::<bool>;
let replay = primary.boot("reimaged-restore", |m| {
let at = format!("127.0.0.1:{}", m.grpc_private);
let st = sess.status_line(m)?;
let catalog = data::cli(&cli, &["control", "repo", "list", "-i", &k, "--at", &at]);
r6b = match (&catalog, st.contains("OPEN")) {
(Err(refusal), true) => Ok(format!(
"FRESH: `{st}`; the old admin's repo list refused ({})",
one_line(refusal)
)),
(Ok(list), true) if repo_names(list).is_empty() => {
Ok(format!("FRESH: `{st}`; no repositories"))
}
_ => Err(format!("NOT fresh: `{st}`; repo list {catalog:?}")),
};
r6b.clone()?;
// ── R9, the INTERRUPTED RESCUE (only when asked for) ─────────
// A rescue is not a happy path: the replay stream breaks. The
// contract (`gunnar-grpc/src/service/bootstrap.rs`, `restore_finish`)
// is that a stream which ends below the head the sender ANNOUNCED
// (`LogChunk.restore_head`) is not a restore: the box stays
// `restoring` and the operator resends. Here the replay client is
// killed mid-stream and the SAME command is run again; it must
// RESUME — proven by `applied` being strictly less than `head_seq`,
// because a box that started over would apply every entry again.
// MEASURED 2026-09-16: the box must NOT be sent a second Restore once one has
// finished, so R9 hands back the answer of whichever replay ended the restore —
// the resume after its cut, or, when the knife missed, the one that got away.
let mut answer = None;
if want_r9 {
let (res, out) = cut_and_resume(&cli, &twin_at, &at, &k);
r9 = res;
answer = out;
eprintln!(" R9: {r9:?}");
}
let out = match answer {
Some(out) => out,
None => data::cli(
&cli,
&[
"control", "twin", "replay", "--twin", &twin_at, "--into", &at, "-i",
&k,
],
)?,
};
let lost = field(&out, "lost").unwrap_or_default();
restart = field(&out, "restart").map(|v| v == "true");
r6c = if lost == "0" && restart.is_some() {
Ok(one_line(&out))
} else {
Err(format!(
"lost `{lost}`, restart {restart:?}: {}",
one_line(&out)
))
};
r6c.clone()?;
if restart == Some(false) {
r6d = Ok("restart false: verified without a restart".into());
r6e = check.verify(&mut sess, m);
if want_r10 {
r10_after = check.verify_half(&sess, m);
}
if r6e.is_ok() {
restore_wall = Ok(format!(
"{:.1} s from the re-image to verified",
restore_t0.elapsed().as_secs_f64()
));
}
let _ = to_twin.send(ToTwin::RestoreDone);
r6f = match recv(&from_twin)? {
FromTwin::Final(s) => Ok(s),
_ => Err("no final".into()),
};
}
Ok("replayed".into())
});
print_boot(&replay);
if let Some(e) = boot_failed(&replay) {
if r6c.is_ok() {
r6c = Err(e);
}
}
if restart == Some(true) && r6c.is_ok() {
let back = primary.boot("restored", |m| {
r6d = Ok(
"restart true: the box was restarted (one boot) and came back serving"
.into(),
);
r6e = check.verify(&mut sess, m);
if want_r10 {
r10_after = check.verify_half(&sess, m);
}
if r6e.is_ok() {
restore_wall = Ok(format!(
"{:.1} s from the re-image to verified",
restore_t0.elapsed().as_secs_f64()
));
}
let _ = to_twin.send(ToTwin::RestoreDone);
r6f = match recv(&from_twin)? {
FromTwin::Final(s) => Ok(s),
_ => Err("no final".into()),
};
Ok("verified".into())
});
print_boot(&back);
if let Some(e) = boot_failed(&back) {
r6d = Err(e);
}
// MEASURED 2026-09-16, run 2: (e) failed with a bare `transport error` and the
// guest's console went to the scrapheap with the work directory, so the one
// question worth asking — what was the box saying at that moment — could not be
// asked. A failed check on the restored box now carries the tail of its console.
if let Err(e) = &r6e {
r6e = Err(format!(
"{e}\n the restored box's console said:\n{}",
rig::tail(&back.serial, 25)
));
}
}
} else {
r6b = r6b_disks.clone();
}
let _ = to_twin.send(ToTwin::Stop);
let twin = twin_thread
.join()
.map_err(|_| "the twin thread panicked".to_string())?;
if !o.keep {
primary.destroy_disks();
twin.destroy_disks();
}
let r6b_all = match (&r6b_disks, &r6b) {
(Ok(d), Ok(f)) => Ok(format!("{d}; {f}")),
(Err(e), _) | (_, Err(e)) => Err(e.clone()),
};
let r10v = if want_r10 {
match &r10 {
Err(e) if e.starts_with(UNMEASURED) => Verdict {
id: "R10",
name: R10_NAME,
status: Status::Unrun,
detail: e.clone(),
},
_ => verdict(
"R10",
R10_NAME,
&[("killed mid-push", &r10), ("what came back", &r10_after)],
),
}
} else {
Verdict { id: "R10", name: R10_NAME, status: Status::Unrun, detail: "R10 is a phase of the H1/R6 estate and was not in --scenarios; add R10 to run it (it costs no extra install)".into() }
};
let r9v = if want_r9 {
match &r9 {
// A race the bench LOST is never a pass: the replay of this estate's
// log finished before the cut could land, so nothing was interrupted.
Err(e) if e.starts_with(UNMEASURED) => Verdict {
id: "R9",
name: R9_NAME,
status: Status::Unrun,
detail: e.clone(),
},
_ => verdict("R9", R9_NAME, &[("cut and resumed", &r9)]),
}
} else {
Verdict { id: "R9", name: R9_NAME, status: Status::Unrun, detail: "R9 is a phase of the H1/R6 estate and was not in --scenarios; add R9 to run it (it costs no extra install)".into() }
};
let r8 = if want_r8 {
match &r8_hole {
Err(e) if e.starts_with(UNMEASURED) => Verdict {
id: "R8",
name: R8_NAME,
status: Status::Unrun,
detail: e.clone(),
},
_ => verdict(
"R8",
R8_NAME,
&[("the hole", &r8_hole), ("closed by a snapshot", &r8_closed)],
),
}
} else {
Verdict { id: "R8", name: R8_NAME, status: Status::Unrun, detail: "R8 is a phase of the H1/R6 estate and was not in --scenarios; add R8 to run it (it costs no extra install)".into() }
};
Ok(vec![
verdict(
"H1",
H1_NAME,
&[
("H1a twin killed", &h1a),
("twin back", &h1a_back),
("H1b primary killed", &h1b),
],
),
r8,
r10v,
r9v,
verdict(
"R6",
R6_NAME,
&[
("(a) twin holds the watermark", &r6a),
("(b) re-imaged FRESH", &r6b_all),
("(c) replay", &r6c),
("(d) restart", &r6d),
("(e) repos + users + catalog", &r6e),
("twin follows the restored primary", &r6f),
("restore wall time", &restore_wall),
],
),
])
})
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn heartbeat_rows_parse_as_the_cli_prints_them() {
let h = "grpc bind eth1:50051\ntwin twin-box SILENT — last seen 1 m ago, lag 0 entries (applied 9 of 9), fill 1.0 % of 19 GiB, from 10.0.2.15\n why no poll for 61 s (bound 60 s)\nprimary 10.0.2.2:4000 ok — last answered 2 s ago, its head 9, applied here 9\n";
assert_eq!(row_state(h, "twin").map(|r| r.0), Some("SILENT"));
assert_eq!(row_state(h, "primary").map(|r| r.0), Some("ok"));
assert!(row_state("grpc bind x", "twin").is_none());
}
#[test]
fn repo_list_is_the_repo_lines_only() {
let l = "repo bench/alpha\nhash sha256\nbranch main\n\nrepo carol/notes\nhash sha1\n";
assert_eq!(
repo_names(l).into_iter().collect::<Vec<_>>(),
vec!["bench/alpha".to_string(), "carol/notes".to_string()]
);
assert!(repo_names("no repositories\n").is_empty());
// What `control repo list` really prints.
let bare = "bench/alpha\nbench/beta\ncarol/notes\nbench/while-twin-down\n";
assert_eq!(repo_names(bare).len(), 4);
assert!(repo_names(bare).contains("carol/notes"));
assert!(repo_names("error: refused (identity)\n").is_empty());
}
#[test]
fn the_bound_is_sixty_seconds_plus_the_slack() {
assert!(within_bound(Duration::from_millis(SILENT_AFTER_MS + 2_000)));
assert!(within_bound(Duration::from_millis(
SILENT_AFTER_MS + SLACK_MS
)));
assert!(!within_bound(Duration::from_millis(
SILENT_AFTER_MS + SLACK_MS + 1
)));
}
}