gunnar-server 3.0.0

//! The phase machine, driven natively over a scripted [`World`].
//!
//! Each test is one row of the torture test (`DATA-SET-GROWTH-FLOW.md`) or
//! one phase of §3, and every assertion is on what the world was ASKED to
//! do and in which order — the same verbs the bench's `--runtime` scenario
//! names against the real box.

use std::collections::VecDeque;
use std::path::PathBuf;
use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
use std::sync::{Arc, Mutex};
use std::time::Duration;

use super::world::{Grown, IdentityFacts, MemberFacts, ResumeFacts, SurveyFacts};
use super::*;

/// What the scripted world will answer, step by step.
#[derive(Default)]
pub(crate) struct Script {
    /// `in_flight()` answers, in order; the last one repeats.
    pub(crate) in_flight: VecDeque<u64>,
    /// `starting()` answers, in order; the last one repeats (`None` = ready).
    pub(crate) starting: VecDeque<Option<String>>,
    /// `survey()` answers, in order; the last one repeats.
    pub(crate) surveys: VecDeque<SurveyFacts>,
    pub(crate) unmount_fails: Option<String>,
    pub(crate) stop_fails: Option<String>,
    pub(crate) suspend_fails: Option<String>,
    /// `grow_and_mount()` answers, in order; the last one repeats.
    pub(crate) grown: VecDeque<Result<Grown, String>>,
    pub(crate) resume_drift: Vec<String>,
    /// `reopen_stores()` refuses with this.
    pub(crate) reopen_fails: Option<String>,
    /// The journal refuses to record this step.
    pub(crate) note_fails: Option<Step>,
}

/// A world that records every verb and answers from its script.
pub(crate) struct Fake {
    pub(crate) script: Mutex<Script>,
    calls: Mutex<Vec<String>>,
    journal: Mutex<Vec<(u64, u64, Step)>>,
    nonces: Mutex<Vec<String>>,
    paused: AtomicBool,
    serving: AtomicBool,
    suspended: AtomicBool,
    stores_closed: AtomicBool,
    surveys_taken: AtomicU64,
    hotplug: AtomicBool,
    /// What [`World::flush_replication`] answers; `None` = the trait default.
    flush: Mutex<Option<crate::grow::world::FlushVerdict>>,
    /// While `true`, `sync_and_unmount` waits (with `/data` still mounted) —
    /// the window the outside actor must not read as "the volumes are free".
    hold_unmount: AtomicBool,
    /// Set while `sync_and_unmount` is inside that wait.
    in_unmount: AtomicBool,
    /// While `true`, `grow_and_mount` waits — the engine's G1–G6 window, in
    /// which the phase reads `growing`.
    hold_grow: AtomicBool,
    dir: tempfile::TempDir,
}

impl Fake {
    pub(crate) fn new(script: Script) -> Arc<Fake> {
        let fake = Fake::without_a_journal(script);
        // An engine-carrying boot stack leaves an EMPTY journal file behind on
        // its first boot (`korp_installer::grow::Journal::open` creates it), so
        // that — and not "after the first growth" — is the ordinary box every
        // other test scripts. `without_a_journal` is the medium that has no
        // grow pass at all.
        std::fs::write(fake.journal_path(), "")
            .expect("an empty journal, as korp-start leaves one");
        fake
    }
    fn without_a_journal(script: Script) -> Arc<Fake> {
        Arc::new(Fake {
            script: Mutex::new(script),
            calls: Mutex::new(Vec::new()),
            journal: Mutex::new(Vec::new()),
            nonces: Mutex::new(Vec::new()),
            paused: AtomicBool::new(false),
            serving: AtomicBool::new(true),
            suspended: AtomicBool::new(false),
            stores_closed: AtomicBool::new(false),
            surveys_taken: AtomicU64::new(0),
            hotplug: AtomicBool::new(true),
            flush: Mutex::new(None),
            hold_unmount: AtomicBool::new(false),
            in_unmount: AtomicBool::new(false),
            hold_grow: AtomicBool::new(false),
            dir: tempfile::tempdir().expect("tempdir"),
        })
    }
    /// What the drain's flush to the standbys answers. `None` — the default —
    /// leaves the trait's own `NoStandby`, which is what every test written
    /// before the flush existed expects.
    fn flush_answers(&self, v: crate::grow::world::FlushVerdict) {
        *self.flush.lock().unwrap() = Some(v);
    }
    fn call(&self, what: &str) {
        self.calls.lock().unwrap().push(what.to_owned());
    }
    pub(crate) fn calls(&self) -> Vec<String> {
        self.calls.lock().unwrap().clone()
    }
    fn steps(&self) -> Vec<Step> {
        self.journal.lock().unwrap().iter().map(|r| r.2).collect()
    }
    fn write_journal_file(&self) {
        // The status reads the journal through the shared reader, so the fake
        // keeps a real file in the engine's line format.
        let text: String = self
            .journal
            .lock()
            .unwrap()
            .iter()
            .map(|(e, t, s)| {
                format!(
                    "epoch={e} target_sectors={t} member=- step={} at=1\n",
                    s.as_str()
                )
            })
            .collect();
        std::fs::write(self.journal_path(), text).unwrap();
    }
}

impl World for Fake {
    fn engine_present(&self) -> bool {
        true
    }
    fn hotplug(&self) -> bool {
        self.hotplug.load(Ordering::SeqCst)
    }
    fn journal_path(&self) -> PathBuf {
        self.dir.path().join("grow-journal")
    }
    fn serving(&self) -> bool {
        self.serving.load(Ordering::SeqCst)
    }
    fn mounted_bytes(&self) -> u64 {
        if self.serving() {
            4096
        } else {
            0
        }
    }
    fn flush_replication(&self, _ceiling: std::time::Duration) -> crate::grow::world::FlushVerdict {
        self.call("flush");
        self.flush.lock().unwrap().clone().unwrap_or_else(|| {
            crate::grow::world::FlushVerdict::NoStandby("the fake was given no standby".to_owned())
        })
    }
    fn pause_admission(&self) {
        self.call("pause");
        self.paused.store(true, Ordering::SeqCst);
    }
    fn unpause_admission(&self) {
        self.call("unpause");
        self.paused.store(false, Ordering::SeqCst);
    }
    fn in_flight(&self) -> u64 {
        let mut s = self.script.lock().unwrap();
        if s.in_flight.len() > 1 {
            s.in_flight.pop_front().unwrap()
        } else {
            s.in_flight.front().copied().unwrap_or(0)
        }
    }
    fn starting(&self) -> Option<String> {
        let mut s = self.script.lock().unwrap();
        if s.starting.len() > 1 {
            s.starting.pop_front().unwrap()
        } else {
            s.starting.front().cloned().unwrap_or(None)
        }
    }
    fn suspend_identity(&self, _why: &str) -> Result<IdentityFacts, String> {
        self.call("suspend_identity");
        if let Some(e) = self.script.lock().unwrap().suspend_fails.clone() {
            return Err(e);
        }
        self.suspended.store(true, Ordering::SeqCst);
        Ok(IdentityFacts {
            documents: 3,
            bytes: 999,
        })
    }
    fn resume_identity(&self) -> Result<ResumeFacts, String> {
        self.call("resume_identity");
        self.suspended.store(false, Ordering::SeqCst);
        Ok(ResumeFacts {
            documents_matched: 3,
            drift: self.script.lock().unwrap().resume_drift.clone(),
            off_stripe_ms: 5,
        })
    }
    fn close_stores(&self) -> Result<(), String> {
        self.call("close_stores");
        self.stores_closed.store(true, Ordering::SeqCst);
        Ok(())
    }
    fn reopen_stores(&self) -> Result<(), String> {
        self.call("reopen_stores");
        if let Some(e) = self.script.lock().unwrap().reopen_fails.clone() {
            return Err(e);
        }
        self.stores_closed.store(false, Ordering::SeqCst);
        Ok(())
    }
    fn sync_and_unmount(&self) -> Result<(), String> {
        self.call("unmount");
        while self.hold_unmount.load(Ordering::SeqCst) {
            self.in_unmount.store(true, Ordering::SeqCst);
            std::thread::sleep(Duration::from_millis(5));
        }
        self.in_unmount.store(false, Ordering::SeqCst);
        if let Some(e) = self.script.lock().unwrap().unmount_fails.clone() {
            return Err(e);
        }
        self.serving.store(false, Ordering::SeqCst);
        Ok(())
    }
    fn stop_array(&self) -> Result<(), String> {
        self.call("stop_array");
        if let Some(e) = self.script.lock().unwrap().stop_fails.clone() {
            return Err(e);
        }
        Ok(())
    }
    fn rescan(&self) -> Result<(), String> {
        self.call("rescan");
        Ok(())
    }
    fn survey(&self) -> SurveyFacts {
        self.surveys_taken.fetch_add(1, Ordering::SeqCst);
        let mut s = self.script.lock().unwrap();
        if s.surveys.len() > 1 {
            s.surveys.pop_front().unwrap()
        } else {
            s.surveys.front().cloned().unwrap_or_default()
        }
    }
    fn grow_and_mount(&self) -> Result<Grown, String> {
        self.call("grow_and_mount");
        while self.hold_grow.load(Ordering::SeqCst) {
            std::thread::sleep(Duration::from_millis(5));
        }
        let mut s = self.script.lock().unwrap();
        let answer = if s.grown.len() > 1 {
            s.grown.pop_front().unwrap()
        } else {
            s.grown
                .front()
                .cloned()
                .unwrap_or_else(|| Err("no script".into()))
        };
        if answer.is_ok() {
            self.serving.store(true, Ordering::SeqCst);
        }
        answer
    }
    fn epoch_for(&self, _target_sectors: u64) -> u64 {
        7
    }
    fn note(&self, epoch: u64, target_sectors: u64, step: Step) -> Result<(), String> {
        if self.script.lock().unwrap().note_fails == Some(step) {
            return Err(format!(
                "the grow journal refused {}: ENOSPC",
                step.as_str()
            ));
        }
        self.journal
            .lock()
            .unwrap()
            .push((epoch, target_sectors, step));
        self.write_journal_file();
        Ok(())
    }
    fn remember_nonce(&self, nonce: &str) -> Result<bool, String> {
        let mut n = self.nonces.lock().unwrap();
        if n.iter().any(|x| x == nonce) {
            return Ok(false);
        }
        n.push(nonce.to_owned());
        Ok(true)
    }
}

pub(crate) fn complete(larger: bool) -> SurveyFacts {
    let member = |i: u32| MemberFacts {
        index: i,
        device: format!("vd{}", (b'b' + i as u8) as char),
        partition: format!("vd{}1", (b'b' + i as u8) as char),
        disk_bytes: if larger { 200 << 20 } else { 100 << 20 },
        set_bytes: 100 << 20,
        offered_bytes: if larger { 200 << 20 } else { 100 << 20 },
        larger,
    };
    SurveyFacts {
        verdict: "complete".into(),
        raid_disks: 4,
        members: (0..4).map(member).collect(),
        eligible: larger,
        target_sectors: larger.then_some(400_000),
        set_uuid: "5d".repeat(16),
        description: if larger {
            "complete (4 of 4): all larger".into()
        } else {
            "complete (4 of 4): as before".into()
        },
    }
}

fn go(target: u64, nonce: &str) -> Vec<u8> {
    GoAhead::operator(target, nonce, gunnar_grpc::bus::now_unix_ms())
}

fn request(target: u64, nonce: &str) -> gunnar_proto::GrowStartRequest {
    gunnar_proto::GrowStartRequest {
        target_sectors_per_member: target,
        go_ahead: go(target, nonce),
        drain_timeout_secs: 2,
        wait_members_timeout_secs: 1,
    }
}

fn wait_for(m: &GrowMachine, phase: Phase) -> gunnar_proto::GrowStatusResponse {
    let deadline = std::time::Instant::now() + Duration::from_secs(20);
    loop {
        if m.phase() == phase {
            return m.status();
        }
        assert!(
            std::time::Instant::now() < deadline,
            "the machine never reached {phase:?}; it is {:?} ({})",
            m.phase(),
            m.status().detail
        );
        std::thread::sleep(Duration::from_millis(20));
    }
}

fn wait_settled(m: &GrowMachine) -> gunnar_proto::GrowStatusResponse {
    let deadline = std::time::Instant::now() + Duration::from_secs(20);
    loop {
        if !m.phase().is_active() {
            return m.status();
        }
        assert!(
            std::time::Instant::now() < deadline,
            "the machine never settled; it is {:?} ({})",
            m.phase(),
            m.status().detail
        );
        std::thread::sleep(Duration::from_millis(20));
    }
}

/// **The whole path, end to end** (§3 steps 5–9): drain → identity to RAM →
/// close → unmount → stop → wait → (members larger) → grow → resume identity
/// → reopen → unpause → done; every verb in that order and every journal
/// record in the engine's order; `Status` readable throughout; the gate
/// paused for exactly the window.
#[test]
fn a_growth_runs_drain_unmount_wait_grow_mount_reopen_in_that_order() {
    let script = Script {
        in_flight: VecDeque::from([2, 1, 0]),
        // Surveys are taken by: the idle `status()` below, `start()`, then the
        // wait loop — which must see one "as before" before the members grow.
        surveys: VecDeque::from([
            complete(false),
            complete(false),
            complete(false),
            complete(true),
        ]),
        grown: VecDeque::from([Ok(Grown {
            grew: true,
            array_bytes: 800 << 20,
            refusal: None,
            xfs: "grown".into(),
        })]),
        ..Script::default()
    };
    let world = Fake::new(script);
    let m = GrowMachine::new(world.clone());
    let announced = Arc::new(Mutex::new(Vec::new()));
    let a = Arc::clone(&announced);
    m.with_announcer(move |x| a.lock().unwrap().push(x));

    let idle = m.status();
    assert_eq!(
        (idle.phase.as_str(), idle.engine_present, idle.serving),
        ("idle", true, true)
    );
    assert_eq!(idle.members.len(), 4);

    let first = m.start(&request(0, "n1")).expect("started");
    assert_eq!(first.phase, "draining");
    assert_eq!(first.go_ahead_signer, "operator");
    let s = wait_for(&m, Phase::WaitingMembers);
    assert!(!s.serving && s.identity_in_ram, "{s:?}");
    assert_eq!(s.epoch, 7);
    assert!(
        s.journal.as_ref().is_some_and(|j| j.phase == "in_progress"),
        "{:?}",
        s.journal
    );
    assert!(
        world.paused.load(Ordering::SeqCst),
        "the gate is paused through the window"
    );
    let done = wait_settled(&m);
    assert_eq!((done.phase.as_str(), done.why.as_str()), ("done", ""));
    assert!(done.serving && !done.identity_in_ram);
    assert!(
        !world.paused.load(Ordering::SeqCst),
        "the gate is open again"
    );
    assert!(!world.suspended.load(Ordering::SeqCst) && !world.stores_closed.load(Ordering::SeqCst));
    assert_eq!(
        world.calls(),
        vec![
            "pause",
            "flush",
            "suspend_identity",
            "close_stores",
            "unmount",
            "stop_array",
            "rescan",
            "rescan",
            "grow_and_mount",
            "resume_identity",
            "reopen_stores",
            "unpause",
        ]
    );
    assert_eq!(
        world.steps(),
        vec![
            Step::Drain,
            Step::Unmount,
            Step::Stop,
            Step::Wait,
            Step::Rescan,
            Step::Reopen
        ]
    );
    assert!(
        done.journal
            .as_ref()
            .is_some_and(|j| j.phase == "in_progress"),
        "the fake writes no G6; the engine would"
    );
    let announced = announced.lock().unwrap();
    assert!(matches!(
        announced.first(),
        Some(Announcement::Draining {
            kind: "growing",
            ..
        })
    ));
    assert!(matches!(announced.last(), Some(Announcement::Ready)));
    // A second growth may start now, with a fresh nonce; the spent one is refused.
    assert!(m
        .start(&request(0, "n1"))
        .unwrap_err()
        .contains("already spent"));
}

/// **T1 — a long push in flight at the drain.** Bounded wait, then ABORT
/// with nothing changed: no journal record, no suspend, no unmount, the gate
/// reopened, `failed{aborted…}` and still serving.
#[test]
fn a_drain_that_times_out_aborts_with_nothing_changed() {
    let script = Script {
        in_flight: VecDeque::from([3]),
        surveys: VecDeque::from([complete(false)]),
        ..Script::default()
    };
    let world = Fake::new(script);
    let m = GrowMachine::new(world.clone());
    let mut req = request(0, "n1");
    req.drain_timeout_secs = 1;
    m.start(&req).expect("started");
    let s = wait_settled(&m);
    assert_eq!(s.phase, "failed");
    assert!(
        s.why.contains("aborted")
            && s.why.contains("3 transfer(s)")
            && s.why.contains("nothing was changed"),
        "{}",
        s.why
    );
    assert!(s.serving && !s.identity_in_ram);
    assert_eq!(world.calls(), vec!["pause", "unpause"]);
    assert!(
        world.steps().is_empty(),
        "nothing on the journal: no epoch was opened"
    );
    // And a new growth may be started.
    script_reset(&world, VecDeque::from([0]));
    assert_eq!(m.start(&request(0, "n2")).unwrap().phase, "draining");
}

/// The measured row: a growth started while the start-up disk test still
/// runs waits for it (its probe file is open under /data and no gate sees
/// it), and one that never finishes is named in the abort.
#[test]
fn the_drain_waits_for_the_start_up_disk_test_and_names_it_when_it_never_ends() {
    let starting = "starting: the start-up disk test is still running".to_owned();
    // Ready on the fourth poll: the drain waits, then proceeds.
    let script = Script {
        in_flight: VecDeque::from([0]),
        starting: VecDeque::from([
            Some(starting.clone()),
            Some(starting.clone()),
            Some(starting.clone()),
            None,
        ]),
        surveys: VecDeque::from([complete(false), complete(true)]),
        grown: VecDeque::from([Ok(Grown {
            grew: true,
            array_bytes: 800 << 20,
            refusal: None,
            xfs: "grown".into(),
        })]),
        ..Script::default()
    };
    let world = Fake::new(script);
    let m = GrowMachine::new(world.clone());
    m.start(&request(0, "n1")).expect("started");
    let s = wait_settled(&m);
    assert_eq!(s.phase, "done", "{}", s.why);
    assert!(
        world.steps().contains(&Step::Drain),
        "the drain closed only after the test finished: {:?}",
        world.steps()
    );
    // Never ready: the abort says so, with nothing changed.
    let script = Script {
        in_flight: VecDeque::from([0]),
        starting: VecDeque::from([Some(starting.clone())]),
        surveys: VecDeque::from([complete(false)]),
        ..Script::default()
    };
    let world = Fake::new(script);
    let m = GrowMachine::new(world.clone());
    let mut req = request(0, "n2");
    req.drain_timeout_secs = 1;
    m.start(&req).expect("started");
    let s = wait_settled(&m);
    assert_eq!(s.phase, "failed");
    assert!(
        s.why.contains("aborted")
            && s.why.contains("still running")
            && s.why.contains("nothing was changed"),
        "{}",
        s.why
    );
    assert!(s.serving && !s.identity_in_ram);
    assert!(world.steps().is_empty(), "nothing on the journal");
}

/// The measured row: the operator asked for 64 GiB volumes, the members came
/// back as 64 GiB volumes, and G1 can carry a little less than 64 GiB on each
/// — the growth must go on, not wait for ever on the exact sector count.
#[test]
fn a_member_a_hair_under_the_asked_volume_size_still_reaches_the_target() {
    let asked: u64 = 64 << 21; // 64 GiB in 512-byte sectors
    assert!(super::target_reached(asked, asked));
    assert!(super::target_reached(asked + 1, asked));
    assert!(
        super::target_reached(asked - 4096, asked),
        "a partition start under the volume size"
    );
    assert!(
        super::target_reached(asked - asked / 100, asked),
        "1 % under"
    );
    assert!(
        !super::target_reached(asked / 2, asked),
        "half the ask is not the ask"
    );
    assert!(
        !super::target_reached(asked - asked / 50, asked),
        "2 % under is a smaller volume"
    );
    let mut back = complete(true);
    back.target_sectors = Some(asked - 4096);
    let script = Script {
        in_flight: VecDeque::from([0]),
        surveys: VecDeque::from([complete(false), back]),
        grown: VecDeque::from([Ok(Grown {
            grew: true,
            array_bytes: 800 << 20,
            refusal: None,
            xfs: "grown".into(),
        })]),
        ..Script::default()
    };
    let world = Fake::new(script);
    let m = GrowMachine::new(world.clone());
    m.start(&request(asked, "n-hair")).expect("started");
    let s = wait_settled(&m);
    assert_eq!(s.phase, "done", "{}", s.why);
}

fn script_reset(world: &Fake, in_flight: VecDeque<u64>) {
    world.script.lock().unwrap().in_flight = in_flight;
}

/// **An unmount that is refused (EBUSY)** after the identity went to RAM:
/// stores reopen, identity back on the stripe, gate open, `ABORT` closes the
/// epoch, nothing on any member changed.
#[test]
fn a_refused_unmount_puts_everything_back_and_aborts() {
    let script = Script {
        in_flight: VecDeque::from([0]),
        surveys: VecDeque::from([complete(false)]),
        unmount_fails: Some(
            "umount /data: EBUSY — 2 descriptor(s) of this process still hold it: [fd 9 → /data/x]"
                .into(),
        ),
        ..Script::default()
    };
    let world = Fake::new(script);
    let m = GrowMachine::new(world.clone());
    m.start(&request(0, "n1")).expect("started");
    let s = wait_settled(&m);
    assert_eq!(s.phase, "failed");
    assert!(
        s.why.contains("aborted") && s.why.contains("EBUSY"),
        "{}",
        s.why
    );
    assert!(s.serving && !s.identity_in_ram);
    assert_eq!(
        world.calls(),
        vec![
            "pause",
            "flush",
            "suspend_identity",
            "close_stores",
            "unmount",
            "reopen_stores",
            "resume_identity",
            "unpause"
        ]
    );
    assert_eq!(world.steps(), vec![Step::Drain, Step::Abort]);
    assert!(
        s.journal.as_ref().is_some_and(|j| j.phase == "aborted"),
        "{:?}",
        s.journal
    );
}

/// **T6 — the members do not come back in time.** The phase STAYS
/// `waiting-members` past the timeout (the outside actor may be working),
/// `Status` keeps answering, and a `Resume` looks again at once; when the
/// members are back larger the growth finishes.
#[test]
fn waiting_for_members_never_fails_on_its_timeout_and_resume_looks_again() {
    let mut gone = SurveyFacts {
        verdict: "incomplete: member 1 of 4 is missing".into(),
        raid_disks: 4,
        ..Default::default()
    };
    gone.description = gone.verdict.clone();
    let script = Script {
        in_flight: VecDeque::from([0]),
        surveys: VecDeque::from([complete(false), gone]),
        grown: VecDeque::from([Ok(Grown {
            grew: true,
            array_bytes: 800 << 20,
            refusal: None,
            xfs: "grown".into(),
        })]),
        ..Script::default()
    };
    let world = Fake::new(script);
    let m = GrowMachine::new(world.clone());
    let mut req = request(0, "n1");
    req.wait_members_timeout_secs = 1;
    m.start(&req).expect("started");
    let s = wait_for(&m, Phase::WaitingMembers);
    assert!(s.members_verdict.starts_with("incomplete"), "{s:?}");
    std::thread::sleep(Duration::from_millis(1500));
    let s = m.status();
    assert_eq!(s.phase, "waiting-members", "a timeout is not a failure");
    assert!(s.detail.contains("timeout passed"), "{}", s.detail);
    assert!(!s.serving && s.identity_in_ram);
    // Resume with the members still gone: looks again, stays waiting.
    let r = m.resume(false).expect("resume applies while waiting");
    assert_eq!(r.phase, "waiting-members");
    // Now they are back and larger: Resume finishes the growth.
    world.script.lock().unwrap().surveys = VecDeque::from([complete(true)]);
    m.resume(false).expect("resume");
    let done = wait_settled(&m);
    assert_eq!(done.phase, "done", "{}", done.detail);
    assert!(world.steps().contains(&Step::Rescan) && world.steps().contains(&Step::Reopen));
    // A late Resume on the finished growth answers `done` — the outside actor's
    // nudge arrived after the box finished by itself (the twin lane, 2026-09-27).
    let late = m
        .resume(false)
        .expect("a resume after done answers the status");
    assert_eq!(late.phase, "done", "{}", late.detail);
}

/// **`Resume{give_up}`** while waiting: stop waiting, assemble what is
/// there, serve — `failed` (nothing grew) but serving, with the reason.
#[test]
fn giving_up_while_waiting_serves_what_is_there() {
    let script = Script {
        in_flight: VecDeque::from([0]),
        surveys: VecDeque::from([complete(false)]),
        grown: VecDeque::from([Ok(Grown {
            grew: false,
            array_bytes: 400 << 20,
            refusal: Some("nothing to do".into()),
            xfs: String::new(),
        })]),
        ..Script::default()
    };
    let world = Fake::new(script);
    let m = GrowMachine::new(world.clone());
    m.start(&request(0, "n1")).expect("started");
    wait_for(&m, Phase::WaitingMembers);
    m.resume(true).expect("give up");
    let s = wait_settled(&m);
    assert_eq!(s.phase, "failed");
    assert!(s.serving && !s.identity_in_ram);
    assert!(s.why.contains("gave up"), "{}", s.why);
    assert!(!world.paused.load(Ordering::SeqCst));
    assert_eq!(world.steps().last(), Some(&Step::Reopen));
    // ★ cov-08 (upgrade drill d5, 2026-10-05): NO member came back larger, so
    // nothing was touched — the epoch is ABORTED (closed, "the set is as it
    // was"), never a FAIL the journal reads as "after a member was touched …
    // the next boot finishes the epoch". MEASURED: the old line wrote FAIL for
    // a growth the spend ceiling refused after main drained, and main carried
    // "epoch 2 FAILED … short 53.1 GiB" into every capacity answer after.
    assert!(
        world.steps().contains(&Step::Abort) && !world.steps().contains(&Step::Fail),
        "{:?}",
        world.steps()
    );
    world.write_journal_file();
    let j = m.status().journal.expect("the journal");
    assert_eq!(j.phase, "aborted", "{}", j.detail);
}

/// **Giving up after SOME member came back larger** (a partial resize): a
/// member was touched by the outside actor, so the epoch is FAILED and stays
/// open — the next boot (or `grow resume`) finishes it once the rest are larger.
#[test]
fn giving_up_after_a_member_came_back_larger_keeps_the_epoch_failed() {
    let mut partial = complete(false);
    partial.members[0].larger = true;
    partial.members[0].disk_bytes = 200 << 20;
    partial.description = "complete (4 of 4): 1 larger".into();
    let script = Script {
        in_flight: VecDeque::from([0]),
        surveys: VecDeque::from([partial]),
        grown: VecDeque::from([Ok(Grown {
            grew: false,
            array_bytes: 400 << 20,
            refusal: Some("the members are unequal".into()),
            xfs: String::new(),
        })]),
        ..Script::default()
    };
    let world = Fake::new(script);
    let m = GrowMachine::new(world.clone());
    m.start(&request(0, "n1")).expect("started");
    wait_for(&m, Phase::WaitingMembers);
    m.resume(true).expect("give up");
    let s = wait_settled(&m);
    assert_eq!(s.phase, "failed");
    assert!(
        world.steps().contains(&Step::Fail) && !world.steps().contains(&Step::Abort),
        "{:?}",
        world.steps()
    );
}

/// **T9 — the engine refuses or the xfs grow fails**: `/data` is served at
/// the old size, `failed{why}` names it, the gate is open, and the epoch
/// stays open (`FAIL`, no `ABORT`) for the next boot or `grow start`.
#[test]
fn an_engine_refusal_serves_at_the_old_size_and_keeps_the_epoch_open() {
    let script = Script {
        in_flight: VecDeque::from([0]),
        surveys: VecDeque::from([complete(false), complete(true)]),
        grown: VecDeque::from([Ok(Grown {
            grew: false,
            array_bytes: 400 << 20,
            refusal: None,
            xfs: "FAILED: XFS_IOC_FSGROWFSDATA: EIO".into(),
        })]),
        ..Script::default()
    };
    let world = Fake::new(script);
    let m = GrowMachine::new(world.clone());
    m.start(&request(0, "n1")).expect("started");
    let s = wait_settled(&m);
    assert_eq!(s.phase, "failed");
    assert!(s.serving, "serving at the old size");
    assert!(s.why.contains("XFS_IOC_FSGROWFSDATA"), "{}", s.why);
    assert!(!world.paused.load(Ordering::SeqCst));
    assert!(
        s.journal.as_ref().is_some_and(|j| j.phase == "failed"),
        "{:?}",
        s.journal
    );
    assert_eq!(
        world.steps(),
        vec![
            Step::Drain,
            Step::Unmount,
            Step::Stop,
            Step::Wait,
            Step::Rescan,
            Step::Fail,
            Step::Reopen
        ],
        "FAIL, then /data is back"
    );
    // A new growth resumes it: same box, fresh nonce.
    world.script.lock().unwrap().surveys = VecDeque::from([complete(true)]);
    world.script.lock().unwrap().grown = VecDeque::from([Ok(Grown {
        grew: true,
        array_bytes: 800 << 20,
        refusal: None,
        xfs: "grown".into(),
    })]);
    m.start(&request(0, "n2"))
        .expect("a failed-but-serving box may try again");
    assert_eq!(wait_settled(&m).phase, "done");
}

/// **T9's fatal arm**: the set cannot be assembled or mounted at all. The
/// machine reports `failed`, `serving: false`, keeps the identity in RAM and
/// the gate paused, and `Status` still answers — the control plane is up.
#[test]
fn a_set_that_cannot_be_mounted_is_fatal_but_the_status_still_answers() {
    let script = Script {
        in_flight: VecDeque::from([0]),
        surveys: VecDeque::from([complete(false), complete(true)]),
        grown: VecDeque::from([Err(
            "korp data set: 3 of 4 members present — member 2 of 4 is missing".into(),
        )]),
        ..Script::default()
    };
    let world = Fake::new(script);
    let m = GrowMachine::new(world.clone());
    m.start(&request(0, "n1")).expect("started");
    let s = wait_settled(&m);
    assert_eq!(s.phase, "failed");
    assert!(!s.serving && s.identity_in_ram, "{s:?}");
    assert!(
        s.why.contains("FATAL") && s.why.contains("member 2 of 4"),
        "{}",
        s.why
    );
    assert!(
        world.paused.load(Ordering::SeqCst),
        "nothing to serve: the gate stays paused"
    );
    assert_eq!(s.total_bytes, 0);
    assert!(m
        .start(&request(0, "n2"))
        .unwrap_err()
        .contains("not being served"));
    assert!(world.steps().contains(&Step::Fail));
}

/// **A stop that fails after the unmount** recovers through the engine's
/// own mount (T9) and reports why.
#[test]
fn an_array_that_will_not_stop_is_mounted_back_and_served() {
    let script = Script {
        in_flight: VecDeque::from([0]),
        surveys: VecDeque::from([complete(false)]),
        stop_fails: Some("md: /sys/block/md_korp_data still exists".into()),
        grown: VecDeque::from([Ok(Grown {
            grew: false,
            array_bytes: 400 << 20,
            refusal: None,
            xfs: String::new(),
        })]),
        ..Script::default()
    };
    let world = Fake::new(script);
    let m = GrowMachine::new(world.clone());
    m.start(&request(0, "n1")).expect("started");
    let s = wait_settled(&m);
    assert_eq!(s.phase, "failed");
    assert!(s.serving && s.why.contains("would not stop"), "{}", s.why);
    assert_eq!(
        world.calls(),
        vec![
            "pause",
            "flush",
            "suspend_identity",
            "close_stores",
            "unmount",
            "stop_array",
            "grow_and_mount",
            "resume_identity",
            "reopen_stores",
            "unpause"
        ]
    );
}

/// **T14 — a growth nobody approved.** An empty, malformed, stale,
/// mismatched-target or spent go-ahead is refused BEFORE anything moves;
/// so is a second `Start` while one runs, and a `Start` on an incomplete
/// set.
#[test]
fn the_go_ahead_and_the_box_are_checked_before_anything_moves() {
    let script = Script {
        in_flight: VecDeque::from([5]),
        surveys: VecDeque::from([complete(false)]),
        ..Script::default()
    };
    let world = Fake::new(script);
    let m = GrowMachine::new(world.clone());
    // A target ABOVE what the fixture's members carry (100 MiB = 204 800
    // sectors each): a target they already carry is finished at once and
    // never drains (`a_growstart_whose_target_the_members_already_carry_…`).
    const T: u64 = 1_000_000;
    let mut r = request(T, "n1");
    r.go_ahead = Vec::new();
    assert!(m.start(&r).unwrap_err().contains("empty"));
    r.go_ahead = go(T - 1, "n1");
    assert!(m.start(&r).unwrap_err().contains("must agree"));
    r.go_ahead = GoAhead::operator(T, "n-old", 1);
    assert!(m.start(&r).unwrap_err().contains("past the"));
    assert!(world.calls().is_empty(), "nothing moved");
    // A running growth refuses a second start.
    r.go_ahead = go(T, "n1");
    r.drain_timeout_secs = 3;
    m.start(&r).expect("started");
    assert!(m
        .start(&request(T, "n2"))
        .unwrap_err()
        .contains("already draining"));
    script_reset(&world, VecDeque::from([0]));
    world.script.lock().unwrap().unmount_fails = Some("x".into());
    wait_settled(&m);
    // An incomplete set refuses a start.
    world.script.lock().unwrap().surveys = VecDeque::from([SurveyFacts {
        verdict: "incomplete: member 3 of 4 is missing".into(),
        ..Default::default()
    }]);
    assert!(m
        .start(&request(T, "n3"))
        .unwrap_err()
        .contains("not a complete md set"));
}

/// **The build without the engine**: `Status` is honest and `Start` is a
/// refusal by name — the credential-free status must never say
/// `Unimplemented` to the outside actor.
#[test]
fn without_the_engine_status_answers_idle_and_start_is_refused() {
    let m = GrowMachine::new(Arc::new(world::NoEngine));
    let s = m.status();
    assert_eq!(
        (s.phase.as_str(), s.engine_present, s.serving),
        ("idle", false, true)
    );
    assert!(s.detail.contains("no runtime grow engine"), "{}", s.detail);
    assert!(m
        .start(&request(0, "n1"))
        .unwrap_err()
        .contains("grow-runtime"));
    assert!(m
        .resume(false)
        .unwrap_err()
        .contains("applies while waiting-members"));
}

/// **T2 — identity drift on resume is reported, never hidden**: the disk is
/// the truth, and the difference lands in `why`.
#[test]
fn identity_drift_on_resume_is_named_in_the_status() {
    let script = Script {
        in_flight: VecDeque::from([0]),
        surveys: VecDeque::from([complete(false), complete(true)]),
        grown: VecDeque::from([Ok(Grown {
            grew: true,
            array_bytes: 800 << 20,
            refusal: None,
            xfs: "grown".into(),
        })]),
        resume_drift: vec!["refs/gunnar/auth: ram aa vs disk bb".into()],
        ..Script::default()
    };
    let world = Fake::new(script);
    let m = GrowMachine::new(world.clone());
    m.start(&request(0, "n1")).expect("started");
    let s = wait_settled(&m);
    assert_eq!(s.phase, "done");
    assert!(
        s.why.contains("identity drift") && s.why.contains("refs/gunnar/auth"),
        "{}",
        s.why
    );
}

/// The nonce ledger on disk: spent once, refused twice, survives a reopen.
#[test]
fn the_nonce_ledger_spends_each_go_ahead_once() {
    let dir = tempfile::tempdir().unwrap();
    let ledger = dir.path().join(world::NONCE_LEDGER_FILE);
    assert!(world::remember_nonce_in(&ledger, "a").unwrap());
    assert!(!world::remember_nonce_in(&ledger, "a").unwrap());
    assert!(world::remember_nonce_in(&ledger, "b").unwrap());
    assert_eq!(std::fs::read_to_string(&ledger).unwrap(), "a\nb\n");
}

/// **Owner's rule (2026-09-15): growth WITHOUT a restart, and a medium whose
/// kernel has no PCI hot-plug refuses a runtime growth BY NAME** — before the
/// go-ahead is even read, so nothing drains and no nonce is spent. The status
/// carries `hotplug` either way.
#[test]
fn a_kernel_without_pci_hotplug_refuses_a_runtime_growth_by_name() {
    let world = Fake::new(Script::default());
    world.hotplug.store(false, Ordering::SeqCst);
    let m = GrowMachine::new(world.clone());
    assert!(!m.status().hotplug);
    let err = m.start(&request(1000, "hp1")).unwrap_err();
    assert!(
        err.contains("no PCI hot-plug") && err.contains("refused"),
        "{err}"
    );
    assert_eq!(m.status().phase, "idle", "nothing moved");
    assert!(
        world.calls().is_empty(),
        "nothing drained: {:?}",
        world.calls()
    );
    world.hotplug.store(true, Ordering::SeqCst);
    assert!(m.status().hotplug);
}

/// **The audit's finding, 2026-09-18: `Capacity` said SEALED and `GrowStart`
/// would have accepted the growth.** A box whose boot stack keeps no grow
/// journal refuses `GrowStart` BY NAME — before the go-ahead is read, so
/// nothing drains, no nonce is spent and the gate is never paused — and the
/// same box with a journal still starts. The two doors now read one file.
#[test]
fn a_box_with_no_grow_journal_refuses_growstart_by_name_and_one_with_a_journal_starts() {
    // No journal: the medium's korp-start has no grow pass at all.
    let script = Script {
        in_flight: VecDeque::from([0]),
        surveys: VecDeque::from([complete(false)]),
        ..Script::default()
    };
    let world = Fake::without_a_journal(script);
    assert!(
        !world.journal_path().exists(),
        "the fixture is a box with no journal file"
    );
    let m = GrowMachine::new(world.clone());
    let err = m.start(&request(0, "nj1")).unwrap_err();
    assert!(
        err.contains("keeps no grow journal") && err.contains("GROWTH_SEALED"),
        "the refusal names the journal and the capacity verdict it agrees with: {err}"
    );
    assert_eq!(m.status().phase, "idle", "nothing moved");
    assert!(
        world.calls().is_empty(),
        "the gate was never paused: {:?}",
        world.calls()
    );
    assert!(
        world.nonces.lock().unwrap().is_empty(),
        "no nonce was spent on a refused start"
    );

    // The same box once its boot stack keeps one — an EMPTY journal, which is
    // what the first boot with the engine leaves, not one a growth wrote.
    std::fs::write(world.journal_path(), "").unwrap();
    let script = Script {
        in_flight: VecDeque::from([0]),
        surveys: VecDeque::from([complete(false), complete(true)]),
        grown: VecDeque::from([Ok(Grown {
            grew: true,
            array_bytes: 800 << 20,
            refusal: None,
            xfs: "grown".into(),
        })]),
        ..Script::default()
    };
    *world.script.lock().unwrap() = script;
    assert_eq!(m.start(&request(0, "nj2")).unwrap().phase, "draining");
    assert_eq!(wait_settled(&m).phase, "done");
}

/// **The metered appliance's third refusal: a REPLAY.** The signature is
/// monetize's and still verifies for ever — that is what a signature is — so
/// what makes a go-ahead good once is the nonce ledger, which `start` spends
/// before anything drains. A go-ahead monetize signed starts a growth; the
/// SAME bytes, replayed after it, are refused by name with nothing paused and
/// nothing drained. (`monetize-embed` and `ed25519-dalek` are dev-dependencies
/// of this crate, so the go-ahead here is signed over the one canonical form,
/// `monetize_embed::signing::go_ahead_message`.)
#[cfg(feature = "metered")]
#[test]
fn a_monetize_signed_go_ahead_starts_a_growth_and_a_replay_of_it_is_refused_by_name() {
    use ed25519_dalek::SigningKey;

    let monetize = SigningKey::from_bytes(&[3u8; 32]);
    let script = Script {
        in_flight: VecDeque::from([0]),
        surveys: VecDeque::from([complete(false), complete(true)]),
        grown: VecDeque::from([Ok(Grown {
            grew: true,
            array_bytes: 800 << 20,
            refusal: None,
            xfs: "grown".into(),
        })]),
        ..Script::default()
    };
    let world = Fake::new(script);
    let m = GrowMachine::with_policy(
        world.clone(),
        Arc::new(
            go_ahead::MonetizePolicy::new(&monetize.verifying_key().to_bytes())
                .expect("a real key"),
        ),
    );

    let bytes =
        go_ahead::metered_tests::signed(&monetize, 0, "mz-1", gunnar_grpc::bus::now_unix_ms());
    let req = gunnar_proto::GrowStartRequest {
        target_sectors_per_member: 0,
        go_ahead: bytes.clone(),
        drain_timeout_secs: 2,
        wait_members_timeout_secs: 1,
    };
    let first = m
        .start(&req)
        .expect("monetize's go-ahead starts the growth");
    assert_eq!(
        (first.phase.as_str(), first.go_ahead_signer.as_str()),
        ("draining", "monetize")
    );
    assert_eq!(wait_settled(&m).phase, "done");

    // The same bytes again: the signature is as valid as it ever was.
    let err = m.start(&req).unwrap_err();
    assert!(
        err.contains("already spent") && err.contains("mz-1"),
        "the replay is refused by name: {err}"
    );
    assert_eq!(m.status().phase, "done", "the replay moved nothing");

    // And an operator go-ahead still works on the same metered box.
    assert_eq!(m.start(&request(0, "op-after")).unwrap().phase, "draining");
    assert_eq!(wait_settled(&m).phase, "done");
}

/// ★ **A STANDBY THAT IS STILL BEHIND WHEN THE GATE SHUTS ABORTS THE GROWTH,
/// AND NOTHING IS TOUCHED.**
///
/// `DATA-SET-GROWTH-FLOW.md` §5 asks the drain to prove two things —
/// *"in_flight == 0; step 0's conjunction holds again"* — and only the first
/// was done. Step 0's proof is the outside actor's, taken before the growth
/// starts; between it and this moment the twin may have been stopped, resized
/// and started again (monetize's `twin-sizing` does exactly that, with this
/// gate still open). A proof taken then is a proof about a different box.
///
/// So the drain takes its own, at the one moment it means anything: the gate
/// is shut, nothing is in flight, and 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.
///
/// The refusal is the drain-timeout's shape: unpause, `failed`, nothing
/// changed. It costs a growth. Not refusing costs whatever ruling 2 would
/// have had to pull back from a twin that did not have it.
#[test]
fn a_standby_still_behind_when_the_gate_shuts_aborts_the_growth_untouched() {
    let script = Script {
        // A complete set that is ready to grow, so the ONLY thing that can stop
        // this growth is the flush.
        surveys: VecDeque::from([complete(false), complete(true)]),
        ..Script::default()
    };
    let world = Fake::new(script);
    world.flush_answers(crate::grow::world::FlushVerdict::Behind {
        pending: 19_264,
        needs_full_resync: false,
        why: "the twin read `applied 0 of 19264` and went quiet".to_owned(),
    });
    let m = GrowMachine::new(world.clone());

    m.start(&request(0, "n1")).expect("started");
    let done = wait_settled(&m);

    assert_eq!(done.phase, "failed", "{done:?}");
    assert!(
        done.why.contains("a standby is still behind"),
        "{}",
        done.why
    );
    assert!(
        done.why.contains("19264"),
        "the refusal must carry the number: {}",
        done.why
    );
    assert!(done.serving, "the box must still be serving: {done:?}");
    assert!(
        !world.paused.load(Ordering::SeqCst),
        "the gate was left shut"
    );

    // **Nothing was touched.** The identity never went to RAM, no store was
    // closed, nothing was unmounted, and the journal has no record — the
    // growth ended before `Step::Drain` was written.
    assert_eq!(
        world.calls(),
        vec!["pause", "flush", "unpause"],
        "something happened after the flush refused"
    );
    assert!(
        world.steps().is_empty(),
        "a journal record was written for a growth that never began: {:?}",
        world.steps()
    );
}

/// ★ **A BOX WITH NO STANDBY GROWS, AND SAYS SO BY NAME.**
///
/// `NoStandby` is not a pass and not a failure. A primary growing with no
/// standby behind it is a choice somebody made in the config, and the run
/// says which rather than implying a twin was checked — the same distinction
/// `CoreControlPlane::replication_flush` makes when it answers
/// FailedPrecondition, and the same one `grow::twin_sizing` makes on the
/// front for a missing twin lane.
#[test]
fn a_box_with_no_standby_grows_and_the_detail_says_there_was_none() {
    let script = Script {
        surveys: VecDeque::from([complete(false), complete(false), complete(true)]),
        grown: VecDeque::from([Ok(Grown {
            grew: true,
            array_bytes: 800 << 20,
            refusal: None,
            xfs: "grown".into(),
        })]),
        ..Script::default()
    };
    let world = Fake::new(script);
    world.flush_answers(crate::grow::world::FlushVerdict::NoStandby(
        "no replicator is running on this box".to_owned(),
    ));
    let m = GrowMachine::new(world.clone());

    m.start(&request(0, "n1")).expect("started");
    let done = wait_settled(&m);

    assert_eq!(
        (done.phase.as_str(), done.why.as_str()),
        ("done", ""),
        "{done:?}"
    );
    assert!(
        world.calls().contains(&"grow_and_mount".to_owned()),
        "{:?}",
        world.calls()
    );
}

/// ★ **A DRAIN THAT COULD NOT LOOK MUST NOT UNMOUNT.**
///
/// `CouldNotLook` is the opposite of `NoStandby` at the only moment either is
/// 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. Folding the second into the first is how `/data` gets unmounted
/// over a twin nobody asked about.
///
/// To see this fail with the defect back, make the `CouldNotLook` arm in
/// `grow/mod.rs` behave like the `NoStandby` arm (warn and carry on).
#[test]
fn a_drain_that_could_not_look_aborts_the_growth_untouched() {
    let script = Script {
        surveys: VecDeque::from([complete(false), complete(true)]),
        ..Script::default()
    };
    let world = Fake::new(script);
    world.flush_answers(crate::grow::world::FlushVerdict::CouldNotLook(
        "twin 'gunnar-twin' has reported no position since this server started".to_owned(),
    ));
    let m = GrowMachine::new(world.clone());

    m.start(&request(0, "n1")).expect("started");
    let done = wait_settled(&m);

    assert_eq!(
        done.phase, "failed",
        "a growth that proved nothing must not reach done: {done:?}"
    );
    assert!(
        done.why.contains("COULD NOT BE ESTABLISHED"),
        "{}",
        done.why
    );
    assert!(
        done.why.contains("gunnar-twin"),
        "the refusal must name the twin: {}",
        done.why
    );
    assert!(done.serving, "the box must still be serving: {done:?}");
    assert!(
        !world.paused.load(Ordering::SeqCst),
        "the gate was left shut"
    );
    // Nothing was touched: no identity to RAM, no close, no unmount, no journal record.
    assert_eq!(
        world.calls(),
        vec!["pause", "flush", "unpause"],
        "something happened after the drain could not look"
    );
    assert!(
        world.steps().is_empty(),
        "a journal record was written for a growth that never began: {:?}",
        world.steps()
    );
}

/// ★ **`lag 0 entries (applied 0 of 0)` IS NOT A FLUSH.**
///
/// The reading that made this green over a dead twin. A twin that has taken
/// nothing reports lag zero, because zero of zero is caught up — and the
/// roster entry beside it read `pending 0 · failures 0 · last success NEVER`,
/// which is `pending == 0`. Both numbers are innocent and both mean nothing
/// happened. MEASURED 2026-09-23 on the t23r mock against a primary at head
/// 318 whose twin had applied 0.
#[test]
fn a_twin_at_zero_of_zero_is_behind_and_never_caught_up() {
    let script = Script {
        surveys: VecDeque::from([complete(false), complete(true)]),
        ..Script::default()
    };
    let world = Fake::new(script);
    world.flush_answers(crate::grow::world::FlushVerdict::Behind {
        pending: 318,
        needs_full_resync: false,
        why: "twin 'gunnar-twin' has applied 0 of this box's head 318".to_owned(),
    });
    let m = GrowMachine::new(world.clone());

    m.start(&request(0, "n1")).expect("started");
    let done = wait_settled(&m);

    assert_eq!(done.phase, "failed", "{done:?}");
    assert!(
        done.why.contains("applied 0 of this box's head 318"),
        "{}",
        done.why
    );
    assert!(
        world.steps().is_empty(),
        "nothing may be journalled: {:?}",
        world.steps()
    );
}

/// **A twin one entry behind is waited for, not refused** (MEASURED loki
/// 2026-09-26: "applied 718 of this box's head 719" aborted a paid growth on
/// one look). Behind until the third look, then caught up: `Ok`. Behind past
/// the ceiling: `Behind`. A twin that cannot be looked at: refused at once.
#[test]
fn a_twin_behind_is_waited_for_within_the_ceiling_and_silence_is_refused_at_once() {
    use crate::grow::world::{await_twins, FlushVerdict};
    use std::time::Duration;
    let behind = || FlushVerdict::Behind {
        pending: 1,
        needs_full_resync: false,
        why: "applied 718 of 719".into(),
    };
    let mut n = 0;
    let got = await_twins(Duration::from_secs(5), Duration::from_millis(1), || {
        n += 1;
        if n < 3 {
            Err(behind())
        } else {
            Ok((1, 719))
        }
    });
    assert!(matches!(got, Ok((1, 719))) && n == 3, "{got:?} after {n}");
    let got = await_twins(Duration::from_millis(20), Duration::from_millis(5), || {
        Err(behind())
    });
    assert!(
        matches!(got, Err(FlushVerdict::Behind { pending: 1, .. })),
        "{got:?}"
    );
    let mut looks = 0;
    let got = await_twins(Duration::from_secs(5), Duration::from_millis(1), || {
        looks += 1;
        Err(FlushVerdict::CouldNotLook("silent".into()))
    });
    assert!(
        matches!(got, Err(FlushVerdict::CouldNotLook(_))) && looks == 1,
        "{got:?} after {looks}"
    );
}

/// ★ **`unmounted` is a PROMISE to the outside actor: the volumes are free.**
///
/// monetize's drive (main's and the twin lane's alike) waits for `unmounted |
/// waiting-members` and then detaches, resizes and re-attaches the volumes —
/// "the appliance is unmounted: the volumes are free". The machine used to
/// enter `unmounted` BEFORE it copied the identity to RAM, closed the stores,
/// synced and unmounted /data and stopped the array, so for that whole window
/// the status said the set was off while it was still mounted.
///
/// MEASURED on the mock 2026-09-27: the twin's FIRST growth in the boot worked;
/// after its SECOND (main 10→22→29 GiB/member) the twin was left with /data
/// unwritable (every audit append failed on monetize's 2 s GrowStatus poll),
/// its heartbeat alive reporting no volume size and its applied head stuck
/// ("behind … 127 entries to go"). The cloud had pulled the member while the
/// twin was still inside that window.
///
/// Two growths in a row on one machine, as on the mock: while the unmount is
/// still under way the phase reads `draining` (never `unmounted`), and the
/// first `unmounted | waiting-members` comes after the array has stopped.
#[test]
fn unmounted_is_not_said_until_the_set_is_off_and_a_second_growth_says_it_the_same_way() {
    let script = Script {
        grown: VecDeque::from([Ok(Grown {
            grew: true,
            array_bytes: 800 << 20,
            refusal: None,
            xfs: "grown".into(),
        })]),
        ..Script::default()
    };
    let world = Fake::new(script);
    let m = GrowMachine::new(world.clone());
    for (round, nonce) in [(1, "twice-1"), (2, "twice-2")] {
        world.script.lock().unwrap().surveys =
            VecDeque::from([complete(false), complete(false), complete(true)]);
        world.hold_unmount.store(true, Ordering::SeqCst);
        let calls_before = world.calls().len();
        let first = m
            .start(&request(0, nonce))
            .unwrap_or_else(|e| panic!("growth {round} refused: {e}"));
        assert_eq!(first.phase, "draining", "growth {round}");

        let deadline = std::time::Instant::now() + Duration::from_secs(20);
        while !world.in_unmount.load(Ordering::SeqCst) {
            assert!(
                std::time::Instant::now() < deadline,
                "growth {round} never reached the unmount; {:?}",
                m.phase()
            );
            std::thread::sleep(Duration::from_millis(5));
        }
        // /data is still mounted here, and the array still runs.
        assert!(
            world.serving(),
            "the fake is still mounted inside the unmount"
        );
        let s = m.status();
        assert_eq!(
            s.phase, "draining",
            "growth {round}: /data is still mounted and the phase already tells the outside actor the volumes are free: {s:?}"
        );
        assert!(
            !matches!(m.phase(), Phase::Unmounted | Phase::WaitingMembers),
            "growth {round}"
        );

        world.hold_unmount.store(false, Ordering::SeqCst);
        loop {
            let p = m.phase();
            if matches!(
                p,
                Phase::Unmounted
                    | Phase::WaitingMembers
                    | Phase::Growing
                    | Phase::Mounting
                    | Phase::Done
            ) {
                let calls = world.calls()[calls_before..].to_vec();
                assert!(
                    calls.iter().any(|c| c == "stop_array"),
                    "growth {round}: {p:?} said before the array stopped: {calls:?}"
                );
                break;
            }
            assert!(
                std::time::Instant::now() < deadline,
                "growth {round} never left draining: {p:?}"
            );
            std::thread::sleep(Duration::from_millis(1));
        }
        let done = wait_settled(&m);
        assert_eq!(done.phase, "done", "growth {round}: {done:?}");
        assert!(done.serving, "growth {round}");
    }
    // A late resume after the second growth answers done, as after the first.
    assert_eq!(m.resume(false).expect("done answers").phase, "done");
}

/// The members as a finished growth leaves them: every one CARRIES
/// `sectors` (the set spans the device), none is larger — the survey a box
/// answers after it grew to `sectors` and served again.
pub(crate) fn carrying(sectors: u64) -> SurveyFacts {
    let mut s = complete(false);
    for m in &mut s.members {
        m.disk_bytes = sectors * 512;
        m.set_bytes = sectors * 512;
        m.offered_bytes = sectors * 512;
    }
    s.description = format!("complete (4 of 4): nothing to grow at {sectors} sectors per member");
    s
}

/// ★ **DEFECT D (release-3 mock, 2026-09-29): a GrowStart whose target the
/// members ALREADY carry finishes at once and never drains or unmounts.**
///
/// MEASURED on the twin: monetize's twin lane grew the twin to 60 GB (done,
/// 12:35:35), lost the answer, and at 12:40 re-drove the same batch with a
/// SECOND `GrowStart` at the same 125 829 120 sectors. The box drained,
/// unmounted /data and waited "for every member to come back larger" — but
/// they were already 60 GiB, so nothing could ever come back larger and /data
/// stayed off: the off-site pull died on `txlog: No such file or directory`,
/// the key re-feed on `Read-only file system`, every audit append failed.
///
/// A growth to a size the set already spans is finished: `done` at once, the
/// gate never paused, /data never unmounted, nothing on the journal. A member
/// a hair under the asked volume size (a partition start, an md data offset)
/// counts as carrying it, exactly as `waiting-members` counts it.
#[test]
fn a_growstart_whose_target_the_members_already_carry_finishes_at_once_and_never_unmounts() {
    const SIXTY_GIB: u64 = 125_829_120;
    for (label, carried) in [
        ("exactly", SIXTY_GIB),
        ("a partition start under", SIXTY_GIB - 67),
    ] {
        let script = Script {
            in_flight: VecDeque::from([0]),
            surveys: VecDeque::from([carrying(carried)]),
            grown: VecDeque::from([Ok(Grown {
                grew: true,
                array_bytes: 800 << 20,
                refusal: None,
                xfs: "grown".into(),
            })]),
            ..Script::default()
        };
        let world = Fake::new(script);
        let m = GrowMachine::new(world.clone());
        let mut req = request(SIXTY_GIB, "twin-grow/gunnar/1790685635332");
        req.wait_members_timeout_secs = 3600;
        let first = m
            .start(&req)
            .unwrap_or_else(|e| panic!("{label}: a growth already carried is not a refusal: {e}"));
        assert_eq!(first.phase, "done", "{label}: finished at once: {first:?}");
        assert!(first.serving, "{label}: /data never left");
        assert!(
            first.detail.contains("already"),
            "{label}: the detail says why nothing moved: {}",
            first.detail
        );
        std::thread::sleep(Duration::from_millis(300));
        assert_eq!(m.phase(), Phase::Done, "{label}");
        assert!(
            world.calls().is_empty(),
            "{label}: nothing drained, nothing unmounted: {:?}",
            world.calls()
        );
        assert!(
            world.steps().is_empty(),
            "{label}: no epoch opened: {:?}",
            world.steps()
        );
        assert!(world.serving(), "{label}");
        // A late resume from the outside actor answers done, as after any growth.
        assert_eq!(
            m.resume(false).expect("done answers").phase,
            "done",
            "{label}"
        );
    }
    // A target the members do NOT carry yet still drains (the ordinary growth).
    let script = Script {
        in_flight: VecDeque::from([0]),
        surveys: VecDeque::from([carrying(SIXTY_GIB / 3)]),
        ..Script::default()
    };
    let world = Fake::new(script);
    let m = GrowMachine::new(world.clone());
    assert_eq!(
        m.start(&request(SIXTY_GIB, "n-grow"))
            .expect("started")
            .phase,
        "draining"
    );
    world.script.lock().unwrap().unmount_fails = Some("stop here".into());
    wait_settled(&m);
}

/// ★ **DEFECT D, the box's half of the race: `GrowResume` while `growing` or
/// `mounting` is progress, not a refusal.**
///
/// MEASURED: the twin's own 2 s poll saw the member come back larger and
/// moved to `growing` 0.6 s before monetize's `GrowResume` arrived; the box
/// answered FailedPrecondition "GrowResume applies while waiting-members; the
/// growth is growing", monetize read that as a failed drive, kept the batch
/// in flight and re-drove it with a second GrowStart. The nudge asks the box
/// to look at the members again; a box already growing HAS looked — the
/// answer is the status.
#[test]
fn growresume_while_growing_or_mounting_answers_the_status_not_a_refusal() {
    let script = Script {
        in_flight: VecDeque::from([0]),
        surveys: VecDeque::from([complete(false), complete(true)]),
        grown: VecDeque::from([Ok(Grown {
            grew: true,
            array_bytes: 800 << 20,
            refusal: None,
            xfs: "grown".into(),
        })]),
        ..Script::default()
    };
    let world = Fake::new(script);
    world.hold_grow.store(true, Ordering::SeqCst);
    let m = GrowMachine::new(world.clone());
    m.start(&request(0, "n1")).expect("started");
    wait_for(&m, Phase::Growing);
    let r = m.resume(false).expect("a resume while growing is progress");
    assert_eq!(r.phase, "growing", "{r:?}");
    let r = m
        .resume(true)
        .expect("give_up while growing changes nothing: the engine is already running");
    assert_eq!(r.phase, "growing", "{r:?}");
    world.hold_grow.store(false, Ordering::SeqCst);
    let done = wait_settled(&m);
    assert_eq!(
        done.phase, "done",
        "a resume during the engine's run neither stops nor repeats it: {done:?}"
    );
    assert_eq!(
        world
            .calls()
            .iter()
            .filter(|c| *c == "grow_and_mount")
            .count(),
        1,
        "{:?}",
        world.calls()
    );
    // Before any growth there is still nothing to resume.
    let idle = GrowMachine::new(Fake::new(Script::default()));
    assert!(idle
        .resume(false)
        .unwrap_err()
        .contains("start one with GrowStart"));
}

/// ★ **DEFECT D's guard: /data never stays unmounted past a bounded time.**
///
/// The wait for the members is bounded by the growth's own
/// `wait_members_timeout` (monetize asks 3600 s). Past it, with every member
/// present but none larger, the set is mounted back at the size the members
/// carry, the growth reads `failed` naming `grow.unmounted_too_long`, and the
/// alarm is on `Admin.Health`. With a member ABSENT nothing can be mounted
/// back — T6 still holds, the wait goes on — but the alarm is raised, so a
/// box that cannot come back is never silent either.
#[test]
fn a_set_left_unmounted_past_its_bound_is_mounted_back_named_and_alarmed() {
    let script = Script {
        in_flight: VecDeque::from([0]),
        surveys: VecDeque::from([complete(false)]),
        grown: VecDeque::from([Ok(Grown {
            grew: false,
            array_bytes: 400 << 20,
            refusal: None,
            xfs: String::new(),
        })]),
        ..Script::default()
    };
    let world = Fake::new(script);
    let m = GrowMachine::new(world.clone());
    assert!(m.alarm().is_none(), "no alarm before anything happened");
    let mut req = request(1_000_000, "n-stuck");
    req.wait_members_timeout_secs = 1;
    m.start(&req).expect("started");
    wait_for(&m, Phase::WaitingMembers);
    let s = wait_settled(&m);
    assert_eq!(s.phase, "failed", "{s:?}");
    assert!(s.serving, "mounted back: {s:?}");
    assert!(!s.identity_in_ram, "{s:?}");
    assert!(
        s.why.contains(UNMOUNTED_TOO_LONG),
        "the failure is named: {}",
        s.why
    );
    assert!(
        !world.paused.load(Ordering::SeqCst),
        "the gate is open again"
    );
    assert!(world.serving(), "the world was asked to mount");
    assert!(world.steps().contains(&Step::Fail), "{:?}", world.steps());
    let alarm = m.alarm().expect("an alarm on Admin.Health");
    assert_eq!(alarm.event, UNMOUNTED_TOO_LONG);
    assert!(alarm.detail.contains("mounted back"), "{}", alarm.detail);

    // A member ABSENT at the bound: nothing to mount back (T6) — still
    // waiting, but alarmed.
    let mut gone = SurveyFacts {
        verdict: "incomplete: member 1 of 4 is missing".into(),
        raid_disks: 4,
        ..Default::default()
    };
    gone.description = gone.verdict.clone();
    let script = Script {
        in_flight: VecDeque::from([0]),
        surveys: VecDeque::from([complete(false), gone]),
        ..Script::default()
    };
    let world = Fake::new(script);
    let m = GrowMachine::new(world.clone());
    let mut req = request(1_000_000, "n-gone");
    req.wait_members_timeout_secs = 1;
    m.start(&req).expect("started");
    wait_for(&m, Phase::WaitingMembers);
    std::thread::sleep(Duration::from_millis(1600));
    assert_eq!(
        m.phase(),
        Phase::WaitingMembers,
        "a set with a member missing cannot be mounted back"
    );
    let alarm = m.alarm().expect("alarmed while it cannot come back");
    assert_eq!(alarm.event, UNMOUNTED_TOO_LONG);
    assert!(alarm.detail.contains("missing"), "{}", alarm.detail);
    assert!(
        !world.calls().iter().any(|c| c == "grow_and_mount"),
        "{:?}",
        world.calls()
    );
}

/// ★ estate-box3 E1 (owner order: EVERY PRODUCT WRITES — a silent product is
/// a FAIL): the growth machine said nothing from its birth until a growth,
/// so on a box that never grows it is indistinguishable from one that never
/// started. Built, it says ONE line on `gunnar::grow`: up, its phase, and
/// whether this build can grow at runtime.
#[test]
fn the_growth_machine_says_it_is_up_once_when_it_is_built() {
    #[derive(Clone, Default)]
    struct Heard(Arc<Mutex<Vec<u8>>>);
    impl std::io::Write for Heard {
        fn write(&mut self, b: &[u8]) -> std::io::Result<usize> {
            self.0.lock().unwrap().extend_from_slice(b);
            Ok(b.len())
        }
        fn flush(&mut self) -> std::io::Result<()> {
            Ok(())
        }
    }
    // tracing's max level is process-global and is recomputed whenever any test thread's
    // subscriber comes or goes, so a concurrent test can briefly hide this thread's one. A
    // capture that heard NOTHING is that race and is tried again; a birth that says the line
    // twice, or says it wrongly, fails at once.
    let mut said = String::new();
    for _ in 0..20 {
        let heard = Heard::default();
        let h = heard.clone();
        let sub = tracing_subscriber::fmt()
            .with_writer(move || h.clone())
            .with_ansi(false)
            .finish();
        tracing::subscriber::with_default(sub, || {
            tracing::callsite::rebuild_interest_cache();
            let _m = GrowMachine::new(Fake::new(Script::default()));
        });
        said = String::from_utf8(heard.0.lock().unwrap().clone()).unwrap();
        if !said.is_empty() {
            break;
        }
        std::thread::sleep(std::time::Duration::from_millis(50));
    }
    let up: Vec<&str> = said
        .lines()
        .filter(|l| l.contains(TARGET) && l.contains("the growth machine is up"))
        .collect();
    assert_eq!(up.len(), 1, "one line, on {TARGET}:\n{said}");
    assert!(
        up[0].contains("idle") && up[0].contains("engine"),
        "its phase and whether it can grow at runtime: {}",
        up[0]
    );
}

/// ★ **cov-13 (2026-10-06): an EMERGENCY growth the approver sent ALONE goes past a
/// standby that is behind.** monetize grows main alone when the set is at ≥ 90 % and the
/// twin is silent past its bound or cannot catch up under the load (MEASURED on the
/// UpCloud mockup: the twin ~93 000 entries behind, the emergency refused six times by
/// the caught-up proof, main refusing pushes). The go-ahead then carries `alone` (the
/// ALERT's words, beside the signed fields); the drain still looks — briefly — and a
/// standby that is behind is SAID, not obeyed: the growth goes on, the detail and the
/// log name it, and the twin catches up from the log afterwards (or takes a snapshot).
/// Without `alone` the same standby still aborts it untouched (the test above).
#[test]
fn cov13_an_alone_go_ahead_grows_past_a_standby_that_is_behind_and_says_so() {
    let script = Script {
        surveys: VecDeque::from([complete(false), complete(false), complete(true)]),
        grown: VecDeque::from([Ok(Grown {
            grew: true,
            array_bytes: 800 << 20,
            refusal: None,
            xfs: "grown".into(),
        })]),
        ..Script::default()
    };
    let world = Fake::new(script);
    world.flush_answers(crate::grow::world::FlushVerdict::Behind {
        pending: 92_775,
        needs_full_resync: false,
        why: "twin 'gunnar-twin' has applied 711743 of this box's head 804518".to_owned(),
    });
    let m = GrowMachine::new(world.clone());
    let mut go: serde_json::Value = serde_json::from_slice(&go(0, "n-alone")).unwrap();
    go["alone"] = serde_json::Value::String(
        "ALERT gunnar: main grown alone: twin gunnar-twin not caught up within the flush bound"
            .into(),
    );
    let req = gunnar_proto::GrowStartRequest {
        go_ahead: serde_json::to_vec(&go).unwrap(),
        ..request(0, "n-alone")
    };

    m.start(&req).expect("started");
    let done = wait_settled(&m);

    assert_eq!(
        (done.phase.as_str(), done.why.as_str()),
        ("done", ""),
        "an emergency sent alone must grow past a lagging standby: {done:?}"
    );
    assert!(
        world.calls().contains(&"grow_and_mount".to_owned()),
        "{:?}",
        world.calls()
    );
    assert!(
        done.detail.contains("ALONE") && done.detail.contains("92775"),
        "the run must say it went alone and how far behind: {}",
        done.detail
    );
}

// ── no traffic against closed stores; no step lost from the journal (stinky rows 10 and 11) ──

/// ★ **An abort whose stores will not reopen keeps the gate shut** (stinky
/// row 10). `abort_after_suspend` did `let _ = reopen_stores()` and then
/// `unpause_admission()`: the box admitted git traffic against stores it had
/// closed for the growth.
#[test]
fn an_abort_whose_stores_will_not_reopen_keeps_the_gate_shut() {
    let script = Script {
        in_flight: VecDeque::from([0]),
        surveys: VecDeque::from([complete(false)]),
        unmount_fails: Some("umount /data: EBUSY".into()),
        reopen_fails: Some("the control store would not open: EIO".into()),
        ..Default::default()
    };
    let world = Fake::new(script);
    let m = GrowMachine::new(world.clone());
    m.start(&request(0, "n1")).expect("started");
    let s = wait_settled(&m);
    assert_eq!(s.phase, "failed");
    assert!(
        world.paused.load(Ordering::SeqCst),
        "the gate opened over closed stores: {:?}",
        world.calls()
    );
    assert!(
        !world.calls().contains(&"unpause".to_owned()),
        "{:?}",
        world.calls()
    );
    assert!(s.why.contains("EIO"), "the refusal is named: {}", s.why);
    assert!(
        !s.serving,
        "a box whose stores are closed does not say it is serving"
    );
}

/// …and so does a growth whose stores will not reopen once `/data` is back:
/// it is no `done`, and it announces no `Ready`.
#[test]
fn a_growth_whose_stores_will_not_reopen_after_the_mount_keeps_the_gate_shut() {
    let script = Script {
        in_flight: VecDeque::from([0]),
        surveys: VecDeque::from([complete(false), complete(true)]),
        grown: VecDeque::from([Ok(Grown {
            grew: true,
            array_bytes: 800 << 20,
            refusal: None,
            xfs: "grown".into(),
        })]),
        reopen_fails: Some("the control store would not open: EIO".into()),
        ..Default::default()
    };
    let world = Fake::new(script);
    let m = GrowMachine::new(world.clone());
    let announced = Arc::new(Mutex::new(Vec::new()));
    let a = Arc::clone(&announced);
    m.with_announcer(move |x| a.lock().unwrap().push(x));
    m.start(&request(0, "n1")).expect("started");
    let s = wait_settled(&m);
    assert_eq!(s.phase, "failed", "{}", s.why);
    assert!(s.why.contains("EIO"), "{}", s.why);
    assert!(
        world.paused.load(Ordering::SeqCst),
        "the gate opened over closed stores: {:?}",
        world.calls()
    );
    assert!(
        !announced
            .lock()
            .unwrap()
            .iter()
            .any(|x| matches!(x, Announcement::Ready)),
        "Ready announced over closed stores"
    );
}

/// ★ **A journal that cannot record a step fails the growth** (stinky row
/// 11). `let _ = self.note(Step::Wait)` went on to wait for members the
/// journal never said it had let go of.
#[test]
fn a_journal_that_cannot_record_the_wait_fails_the_growth() {
    let script = Script {
        in_flight: VecDeque::from([0]),
        surveys: VecDeque::from([complete(false), complete(true)]),
        grown: VecDeque::from([Ok(Grown {
            grew: false,
            array_bytes: 400 << 20,
            refusal: None,
            xfs: "as before".into(),
        })]),
        note_fails: Some(Step::Wait),
        ..Default::default()
    };
    let world = Fake::new(script);
    let m = GrowMachine::new(world.clone());
    m.start(&request(0, "n1")).expect("started");
    let s = wait_settled(&m);
    assert_eq!(s.phase, "failed", "{}", s.why);
    assert!(
        s.why.contains("journal"),
        "the journal's refusal is named: {}",
        s.why
    );
    assert!(
        !world.steps().contains(&Step::Rescan),
        "the growth went on past a step the journal lost: {:?}",
        world.steps()
    );
}

/// …a reopen it cannot record is no `done`…
#[test]
fn a_journal_that_cannot_record_the_reopen_fails_the_growth() {
    let script = Script {
        in_flight: VecDeque::from([0]),
        surveys: VecDeque::from([complete(false), complete(true)]),
        grown: VecDeque::from([Ok(Grown {
            grew: true,
            array_bytes: 800 << 20,
            refusal: None,
            xfs: "grown".into(),
        })]),
        note_fails: Some(Step::Reopen),
        ..Default::default()
    };
    let world = Fake::new(script);
    let m = GrowMachine::new(world.clone());
    m.start(&request(0, "n1")).expect("started");
    let s = wait_settled(&m);
    assert_eq!(s.phase, "failed", "{}", s.why);
    assert!(
        s.why.contains("journal") && s.why.contains("REOPEN"),
        "{}",
        s.why
    );
}

/// …and an abort it cannot record says so in the failure.
#[test]
fn a_journal_that_cannot_record_the_abort_says_so() {
    let script = Script {
        in_flight: VecDeque::from([0]),
        surveys: VecDeque::from([complete(false)]),
        unmount_fails: Some("umount /data: EBUSY".into()),
        note_fails: Some(Step::Abort),
        ..Default::default()
    };
    let world = Fake::new(script);
    let m = GrowMachine::new(world.clone());
    m.start(&request(0, "n1")).expect("started");
    let s = wait_settled(&m);
    assert_eq!(s.phase, "failed");
    assert!(
        s.why.contains("journal") && s.why.contains("ABORT"),
        "{}",
        s.why
    );
}