crates/appliance-grow/src/world.rs
//! **The box, as the growth sees it** — one trait, a product per implementation.
//!
//! [`World`] is everything [`crate::GrowMachine`] does to the machine:
//! pause the gate, count transfers, copy the identity to RAM, close and
//! reopen stores, unmount, stop the array, survey the members, run the
//! engine. Its shape is what makes the phase machine, the journal discipline
//! and the failure policy testable natively (`crate::tests` scripts a world)
//! and what makes a bench's `--runtime` scenario a script over the same verbs.
//!
//! A product's world is its appliance (gunnar's `BoxWorld`: its admission
//! gate, `ServerStores`, the `ControlDb`; holger's: its write hold, its
//! releasable handles), and the volume half is shared: [`crate::md::MdVolumes`]
//! (feature `md`) over korp-installer's `grow::runtime`. [`NoEngine`] is every
//! build without it: `engine_present() == false`, `start` is refused by name
//! and `status` still answers.
use std::path::PathBuf;
use crate::Step;
/// One member, as the last survey saw it (`GrowMember` on the wire).
#[derive(Debug, Clone, PartialEq, Eq, Default)]
pub struct MemberFacts {
pub index: u32,
pub device: String,
pub partition: String,
pub disk_bytes: u64,
pub set_bytes: u64,
pub offered_bytes: u64,
pub larger: bool,
}
/// The members, surveyed.
#[derive(Debug, Clone, PartialEq, Eq, Default)]
pub struct SurveyFacts {
/// `complete`, `incomplete: <why>`, `none`.
pub verdict: String,
pub raid_disks: u32,
pub members: Vec<MemberFacts>,
/// Every member present, larger, equal within 1 % — the engine would grow.
pub eligible: bool,
/// The per-member target G1 would grow to, when eligible.
pub target_sectors: Option<u64>,
/// The set's md uuid, hex.
pub set_uuid: String,
/// One line.
pub description: String,
}
impl SurveyFacts {
pub fn describe(&self) -> String {
if self.description.is_empty() {
format!("{} ({} of {})", self.verdict, self.members.len(), self.raid_disks)
} else {
self.description.clone()
}
}
}
/// What the RAM copy holds, for the log.
#[derive(Debug, Clone, PartialEq, Eq, Default)]
pub struct IdentityFacts {
pub documents: usize,
pub bytes: u64,
}
/// What the resume found.
#[derive(Debug, Clone, PartialEq, Eq, Default)]
pub struct ResumeFacts {
pub documents_matched: usize,
pub drift: Vec<String>,
pub off_stripe_ms: i64,
}
/// What the engine left behind after `grow_and_mount`.
#[derive(Debug, Clone, PartialEq, Eq, Default)]
pub struct Grown {
/// The epoch closed at its target (G6 on record).
pub grew: bool,
pub array_bytes: u64,
/// The engine's refusal, by name, when it refused.
pub refusal: Option<String>,
/// The xfs step's outcome, in words.
pub xfs: String,
}
impl Grown {
pub fn describe(&self) -> String {
let mut parts = vec![format!("array {} B", self.array_bytes)];
if self.grew {
parts.push("grown (G6)".into());
}
if let Some(r) = &self.refusal {
parts.push(format!("engine refused: {r}"));
}
if !self.xfs.is_empty() {
parts.push(format!("xfs: {}", self.xfs));
}
parts.join("; ")
}
}
/// **Everything the growth does to the box.** Every method is a verb a
/// bench's runtime scenario can name, and every one is a step a scripted
/// world can make fail.
pub trait World: Send + Sync + 'static {
/// Does this build link the engine?
fn engine_present(&self) -> bool;
/// **Does the RUNNING kernel have PCI hot-plug?** A member detached and
/// re-attached while the server runs is invisible without it, so a runtime
/// growth is refused by name ([`crate::probe::pci_hotplug`]).
fn hotplug(&self) -> bool {
crate::probe::pci_hotplug().present
}
/// `/serialdat/grow-journal`.
fn journal_path(&self) -> PathBuf;
/// **Is that journal FILE on this box?** — the one measurement that says
/// whether this MEDIUM's boot stack grows the set, as opposed to whether
/// this BINARY links the engine (`engine_present`). The two are different
/// facts and the box carries both: `korp_installer::grow::linux::open_journal`
/// is called from `datahome`'s assemble pass on every boot that carries the
/// engine, and `Journal::open` opens with `.create(true)` — so the file
/// exists from the FIRST such boot, before any growth has run, and is
/// absent exactly when korp-start never had a grow pass (or found no
/// off-stripe home to keep it on).
///
/// MEASURED on gunnar's live appliance 2026-09-18: `no journal: this boot
/// stack does not grow the data set` with `/serialdat` a real mounted P3 —
/// the medium's korp-start predated the engine. The capacity ladder reads
/// the same file ([`crate::capacity::decide_growth`] says `GROWTH_SEALED`
/// on it); [`crate::GrowMachine::start`] refuses on it, so the two doors
/// cannot disagree.
fn journal_present(&self) -> bool {
self.journal_path().exists()
}
/// Is `/data` mounted and are the stores open?
fn serving(&self) -> bool;
/// `/data`'s size as mounted, 0 when not.
fn mounted_bytes(&self) -> u64;
// ── the drain ──
fn pause_admission(&self);
fn unpause_admission(&self);
/// **Push everything to every standby and say whether they took it**,
/// bounded by `ceiling`. Called once, by the drain, after `in_flight`
/// reaches 0 — see [`FlushVerdict`].
///
/// The default is `NoStandby`: a build with no replicator linked has
/// nothing to flush, and saying so by name is the honest answer. A box
/// that HAS a standby overrides this, and a `Behind` from it aborts the
/// growth with nothing changed. **A twin answers `NoStandby`** — its set
/// is a copy of its primary's, which is where it is pulled back from
/// (gunnar b0922aa: a twin that asked about itself refused every growth).
fn flush_replication(&self, ceiling: std::time::Duration) -> FlushVerdict {
let _ = ceiling;
FlushVerdict::NoStandby("this build links no replicator, so there is no standby to flush to".to_owned())
}
fn in_flight(&self) -> u64;
/// **Is the box still STARTING** — a start-up disk test running on its
/// own thread with its probe file open under `/data`? `Some(why)` while
/// it is. The drain waits for it like a transfer: MEASURED 2026-09-15
/// under KVM (gunnar bench R1), a growth started 13 s after the banner
/// unmounted straight into `EBUSY — fd 18 → /data/.korp-disktest/probe`.
fn starting(&self) -> Option<String> {
None
}
// ── the identity (T2) ──
fn suspend_identity(&self, why: &str) -> Result<IdentityFacts, String>;
fn resume_identity(&self) -> Result<ResumeFacts, String>;
// ── the stores and the mount ──
fn close_stores(&self) -> Result<(), String>;
fn reopen_stores(&self) -> Result<(), String>;
fn sync_and_unmount(&self) -> Result<(), String>;
fn stop_array(&self) -> Result<(), String>;
// ── the members ──
fn rescan(&self) -> Result<(), String>;
fn survey(&self) -> SurveyFacts;
// ── the engine ──
/// G1–G6 and the mount, with the serve-at-old-size fallback inside;
/// `Err` only when nothing could be assembled or mounted.
fn grow_and_mount(&self) -> Result<Grown, String>;
// ── the journal ──
fn epoch_for(&self, target_sectors: u64) -> u64;
fn note(&self, epoch: u64, target_sectors: u64, step: Step) -> Result<(), String>;
// ── the go-ahead ledger ──
/// `Ok(true)` when `nonce` was never seen on this box and is now
/// recorded; `Ok(false)` when it was spent before.
fn remember_nonce(&self, nonce: &str) -> Result<bool, String>;
}
/// **What a flush to the standbys found**, at the one moment it matters: the
/// gate is paused and `in_flight == 0`, so what the twin does not have now it
/// will not be given before the set is unmounted.
///
/// `DATA-SET-GROWTH-FLOW.md` §5, "Drain", must prove: *"in_flight == 0; step
/// 0's conjunction holds again"*. Step 0's proof is taken by the OUTSIDE actor
/// before the growth starts, and between it and this moment the twin may have
/// been stopped, resized and started again — monetize's `twin-sizing` does
/// exactly that. A proof taken then is a proof about a different box.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum FlushVerdict {
/// **No standby is configured on this box** (or this box IS the twin).
/// Not a pass and not a failure: there is nothing to prove, and the growth
/// says so rather than implying a twin was checked.
NoStandby(String),
/// Every standby has taken everything.
CaughtUp { standbys: u64, waited_ms: u64 },
/// At least one standby is behind, and will stay behind: the gate is
/// already shut, so nothing more is coming that would help it.
Behind { pending: u64, needs_full_resync: bool, why: String },
/// **The question could not be asked**, so nothing is known — and a drain
/// that knows nothing must not report a flush.
///
/// Its own arm, and NOT folded into [`FlushVerdict::NoStandby`], because
/// the two are opposites at the only moment they are read: `NoStandby`
/// means "there is provably nothing to prove" and the growth continues;
/// this means "there is something to prove and it could not be looked at",
/// and the growth must abort with nothing changed. Silently treating the
/// second as the first is how `/data` gets unmounted over a twin nobody
/// asked about.
CouldNotLook(String),
}
/// **One enrolled twin, as its primary last heard from it** — the neutral
/// half of gunnar's `TwinHeartbeat`: WHO, how far it has applied this box's
/// sequence, and how fresh that word is.
#[derive(Debug, Clone, PartialEq, Eq, Default)]
pub struct TwinPosition {
pub principal: String,
/// How far into THIS box's sequence the twin has applied.
pub applied_seq: u64,
/// Wall clock of the twin's last report; 0 = never since this server started.
pub last_seen_unix_ms: i64,
/// `now - last_seen`, ms.
pub seen_ago_ms: u64,
/// A report older than this says nothing about now.
pub silent_after_ms: u64,
}
/// **The drain's verdict on the twins, as a function of what they reported.**
///
/// Pure and outside any engine gate, so the rule in front of the unmount can
/// be tested without a box, a store or a mounted set. A product's world reads
/// the barrier and the rows and calls THIS; there is no second copy of the
/// rule.
///
/// `Ok((n, barrier))` = `n` enrolled twins have each applied up to `head`.
/// Every other outcome is a refusal that names the twin.
pub fn twin_verdict(members: &[String], rows: &[TwinPosition], head: u64) -> Result<(u64, u64), FlushVerdict> {
if members.is_empty() {
return Ok((0, 0));
}
for m in members {
let Some(r) = rows.iter().find(|r| &r.principal == m) else {
return Err(FlushVerdict::CouldNotLook(format!(
"twin '{m}' is enrolled and has no heartbeat row at all"
)));
};
// Never heard since this server started. The registry is in memory,
// so a restart makes every position unknown — and unknown is the one
// window this refusal exists for.
if r.last_seen_unix_ms == 0 {
return Err(FlushVerdict::CouldNotLook(format!(
"twin '{m}' has reported no position since this server started, so what it has applied is \
unknown; nothing says it holds this set. Wait for its next poll (every few seconds while it \
follows)"
)));
}
if r.seen_ago_ms > r.silent_after_ms {
return Err(FlushVerdict::CouldNotLook(format!(
"twin '{m}' last reported {} s ago (bound {} s), so its position is stale and says nothing about \
the set as it stands now",
r.seen_ago_ms / 1000,
r.silent_after_ms / 1000
)));
}
// **BOTH clauses, and neither is the lag.** `head > 0` because
// `applied 0 of 0` is lag-zero containing nothing, and `applied >=
// head` because behind is not caught up. `head` is the drain's BARRIER
// (the head when the gate shut), so a twin already past it — the live
// head moved on — holds everything it must (gunnar 364a4f5).
if head == 0 || r.applied_seq < head {
return Err(FlushVerdict::Behind {
pending: head.saturating_sub(r.applied_seq),
needs_full_resync: false,
why: format!(
"twin '{m}' has applied {} of this box's head {head}; a twin that has not taken the log cannot \
restore what is about to be unmounted",
r.applied_seq
),
});
}
}
Ok((members.len() as u64, head))
}
/// **Wait for the twins to reach the barrier**, within `ceiling`: `look` is
/// asked again every `every` while it answers [`FlushVerdict::Behind`] (a twin
/// that follows catches up at its next poll; the gate is shut, so the barrier
/// stands still). Any other refusal — a twin that cannot be looked at — is
/// answered at once; `Behind` at the ceiling is answered as it stands.
pub fn await_twins(ceiling: std::time::Duration, every: std::time::Duration, mut look: impl FnMut() -> Result<(u64, u64), FlushVerdict>) -> Result<(u64, u64), FlushVerdict> {
let started = std::time::Instant::now();
loop {
match look() {
Err(FlushVerdict::Behind { .. }) if started.elapsed() < ceiling => std::thread::sleep(every),
other => return other,
}
}
}
/// The build without the engine: honest `idle`, refusals by name.
#[derive(Debug, Default)]
pub struct NoEngine;
impl World for NoEngine {
fn engine_present(&self) -> bool {
false
}
fn journal_path(&self) -> PathBuf {
PathBuf::from(crate::journal::JOURNAL_PATH)
}
fn serving(&self) -> bool {
true
}
fn mounted_bytes(&self) -> u64 {
0
}
fn pause_admission(&self) {}
fn unpause_admission(&self) {}
fn in_flight(&self) -> u64 {
0
}
fn suspend_identity(&self, _why: &str) -> Result<IdentityFacts, String> {
Err("no engine".into())
}
fn resume_identity(&self) -> Result<ResumeFacts, String> {
Err("no engine".into())
}
fn close_stores(&self) -> Result<(), String> {
Err("no engine".into())
}
fn reopen_stores(&self) -> Result<(), String> {
Ok(())
}
fn sync_and_unmount(&self) -> Result<(), String> {
Err("no engine".into())
}
fn stop_array(&self) -> Result<(), String> {
Err("no engine".into())
}
fn rescan(&self) -> Result<(), String> {
Ok(())
}
fn survey(&self) -> SurveyFacts {
SurveyFacts { verdict: "none".into(), description: "no runtime grow engine in this build".into(), ..Default::default() }
}
fn grow_and_mount(&self) -> Result<Grown, String> {
Err("no engine".into())
}
fn epoch_for(&self, _target_sectors: u64) -> u64 {
0
}
fn note(&self, _epoch: u64, _target_sectors: u64, _step: Step) -> Result<(), String> {
Err("no engine".into())
}
fn remember_nonce(&self, _nonce: &str) -> Result<bool, String> {
Err("no engine".into())
}
}
/// **The volume half of a [`World`], as its own seam** — everything the growth
/// does to the DATA DISKS and nothing it does to the product. A product's world
/// holds an `Arc<dyn Volumes>` and delegates these verbs to it: on the box
/// [`crate::md::MdVolumes`] (feature `md`), in a product's in-process tests a
/// fake over a tempdir, and in a build without the engine [`NoEngine`]. The
/// product's own half (its gate, its stores, its twin) is then tested against
/// the fake with the REAL machine driving it.
pub trait Volumes: Send + Sync + 'static {
/// Does this layer carry the engine (G1–G6)?
fn engine_present(&self) -> bool;
/// `/serialdat/grow-journal`.
fn journal_path(&self) -> PathBuf;
/// Is the set mounted?
fn is_mounted(&self) -> bool;
/// The mounted size, 0 when not.
fn mounted_bytes(&self) -> u64;
fn sync_and_unmount(&self) -> Result<(), String>;
fn stop_array(&self) -> Result<(), String>;
fn rescan(&self) -> Result<(), String>;
fn survey(&self) -> SurveyFacts;
fn grow_and_mount(&self) -> Result<Grown, String>;
fn epoch_for(&self, target_sectors: u64) -> u64;
fn note(&self, epoch: u64, target_sectors: u64, step: Step) -> Result<(), String>;
fn remember_nonce(&self, nonce: &str) -> Result<bool, String>;
}
impl Volumes for NoEngine {
fn engine_present(&self) -> bool {
false
}
fn journal_path(&self) -> PathBuf {
World::journal_path(self)
}
fn is_mounted(&self) -> bool {
true
}
fn mounted_bytes(&self) -> u64 {
0
}
fn sync_and_unmount(&self) -> Result<(), String> {
World::sync_and_unmount(self)
}
fn stop_array(&self) -> Result<(), String> {
World::stop_array(self)
}
fn rescan(&self) -> Result<(), String> {
Ok(())
}
fn survey(&self) -> SurveyFacts {
World::survey(self)
}
fn grow_and_mount(&self) -> Result<Grown, String> {
World::grow_and_mount(self)
}
fn epoch_for(&self, target_sectors: u64) -> u64 {
World::epoch_for(self, target_sectors)
}
fn note(&self, epoch: u64, target_sectors: u64, step: Step) -> Result<(), String> {
World::note(self, epoch, target_sectors, step)
}
fn remember_nonce(&self, nonce: &str) -> Result<bool, String> {
World::remember_nonce(self, nonce)
}
}
/// Where the spent go-ahead nonces are kept, beside the journal.
pub const NONCE_LEDGER_FILE: &str = "grow-go-aheads";
/// **The nonce ledger on disk**: one line per accepted go-ahead, appended
/// and fsynced, read back on every check. Beside the journal on
/// `/serialdat`, off the stripe, so it is readable and writable while
/// `/data` is unmounted and survives the growth.
pub fn remember_nonce_in(ledger: &std::path::Path, nonce: &str) -> Result<bool, String> {
use std::io::Write as _;
let text = std::fs::read_to_string(ledger).unwrap_or_default();
if text.lines().any(|l| l.trim() == nonce) {
return Ok(false);
}
let mut f = std::fs::OpenOptions::new()
.create(true)
.append(true)
.open(ledger)
.map_err(|e| format!("open {}: {e}", ledger.display()))?;
f.write_all(format!("{nonce}\n").as_bytes())
.and_then(|_| f.sync_all())
.map_err(|e| format!("append {}: {e}", ledger.display()))?;
Ok(true)
}
#[cfg(test)]
mod twin_verdict_tests {
use super::*;
fn row(principal: &str, applied: u64, seen_ago_ms: u64, last_seen: i64) -> TwinPosition {
TwinPosition { principal: principal.to_owned(), applied_seq: applied, last_seen_unix_ms: last_seen, seen_ago_ms, silent_after_ms: 60_000 }
}
fn twin() -> Vec<String> {
vec!["the-twin".to_owned()]
}
/// ★ **THE READING THAT UNMOUNTED /data OVER A DEAD TWIN** (gunnar,
/// MEASURED 2026-09-23): the primary at head 318, the twin reporting
/// `applied 0 of 0`. Zero means nothing ever happened.
#[test]
fn a_twin_reporting_zero_is_never_caught_up() {
let rows = [row("the-twin", 0, 500, 1_790_000_000_000)];
match twin_verdict(&twin(), &rows, 318) {
Err(FlushVerdict::Behind { pending, why, .. }) => {
assert_eq!(pending, 318);
assert!(why.contains("applied 0 of this box's head 318"), "{why}");
}
other => panic!("a twin holding nothing must never be caught up: {other:?}"),
}
}
/// A twin PAST the barrier (the live head moved on while it followed) holds
/// everything up to it and passes (gunnar 364a4f5, MEASURED on the mock
/// 2026-09-27: a twin always 1-2 behind the LIVE head kept main from ever
/// growing).
#[test]
fn a_twin_past_the_barrier_is_caught_up_and_one_short_is_behind() {
let rows = [row("the-twin", 320, 500, 1_790_000_000_000)];
assert_eq!(twin_verdict(&twin(), &rows, 318), Ok((1, 318)));
let behind = [row("the-twin", 317, 500, 1_790_000_000_000)];
assert!(matches!(twin_verdict(&twin(), &behind, 318), Err(FlushVerdict::Behind { pending: 1, .. })));
let at = [row("the-twin", 318, 500, 1_790_000_000_000)];
assert_eq!(twin_verdict(&twin(), &at, 318), Ok((1, 318)));
}
/// ★ **NEVER HEARD IS NOT CAUGHT UP, AND NOT `NoStandby` EITHER** — the
/// state right after a restart; a stale word and a missing row alike.
#[test]
fn silence_staleness_and_absence_could_not_be_looked_at() {
match twin_verdict(&twin(), &[row("the-twin", 0, 0, 0)], 318) {
Err(FlushVerdict::CouldNotLook(why)) => assert!(why.contains("reported no position since this server started") && why.contains("the-twin"), "{why}"),
other => panic!("expected CouldNotLook: {other:?}"),
}
assert!(matches!(twin_verdict(&twin(), &[row("the-twin", 318, 120_000, 1_790_000_000_000)], 318), Err(FlushVerdict::CouldNotLook(_))));
assert!(matches!(twin_verdict(&twin(), &[], 318), Err(FlushVerdict::CouldNotLook(_))));
}
/// ★ **A HEAD OF ZERO PROVES NOTHING**; no twin enrolled proves nothing and
/// pretends nothing; every enrolled twin must be caught up, not one of them.
#[test]
fn a_zero_head_no_twin_and_one_laggard_among_two() {
assert!(twin_verdict(&twin(), &[row("the-twin", 0, 500, 1_790_000_000_000)], 0).is_err());
assert_eq!(twin_verdict(&[], &[], 318), Ok((0, 0)));
let members = vec!["twin-a".to_owned(), "twin-b".to_owned()];
let rows = [row("twin-a", 318, 500, 1_790_000_000_000), row("twin-b", 12, 500, 1_790_000_000_000)];
match twin_verdict(&members, &rows, 318) {
Err(FlushVerdict::Behind { why, .. }) => assert!(why.contains("twin-b"), "{why}"),
other => panic!("expected twin-b to refuse: {other:?}"),
}
}
}