gunnar-server 3.0.0

//! **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:?}"),
        }
    }
}