crates/appliance-grow/src/lib.rs
//! **The runtime data-set growth** — `/data` grows WITHOUT a restart
//! (`private-gunnar-ops/DATA-SET-GROWTH-FLOW.md` §0 ruling 1, §3; the
//! torture test's T1, T2, T6, T9, T11 are binding).
//!
//! Moved AS IS out of gunnar-server's `src/grow/` (gunnar `gunnar-live`
//! 18a44e3) so every appliance that sells a growing disk grows it the SAME
//! way; the product names are gone and nothing else is. What each product
//! still owns is its [`World`] — its admission gate, its stores, its twin.
//!
//! ```text
//! GrowStart{target, go_ahead}
//! │ (the product's admin door; the go-ahead's structure verified, its signer policy pluggable)
//! ▼
//! draining ──── pause the admission gate (every front says the same sentence),
//! │ wait ≤ drain_timeout for in_flight == 0 T1: never kill a transfer
//! │ timeout ─▶ unpause, failed{aborted}, NOTHING changed
//! │ then the twins, held to the BARRIER (the head when the gate shut)
//! ▼ journal DRAIN
//! (still draining) identity → RAM, close every store, sync, umount /data,
//! │ a read-only tmpfs over the empty mountpoint, md array_state ← clear
//! │ any failure ─▶ mount back, serve, failed{why} T9
//! ▼ journal UNMOUNT, STOP
//! unmounted ─── the set is OFF: the outside actor may take the volumes now
//! ▼ journal WAIT
//! waiting-members ── poll the block devices (pci rescan + survey) until every
//! │ member is back AND larger AND equal; GrowResume looks again;
//! │ past the bound (wait timeout) with every member present /data
//! │ is MOUNTED BACK, failed{grow.unmounted_too_long}, alarmed; with
//! │ a member absent it stays here, alarmed T6
//! ▼ journal RESCAN
//! growing ───── the SAME engine as at boot: G1–G3 → assemble → G4 → mount → G5 → G6
//! │ (korp-installer::grow::runtime::grow_and_mount); on any refusal
//! │ the set is mounted at the size the members still share T9
//! ▼
//! mounting ──── identity back (RAM copy == disk, verified), stores reopen,
//! │ the gate unpaused
//! ▼ journal REOPEN
//! done | failed{why} (failed still SERVES; only cannot-assemble/mount is
//! fatal, and even then the control plane and the
//! credential-free status keep answering)
//! ```
//!
//! # Where the machine ends and the box begins
//!
//! [`GrowMachine`] is the phase machine, the journal discipline, the go-ahead
//! check and the failure policy, written over the [`World`] seam so every
//! transition is driven natively (`tests` below, with a scripted world). A
//! product's world is its appliance: its admission gate, its stores, its
//! identity table and — feature `md` — [`md::MdVolumes`], the volume half over
//! korp-installer's `grow::runtime`. Without the engine the machine still
//! exists — `status` answers `idle` with `engine_present: false`, `start` is
//! refused by name — so a product's grow status tells the truth on every build.
//!
//! # The journal is the boot path's, and that is the crash story
//!
//! Every runtime step is a record in `/serialdat/grow-journal` in the exact
//! line format `korp-start` reads (`korp_installer::grow::Step`'s runtime
//! steps, `DRAIN … REOPEN`, beside the engine's own `G1 … G6`). The journal
//! decides nothing: at the next boot the engine looks at the DISK — members
//! present? larger? partitions extended? superblocks agreed? — and finishes
//! whatever is left. A crash while the members are detached halts that boot
//! on a missing member (`datahome.rs`), which is T6, and why the phase is
//! readable with no credential: the actor outside re-attaches, then reboots
//! or calls `GrowResume`.
//!
//! # What is refused during the window, and how
//!
//! The product's, through its [`World`]: every transfer through its ONE
//! admission gate with one sentence, control-plane writes refused (never
//! queued — the caller retries after the growth), repository opens refused so
//! a background worker meets an error rather than pinning the mount.
//!
//! # `Draining{kind: growing}` on the presence stream
//!
//! [`GrowMachine::with_announcer`] is the hook: whatever the product's stream
//! fan-out is, it registers a closure and receives [`Announcement::Draining`]
//! before the pause and [`Announcement::Ready`] after the reopen. The same two
//! facts are always `tracing` rows on [`TARGET`] (`event = "presence.draining"`
//! / `"presence.ready"`).
pub mod capacity;
pub mod go_ahead;
pub mod journal;
#[cfg(feature = "md")]
pub mod md;
pub mod probe;
pub mod world;
use std::sync::{Arc, Condvar, Mutex};
use std::time::{Duration, Instant};
pub use go_ahead::{GoAhead, GoAheadPolicy, OperatorPolicy};
#[cfg(feature = "monetize")]
pub use go_ahead::MonetizePolicy;
pub use world::{FlushVerdict, Grown, MemberFacts, SurveyFacts, TwinPosition, Volumes, World};
/// The log target every row of the growth is written on.
pub const TARGET: &str = "appliance::grow";
/// Wall-clock milliseconds since the epoch — the one clock every status
/// field and the go-ahead's age window read.
pub fn now_unix_ms() -> i64 {
std::time::SystemTime::now().duration_since(std::time::UNIX_EPOCH).map(|d| d.as_millis() as i64).unwrap_or(0)
}
/// The drain's default ceiling: how long in-flight transfers may take to
/// finish before the growth ABORTS with nothing changed (T1).
pub const DEFAULT_DRAIN_TIMEOUT: Duration = Duration::from_secs(600);
/// How long `waiting-members` polls before it says so and slows down — and
/// **the bound on `/data` staying unmounted**: past it, with every member
/// present, the set is mounted back at the size the members carry and the
/// growth fails as [`UNMOUNTED_TOO_LONG`] (DEFECT D, 2026-09-29). With a member
/// absent nothing can be mounted, so the wait goes on (T6), alarmed.
pub const DEFAULT_WAIT_MEMBERS_TIMEOUT: Duration = Duration::from_secs(3600);
/// The alarm word for a set kept unmounted past its bound.
pub const UNMOUNTED_TOO_LONG: &str = "grow.unmounted_too_long";
/// **An alarm the growth raises for the product's health door** — gunnar's
/// `HealthAlarm` field for field, so a product maps it 1:1.
#[derive(Debug, Clone, PartialEq, Eq, Default)]
pub struct GrowAlarm {
pub event: String,
pub count: u64,
pub since_unix_ms: i64,
pub detail: String,
}
/// How often the drain re-reads the in-flight count.
pub const DRAIN_POLL: Duration = Duration::from_millis(200);
/// Has G1's per-member target `have` reached the plan's `asked` volume size?
/// Within 1 % under, yes — the member's usable sectors sit a partition start
/// and an md data offset below the volume's, and G1 itself allows that much
/// between members. Exactly `asked` or more is always yes.
pub fn target_reached(have: u64, asked: u64) -> bool {
have >= asked || have.saturating_mul(100) >= asked.saturating_mul(99)
}
/// How often the members are surveyed while waiting, before the timeout…
pub const WAIT_POLL: Duration = Duration::from_secs(2);
/// …and after it (still waiting, more slowly).
pub const WAIT_POLL_AFTER_TIMEOUT: Duration = Duration::from_secs(15);
/// The phase, as the status spells it — the words monetize's
/// `GrowPhase::parse` reads, so they are frozen.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum Phase {
Idle,
Draining,
Unmounted,
WaitingMembers,
Growing,
Mounting,
Done,
Failed,
}
impl Phase {
pub fn as_str(self) -> &'static str {
match self {
Phase::Idle => "idle",
Phase::Draining => "draining",
Phase::Unmounted => "unmounted",
Phase::WaitingMembers => "waiting-members",
Phase::Growing => "growing",
Phase::Mounting => "mounting",
Phase::Done => "done",
Phase::Failed => "failed",
}
}
/// A growth is under way: `Start` is refused, `Resume` may apply.
pub fn is_active(self) -> bool {
matches!(
self,
Phase::Draining | Phase::Unmounted | Phase::WaitingMembers | Phase::Growing | Phase::Mounting
)
}
}
/// The runtime journal steps, as the machine names them; the world spells
/// them into `korp_installer::grow::Step` (or a test's list). The words are
/// the journal's, fixed by the engine's own test.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum Step {
Drain,
Unmount,
Stop,
Wait,
Rescan,
Reopen,
Abort,
Fail,
}
impl Step {
pub fn as_str(self) -> &'static str {
match self {
Step::Drain => "DRAIN",
Step::Unmount => "UNMOUNT",
Step::Stop => "STOP",
Step::Wait => "WAIT",
Step::Rescan => "RESCAN",
Step::Reopen => "REOPEN",
Step::Abort => "ABORT",
Step::Fail => "FAIL",
}
}
}
/// What the stream (or the log) is told.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum Announcement {
/// `Draining{kind: growing}`: before the gate is paused.
Draining { kind: &'static str, reason: String },
/// `Ready`: `/data` is back and the gate is open.
Ready,
}
/// **A `GrowStart`, as the product's wire hands it over** — gunnar's
/// `GrowStartRequest` field for field, so a product's RPC maps onto it 1:1.
#[derive(Debug, Clone, PartialEq, Eq, Default)]
pub struct StartRequest {
/// Per-member target in sectors; 0 = whatever every member comes back
/// larger at.
pub target_sectors_per_member: u64,
/// The go-ahead's JSON bytes ([`go_ahead`]).
pub go_ahead: Vec<u8>,
/// 0 = [`DEFAULT_DRAIN_TIMEOUT`].
pub drain_timeout_secs: u32,
/// 0 = [`DEFAULT_WAIT_MEMBERS_TIMEOUT`].
pub wait_members_timeout_secs: u32,
}
/// **The status, product-neutral** — gunnar's `GrowStatusResponse` field for
/// field (the same fields monetize's `GrowStatus` carries), so a product's
/// wire maps from it 1:1 and monetize reads the same facts from every box.
#[derive(Debug, Clone, PartialEq, Eq, Default)]
pub struct GrowStatus {
pub phase: String,
pub why: String,
pub epoch: u64,
pub target_sectors_per_member: u64,
pub members: Vec<MemberFacts>,
pub members_verdict: String,
/// The boot stack's journal, read by the same reader the capacity door uses.
pub journal: Option<journal::Summary>,
pub engine_present: bool,
pub hotplug: bool,
/// **`/data` is mounted and served.** monetize's restore decision reads
/// exactly this (`serving == false` on a failed run orders a recharge), so
/// it is true only when the set is really mounted and the stores are open.
pub serving: bool,
pub identity_in_ram: bool,
pub in_flight: u64,
pub since_unix_ms: i64,
pub server_unix_ms: i64,
pub detail: String,
pub go_ahead_signer: String,
pub set_uuid: String,
/// `/data`'s size as mounted; 0 when not serving.
pub total_bytes: u64,
}
/// What a `GrowStart` asks for, after the request was checked.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct Plan {
/// Per-member target in sectors; 0 = whatever every member comes back
/// larger at.
pub target_sectors: u64,
pub drain_timeout: Duration,
pub wait_timeout: Duration,
/// Who signed the go-ahead.
pub signer: String,
}
/// A `GrowResume`.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum Resume {
LookAgain,
GiveUp,
}
/// How `waiting-members` ended.
#[derive(Debug, Clone, PartialEq, Eq)]
enum Ended {
/// Every member is back larger: the engine grows.
Eligible,
/// `GrowResume{give_up}`: assemble and serve what is there.
GaveUp,
/// The bound passed with every member present: mount back, named.
Bound(String),
}
/// **Does every member already CARRY `target` sectors?** — the set spans it
/// (within G1's 1 %, the same allowance `waiting-members` grants), so a
/// growth to it has nothing left to do.
pub fn carries(survey: &SurveyFacts, target: u64) -> bool {
survey.verdict == "complete" && !survey.members.is_empty() && survey.members.iter().all(|m| target_reached(m.set_bytes / 512, target))
}
/// The machine's state, behind one mutex.
#[derive(Debug, Clone)]
struct State {
phase: Phase,
since_unix_ms: i64,
why: String,
detail: String,
epoch: u64,
target_sectors: u64,
signer: String,
members: SurveyFacts,
identity_in_ram: bool,
serving: bool,
in_flight: u64,
resume: Option<Resume>,
/// `waiting-members` passed its timeout and is still waiting.
wait_timed_out: bool,
/// The last thing the engine said, for `detail` after `done`/`failed`.
grown: Option<Grown>,
/// **The growth's alarm** ([`UNMOUNTED_TOO_LONG`], or a fatal mount),
/// on the product's health door until the next growth starts.
alarm: Option<(String, i64, String)>,
}
impl State {
fn idle(serving: bool) -> Self {
State {
phase: Phase::Idle,
since_unix_ms: now_unix_ms(),
why: String::new(),
detail: String::new(),
epoch: 0,
target_sectors: 0,
signer: String::new(),
members: SurveyFacts::default(),
identity_in_ram: false,
serving,
in_flight: 0,
resume: None,
wait_timed_out: false,
grown: None,
alarm: None,
}
}
}
/// **The growth, as one object the control plane holds.** Cheap to share
/// (`Arc`); the run happens on a thread of its own so the RPC that started
/// it returns at once with the first status.
pub struct GrowMachine {
world: Arc<dyn World>,
policy: Arc<dyn GoAheadPolicy>,
state: Mutex<State>,
/// Woken by `GrowResume` while `waiting-members`.
wake: Condvar,
announcer: Mutex<Option<Box<dyn Fn(Announcement) + Send + Sync>>>,
}
impl std::fmt::Debug for GrowMachine {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.debug_struct("GrowMachine").field("phase", &self.phase()).finish()
}
}
impl GrowMachine {
/// A machine over `world`, with the operator go-ahead policy.
pub fn new(world: Arc<dyn World>) -> Arc<Self> {
Self::with_policy(world, Arc::new(OperatorPolicy::default()))
}
/// A machine over `world` with an explicit go-ahead policy (a metered
/// build plugs monetize's key in here — see [`go_ahead`]).
pub fn with_policy(world: Arc<dyn World>, policy: Arc<dyn GoAheadPolicy>) -> Arc<Self> {
let serving = world.serving();
Arc::new(GrowMachine {
world,
policy,
state: Mutex::new(State::idle(serving)),
wake: Condvar::new(),
announcer: Mutex::new(None),
})
}
/// Register the presence stream's fan-out (see the module docs).
pub fn with_announcer(self: &Arc<Self>, f: impl Fn(Announcement) + Send + Sync + 'static) {
*self.announcer.lock().unwrap_or_else(|p| p.into_inner()) = Some(Box::new(f));
}
fn announce(&self, a: Announcement) {
match &a {
Announcement::Draining { kind, reason } => tracing::warn!(
target: TARGET,
event = "presence.draining",
kind,
reason = %reason,
"Draining{{kind: {kind}}}: {reason}"
),
Announcement::Ready => tracing::info!(target: TARGET, event = "presence.ready", "Ready: /data is back and the gate is open"),
}
if let Some(f) = self.announcer.lock().unwrap_or_else(|p| p.into_inner()).as_ref() {
f(a);
}
}
fn lock(&self) -> std::sync::MutexGuard<'_, State> {
self.state.lock().unwrap_or_else(|p| p.into_inner())
}
/// The growth's alarm for the product's health door, when it has one.
pub fn alarm(&self) -> Option<GrowAlarm> {
self.lock().alarm.clone().map(|(event, since_unix_ms, detail)| GrowAlarm { event, count: 1, since_unix_ms, detail })
}
/// Raise the growth's alarm: an `error` row on the log and the alarm on
/// the product's health door. Said on the log the first time (or `loud`); the
/// detail follows the latest.
fn raise(&self, event: &str, detail: String, loud: bool) {
let mut s = self.lock();
if loud || s.alarm.as_ref().is_none_or(|(e, _, _)| e != event) {
tracing::error!(target: TARGET, event, "{detail}");
}
let since = match &s.alarm {
Some((e, since, _)) if e == event => *since,
_ => now_unix_ms(),
};
s.alarm = Some((event.to_owned(), since, detail));
}
/// The phase right now.
pub fn phase(&self) -> Phase {
self.lock().phase
}
fn enter(&self, phase: Phase, detail: impl Into<String>) {
let mut s = self.lock();
let detail = detail.into();
tracing::info!(target: TARGET, event = "grow.phase", from = s.phase.as_str(), to = phase.as_str(), "{detail}");
s.phase = phase;
s.since_unix_ms = now_unix_ms();
s.detail = detail;
}
fn note(&self, step: Step) -> Result<(), String> {
let (epoch, target) = {
let s = self.lock();
(s.epoch, s.target_sectors)
};
self.world.note(epoch, target, step)
}
// ── the verbs ───────────────────────────────────────────────────────
/// **`GrowStart`.** Checks the go-ahead and the box, moves to
/// `draining`, and runs the growth on its own thread. Refused by name
/// when the engine is absent, a growth is active, or the go-ahead does
/// not verify — nothing moves in any of those cases.
pub fn start(self: &Arc<Self>, request: &StartRequest) -> Result<GrowStatus, String> {
if !self.world.engine_present() {
return Err("this server was built without the runtime grow engine (feature `grow-runtime`); the set grows only at a restart".into());
}
// Owner's rule (2026-09-15): growth WITHOUT a restart, and a medium
// whose kernel cannot see a member leave and come back refuses it BY
// NAME rather than draining the box into a `waiting-members` that only
// a reboot ends. UpCloud resizes only detached volumes.
if !self.world.hotplug() {
return Err(format!(
"the running kernel has no PCI hot-plug ({}): a member detached, resized and re-attached while the server runs is invisible to it, so a runtime growth is refused — re-image onto a medium whose kernel carries CONFIG_HOTPLUG_PCI, or enlarge every member and restart (the set grows at the next boot)",
crate::probe::pci_hotplug().evidence
));
}
// **The capacity door and this one must say the same thing.** AUDITED
// 2026-09-18 (gunnar): the capacity ladder reported `GROWTH_SEALED —
// serving more needs a fresh install` whenever the grow journal file
// was absent, and its own sentence claimed this machine "refuses by
// name" — while `start` had no journal precondition at all. A box
// whose medium carries no grow engine therefore ACCEPTED a
// `GrowStart`, paused the admission gate and drained live sessions,
// and only met the problem four steps later at `note(Step::Drain)`.
// The refusal sits here, before anything moves, and it is the SAME
// file the capacity ladder measures — one fact, two doors, one answer.
// See `World::journal_present` for why the file's presence is the
// medium's grow engine.
if !self.world.journal_present() {
return Err(format!(
"this box's boot stack keeps no grow journal ({}): korp-start never opened one, so this medium's boot stack does not grow the data set and a runtime growth has nowhere to record the steps a crash would be finished from — serving more needs a fresh install onto a larger set, not another disk. This is the same fact the capacity door reports as GROWTH_SEALED",
self.world.journal_path().display()
));
}
let go = GoAhead::parse(&request.go_ahead)?;
if go.target_sectors != request.target_sectors_per_member {
return Err(format!(
"the go-ahead names target_sectors {} and the request {}; they must agree",
go.target_sectors, request.target_sectors_per_member
));
}
let now = now_unix_ms();
self.policy.verify(&go, now)?;
if !self.world.remember_nonce(&go.nonce)? {
return Err(format!("the go-ahead's nonce {:?} was already spent on this box", go.nonce));
}
let plan = Plan {
target_sectors: request.target_sectors_per_member,
drain_timeout: if request.drain_timeout_secs == 0 { DEFAULT_DRAIN_TIMEOUT } else { Duration::from_secs(u64::from(request.drain_timeout_secs)) },
wait_timeout: if request.wait_members_timeout_secs == 0 { DEFAULT_WAIT_MEMBERS_TIMEOUT } else { Duration::from_secs(u64::from(request.wait_members_timeout_secs)) },
signer: go.signer.clone(),
};
// Read before the lock: the world answers from its own state.
let world_serving = self.world.serving();
{
let mut s = self.lock();
if s.phase.is_active() {
return Err(format!("a growth is already {} (since {}); GrowResume looks again, and a new growth waits for this one", s.phase.as_str(), s.since_unix_ms));
}
// No growth in flight: the machine holds nothing of /data, and the
// world's word is the answer — not the word at the machine's birth
// (a recharged main is built awaiting its twin and serves minutes
// later; MEASURED holger ops rung 3, 2026-10-05).
s.serving = world_serving;
if !s.serving {
return Err(format!("/data is not being served ({}); a growth cannot start from here — reboot, or re-attach the members and GrowResume", s.why));
}
let survey = self.world.survey();
if survey.verdict != "complete" {
return Err(format!("the data set is not a complete md set right now ({}); nothing to grow", survey.verdict));
}
// **DEFECT D (release-3 mock, 2026-09-29): GrowStart is idempotent.**
// A growth to a size every member ALREADY carries is finished: it
// is `done` at once, and nothing is paused, drained or unmounted.
// MEASURED: monetize's twin lane re-drove a finished growth with a
// second GrowStart at the same 60 GiB; the box drained, unmounted
// /data and waited for members to come back LARGER than 60 GiB —
// for ever, with /data off (the off-site pull, the key re-feed and
// every audit append failing behind it).
if plan.target_sectors > 0 && carries(&survey, plan.target_sectors) {
let described = survey.describe();
*s = State::idle(true);
s.signer = plan.signer.clone();
s.target_sectors = plan.target_sectors;
s.members = survey;
s.phase = Phase::Done;
s.since_unix_ms = now;
s.detail = format!(
"done: every member already carries the {} sectors asked ({described}); nothing was drained and /data never left",
plan.target_sectors
);
drop(s);
tracing::warn!(
target: TARGET,
event = "grow.already_met",
signer = %plan.signer,
nonce = %go.nonce,
target_sectors = plan.target_sectors,
"GrowStart for a size the members already carry: done at once, nothing drained or unmounted"
);
return Ok(self.status());
}
*s = State::idle(true);
s.signer = plan.signer.clone();
s.target_sectors = plan.target_sectors;
s.members = survey;
s.phase = Phase::Draining;
s.since_unix_ms = now;
s.detail = format!("draining: the gate is paused, waiting up to {} s for {} transfer(s) in flight", plan.drain_timeout.as_secs(), self.world.in_flight());
}
tracing::warn!(
target: TARGET,
event = "grow.start",
signer = %plan.signer,
nonce = %go.nonce,
target_sectors = plan.target_sectors,
drain_timeout_s = plan.drain_timeout.as_secs(),
wait_timeout_s = plan.wait_timeout.as_secs(),
"a runtime data-set growth begins"
);
let me = Arc::clone(self);
// A32-14, stays a thread: the one growth run a verified go starts (drain, unmount, wait, grow), reporting each phase; the crate holds no job ledger.
std::thread::Builder::new()
.name("appliance-grow".into())
.spawn(move || me.run(plan))
.map_err(|e| format!("spawning the growth thread: {e}"))?;
Ok(self.status())
}
/// **`GrowResume`.** While `waiting-members`: look again now
/// (`give_up`: stop waiting, assemble what is there, serve). On a growth
/// that is already `done` it answers the status — the resize it would have
/// waited for has happened and been finished, so there is nothing left to
/// nudge and nothing wrong. In any other phase a refusal that names the
/// phase.
///
/// MEASURED on the mock 2026-09-27 (gunnar): monetize's twin lane resized
/// the twin's volume and asked GrowResume; the twin had ALREADY seen the
/// larger member, grown (G6) and served again, answered FailedPrecondition
/// "the growth is done", and monetize kept the finished batch open and
/// drove it again — main's growth waited for it at twin-sizing forever.
pub fn resume(&self, give_up: bool) -> Result<GrowStatus, String> {
if self.lock().phase == Phase::Done {
tracing::info!(target: TARGET, event = "grow.resume.done", give_up, "GrowResume on a growth that is already done: nothing to resume");
return Ok(self.status());
}
{
let s = self.lock();
// **DEFECT D: a nudge that arrives after the box moved on is
// progress.** MEASURED on the release-3 mock: the box's own 2 s
// poll saw the member come back larger and moved to `growing`
// 0.6 s before the outside actor's GrowResume; the refusal it
// answered was read as a failed drive and the growth re-driven
// with a second GrowStart. A box that is growing or mounting HAS
// looked at the members — the answer is its status. (`give_up`
// changes nothing either: the engine is already running.)
if matches!(s.phase, Phase::Growing | Phase::Mounting) {
drop(s);
tracing::info!(target: TARGET, event = "grow.resume.progress", give_up, "GrowResume while the engine runs: the members were already seen; nothing to look at again");
return Ok(self.status());
}
}
{
let mut s = self.lock();
// `unmounted` is the instant before `waiting-members`: the flag is
// read by the wait's first look.
if s.phase != Phase::WaitingMembers && s.phase != Phase::Unmounted {
return Err(format!(
"GrowResume applies while waiting-members; the growth is {} ({})",
s.phase.as_str(),
if s.phase == Phase::Idle { "start one with GrowStart" } else { s.detail.as_str() }
));
}
s.resume = Some(if give_up { Resume::GiveUp } else { Resume::LookAgain });
}
tracing::info!(target: TARGET, event = "grow.resume", give_up, "GrowResume: looking at the members again");
self.wake.notify_all();
Ok(self.status())
}
/// **`GrowStatus`** (gated and credential-free alike). The state, the last
/// survey (a fresh one when idle, since it is cheap and nothing else
/// refreshes it), and the journal as the same reader the capacity door uses.
pub fn status(&self) -> GrowStatus {
let world_serving = self.world.serving();
let mut s = self.lock().clone();
if !s.phase.is_active() {
// Idle, done or failed: /data is the world's to say (see `start`).
s.serving = world_serving;
}
if !s.phase.is_active() && self.world.engine_present() {
s.members = self.world.survey();
}
if s.phase == Phase::Draining {
s.in_flight = self.world.in_flight();
}
let journal = journal::Summary::read(&self.world.journal_path());
let detail = if s.detail.is_empty() {
match s.phase {
Phase::Idle if !self.world.engine_present() => "no runtime grow engine in this build; the set grows only at a restart".to_owned(),
Phase::Idle => format!("idle; members: {}", s.members.describe()),
_ => String::new(),
}
} else {
s.detail.clone()
};
GrowStatus {
phase: s.phase.as_str().to_owned(),
why: s.why.clone(),
epoch: s.epoch,
target_sectors_per_member: s.target_sectors,
members: s.members.members.clone(),
members_verdict: s.members.verdict.clone(),
journal,
engine_present: self.world.engine_present(),
hotplug: self.world.hotplug(),
serving: s.serving,
identity_in_ram: s.identity_in_ram,
in_flight: s.in_flight,
since_unix_ms: s.since_unix_ms,
server_unix_ms: now_unix_ms(),
detail,
go_ahead_signer: s.signer.clone(),
set_uuid: s.members.set_uuid.clone(),
total_bytes: if s.serving { self.world.mounted_bytes() } else { 0 },
}
}
// ── the run ─────────────────────────────────────────────────────────
fn run(self: Arc<Self>, plan: Plan) {
// 1. DRAIN (T1). Pause first, so nothing new is admitted while the
// count is read; abort with nothing changed if it does not reach 0.
self.announce(Announcement::Draining { kind: "growing", reason: "the data set is being grown".into() });
self.world.pause_admission();
let started = Instant::now();
loop {
let in_flight = self.world.in_flight();
self.lock().in_flight = in_flight;
// The start-up disk test is the product's own transfer: it holds a
// probe open under /data until it is done, and the gate does not
// see it. It is waited for like one (gunnar bench R1, 2026-09-15).
let starting = self.world.starting();
if in_flight == 0 && starting.is_none() {
break;
}
if started.elapsed() >= plan.drain_timeout {
self.world.unpause_admission();
let what = match &starting {
Some(why) => format!("{in_flight} transfer(s) still in flight and the box still {why}"),
None => format!("{in_flight} transfer(s) still in flight"),
};
self.fail_untouched(format!(
"aborted: {what} after the {} s drain; nothing was changed — retry when the box is quieter, or raise drain_timeout_secs",
plan.drain_timeout.as_secs()
));
return;
}
std::thread::sleep(DRAIN_POLL);
}
// 1b. **THE LAST WRITE HAS HAPPENED — prove the twin has it.**
//
// `DATA-SET-GROWTH-FLOW.md` §5 asks the drain to prove "in_flight ==
// 0; step 0's conjunction holds again". Step 0's proof belongs to the
// outside actor and is taken before the growth starts; between it and
// this line the twin may have been stopped, resized and started again
// — monetize's `twin-sizing` phase does exactly that, and while the
// twin is down the gate here is still open. A proof taken then is a
// proof about a box that no longer exists.
//
// Here it is not. The gate is paused and nothing is in flight, so
// what the standby does not hold now, it never will: the next thing
// that happens to /data is that it is unmounted and its members are
// resized underneath it.
//
// A standby that is BEHIND aborts with nothing changed, exactly as
// the drain timeout above does. Refusing here costs a growth; not
// refusing costs whatever ruling 2 would have pulled back.
match self.world.flush_replication(plan.drain_timeout) {
FlushVerdict::CaughtUp { standbys, waited_ms } => {
tracing::info!(target: TARGET, event = "grow.flushed", standbys, waited_ms, "every standby has taken everything; the set may be unmounted");
self.lock().detail = format!("drained: nothing in flight and {standbys} standby(s) caught up after {waited_ms} ms");
}
FlushVerdict::NoStandby(why) => {
// Not a pass and not a failure. A primary growing with no
// standby at all is a choice somebody made in the config — and
// a TWIN has no standby of its own — and the run says so rather
// than implying a twin was checked.
tracing::warn!(target: TARGET, event = "grow.no_standby", %why, "the set is being grown with no standby behind it");
self.lock().detail = format!("drained: nothing in flight, and {why}");
}
FlushVerdict::CouldNotLook(why) => {
// **NOT `NoStandby`.** There is something to prove and it
// could not be looked at, so nothing is known — and the next
// thing that happens to /data is that it is unmounted.
self.world.unpause_admission();
self.fail_untouched(format!(
"aborted: the gate is shut and nothing is in flight, but whether the twin holds this set COULD NOT BE \
ESTABLISHED: {why} \u{2014} nothing was changed. A set unmounted now could only be pulled back from a twin, \
and nothing here says there is one that could."
));
return;
}
FlushVerdict::Behind { pending, needs_full_resync, why } => {
self.world.unpause_admission();
self.fail_untouched(format!(
"aborted: the gate is shut and nothing is in flight, but a standby is still behind ({pending} pending, needs_full_resync {needs_full_resync}): {why} \u{2014} nothing was changed. Nothing more is coming that would help it, and a set unmounted now could only be pulled back from a twin that does not have it."
));
return;
}
}
let epoch = self.world.epoch_for(plan.target_sectors);
self.lock().epoch = epoch;
if let Err(e) = self.note(Step::Drain) {
self.world.unpause_admission();
self.fail_untouched(format!("aborted: {e}; nothing was changed"));
return;
}
// 2. IDENTITY TO RAM (T2), CLOSE, UNMOUNT, STOP.
//
// **Still `draining` until the set is OFF.** `unmounted` is read by the
// outside actor as "the volumes are free" (monetize: main's drive and
// the twin lane both detach → resize → attach on `unmounted |
// waiting-members`). It used to be entered HERE, before the identity
// copy, the store close, the sync, the unmount and the array stop, so
// the cloud could pull a member out from under a mounted /data.
// MEASURED on the mock 2026-09-27 (gunnar 18a44e3): the twin's second
// growth in one boot left it with /data unwritable, no volume size on
// its heartbeat and its applied head stuck. The phase changes below,
// once the unmount and the stop are done and on record.
self.lock().detail = "draining → unmounting: the identity table is served from RAM, the stores close, /data unmounts, the array stops; the volumes are NOT free yet".to_owned();
tracing::info!(target: TARGET, event = "grow.unmounting", "unmounting: identity to RAM, stores close, /data unmounts, the array stops (the phase stays draining until the set is off)");
match self.world.suspend_identity("draining for the data-set growth") {
Ok(facts) => {
self.lock().identity_in_ram = true;
tracing::info!(target: TARGET, event = "grow.identity_in_ram", documents = facts.documents, bytes = facts.bytes, "the control store is served from its RAM copy");
}
Err(e) => {
self.world.unpause_admission();
let _ = self.note(Step::Abort);
self.fail_untouched(format!("aborted: the identity table could not be copied into RAM ({e}); nothing was changed"));
return;
}
}
if let Err(e) = self.world.close_stores() {
self.abort_after_suspend(format!("aborted: the stores would not close ({e})"));
return;
}
if let Err(e) = self.world.sync_and_unmount() {
self.abort_after_suspend(format!("aborted: {e}"));
return;
}
self.lock().serving = false;
if let Err(e) = self.note(Step::Unmount) {
self.recover_serve(format!("{e} after the unmount"));
return;
}
if let Err(e) = self.world.stop_array() {
self.recover_serve(format!("the array would not stop ({e})"));
return;
}
if let Err(e) = self.note(Step::Stop) {
self.recover_serve(format!("{e} after the array stopped"));
return;
}
self.enter(Phase::Unmounted, "unmounted: /data is off, the stores are closed, the array is stopped; the volumes are free");
// 3. WAIT FOR THE MEMBERS (T6-aware: a timeout keeps waiting).
let _ = self.note(Step::Wait);
self.enter(Phase::WaitingMembers, format!(
"waiting for every member to come back larger{}; the outside actor detaches, resizes and re-attaches the volumes; GrowResume looks again",
if plan.target_sectors > 0 { format!(" (≥ {} sectors each)", plan.target_sectors) } else { String::new() }
));
let waited_from = Instant::now();
let ended = loop {
// The timeout is noticed BEFORE this iteration's detail is written,
// so the "still waiting" words land with the survey they describe.
let timed_out = waited_from.elapsed() >= plan.wait_timeout;
if timed_out {
let mut s = self.lock();
if !s.wait_timed_out {
s.wait_timed_out = true;
tracing::warn!(target: TARGET, event = "grow.wait_timeout", waited_s = waited_from.elapsed().as_secs(), "the members did not come back larger within this growth's bound: with every member present /data is mounted back now at the size they carry; with one missing the wait goes on, alarmed — GrowResume looks again, GrowResume{{give_up}} serves what is there");
}
}
let _ = self.world.rescan();
let survey = self.world.survey();
// The plan's target is the VOLUME size asked of the cloud; G1's
// target is what a member can carry, always a little under it
// (partition start, md data offset). MEASURED under KVM (gunnar
// bench R1 and R5, 2026-09-15): four 64 GiB volumes back, the survey
// saying "eligible: 32.0GiB → 64.0GiB per member", and this line
// waiting for ever on the exact sector count. The same 1 % the
// engine's own G1 grants between members is granted here.
let eligible = survey.eligible && survey.target_sectors.is_some_and(|t| plan.target_sectors == 0 || target_reached(t, plan.target_sectors));
let same_set = survey.set_uuid.is_empty() || {
let s = self.lock();
s.members.set_uuid.is_empty() || s.members.set_uuid == survey.set_uuid
};
{
let mut s = self.lock();
let described = survey.describe();
s.members = survey;
s.detail = format!(
"waiting-members: {described}{}",
if !same_set { " — NOT THE SAME SET (uuid differs); refusing to grow it" } else if s.wait_timed_out { " — the wait timeout passed; still waiting for GrowResume or the members" } else { "" }
);
}
if eligible && same_set {
break Ended::Eligible;
}
// **DEFECT D's guard: /data is never left unmounted past a bound.**
// The bound is this growth's own wait timeout (monetize asks
// 3600 s). Until 2026-09-29 it was only logged and the wait went on
// for ever (T6), which is how a GrowStart for a size the members
// already had kept the twin's /data off indefinitely. Past it:
// every member present and the same set → mount it back at the size
// the members carry (the engine's own fallback), `failed` naming
// UNMOUNTED_TOO_LONG, alarmed. A member ABSENT → nothing can be
// mounted (T6 still holds, the outside actor may be re-attaching)
// — keep waiting, but alarmed, so it is never silent.
if timed_out {
let waited = waited_from.elapsed().as_secs();
let described = self.lock().members.describe();
let complete = self.lock().members.verdict == "complete";
if complete && same_set {
break Ended::Bound(format!(
"{UNMOUNTED_TOO_LONG}: /data stayed unmounted {waited} s, past this growth's {} s bound, waiting for members that never came back larger{} ({described}); mounted back at the size the members carry",
plan.wait_timeout.as_secs(),
if plan.target_sectors > 0 { format!(" (≥ {} sectors each)", plan.target_sectors) } else { String::new() }
));
}
self.raise(
UNMOUNTED_TOO_LONG,
format!(
"/data has been unmounted {waited} s, past this growth's {} s bound, and cannot be mounted back: {} ({described}) — re-attach the members (GrowResume looks again) or reboot",
plan.wait_timeout.as_secs(),
if same_set { "a member is missing" } else { "the members are NOT the set that was unmounted" }
),
false,
);
}
let resume = self.lock().resume.take();
match resume {
Some(Resume::GiveUp) => break Ended::GaveUp,
Some(Resume::LookAgain) => continue,
None => {}
}
// Sleep until the next poll — or until the timeout, whichever is
// first, so the "still waiting" row lands on time and not a poll
// late.
let poll = if timed_out {
WAIT_POLL_AFTER_TIMEOUT
} else {
WAIT_POLL.min(plan.wait_timeout.saturating_sub(waited_from.elapsed()).max(Duration::from_millis(50)))
};
let guard = self.lock();
let (guard, _) = self.wake.wait_timeout(guard, poll).unwrap_or_else(|p| p.into_inner());
drop(guard);
};
if let Ended::Bound(why) = ended {
self.raise(UNMOUNTED_TOO_LONG, why.clone(), true);
self.recover_serve(why);
return;
}
let give_up = matches!(ended, Ended::GaveUp);
if let Err(e) = self.note(Step::Rescan) {
tracing::warn!(target: TARGET, event = "grow.journal", %e, "RESCAN not recorded; going on (the disk is the truth)");
}
// 4. GROW + MOUNT (the engine; T9 inside it), then REOPEN.
self.enter(Phase::Growing, if give_up { "growing: the operator gave up waiting; assembling and mounting what is there" } else { "growing: every member is back and larger; the engine runs G1–G6" });
match self.world.grow_and_mount() {
Ok(grown) => {
self.lock().serving = true;
self.lock().grown = Some(grown.clone());
if grown.grew {
self.reopen();
self.enter(Phase::Done, format!("done: {}", grown.describe()));
tracing::warn!(target: TARGET, event = "grow.done", array_bytes = grown.array_bytes, "the data set grew and is being served");
} else {
// FAIL before REOPEN: the journal reads "it failed, and
// /data is back", in that order.
let why = if give_up { format!("gave up waiting; serving at the size the members share: {}", grown.describe()) } else { format!("the engine did not grow the set; serving at the size the members share: {}", grown.describe()) };
let _ = self.note(Step::Fail);
self.lock().why = why.clone();
self.reopen();
self.enter(Phase::Failed, format!("failed (serving): {why}"));
}
}
Err(e) => {
// FATAL: the set could not be assembled or mounted at all. The
// control plane and the status stay up; the identity stays in
// RAM; the gate stays paused (there is nothing to serve).
let _ = self.note(Step::Fail);
let why = format!("FATAL: /data could not be assembled or mounted ({e}); the control plane and GrowStatus keep answering, writes are paused, the identity table is served from RAM; re-attach the members and GrowResume is not possible from here — reboot to let the boot path finish the epoch, or reload from the twin (a separate verb)");
self.lock().why = why.clone();
self.raise("grow.fatal", why.clone(), true);
self.enter(Phase::Failed, why);
tracing::error!(target: TARGET, event = "grow.fatal", "{e}");
}
}
}
/// Nothing was touched: back to serving as before, `failed{why}`.
fn fail_untouched(&self, why: String) {
tracing::warn!(target: TARGET, event = "grow.aborted", %why, "the growth aborted with nothing changed");
let mut s = self.lock();
s.why = why.clone();
s.serving = true;
s.identity_in_ram = false;
drop(s);
self.enter(Phase::Failed, why);
}
/// The identity is in RAM and the stores may be closed, but `/data` is
/// still mounted: put everything back and abort.
fn abort_after_suspend(&self, why: String) {
let _ = self.world.reopen_stores();
match self.world.resume_identity() {
Ok(r) => self.report_resume(&r),
Err(e) => tracing::error!(target: TARGET, event = "grow.identity_resume_failed", %e, "the control store could not be put back on the stripe; it is still served from RAM"),
}
let _ = self.note(Step::Abort);
self.world.unpause_admission();
self.fail_untouched(why);
}
/// Something failed after `/data` was unmounted: assemble and mount what
/// is there (the engine's own fallback), reopen, serve, `failed{why}`
/// (T9). Only "cannot assemble/mount at all" is fatal.
fn recover_serve(&self, why: String) {
tracing::warn!(target: TARGET, event = "grow.recover", %why, "recovering: mounting what is there and serving at the size it has");
match self.world.grow_and_mount() {
Ok(grown) => {
self.lock().serving = true;
self.lock().grown = Some(grown);
let _ = self.note(Step::Fail);
self.lock().why = why.clone();
self.reopen();
self.enter(Phase::Failed, format!("failed (serving at the old size): {why}"));
}
Err(e) => {
let _ = self.note(Step::Fail);
let full = format!("FATAL: {why}; and /data could not be mounted back ({e}); the control plane and GrowStatus keep answering, writes are paused, the identity table is served from RAM — reboot to let the boot path finish the epoch");
self.lock().why = full.clone();
self.raise("grow.fatal", full.clone(), true);
self.enter(Phase::Failed, full);
tracing::error!(target: TARGET, event = "grow.fatal", "{e}");
}
}
}
/// `/data` is mounted: identity back on the stripe, stores open, gate
/// open, `REOPEN` on record, `Ready` announced.
fn reopen(&self) {
self.enter(Phase::Mounting, "mounting: /data is back; the identity table returns to the stripe and the stores reopen");
match self.world.resume_identity() {
Ok(r) => {
self.lock().identity_in_ram = false;
self.report_resume(&r);
}
Err(e) => tracing::error!(target: TARGET, event = "grow.identity_resume_failed", %e, "the control store could not be put back on the stripe; it is still served from RAM and every write is refused — restart to recover"),
}
if let Err(e) = self.world.reopen_stores() {
tracing::error!(target: TARGET, event = "grow.reopen_failed", %e, "the stores did not reopen");
}
self.world.unpause_admission();
let _ = self.note(Step::Reopen);
self.announce(Announcement::Ready);
}
fn report_resume(&self, r: &world::ResumeFacts) {
if r.drift.is_empty() {
tracing::info!(target: TARGET, event = "grow.identity_resumed", documents = r.documents_matched, off_stripe_ms = r.off_stripe_ms, "the control store is back on the stripe; the RAM copy and the disk agree");
} else {
tracing::error!(target: TARGET, event = "grow.identity_drift", drift = ?r.drift, "the control store on disk does NOT match the RAM copy that stood in for it — the disk is the truth and is what is served now; these documents changed underneath the growth");
let mut s = self.lock();
s.why = format!("{}{}identity drift on resume: {}", s.why, if s.why.is_empty() { "" } else { "; " }, r.drift.join(", "));
}
}
}
#[cfg(test)]
mod tests;