gunnar-server 2.0.0

//! **The runtime data-set growth** — `/data` grows WITHOUT a restart
//! (`private-gunnar-ops/DATA-SET-GROWTH-FLOW.md` §0 ruling 1, §3; the
//! torture test's T1, T2, T6, T9, T11 are binding).
//!
//! ```text
//!   Admin.GrowStart{target, go_ahead}
//!        │ (gunnar-admin; the go-ahead's structure verified, its signer policy pluggable)
//!        ▼
//!   draining ──── pause the admission gate (every front: "gunnar.rs is growing…"),
//!        │        wait ≤ drain_timeout for in_flight == 0             T1: never kill a transfer
//!        │   timeout ─▶ unpause, failed{aborted}, NOTHING changed
//!        ▼  journal DRAIN
//!   (still draining) identity + catalog + sealed secrets → RAM (ControlDb::suspend)   T2
//!        │        close every store, sync, umount /data, a read-only tmpfs over the
//!        │        empty mountpoint, md array_state ← clear
//!        │   any failure ─▶ mount back, serve, failed{why}                         T9
//!        ▼  journal UNMOUNT, STOP
//!   unmounted ─── the set is OFF: the outside actor may take the volumes now
//!        ▼  journal WAIT
//!   waiting-members ── poll the block devices (pci rescan + survey) until every
//!        │        member is back AND larger AND equal; GrowResume looks again;
//!        │        past the bound (wait timeout) with every member present /data
//!        │        is MOUNTED BACK, failed{grow.unmounted_too_long}, alarmed; with
//!        │        a member absent it stays here, alarmed                           T6
//!        ▼  journal RESCAN
//!   growing ───── the SAME engine as at boot: G1–G3 → assemble → G4 → mount → G5 → G6
//!        │        (korp-installer::grow::runtime::grow_and_mount); on any refusal
//!        │        the set is mounted at the size the members still share           T9
//!        ▼
//!   mounting ──── ControlDb::resume (RAM copy == disk, verified), stores reopen,
//!        │        the gate unpaused
//!        ▼  journal REOPEN
//!   done | failed{why}   (failed still SERVES; only cannot-assemble/mount is
//!                         fatal, and even then the control plane and
//!                         Bootstrap.GrowStatus keep answering)
//! ```
//!
//! # Where the machine ends and the box begins
//!
//! [`GrowMachine`] is the phase machine, the journal discipline, the go-ahead
//! check and the failure policy, written over the [`World`] seam so every
//! transition is driven natively (`tests` below, with a scripted world).
//! [`world::BoxWorld`] is the appliance: the admission gate, `ServerStores`,
//! the `ControlDb` handle and edda's `korp-installer::grow::runtime`,
//! compiled only with the `grow-runtime` feature. Without the feature the
//! machine still exists — `Status` answers `idle` with `engine_present:
//! false`, `Start` is refused by name — so `gunnar control grow status`
//! tells the truth on every build.
//!
//! # The journal is the boot path's, and that is the crash story
//!
//! Every runtime step is a record in `/serialdat/grow-journal` in the exact
//! line format `korp-start` reads (`korp_installer::grow::Step`'s runtime
//! steps, `DRAIN … REOPEN`, beside the engine's own `G1 … G6`). The journal
//! decides nothing: at the next boot the engine looks at the DISK — members
//! present? larger? partitions extended? superblocks agreed? — and finishes
//! whatever is left. A crash while the members are detached halts that boot
//! on a missing member (`datahome.rs`), which is T6, and why the phase is
//! readable with no credential: the actor outside re-attaches, then reboots
//! or calls `GrowResume`.
//!
//! # What is refused during the window, and how
//!
//! * git transfers: [`gunnar_ssh::Pause::Growing`] through the one admission
//!   gate, so SSH, smart HTTP and the gRPC archive all say the same sentence;
//! * control-plane WRITES: `gunnar_core::Error::OffStripe` from the suspended
//!   store → `UNAVAILABLE` with the phase in the words (D7: refuse, never
//!   queue — the caller retries after the growth);
//! * repository opens: `ServerStores::repo_store` refuses while the stores are
//!   closed, so a background worker (gc, purge, the replicator) meets an
//!   error and retries rather than pinning the mount.
//!
//! # `Draining{kind: growing}` on the presence stream
//!
//! The shop-presence contract's `Presence.Watch` stream does not exist in
//! this tree yet. The hook is [`GrowMachine::with_announcer`]: whatever the
//! stream's fan-out is, it registers a closure and receives
//! [`Announcement::Draining`] before the pause and [`Announcement::Ready`]
//! after the reopen. Until then the same two facts are `tracing::warn!` rows
//! on `gunnar::grow` (`event = "presence.draining"` / `"presence.ready"`).

pub mod go_ahead;
pub mod world;

use std::sync::{Arc, Condvar, Mutex};
use std::time::{Duration, Instant};

pub use go_ahead::{GoAhead, GoAheadPolicy, OperatorPolicy};
#[cfg(feature = "metered")]
pub use go_ahead::MonetizePolicy;
pub use world::{Grown, MemberFacts, SurveyFacts, World};

/// The log target every row of the growth is written on.
pub const TARGET: &str = "gunnar::grow";

/// The drain's default ceiling: how long in-flight transfers may take to
/// finish before the growth ABORTS with nothing changed (T1).
pub const DEFAULT_DRAIN_TIMEOUT: Duration = Duration::from_secs(600);
/// How long `waiting-members` polls before it says so and slows down — and
/// **the bound on `/data` staying unmounted**: past it, with every member
/// present, the set is mounted back at the size the members carry and the
/// growth fails as [`UNMOUNTED_TOO_LONG`] (DEFECT D, 2026-09-29). With a member
/// absent nothing can be mounted, so the wait goes on (T6), alarmed.
pub const DEFAULT_WAIT_MEMBERS_TIMEOUT: Duration = Duration::from_secs(3600);
/// The alarm word for a set kept unmounted past its bound.
pub const UNMOUNTED_TOO_LONG: &str = "grow.unmounted_too_long";
/// How often the drain re-reads the in-flight count.
pub const DRAIN_POLL: Duration = Duration::from_millis(200);

/// Has G1's per-member target `have` reached the plan's `asked` volume size?
/// Within 1 % under, yes — the member's usable sectors sit a partition start
/// and an md data offset below the volume's, and G1 itself allows that much
/// between members. Exactly `asked` or more is always yes.
pub fn target_reached(have: u64, asked: u64) -> bool {
    have >= asked || have.saturating_mul(100) >= asked.saturating_mul(99)
}
/// How often the members are surveyed while waiting, before the timeout…
pub const WAIT_POLL: Duration = Duration::from_secs(2);
/// …and after it (still waiting, more slowly).
pub const WAIT_POLL_AFTER_TIMEOUT: Duration = Duration::from_secs(15);

/// The phase, as `Status.phase` spells it.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum Phase {
    Idle,
    Draining,
    Unmounted,
    WaitingMembers,
    Growing,
    Mounting,
    Done,
    Failed,
}

impl Phase {
    pub fn as_str(self) -> &'static str {
        match self {
            Phase::Idle => "idle",
            Phase::Draining => "draining",
            Phase::Unmounted => "unmounted",
            Phase::WaitingMembers => "waiting-members",
            Phase::Growing => "growing",
            Phase::Mounting => "mounting",
            Phase::Done => "done",
            Phase::Failed => "failed",
        }
    }
    /// A growth is under way: `Start` is refused, `Resume` may apply.
    pub fn is_active(self) -> bool {
        matches!(
            self,
            Phase::Draining | Phase::Unmounted | Phase::WaitingMembers | Phase::Growing | Phase::Mounting
        )
    }
}

/// The runtime journal steps, as the machine names them; the world spells
/// them into `korp_installer::grow::Step` (or a test's list). The words are
/// the journal's, fixed by the engine's own test.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum Step {
    Drain,
    Unmount,
    Stop,
    Wait,
    Rescan,
    Reopen,
    Abort,
    Fail,
}

impl Step {
    pub fn as_str(self) -> &'static str {
        match self {
            Step::Drain => "DRAIN",
            Step::Unmount => "UNMOUNT",
            Step::Stop => "STOP",
            Step::Wait => "WAIT",
            Step::Rescan => "RESCAN",
            Step::Reopen => "REOPEN",
            Step::Abort => "ABORT",
            Step::Fail => "FAIL",
        }
    }
}

/// What the stream (or the log) is told.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum Announcement {
    /// `Draining{kind: growing}`: before the gate is paused.
    Draining { kind: &'static str, reason: String },
    /// `Ready`: `/data` is back and the gate is open.
    Ready,
}

/// What a `GrowStart` asks for, after the request was checked.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct Plan {
    /// Per-member target in sectors; 0 = whatever every member comes back
    /// larger at.
    pub target_sectors: u64,
    pub drain_timeout: Duration,
    pub wait_timeout: Duration,
    /// Who signed the go-ahead.
    pub signer: String,
    /// ★ Non-empty: the approver's reason to grow WITHOUT a caught-up standby (cov-13;
    /// [`go_ahead::GoAhead::alone`]), one line, at most [`ALONE_MAX`] bytes.
    pub alone: String,
}

/// ★ The most of an `alone` reason that is kept (it is said in the detail and the log).
pub const ALONE_MAX: usize = 600;

/// ★ How long the drain looks for a standby to catch up when the growth was sent ALONE:
/// long enough for one that is a heartbeat behind, never the whole drain budget (the gate
/// is shut while it looks, and an emergency is a set that is refusing pushes).
pub const ALONE_FLUSH_WAIT: Duration = Duration::from_secs(5);

/// A `GrowResume`.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum Resume {
    LookAgain,
    GiveUp,
}

/// How `waiting-members` ended.
#[derive(Debug, Clone, PartialEq, Eq)]
enum Ended {
    /// Every member is back larger: the engine grows.
    Eligible,
    /// `GrowResume{give_up}`: assemble and serve what is there.
    GaveUp,
    /// The bound passed with every member present: mount back, named.
    Bound(String),
}

/// **Does every member already CARRY `target` sectors?** — the set spans it
/// (within G1's 1 %, the same allowance `waiting-members` grants), so a
/// growth to it has nothing left to do.
pub fn carries(survey: &SurveyFacts, target: u64) -> bool {
    survey.verdict == "complete" && !survey.members.is_empty() && survey.members.iter().all(|m| target_reached(m.set_bytes / 512, target))
}

/// The machine's state, behind one mutex.
#[derive(Debug, Clone)]
struct State {
    phase: Phase,
    since_unix_ms: i64,
    why: String,
    detail: String,
    epoch: u64,
    target_sectors: u64,
    signer: String,
    members: SurveyFacts,
    identity_in_ram: bool,
    serving: bool,
    in_flight: u64,
    resume: Option<Resume>,
    /// `waiting-members` passed its timeout and is still waiting.
    wait_timed_out: bool,
    /// The last thing the engine said, for `detail` after `done`/`failed`.
    grown: Option<Grown>,
    /// **The growth's alarm** ([`UNMOUNTED_TOO_LONG`], or a fatal mount),
    /// on `Admin.Health` until the next growth starts.
    alarm: Option<(String, i64, String)>,
    /// ★ cov-13: the drain went on although a standby was behind — the approver sent
    /// this growth ALONE. Said in `done`'s detail.
    alone: String,
}

impl State {
    fn idle(serving: bool) -> Self {
        State {
            phase: Phase::Idle,
            since_unix_ms: gunnar_grpc::bus::now_unix_ms(),
            why: String::new(),
            detail: String::new(),
            epoch: 0,
            target_sectors: 0,
            signer: String::new(),
            members: SurveyFacts::default(),
            identity_in_ram: false,
            serving,
            in_flight: 0,
            resume: None,
            wait_timed_out: false,
            grown: None,
            alarm: None,
            alone: String::new(),
        }
    }
}

/// Told of every [`Announcement`] a growth makes.
type Announcer = Box<dyn Fn(Announcement) + Send + Sync>;

/// **The growth, as one object the control plane holds.** Cheap to share
/// (`Arc`); the run happens on a thread of its own so the RPC that started
/// it returns at once with the first status.
pub struct GrowMachine {
    world: Arc<dyn World>,
    policy: Arc<dyn GoAheadPolicy>,
    state: Mutex<State>,
    /// Woken by `GrowResume` while `waiting-members`.
    wake: Condvar,
    announcer: Mutex<Option<Announcer>>,
}

impl std::fmt::Debug for GrowMachine {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("GrowMachine").field("phase", &self.phase()).finish()
    }
}

impl GrowMachine {
    /// A machine over `world`, with the operator go-ahead policy.
    pub fn new(world: Arc<dyn World>) -> Arc<Self> {
        Self::with_policy(world, Arc::new(OperatorPolicy::default()))
    }

    /// A machine over `world` with an explicit go-ahead policy (a metered
    /// build plugs monetize's key in here — see [`go_ahead`]).
    pub fn with_policy(world: Arc<dyn World>, policy: Arc<dyn GoAheadPolicy>) -> Arc<Self> {
        let serving = world.serving();
        // Said ONCE, at birth (estate-box3 E1): a machine that is up and idle
        // says so, so a box that never grows still shows it has one.
        tracing::info!(
            target: TARGET,
            event = "grow.up",
            engine_present = world.engine_present(),
            serving,
            "the growth machine is up — idle; {}",
            if world.engine_present() { "this build grows /data at runtime (Admin.GrowStart)" } else { "no runtime grow engine in this build: the set grows only at a restart" }
        );
        Arc::new(GrowMachine {
            world,
            policy,
            state: Mutex::new(State::idle(serving)),
            wake: Condvar::new(),
            announcer: Mutex::new(None),
        })
    }

    /// Register the presence stream's fan-out (see the module docs).
    pub fn with_announcer(self: &Arc<Self>, f: impl Fn(Announcement) + Send + Sync + 'static) {
        *self.announcer.lock().unwrap_or_else(|p| p.into_inner()) = Some(Box::new(f));
    }

    fn announce(&self, a: Announcement) {
        match &a {
            Announcement::Draining { kind, reason } => tracing::warn!(
                target: TARGET,
                event = "presence.draining",
                kind,
                reason = %reason,
                "Draining{{kind: {kind}}}: {reason}"
            ),
            Announcement::Ready => tracing::info!(target: TARGET, event = "presence.ready", "Ready: /data is back and the gate is open"),
        }
        if let Some(f) = self.announcer.lock().unwrap_or_else(|p| p.into_inner()).as_ref() {
            f(a);
        }
    }

    fn lock(&self) -> std::sync::MutexGuard<'_, State> {
        self.state.lock().unwrap_or_else(|p| p.into_inner())
    }

    /// The growth's alarm for `Admin.Health`, when it has one.
    pub fn alarm(&self) -> Option<gunnar_proto::HealthAlarm> {
        self.lock().alarm.clone().map(|(event, since_unix_ms, detail)| gunnar_proto::HealthAlarm { event, count: 1, since_unix_ms, detail })
    }

    /// Raise the growth's alarm: an `error` row on the log and the alarm on
    /// `Admin.Health`. Said on the log the first time (or `loud`); the
    /// detail follows the latest.
    fn raise(&self, event: &str, detail: String, loud: bool) {
        let mut s = self.lock();
        if loud || s.alarm.as_ref().is_none_or(|(e, _, _)| e != event) {
            tracing::error!(target: TARGET, event, "{detail}");
        }
        let since = match &s.alarm {
            Some((e, since, _)) if e == event => *since,
            _ => gunnar_grpc::bus::now_unix_ms(),
        };
        s.alarm = Some((event.to_owned(), since, detail));
    }

    /// The phase right now.
    pub fn phase(&self) -> Phase {
        self.lock().phase
    }

    fn enter(&self, phase: Phase, detail: impl Into<String>) {
        let mut s = self.lock();
        let detail = detail.into();
        tracing::info!(target: TARGET, event = "grow.phase", from = s.phase.as_str(), to = phase.as_str(), "{detail}");
        s.phase = phase;
        s.since_unix_ms = gunnar_grpc::bus::now_unix_ms();
        s.detail = detail;
    }

    fn note(&self, step: Step) -> Result<(), String> {
        let (epoch, target) = {
            let s = self.lock();
            (s.epoch, s.target_sectors)
        };
        self.world.note(epoch, target, step)
    }

    /// **A step on the way to `failed`, recorded — or the failure says it is not** (stinky row
    /// 11, 2026-10-10): `let _ = self.note(..)` lost ABORT and FAIL silently, so the journal a
    /// crash is finished from said less than the box had done. The growth is failing anyway;
    /// what the journal did not take is said at ERROR and in `why`.
    fn note_failing(&self, step: Step, why: &mut String) {
        if let Err(e) = self.note(step) {
            tracing::error!(target: TARGET, event = "grow.journal", step = step.as_str(), %e, "the grow journal did not record a step of a failing growth");
            why.push_str(&format!("; the grow journal did not record {}: {e}", step.as_str()));
        }
    }

    // ── the verbs ───────────────────────────────────────────────────────

    /// **`Admin.GrowStart`.** Checks the go-ahead and the box, moves to
    /// `draining`, and runs the growth on its own thread. Refused by name
    /// when the engine is absent, a growth is active, or the go-ahead does
    /// not verify — nothing moves in any of those cases.
    pub fn start(self: &Arc<Self>, request: &gunnar_proto::GrowStartRequest) -> Result<gunnar_proto::GrowStatusResponse, String> {
        if !self.world.engine_present() {
            return Err("this gunnar-server was built without the runtime grow engine (feature `grow-runtime`); the set grows only at a restart".into());
        }
        // Owner's rule (2026-09-15): growth WITHOUT a restart, and a medium
        // whose kernel cannot see a member leave and come back refuses it BY
        // NAME rather than draining the box into a `waiting-members` that only
        // a reboot ends. UpCloud resizes only detached volumes.
        if !self.world.hotplug() {
            return Err(format!(
                "the running kernel has no PCI hot-plug ({}): a member detached, resized and re-attached while gunnar runs is invisible to it, so a runtime growth is refused — re-image onto a medium whose kernel carries CONFIG_HOTPLUG_PCI, or enlarge every member and restart (the set grows at the next boot)",
                crate::dashboard::probe::pci_hotplug().evidence
            ));
        }
        // **The capacity door and this one must say the same thing.** AUDITED
        // 2026-09-18: `capacity::decide_growth` reported `GROWTH_SEALED —
        // serving more needs a fresh install` whenever the grow journal file
        // was absent, and its own sentence claimed this machine "refuses by
        // name" — while `start` had no journal precondition at all. A box
        // whose medium carries no grow engine therefore ACCEPTED a
        // `GrowStart`, paused the admission gate and drained live git
        // sessions, and only met the problem four steps later at
        // `note(Step::Drain)`. Nothing was ever at risk (that abort unpauses
        // with nothing touched), but an operator who had just read SEALED had
        // been told the box refuses, and it did not. The refusal moves here,
        // before anything moves, and it is the SAME file `Capacity` measures —
        // one fact, two doors, one answer. See `World::journal_present` for why
        // the file's presence is the medium's grow engine.
        if !self.world.journal_present() {
            return Err(format!(
                "this box's boot stack keeps no grow journal ({}): korp-start never opened one, so this medium's boot stack does not grow the data set and a runtime growth has nowhere to record the steps a crash would be finished from — serving more needs a fresh install onto a larger set, not another disk. This is the same fact `Admin.Capacity` reports as GROWTH_SEALED",
                self.world.journal_path().display()
            ));
        }
        let go = GoAhead::parse(&request.go_ahead)?;
        if go.target_sectors != request.target_sectors_per_member {
            return Err(format!(
                "the go-ahead names target_sectors {} and the request {}; they must agree",
                go.target_sectors, request.target_sectors_per_member
            ));
        }
        let now = gunnar_grpc::bus::now_unix_ms();
        self.policy.verify(&go, now)?;
        if !self.world.remember_nonce(&go.nonce)? {
            return Err(format!("the go-ahead's nonce {:?} was already spent on this box", go.nonce));
        }
        let plan = Plan {
            target_sectors: request.target_sectors_per_member,
            drain_timeout: if request.drain_timeout_secs == 0 { DEFAULT_DRAIN_TIMEOUT } else { Duration::from_secs(u64::from(request.drain_timeout_secs)) },
            wait_timeout: if request.wait_members_timeout_secs == 0 { DEFAULT_WAIT_MEMBERS_TIMEOUT } else { Duration::from_secs(u64::from(request.wait_members_timeout_secs)) },
            signer: go.signer.clone(),
            alone: {
                let one_line = go.alone.replace(['\n', '\r'], " ");
                let mut cut = one_line.trim().to_owned();
                if cut.len() > ALONE_MAX {
                    let mut at = ALONE_MAX;
                    while !cut.is_char_boundary(at) {
                        at -= 1;
                    }
                    cut.truncate(at);
                }
                cut
            },
        };
        {
            let mut s = self.lock();
            if s.phase.is_active() {
                return Err(format!("a growth is already {} (since {}); GrowResume looks again, and a new growth waits for this one", s.phase.as_str(), s.since_unix_ms));
            }
            if !s.serving {
                return Err(format!("/data is not being served ({}); a growth cannot start from here — reboot, or re-attach the members and GrowResume", s.why));
            }
            let survey = self.world.survey();
            if survey.verdict != "complete" {
                return Err(format!("the data set is not a complete md set right now ({}); nothing to grow", survey.verdict));
            }
            // **DEFECT D (release-3 mock, 2026-09-29): GrowStart is idempotent.**
            // A growth to a size every member ALREADY carries is finished: it
            // is `done` at once, and nothing is paused, drained or unmounted.
            // MEASURED: monetize's twin lane re-drove a finished growth with a
            // second GrowStart at the same 60 GiB; the box drained, unmounted
            // /data and waited for members to come back LARGER than 60 GiB —
            // for ever, with /data off (the off-site pull, the key re-feed and
            // every audit append failing behind it).
            if plan.target_sectors > 0 && carries(&survey, plan.target_sectors) {
                let described = survey.describe();
                *s = State::idle(true);
                s.signer = plan.signer.clone();
                s.target_sectors = plan.target_sectors;
                s.members = survey;
                s.phase = Phase::Done;
                s.since_unix_ms = now;
                s.detail = format!(
                    "done: every member already carries the {} sectors asked ({described}); nothing was drained and /data never left",
                    plan.target_sectors
                );
                drop(s);
                tracing::warn!(
                    target: TARGET,
                    event = "grow.already_met",
                    signer = %plan.signer,
                    nonce = %go.nonce,
                    target_sectors = plan.target_sectors,
                    "GrowStart for a size the members already carry: done at once, nothing drained or unmounted"
                );
                return Ok(self.status());
            }
            *s = State::idle(true);
            s.signer = plan.signer.clone();
            s.target_sectors = plan.target_sectors;
            s.members = survey;
            s.phase = Phase::Draining;
            s.since_unix_ms = now;
            s.detail = format!("draining: the gate is paused, waiting up to {} s for {} transfer(s) in flight", plan.drain_timeout.as_secs(), self.world.in_flight());
        }
        tracing::warn!(
            target: TARGET,
            event = "grow.start",
            signer = %plan.signer,
            nonce = %go.nonce,
            target_sectors = plan.target_sectors,
            drain_timeout_s = plan.drain_timeout.as_secs(),
            wait_timeout_s = plan.wait_timeout.as_secs(),
            "a runtime data-set growth begins"
        );
        // The answer is the state this call PUT the machine in (draining), read before the thread
        // exists. Read after the spawn it was whatever the growth thread had reached by then — on a
        // fast fake world, or a loaded box that ran the thread first, already `done` (seen
        // 2026-10-16 at load 25: `a_box_with_no_grow_journal_…` got "done" for "draining").
        let started = self.status();
        let me = Arc::clone(self);
        // A28 M18, stays a thread: one growth, whose status and journal are its record (Admin.GrowStatus); its start failure is answered.
        std::thread::Builder::new()
            .name("gunnar-grow".into())
            .spawn(move || me.run(plan))
            .map_err(|e| format!("spawning the growth thread: {e}"))?;
        Ok(started)
    }

    /// **`Admin.GrowResume`.** While `waiting-members`: look again now
    /// (`give_up`: stop waiting, assemble what is there, serve). On a growth
    /// that is already `done` it answers the status — the resize it would have
    /// waited for has happened and been finished, so there is nothing left to
    /// nudge and nothing wrong. In any other phase a refusal that names the
    /// phase.
    ///
    /// MEASURED on the mock 2026-09-27: monetize's twin lane resized the
    /// twin's volume and asked GrowResume; the twin had ALREADY seen the
    /// larger member, grown (G6) and served again, answered FailedPrecondition
    /// "the growth is done", and monetize kept the finished batch open and
    /// drove it again — main's growth waited for it at twin-sizing forever.
    pub fn resume(&self, give_up: bool) -> Result<gunnar_proto::GrowStatusResponse, String> {
        if self.lock().phase == Phase::Done {
            tracing::info!(target: TARGET, event = "grow.resume.done", give_up, "GrowResume on a growth that is already done: nothing to resume");
            return Ok(self.status());
        }
        {
            let s = self.lock();
            // **DEFECT D: a nudge that arrives after the box moved on is
            // progress.** MEASURED on the release-3 mock: the box's own 2 s
            // poll saw the member come back larger and moved to `growing`
            // 0.6 s before the outside actor's GrowResume; the refusal it
            // answered was read as a failed drive and the growth re-driven
            // with a second GrowStart. A box that is growing or mounting HAS
            // looked at the members — the answer is its status. (`give_up`
            // changes nothing either: the engine is already running.)
            if matches!(s.phase, Phase::Growing | Phase::Mounting) {
                drop(s);
                tracing::info!(target: TARGET, event = "grow.resume.progress", give_up, "GrowResume while the engine runs: the members were already seen; nothing to look at again");
                return Ok(self.status());
            }
        }
        {
            let mut s = self.lock();
            // `unmounted` is the instant before `waiting-members`: the flag is
            // read by the wait's first look.
            if s.phase != Phase::WaitingMembers && s.phase != Phase::Unmounted {
                return Err(format!(
                    "GrowResume applies while waiting-members; the growth is {} ({})",
                    s.phase.as_str(),
                    if s.phase == Phase::Idle { "start one with GrowStart" } else { s.detail.as_str() }
                ));
            }
            s.resume = Some(if give_up { Resume::GiveUp } else { Resume::LookAgain });
        }
        tracing::info!(target: TARGET, event = "grow.resume", give_up, "GrowResume: looking at the members again");
        self.wake.notify_all();
        Ok(self.status())
    }

    /// **`Admin.GrowStatus` / `Bootstrap.GrowStatus`.** The state, the last
    /// survey (a fresh one when idle, since it is cheap and nothing else
    /// refreshes it), and the journal as the same reader `Capacity` uses.
    pub fn status(&self) -> gunnar_proto::GrowStatusResponse {
        let mut s = self.lock().clone();
        if !s.phase.is_active() && self.world.engine_present() {
            s.members = self.world.survey();
        }
        if s.phase == Phase::Draining {
            s.in_flight = self.world.in_flight();
        }
        let journal = gunnar_grpc::grow_journal::Summary::read(&self.world.journal_path()).map(|j| j.to_wire());
        let detail = if s.detail.is_empty() {
            match s.phase {
                Phase::Idle if !self.world.engine_present() => "no runtime grow engine in this build; the set grows only at a restart".to_owned(),
                Phase::Idle => format!("idle; members: {}", s.members.describe()),
                _ => String::new(),
            }
        } else {
            s.detail.clone()
        };
        gunnar_proto::GrowStatusResponse {
            phase: s.phase.as_str().to_owned(),
            why: s.why.clone(),
            epoch: s.epoch,
            target_sectors_per_member: s.target_sectors,
            members: s
                .members
                .members
                .iter()
                .map(|m| gunnar_proto::GrowMember {
                    index: m.index,
                    device: m.device.clone(),
                    partition: m.partition.clone(),
                    disk_bytes: m.disk_bytes,
                    set_bytes: m.set_bytes,
                    offered_bytes: m.offered_bytes,
                    larger: m.larger,
                })
                .collect(),
            members_verdict: s.members.verdict.clone(),
            journal,
            engine_present: self.world.engine_present(),
            hotplug: self.world.hotplug(),
            serving: s.serving,
            identity_in_ram: s.identity_in_ram,
            in_flight: s.in_flight,
            since_unix_ms: s.since_unix_ms,
            server_unix_ms: gunnar_grpc::bus::now_unix_ms(),
            detail,
            go_ahead_signer: s.signer.clone(),
            set_uuid: s.members.set_uuid.clone(),
            total_bytes: if s.serving { self.world.mounted_bytes() } else { 0 },
        }
    }

    // ── the run ─────────────────────────────────────────────────────────

    fn run(self: Arc<Self>, plan: Plan) {
        // 1. DRAIN (T1). Pause first, so nothing new is admitted while the
        //    count is read; abort with nothing changed if it does not reach 0.
        self.announce(Announcement::Draining { kind: "growing", reason: "the data set is being grown".into() });
        self.world.pause_admission();
        let started = Instant::now();
        loop {
            let in_flight = self.world.in_flight();
            self.lock().in_flight = in_flight;
            // The start-up disk test is gunnar's own transfer: it holds a
            // probe open under /data until it is done, and the gate does not
            // see it. It is waited for like one (bench R1, 2026-09-15).
            let starting = self.world.starting();
            if in_flight == 0 && starting.is_none() {
                break;
            }
            if started.elapsed() >= plan.drain_timeout {
                self.world.unpause_admission();
                let what = match &starting {
                    Some(why) => format!("{in_flight} transfer(s) still in flight and the box still {why}"),
                    None => format!("{in_flight} transfer(s) still in flight"),
                };
                self.fail_untouched(format!(
                    "aborted: {what} after the {} s drain; nothing was changed — retry when the box is quieter, or raise drain_timeout_secs",
                    plan.drain_timeout.as_secs()
                ));
                return;
            }
            std::thread::sleep(DRAIN_POLL);
        }
        // 1b. **THE LAST WRITE HAS HAPPENED — prove the twin has it.**
        //
        // `DATA-SET-GROWTH-FLOW.md` §5 asks the drain to prove "in_flight ==
        // 0; step 0's conjunction holds again", and until now only the first
        // half was done. Step 0's proof belongs to the outside actor and is
        // taken before the growth starts; between it and this line the twin
        // may have been stopped, resized and started again — monetize's
        // `twin-sizing` phase does exactly that, and while the twin is down
        // the gate here is still open. A proof taken then is a proof about a
        // box that no longer exists.
        //
        // Here it is not. The gate is paused and nothing is in flight, so
        // what the standby does not hold now, it never will: the next thing
        // that happens to /data is that it is unmounted and its members are
        // resized underneath it. This is the only moment at which "everything
        // is drained to the twin" can be a true sentence.
        //
        // A standby that is BEHIND aborts with nothing changed, exactly as
        // the drain timeout above does — the gate is unpaused and the box
        // serves as before. Refusing here costs a growth; not refusing costs
        // whatever ruling 2 would have pulled back.
        // ★ cov-13: a growth sent ALONE looks briefly and goes on past a standby that is
        // behind (or that could not be looked at) — said, not obeyed. What that gives up:
        // ruling 2's re-lay for THIS growth (the front re-proves the twin before any re-lay
        // and refuses with one that is behind). What it keeps: nothing here destroys the set
        // (a refused or partial growth serves at the old size), and this box's transaction
        // log holds every entry the twin has not applied — it catches up from it after the
        // remount, or takes a snapshot if the byte-bounded retention had to trim past it.
        let flush_wait = if plan.alone.is_empty() { plan.drain_timeout } else { plan.drain_timeout.min(ALONE_FLUSH_WAIT) };
        let verdict = match self.world.flush_replication(flush_wait) {
            world::FlushVerdict::Behind { pending, needs_full_resync, why } if !plan.alone.is_empty() => {
                let said = format!("a standby is behind ({pending} pending, needs_full_resync {needs_full_resync}): {why}");
                tracing::warn!(target: TARGET, event = "grow.alone", pending, %why, alone = %plan.alone, "the growth goes on ALONE: {said}");
                world::FlushVerdict::NoStandby(format!("the growth goes on ALONE ({}) — {said}; the twin catches up from the log after the remount", plan.alone))
            }
            world::FlushVerdict::CouldNotLook(why) if !plan.alone.is_empty() => {
                tracing::warn!(target: TARGET, event = "grow.alone", %why, alone = %plan.alone, "the growth goes on ALONE: the standby could not be looked at");
                world::FlushVerdict::NoStandby(format!("the growth goes on ALONE ({}) — the standby could not be looked at: {why}", plan.alone))
            }
            other => other,
        };
        if !plan.alone.is_empty() {
            if let world::FlushVerdict::NoStandby(why) = &verdict {
                self.lock().alone = why.clone();
            }
        }
        match verdict {
            world::FlushVerdict::CaughtUp { standbys, waited_ms } => {
                tracing::info!(target: TARGET, event = "grow.flushed", standbys, waited_ms, "every standby has taken everything; the set may be unmounted");
                self.lock().detail = format!("drained: nothing in flight and {standbys} standby(s) caught up after {waited_ms} ms");
            }
            world::FlushVerdict::NoStandby(why) => {
                // Not a pass and not a failure. A primary growing with no
                // standby at all is a choice somebody made in the config, and
                // the run says so rather than implying a twin was checked.
                tracing::warn!(target: TARGET, event = "grow.no_standby", %why, "the set is being grown with no standby behind it");
                self.lock().detail = format!("drained: nothing in flight, and {why}");
            }
            world::FlushVerdict::CouldNotLook(why) => {
                // **NOT `NoStandby`.** There is something to prove and it
                // could not be looked at, so nothing is known — and the next
                // thing that happens to /data is that it is unmounted. A
                // growth that proceeds here is a growth that reported a flush
                // it never measured.
                self.world.unpause_admission();
                self.fail_untouched(format!(
                    "aborted: the gate is shut and nothing is in flight, but whether the twin holds this set COULD NOT BE \
                     ESTABLISHED: {why} \u{2014} nothing was changed. A set unmounted now could only be pulled back from a twin, \
                     and nothing here says there is one that could."
                ));
                return;
            }
            world::FlushVerdict::Behind { pending, needs_full_resync, why } => {
                self.world.unpause_admission();
                self.fail_untouched(format!(
                    "aborted: the gate is shut and nothing is in flight, but a standby is still behind ({pending} pending, needs_full_resync {needs_full_resync}): {why} \u{2014} nothing was changed. Nothing more is coming that would help it, and a set unmounted now could only be pulled back from a twin that does not have it."
                ));
                return;
            }
        }

        let epoch = self.world.epoch_for(plan.target_sectors);
        self.lock().epoch = epoch;
        if let Err(e) = self.note(Step::Drain) {
            self.world.unpause_admission();
            self.fail_untouched(format!("aborted: {e}; nothing was changed"));
            return;
        }

        // 2. IDENTITY TO RAM (T2), CLOSE, UNMOUNT, STOP.
        //
        // **Still `draining` until the set is OFF.** `unmounted` is read by the
        // outside actor as "the volumes are free" (monetize: main's drive and
        // the twin lane both detach → resize → attach on `unmounted |
        // waiting-members`). It used to be entered HERE, before the identity
        // copy, the store close, the sync, the unmount and the array stop, so
        // the cloud could pull a member out from under a mounted /data.
        // MEASURED on the mock 2026-09-27: the twin's second growth in one boot
        // left it with /data unwritable (every audit append failing), no volume
        // size on its heartbeat and its applied head stuck. The phase changes
        // below, once the unmount and the stop are done and on record.
        self.lock().detail = "draining → unmounting: the identity table is served from RAM, the stores close, /data unmounts, the array stops; the volumes are NOT free yet".to_owned();
        tracing::info!(target: TARGET, event = "grow.unmounting", "unmounting: identity to RAM, stores close, /data unmounts, the array stops (the phase stays draining until the set is off)");
        match self.world.suspend_identity("draining for the data-set growth") {
            Ok(facts) => {
                self.lock().identity_in_ram = true;
                tracing::info!(target: TARGET, event = "grow.identity_in_ram", documents = facts.documents, bytes = facts.bytes, "the control store is served from its RAM copy");
            }
            Err(e) => {
                self.world.unpause_admission();
                let mut why = format!("aborted: the identity table could not be copied into RAM ({e}); nothing was changed");
                self.note_failing(Step::Abort, &mut why);
                self.fail_untouched(why);
                return;
            }
        }
        if let Err(e) = self.world.close_stores() {
            self.abort_after_suspend(format!("aborted: the stores would not close ({e})"));
            return;
        }
        if let Err(e) = self.world.sync_and_unmount() {
            self.abort_after_suspend(format!("aborted: {e}"));
            return;
        }
        self.lock().serving = false;
        if let Err(e) = self.note(Step::Unmount) {
            self.recover_serve(format!("{e} after the unmount"));
            return;
        }
        if let Err(e) = self.world.stop_array() {
            self.recover_serve(format!("the array would not stop ({e})"));
            return;
        }
        if let Err(e) = self.note(Step::Stop) {
            self.recover_serve(format!("{e} after the array stopped"));
            return;
        }
        self.enter(Phase::Unmounted, "unmounted: /data is off, the stores are closed, the array is stopped; the volumes are free");

        // 3. WAIT FOR THE MEMBERS (T6-aware: a timeout keeps waiting).
        // A wait the journal did not take is no wait: the members are not let go of.
        if let Err(e) = self.note(Step::Wait) {
            self.recover_serve(format!("the grow journal did not record WAIT ({e}); the members were not let go of"));
            return;
        }
        self.enter(Phase::WaitingMembers, format!(
            "waiting for every member to come back larger{}; the outside actor detaches, resizes and re-attaches the volumes; GrowResume looks again",
            if plan.target_sectors > 0 { format!(" (≥ {} sectors each)", plan.target_sectors) } else { String::new() }
        ));
        let waited_from = Instant::now();
        let ended = loop {
            // The timeout is noticed BEFORE this iteration's detail is written,
            // so the "still waiting" words land with the survey they describe.
            let timed_out = waited_from.elapsed() >= plan.wait_timeout;
            if timed_out {
                let mut s = self.lock();
                if !s.wait_timed_out {
                    s.wait_timed_out = true;
                    tracing::warn!(target: TARGET, event = "grow.wait_timeout", waited_s = waited_from.elapsed().as_secs(), "the members did not come back larger within this growth's bound: with every member present /data is mounted back now at the size they carry; with one missing the wait goes on, alarmed — GrowResume looks again, GrowResume{{give_up}} serves what is there");
                }
            }
            let _ = self.world.rescan();
            let survey = self.world.survey();
            // The plan's target is the VOLUME size asked of the cloud; G1's
            // target is what a member can carry, always a little under it
            // (partition start, md data offset). MEASURED under KVM (bench R1
            // and R5, 2026-09-15): four 64 GiB volumes back, the survey
            // saying "eligible: 32.0GiB → 64.0GiB per member", and this line
            // waiting for ever on the exact sector count. The same 1 % the
            // engine's own G1 grants between members is granted here.
            let eligible = survey.eligible && survey.target_sectors.is_some_and(|t| plan.target_sectors == 0 || target_reached(t, plan.target_sectors));
            let same_set = survey.set_uuid.is_empty() || {
                let s = self.lock();
                s.members.set_uuid.is_empty() || s.members.set_uuid == survey.set_uuid
            };
            {
                let mut s = self.lock();
                let described = survey.describe();
                s.members = survey;
                s.detail = format!(
                    "waiting-members: {described}{}",
                    if !same_set { " — NOT THE SAME SET (uuid differs); refusing to grow it" } else if s.wait_timed_out { " — the wait timeout passed; still waiting for GrowResume or the members" } else { "" }
                );
            }
            if eligible && same_set {
                break Ended::Eligible;
            }
            // **DEFECT D's guard: /data is never left unmounted past a bound.**
            // The bound is this growth's own wait timeout (monetize asks
            // 3600 s). Until 2026-09-29 it was only logged and the wait went on
            // for ever (T6), which is how a GrowStart for a size the members
            // already had kept the twin's /data off indefinitely. Past it:
            // every member present and the same set → mount it back at the size
            // the members carry (the engine's own fallback), `failed` naming
            // UNMOUNTED_TOO_LONG, alarmed. A member ABSENT → nothing can be
            // mounted (T6 still holds, the outside actor may be re-attaching)
            // — keep waiting, but alarmed, so it is never silent.
            if timed_out {
                let waited = waited_from.elapsed().as_secs();
                let described = self.lock().members.describe();
                let complete = self.lock().members.verdict == "complete";
                if complete && same_set {
                    break Ended::Bound(format!(
                        "{UNMOUNTED_TOO_LONG}: /data stayed unmounted {waited} s, past this growth's {} s bound, waiting for members that never came back larger{} ({described}); mounted back at the size the members carry",
                        plan.wait_timeout.as_secs(),
                        if plan.target_sectors > 0 { format!(" (≥ {} sectors each)", plan.target_sectors) } else { String::new() }
                    ));
                }
                self.raise(
                    UNMOUNTED_TOO_LONG,
                    format!(
                        "/data has been unmounted {waited} s, past this growth's {} s bound, and cannot be mounted back: {} ({described}) — re-attach the members (GrowResume looks again) or reboot",
                        plan.wait_timeout.as_secs(),
                        if same_set { "a member is missing" } else { "the members are NOT the set that was unmounted" }
                    ),
                    false,
                );
            }
            let resume = self.lock().resume.take();
            match resume {
                Some(Resume::GiveUp) => break Ended::GaveUp,
                Some(Resume::LookAgain) => continue,
                None => {}
            }
            // Sleep until the next poll — or until the timeout, whichever is
            // first, so the "still waiting" row lands on time and not a poll
            // late.
            let poll = if timed_out {
                WAIT_POLL_AFTER_TIMEOUT
            } else {
                WAIT_POLL.min(plan.wait_timeout.saturating_sub(waited_from.elapsed()).max(Duration::from_millis(50)))
            };
            let guard = self.lock();
            let (guard, _) = self.wake.wait_timeout(guard, poll).unwrap_or_else(|p| p.into_inner());
            drop(guard);
        };
        if let Ended::Bound(why) = ended {
            self.raise(UNMOUNTED_TOO_LONG, why.clone(), true);
            self.recover_serve(why);
            return;
        }
        let give_up = matches!(ended, Ended::GaveUp);
        if let Err(e) = self.note(Step::Rescan) {
            tracing::warn!(target: TARGET, event = "grow.journal", %e, "RESCAN not recorded; going on (the disk is the truth)");
        }

        // 4. GROW + MOUNT (the engine; T9 inside it), then REOPEN.
        self.enter(Phase::Growing, if give_up { "growing: the operator gave up waiting; assembling and mounting what is there" } else { "growing: every member is back and larger; the engine runs G1–G6" });
        match self.world.grow_and_mount() {
            Ok(grown) => {
                self.lock().serving = true;
                self.lock().grown = Some(grown.clone());
                if grown.grew {
                    if let Err(why) = self.reopen() {
                        let why = format!("the data set grew ({}), but {why}", grown.describe());
                        self.lock().why = why.clone();
                        self.enter(Phase::Failed, format!("failed: {why}"));
                        return;
                    }
                    let alone = std::mem::take(&mut self.lock().alone);
                    self.enter(Phase::Done, if alone.is_empty() { format!("done: {}", grown.describe()) } else { format!("done: {}; grown {alone}", grown.describe()) });
                    tracing::warn!(target: TARGET, event = "grow.done", array_bytes = grown.array_bytes, "the data set grew and is being served");
                } else {
                    // FAIL before REOPEN: the journal reads "it failed, and
                    // /data is back", in that order. ★ cov-08: given up with
                    // NO member back larger, nothing was touched — the epoch
                    // is ABORTED (closed, "the set is as it was"), not a FAIL
                    // the journal reads as "after a member was touched … the
                    // next boot finishes the epoch" (the WireGuard upgrade
                    // drill d5, 2026-10-05: the spend ceiling refused the
                    // cloud's resize after main drained).
                    let untouched = give_up && self.lock().members.members.iter().all(|m| !m.larger);
                    let why = if give_up {
                        format!("gave up waiting{}; serving at the size the members share: {}", if untouched { " with no member back larger (nothing was touched)" } else { "" }, grown.describe())
                    } else {
                        format!("the engine did not grow the set; serving at the size the members share: {}", grown.describe())
                    };
                    let mut why = why;
                    self.note_failing(if untouched { Step::Abort } else { Step::Fail }, &mut why);
                    self.lock().why = why.clone();
                    match self.reopen() {
                        Ok(()) => self.enter(Phase::Failed, format!("failed (serving): {why}")),
                        Err(more) => {
                            let why = format!("{why}; and {more}");
                            self.lock().why = why.clone();
                            self.enter(Phase::Failed, format!("failed: {why}"));
                        }
                    }
                }
            }
            Err(e) => {
                // FATAL: the set could not be assembled or mounted at all. The
                // control plane and the status stay up; the identity stays in
                // RAM; the gate stays paused (there is nothing to serve).
                let mut why = format!("FATAL: /data could not be assembled or mounted ({e}); the control plane and GrowStatus keep answering, git is paused, the identity table is served from RAM; re-attach the members and GrowResume is not possible from here — reboot to let the boot path finish the epoch, or reload from the twin (a separate verb)");
                self.note_failing(Step::Fail, &mut why);
                self.lock().why = why.clone();
                self.raise("grow.fatal", why.clone(), true);
                self.enter(Phase::Failed, why);
                tracing::error!(target: TARGET, event = "grow.fatal", "{e}");
            }
        }
    }

    /// Nothing was touched: back to serving as before, `failed{why}`.
    fn fail_untouched(&self, why: String) {
        tracing::warn!(target: TARGET, event = "grow.aborted", %why, "the growth aborted with nothing changed");
        let mut s = self.lock();
        s.why = why.clone();
        s.serving = true;
        s.identity_in_ram = false;
        drop(s);
        self.enter(Phase::Failed, why);
    }

    /// The identity is in RAM and the stores may be closed, but `/data` is
    /// still mounted: put everything back and abort.
    ///
    /// **The gate opens only over stores that reopened** (stinky row 10, 2026-10-10): this used
    /// to `let _ = reopen_stores()` and unpause anyway, admitting git against closed stores.
    fn abort_after_suspend(&self, why: String) {
        let reopened = self.world.reopen_stores();
        match self.world.resume_identity() {
            Ok(r) => self.report_resume(&r),
            Err(e) => tracing::error!(target: TARGET, event = "grow.identity_resume_failed", %e, "the control store could not be put back on the stripe; it is still served from RAM"),
        }
        let mut why = why;
        self.note_failing(Step::Abort, &mut why);
        match reopened {
            Ok(()) => {
                self.world.unpause_admission();
                self.fail_untouched(why);
            }
            Err(e) => self.stores_stay_closed(format!("{why}; and the stores would not reopen ({e})")),
        }
    }

    /// **The stores did not reopen: git stays paused** — nothing is admitted against closed
    /// stores, the box says it is not serving, and the alarm says what to do.
    fn stores_stay_closed(&self, why: String) {
        let why = format!("{why}: git stays paused over closed stores — restart the box to recover");
        {
            let mut s = self.lock();
            s.why = why.clone();
            s.serving = false;
        }
        self.raise("grow.reopen_failed", why.clone(), true);
        self.enter(Phase::Failed, why);
    }

    /// Something failed after `/data` was unmounted: assemble and mount what
    /// is there (the engine's own fallback), reopen, serve, `failed{why}`
    /// (T9). Only "cannot assemble/mount at all" is fatal.
    fn recover_serve(&self, why: String) {
        tracing::warn!(target: TARGET, event = "grow.recover", %why, "recovering: mounting what is there and serving at the size it has");
        match self.world.grow_and_mount() {
            Ok(grown) => {
                self.lock().serving = true;
                self.lock().grown = Some(grown);
                let mut why = why;
                self.note_failing(Step::Fail, &mut why);
                self.lock().why = why.clone();
                match self.reopen() {
                    Ok(()) => self.enter(Phase::Failed, format!("failed (serving at the old size): {why}")),
                    Err(more) => {
                        let why = format!("{why}; and {more}");
                        self.lock().why = why.clone();
                        self.enter(Phase::Failed, format!("failed: {why}"));
                    }
                }
            }
            Err(e) => {
                let mut full = format!("FATAL: {why}; and /data could not be mounted back ({e}); the control plane and GrowStatus keep answering, git is paused, the identity table is served from RAM — reboot to let the boot path finish the epoch");
                self.note_failing(Step::Fail, &mut full);
                self.lock().why = full.clone();
                self.raise("grow.fatal", full.clone(), true);
                self.enter(Phase::Failed, full);
                tracing::error!(target: TARGET, event = "grow.fatal", "{e}");
            }
        }
    }

    /// `/data` is mounted: identity back on the stripe, stores open, gate
    /// open, `REOPEN` on record, `Ready` announced.
    ///
    /// `Err` names what went wrong (stinky rows 10 and 11): stores that did not reopen keep the
    /// gate SHUT and announce no `Ready` ([`GrowMachine::stores_stay_closed`] raised the alarm);
    /// a `REOPEN` the journal did not take is no `done`.
    fn reopen(&self) -> Result<(), String> {
        self.enter(Phase::Mounting, "mounting: /data is back; the identity table returns to the stripe and the stores reopen");
        match self.world.resume_identity() {
            Ok(r) => {
                self.lock().identity_in_ram = false;
                self.report_resume(&r);
            }
            Err(e) => tracing::error!(target: TARGET, event = "grow.identity_resume_failed", %e, "the control store could not be put back on the stripe; it is still served from RAM and every write is refused — restart to recover"),
        }
        if let Err(e) = self.world.reopen_stores() {
            let why = format!("the stores did not reopen ({e}): git stays paused over closed stores — restart the box to recover");
            self.lock().serving = false;
            self.raise("grow.reopen_failed", why.clone(), true);
            return Err(why);
        }
        let noted = self.note(Step::Reopen);
        self.world.unpause_admission();
        self.announce(Announcement::Ready);
        noted.map_err(|e| {
            tracing::error!(target: TARGET, event = "grow.journal", step = Step::Reopen.as_str(), %e, "the grow journal did not record REOPEN");
            format!("the grow journal did not record REOPEN ({e}); the stores are open and serving")
        })
    }

    fn report_resume(&self, r: &world::ResumeFacts) {
        if r.drift.is_empty() {
            tracing::info!(target: TARGET, event = "grow.identity_resumed", documents = r.documents_matched, off_stripe_ms = r.off_stripe_ms, "the control store is back on the stripe; the RAM copy and the disk agree");
        } else {
            tracing::error!(target: TARGET, event = "grow.identity_drift", drift = ?r.drift, "the control store on disk does NOT match the RAM copy that stood in for it — the disk is the truth and is what is served now; these documents changed underneath the growth");
            let mut s = self.lock();
            s.why = format!("{}{}identity drift on resume: {}", s.why, if s.why.is_empty() { "" } else { "; " }, r.drift.join(", "));
        }
    }
}

#[cfg(test)]
pub(crate) mod tests;