crates/gunnar-server/src/grow/tests.rs
//! 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
);
}