crates/gunnar-server/src/grow/world.rs
//! **The box, as the growth sees it** — one trait, two implementations.
//!
//! [`World`] is everything [`super::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 ([`super::tests`] scripts a
//! world) and what makes the bench's `--runtime` scenario a script over the
//! same verbs.
//!
//! [`BoxWorld`] is the appliance, compiled only with `grow-runtime`: the
//! `Arc<Admission>` all three fronts share, `ServerStores`, the `ControlDb`
//! handle, and edda's `korp-installer::grow::runtime`. [`NoEngine`] is every
//! other build: `engine_present() == false`, `Start` is refused by name and
//! `Status` still answers.
use std::path::PathBuf;
use super::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 the
/// 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 gunnar runs is invisible without it, so a runtime
/// growth is refused by name ([`crate::dashboard::probe::pci_hotplug`]).
fn hotplug(&self) -> bool {
crate::dashboard::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`
/// (0.1.12, `src/grow.rs:1072`) is called from `datahome`'s assemble pass
/// on every boot that carries the engine, and `Journal::open`
/// (`src/grow.rs:277`) 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, in which case the engine went `Journal::volatile` and this
/// box keeps no crash record either way).
///
/// MEASURED on the live appliance 2026-09-18: `data grow no journal: this
/// boot stack does not grow the data set` with `/serialdat` a real mounted
/// P3 (`boot.p3_dev /dev/vda3`, 47.9 GiB) — the medium's korp-start
/// predates the engine. `Admin.Capacity` reads the same file through
/// `gunnar_grpc::grow_journal::Summary::measure` and says `GROWTH_SEALED`
/// on it; [`super::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.
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** — the 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 (bench R1), a growth started 13 s after the banner unmounted
/// straight into `EBUSY — fd 18 → /data/.korp-disktest/probe` and
/// aborted, because the test is gunnar's own I/O and no gate pauses it.
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.** Not a pass and not a
/// failure: there is nothing to prove, and the growth says so rather than
/// implying a twin was checked. The same distinction
/// `replication_flush` makes when it answers FailedPrecondition.
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 — the exact shape of the failure this whole drain exists
/// to prevent. The same distinction `Mirror.LogRetain` already makes when
/// a twin has reported no position (`retention-twin-position-unknown`).
CouldNotLook(String),
}
/// **The drain's verdict on the twins, as a function of what they reported.**
///
/// Pure and outside the engine's feature gate, so the rule in front of the
/// unmount can be tested without a box, a store or a mounted set. `BoxWorld`
/// reads `head` and the rows and calls THIS; there is no second copy of the
/// rule.
///
/// `Ok(n)` = `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: &[gunnar_proto::TwinHeartbeat],
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 in group `twin` and has no heartbeat row at all"
)));
};
// Never polled 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. `Mirror.LogRetain`'s rule, reused.
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 by control-plane entries — holds everything it must.
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 head**, 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 head 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(gunnar_grpc::grow_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())
}
}
/// 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)
}
/// **The appliance.** Compiled only with `grow-runtime`.
#[cfg(feature = "grow-runtime")]
pub use real::BoxWorld;
#[cfg(feature = "grow-runtime")]
mod real {
use super::*;
use std::sync::{Arc, Mutex};
use korp_installer::grow::{self as engine, runtime};
/// The real world: the gate, the stores, the control database and the
/// engine.
pub struct BoxWorld {
admission: Arc<gunnar_ssh::Admission>,
stores: Arc<crate::stores::ServerStores>,
db: Arc<gunnar_core::ControlDb>,
/// **The standbys, for the flush at the end of the drain.** `None`
/// leaves [`World::flush_replication`]'s default in place, which
/// answers `NoStandby` — honest for a build with no slot, and the
/// reason this is an option rather than a required argument.
replication: Option<Arc<crate::mirror::ReplicationSlot>>,
/// **The twin heartbeat registry the gRPC plane records into**, shared
/// with it rather than a second copy: what a twin has applied is
/// reported by the twin's own poll, and this is the one place on this
/// box that holds it. `None` means the drain cannot look, which is
/// [`FlushVerdict::CouldNotLook`] and not a pass.
heartbeats: Option<Arc<gunnar_grpc::heartbeat::Registry>>,
/// The journal, opened at the first record and held for the growth's
/// whole life (one writer per box, like `korp-start`'s).
journal: Mutex<Option<engine::Journal>>,
mountpoint: PathBuf,
}
impl BoxWorld {
pub fn new(
admission: Arc<gunnar_ssh::Admission>,
stores: Arc<crate::stores::ServerStores>,
) -> Self {
let db = stores.control_db();
BoxWorld {
admission,
stores,
db,
replication: None,
heartbeats: None,
journal: Mutex::new(None),
mountpoint: runtime::data_mountpoint().to_path_buf(),
}
}
/// **Hand this world the live replication slot**, so the drain can
/// prove the standbys are caught up at the one moment that proves
/// anything. Called by `Server::start` before the world is wrapped in
/// an `Arc`, the same shape as `CoreControlPlane::with_webhooks`.
pub fn with_replication(mut self, slot: Arc<crate::mirror::ReplicationSlot>) -> Self {
self.replication = Some(slot);
self
}
/// **Hand this world the SAME heartbeat registry the gRPC plane
/// records twin polls into.** The drain's proof that a twin holds the
/// set is the twin's own reported position, and nothing else on this
/// box knows it.
pub fn with_heartbeats(mut self, registry: Arc<gunnar_grpc::heartbeat::Registry>) -> Self {
self.heartbeats = Some(registry);
self
}
/// **Has every enrolled twin applied up to this box's head?**
///
/// THE DIRECTION MATTERS. The replication roster names
/// `ssh://git@<twin>:2222` and the drain used to prove itself by
/// pushing there — but a twin refuses all git by design
/// (`gunnar_ssh` `Dispatch::authorise` answers `DispatchError::Twin`
/// before the ACL, before the catalogue, before a byte), so that send
/// can never succeed and `pending` can never fall because anything
/// arrived. It falls because nothing was ever enqueued. MEASURED
/// 2026-09-23: a roster entry reading `pending 0 · failures 0 · last
/// success NEVER` — which is `pending == 0`, and was read as caught
/// up, over a twin holding nothing at all.
///
/// The twin does hold everything, and it got it the other way round:
/// it PULLED the log over the mutual replication listener. So that is
/// what is asked about here. `head > 0` and `applied == head` are both
/// required: `lag 0 entries (applied 0 of 0)` is lag-zero containing
/// nothing, and it is exactly the reading that made this green before.
///
/// The membership read and the "has not reported" refusal are
/// `Mirror.LogRetain`'s, deliberately — one rule about when a twin's
/// position is unknown, not two.
fn twins_caught_up(&self, barrier: u64) -> Result<(u64, u64), FlushVerdict> {
let Some(hb) = self.heartbeats.as_ref() else {
return Err(FlushVerdict::CouldNotLook(
"no twin heartbeat registry was handed to the grow world, so what any twin has applied cannot be read on this box"
.to_owned(),
));
};
let members = self.stores.auth().members_of(crate::twin::TWIN_GROUP).map_err(|e| {
FlushVerdict::CouldNotLook(format!(
"the identity table could not be read for group `{}`, so which twins exist is unknown: {e}",
crate::twin::TWIN_GROUP
))
})?;
if members.is_empty() {
return Ok((0, 0));
}
// THE BARRIER, not the live head (see flush_replication).
let head = barrier;
let now = estate_clock::now_unix_ms();
let rows = hb.view(&members, now);
twin_verdict(&members, &rows, head)
}
fn with_journal<T>(
&self,
f: impl FnOnce(&mut engine::Journal) -> Result<T, String>,
) -> Result<T, String> {
let mut guard = self.journal.lock().unwrap_or_else(|p| p.into_inner());
if guard.is_none() {
let path = engine::journal_path();
let j = engine::Journal::open(&path).map_err(|e| {
format!(
"open the grow journal {}: {e} — is /serialdat mounted?",
path.display()
)
})?;
*guard = Some(j);
}
f(guard.as_mut().expect("opened above"))
}
fn log(line: &str) {
tracing::info!(target: super::super::TARGET, event = "grow.engine", "{line}");
}
}
impl World for BoxWorld {
fn engine_present(&self) -> bool {
true
}
fn journal_path(&self) -> PathBuf {
engine::journal_path()
}
fn serving(&self) -> bool {
runtime::is_mounted(&self.mountpoint) && !self.stores.suspended()
}
/// The same loop `CoreControlPlane::replication_flush` runs, at the
/// one moment the gate is shut: wait for every standby to reach
/// `pending == 0`, then report what is actually true. No `resync` is
/// attempted here — a full resync under a paused gate is a different,
/// much longer operation, and the growth's answer to "the twin cannot
/// catch up" is to abort with nothing changed, not to start one.
fn flush_replication(&self, ceiling: std::time::Duration) -> FlushVerdict {
// **A TWIN HAS NO STANDBY OF ITS OWN.** Its set is a COPY of its
// primary's — the follower it just drained (`pause_admission`) is its
// only writer — so what it holds is pulled back from the primary, not
// from a twin of the twin. The twin roster it reads here is the one it
// REPLICATED from its primary, and it names THIS box: asked, it waited
// for its own position and refused every growth. MEASURED on the mock
// 2026-09-27 (monetize's twin lane, `twin 0.6 %, 20 GB → 88 GB`):
// "whether the twin holds this set COULD NOT BE ESTABLISHED: twin
// 'gunnar-twin' has reported no position" — so the twin never grew,
// and main's growth waited for it at `twin-sizing` forever.
if self.stores.is_twin() {
return FlushVerdict::NoStandby("this box is a twin: its set is a copy of its primary's, which is where it is pulled back from".to_owned());
}
// **THE TWIN'S OWN POSITION, FIRST AND UNCONDITIONALLY.**
//
// Before anything is asked about the push roster, because the
// roster cannot answer this question: its entry points at
// `ssh://git@<twin>:2222`, and a twin serves no git to anyone.
// Whatever `pending` says there, it is not evidence that the twin
// holds the set — and this is the last moment before /data is
// unmounted, so "no evidence" must not read as "flushed".
//
// Asked even when there is no replication slot at all: a twin
// follows over its own channel, so a box with an enrolled twin and
// no roster still has a twin whose position decides whether this
// set can be pulled back.
// WAITED FOR, within this flush's ceiling: the gate is shut, so a twin
// that is BEHIND can only catch up — its next poll is seconds away.
// MEASURED loki 2026-09-26 (the pro kit's paid growth on the mock): one
// look aborted the growth at "applied 718 of this box's head 719", the
// one entry the growth itself had just written. Silence or a stale
// position is still refused at once ([`await_twins`]).
// **THE BARRIER: the head as it stands NOW**, with the gate shut and
// nothing in flight. The twin must hold everything up to it; the head
// keeps moving by a few control-plane entries a minute (MEASURED on the
// mock 2026-09-27: one entry every ~4 s, the twin always 1-2 behind the
// LIVE head, so a comparison against it never settled and main never
// grew: "twin applied 1153 of this box's head 1154"). Those later
// entries are not the set being unmounted; the twin takes them after the
// remount, as it takes every entry.
let barrier = self.stores.txlog().head_seq();
let twins = match await_twins(ceiling, std::time::Duration::from_millis(250), || {
self.twins_caught_up(barrier)
}) {
Ok(t) => t,
Err(verdict) => return verdict,
};
let Some(slot) = self.replication.as_ref() else {
if twins.0 > 0 {
return FlushVerdict::CaughtUp {
standbys: twins.0,
waited_ms: 0,
};
}
return FlushVerdict::NoStandby(
"this box has no replication slot, so there is no standby to flush to"
.to_owned(),
);
};
let Some(replicator) = slot.replicator() else {
if twins.0 > 0 {
return FlushVerdict::CaughtUp {
standbys: twins.0,
waited_ms: 0,
};
}
return FlushVerdict::NoStandby(
"no replicator is running on this box (no standby configured, or this box is not a primary); there is nothing to flush and nothing to prove".to_owned(),
);
};
let started = std::time::Instant::now();
let deadline = started + ceiling;
loop {
if replicator.status().iter().all(|s| s.pending == 0)
|| std::time::Instant::now() >= deadline
{
break;
}
std::thread::sleep(std::time::Duration::from_millis(100));
}
let status = replicator.status();
let waited_ms = started.elapsed().as_millis() as u64;
let pending: u64 = status.iter().map(|s| s.pending as u64).sum();
let needs_full_resync = status.iter().any(|s| s.needs_full_resync);
if pending == 0 && !needs_full_resync {
// `twins_caught_up` has already proved the twin's side above;
// this only adds what the push roster is entitled to say.
return FlushVerdict::CaughtUp {
standbys: status.len() as u64 + twins.0,
waited_ms,
};
}
FlushVerdict::Behind {
pending,
needs_full_resync,
why: format!(
"waited {waited_ms} ms of the {} ms the drain allows, over {} standby(s)",
ceiling.as_millis(),
status.len()
),
}
}
fn mounted_bytes(&self) -> u64 {
if !runtime::is_mounted(&self.mountpoint) {
return 0;
}
rustix::fs::statvfs(&self.mountpoint)
.map(|s| s.f_blocks * s.f_frsize)
.unwrap_or(0)
}
fn pause_admission(&self) {
self.admission.pause(gunnar_ssh::Pause::Growing);
// **A twin drains its FOLLOWER** (T21, owner ruling 2026-09-22: the twin grows exactly
// like main, never restarted). The follower is the twin's only writer under /data; this
// returns once no fetch/append/apply is in flight, and the twin says `maintenance`.
if self.stores.is_twin() {
crate::twin::maintenance::hold(self.stores.root());
}
}
fn unpause_admission(&self) {
self.admission.unpause();
if self.stores.is_twin() {
crate::twin::maintenance::release(self.stores.root());
}
}
fn in_flight(&self) -> u64 {
self.admission.in_flight()
}
fn starting(&self) -> Option<String> {
(!crate::readiness::is_ready()).then(crate::readiness::detail)
}
fn suspend_identity(&self, why: &str) -> Result<IdentityFacts, String> {
let copy = self.db.suspend(why).map_err(|e| e.to_string())?;
Ok(IdentityFacts {
documents: copy.documents(),
bytes: copy.bytes(),
})
}
fn resume_identity(&self) -> Result<ResumeFacts, String> {
let r = self.db.resume().map_err(|e| e.to_string())?;
Ok(ResumeFacts {
documents_matched: r.documents_matched,
drift: r.drift,
off_stripe_ms: r.off_stripe_ms,
})
}
fn close_stores(&self) -> Result<(), String> {
self.stores.suspend_repos("the data set is being grown");
// The transaction log keeps its open segment's `.log` + `.idx` under /data for the
// server's whole life. MEASURED 2026-09-15 (growbench R5 on the metered medium):
// they were the only descriptors left after the stores closed, and the unmount
// answered EBUSY on every runtime growth. Given up here, taken back in
// `reopen_stores` — which every abort after the drain also reaches.
self.stores.txlog().release().map_err(|e| {
format!(
"release the transaction log's files under {}: {e}",
self.mountpoint.display()
)
})
}
fn reopen_stores(&self) -> Result<(), String> {
let taken = self.stores.txlog().reacquire().map_err(|e| {
format!(
"take the transaction log's files back under {}: {e}",
self.mountpoint.display()
)
});
self.stores.resume_repos();
taken
}
fn sync_and_unmount(&self) -> Result<(), String> {
runtime::unmount(&self.mountpoint)?;
// A read-only tmpfs over the empty mountpoint: anything that still
// writes by path under /data during the window meets EROFS —
// an error it can report — instead of a file in the root
// filesystem's /data directory that the remount would hide.
nix::mount::mount(
Some("none"),
&self.mountpoint,
Some("tmpfs"),
nix::mount::MsFlags::MS_RDONLY,
Some("size=64k"),
)
.map_err(|e| format!("guard tmpfs over {}: {e}", self.mountpoint.display()))
}
fn stop_array(&self) -> Result<(), String> {
runtime::stop_array(std::time::Duration::from_secs(10))
}
fn rescan(&self) -> Result<(), String> {
runtime::pci_rescan()
}
fn survey(&self) -> SurveyFacts {
let s = runtime::survey();
SurveyFacts {
verdict: s.verdict.clone(),
raid_disks: s.raid_disks,
eligible: s.eligible(),
target_sectors: s.target_sectors(),
set_uuid: s.set_uuid.clone(),
description: s.describe(),
members: s
.members
.iter()
.map(|m| MemberFacts {
index: m.index,
device: m.disk.clone(),
partition: m.partition.clone(),
disk_bytes: m.disk_sectors * 512,
set_bytes: m.set_sectors * 512,
offered_bytes: m.device_offers * 512,
larger: m.larger,
})
.collect(),
}
}
fn grow_and_mount(&self) -> Result<Grown, String> {
// The guard tmpfs goes first; a mount over a mount would hide it
// and leave it to pin nothing forever.
if runtime::is_mounted(&self.mountpoint) {
nix::mount::umount2(&self.mountpoint, nix::mount::MntFlags::empty())
.map_err(|e| format!("unguard {}: {e}", self.mountpoint.display()))?;
}
let mountpoint = self.mountpoint.clone();
let mounted =
self.with_journal(|j| runtime::grow_and_mount(j, &mountpoint, &mut Self::log))?;
Ok(Grown {
grew: mounted.grew,
array_bytes: mounted.array_bytes,
refusal: mounted.refusal,
xfs: match &mounted.xfs {
None => String::new(),
Some(Ok((plan, after))) => format!("{plan:?} → {after} blocks"),
Some(Err(e)) => format!("FAILED: {e}"),
},
})
}
fn epoch_for(&self, target_sectors: u64) -> u64 {
self.with_journal(|j| Ok(j.epoch_for(target_sectors)))
.unwrap_or(0)
}
fn note(&self, epoch: u64, target_sectors: u64, step: Step) -> Result<(), String> {
let step = match step {
Step::Drain => engine::Step::Drain,
Step::Unmount => engine::Step::Unmount,
Step::Stop => engine::Step::Stop,
Step::Wait => engine::Step::Wait,
Step::Rescan => engine::Step::Rescan,
Step::Reopen => engine::Step::Reopen,
Step::Abort => engine::Step::Abort,
Step::Fail => engine::Step::Fail,
};
self.with_journal(|j| runtime::note(j, epoch, target_sectors, step).map(|_| ()))
}
fn remember_nonce(&self, nonce: &str) -> Result<bool, String> {
let ledger = engine::journal_path().with_file_name(NONCE_LEDGER_FILE);
remember_nonce_in(&ledger, nonce)
}
}
}
#[cfg(test)]
mod twin_verdict_tests {
use super::*;
fn row(
principal: &str,
applied: u64,
head: u64,
seen_ago_ms: u64,
last_seen: i64,
) -> gunnar_proto::TwinHeartbeat {
gunnar_proto::TwinHeartbeat {
principal: principal.to_owned(),
applied_seq: applied,
head_seq: head,
lag_entries: head.saturating_sub(applied),
last_seen_unix_ms: last_seen,
seen_ago_ms,
silent_after_ms: 60_000,
..Default::default()
}
}
fn twin() -> Vec<String> {
vec!["gunnar-twin".to_owned()]
}
/// ★ **THE READING THAT UNMOUNTED /data OVER A DEAD TWIN.**
///
/// MEASURED 2026-09-23 on the t23r mock: the primary at head 318, the twin
/// reporting `lag 0 entries (applied 0 of 0)`, and the roster entry beside
/// it reading `pending 0 · failures 0 · last success NEVER`. Both numbers
/// are zero and both mean nothing ever happened.
///
/// To see this fail with the defect back, drop the `head == 0` clause from
/// `twin_verdict`, or decide on `r.lag_entries == 0` instead.
#[test]
fn a_twin_reporting_zero_of_zero_is_never_caught_up() {
let rows = [row("gunnar-twin", 0, 0, 500, 1_790_000_000_000)];
// The twin's own lag is zero. It holds nothing.
assert_eq!(rows[0].lag_entries, 0, "the trap: lag zero");
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. MEASURED on the mock 2026-09-27: main's
/// head moved an entry every ~4 s and a twin always 1-2 behind the LIVE head
/// kept main from ever growing.
#[test]
fn a_twin_past_the_barrier_is_caught_up() {
let rows = [row("gunnar-twin", 320, 321, 500, 1_790_000_000_000)];
assert_eq!(twin_verdict(&twin(), &rows, 318), Ok((1, 318)));
let behind = [row("gunnar-twin", 317, 321, 500, 1_790_000_000_000)];
assert!(matches!(
twin_verdict(&twin(), &behind, 318),
Err(FlushVerdict::Behind { pending: 1, .. })
));
}
/// ★ A twin that has genuinely taken the log passes, and only then.
#[test]
fn a_twin_applied_up_to_the_head_is_caught_up() {
let rows = [row("gunnar-twin", 318, 318, 500, 1_790_000_000_000)];
assert_eq!(twin_verdict(&twin(), &rows, 318), Ok((1, 318)));
}
/// ★ Behind by one entry is behind.
#[test]
fn a_twin_one_entry_behind_is_behind() {
let rows = [row("gunnar-twin", 317, 318, 500, 1_790_000_000_000)];
assert!(matches!(
twin_verdict(&twin(), &rows, 318),
Err(FlushVerdict::Behind { pending: 1, .. })
));
}
/// ★ **NEVER POLLED IS NOT CAUGHT UP, AND NOT `NoStandby` EITHER.**
/// The registry is in memory, so this is the state right after a restart.
#[test]
fn a_twin_that_never_reported_could_not_be_looked_at() {
let rows = [row("gunnar-twin", 0, 0, 0, 0)];
match twin_verdict(&twin(), &rows, 318) {
Err(FlushVerdict::CouldNotLook(why)) => {
assert!(
why.contains("reported no position since this server started"),
"{why}"
);
assert!(why.contains("gunnar-twin"), "{why}");
}
other => panic!("expected CouldNotLook: {other:?}"),
}
}
/// ★ A position older than the silence bound says nothing about now.
#[test]
fn a_stale_position_could_not_be_looked_at() {
let rows = [row("gunnar-twin", 318, 318, 120_000, 1_790_000_000_000)];
assert!(matches!(
twin_verdict(&twin(), &rows, 318),
Err(FlushVerdict::CouldNotLook(_))
));
}
/// ★ An enrolled twin with no row at all is a refusal, never a pass.
#[test]
fn an_enrolled_twin_with_no_row_could_not_be_looked_at() {
assert!(matches!(
twin_verdict(&twin(), &[], 318),
Err(FlushVerdict::CouldNotLook(_))
));
}
/// ★ **A HEAD OF ZERO PROVES NOTHING**, even against a twin reporting zero.
/// Equality alone would call this caught up.
#[test]
fn a_head_of_zero_is_not_a_flush() {
let rows = [row("gunnar-twin", 0, 0, 500, 1_790_000_000_000)];
assert!(
twin_verdict(&twin(), &rows, 0).is_err(),
"applied == head == 0 must not pass"
);
}
/// ★ No twin enrolled: nothing to prove, and it does not pretend one was checked.
#[test]
fn no_enrolled_twin_is_nothing_to_prove() {
assert_eq!(twin_verdict(&[], &[], 318), Ok((0, 0)));
}
/// ★ Every enrolled twin must be caught up, not merely one of them.
#[test]
fn one_lagging_twin_among_two_refuses() {
let members = vec!["twin-a".to_owned(), "twin-b".to_owned()];
let rows = [
row("twin-a", 318, 318, 500, 1_790_000_000_000),
row("twin-b", 12, 318, 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:?}"),
}
}
}
#[cfg(all(test, feature = "grow-runtime"))]
mod twin_flush_tests {
use super::*;
use std::sync::Arc;
/// **A twin's drain proves nothing about a standby, because it has none.**
/// MEASURED on the mock 2026-09-27: the twin read the roster it replicated
/// from main (which names the twin), waited for its OWN position and
/// refused every growth ("COULD NOT BE ESTABLISHED"). A twin answers
/// `NoStandby` — its primary is where its set is pulled back from.
#[test]
fn a_twin_has_no_standby_to_flush_to() {
let d = tempfile::tempdir().unwrap();
let stores = Arc::new(crate::stores::ServerStores::open(d.path()).unwrap());
stores.mark_twin();
let w = BoxWorld::new(Arc::new(gunnar_ssh::Admission::new(None, None)), stores);
match w.flush_replication(std::time::Duration::from_millis(50)) {
FlushVerdict::NoStandby(why) => assert!(why.contains("this box is a twin"), "{why}"),
other => panic!("a twin must answer NoStandby: {other:?}"),
}
}
}