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