server/lib/src/grow/plane.rs
//! **The growth plane** (`estate`): edda's `appliance-grow` machine over holger's
//! world — the [`TwinRoster`], [`HolgerWorld`], the go-ahead policies, the
//! [`GrowPlane`] the `GrowService` door drives and the capacity ladder. The
//! [`super::Hold`] and the releases stay in every build (every write counts in
//! and out of the hold); a build without `estate` answers the door from
//! `plane_off.rs` and has nothing to grow.
use super::*;
use std::collections::BTreeMap;
use std::path::PathBuf;
use appliance_grow::world::{await_twins, twin_verdict, IdentityFacts, ResumeFacts};
use appliance_grow::{
FlushVerdict, GrowMachine, GrowStatus, Grown, Step, SurveyFacts, TwinPosition, Volumes, World,
};
#[derive(Debug, Clone, Default)]
struct TwinHeard {
address: String,
/// Where the twin said it applied to, and since when it repeats it
/// ([`crate::follow_point`]: the one rule twins and paired downstreams share).
point: crate::follow_point::FollowPoint,
total: u64,
used: u64,
/// What the twin last said it holds (`TwinReportRequest` 7-9): the root of
/// the estate manifest it holds whole, its file count, its newest writes.
estate_root: String,
estate_files: u64,
held: Vec<pb::HeldArtifact>,
/// Today's pulls and the last one — a [`crate::peers::PeerCount`], the
/// downstream's tally: the twin reports once per whole follow cycle, so one
/// report is one pull (`day` is the UTC day the count belongs to).
pulls: crate::peers::PeerCount,
day: u64,
/// The heads this primary answered the twin with, newest last: `(boot, head,
/// unix ms)` — what turns an applied SEQUENCE into a MOMENT (stinky SB #35).
answered: std::collections::VecDeque<(u64, u64, i64)>,
/// Everything this primary admitted before this moment (Unix ms) the twin
/// holds: the last answer whose head the twin has since applied. Persisted in
/// the roster file, so the retention executor — another process — reads it.
followed_to_ms: Option<i64>,
}
/// How many answered heads a twin's row keeps (a cycle reports twice).
const ANSWERS_KEPT: usize = 16;
/// **Every twin that has ever reported to this primary** — enrolled on its
/// first report (a file under `<data_root>/holger-twins/`, so a restart does
/// not forget it) and heard in memory (so a restart DOES forget its position:
/// until it reports again, it is `CouldNotLook`, never caught up and never
/// absent — gunnar's `Mirror.LogRetain` rule).
pub struct TwinRoster {
dir: Option<PathBuf>,
heard: Mutex<BTreeMap<String, TwinHeard>>,
silent_after_ms: u64,
stalled_after_ms: u64,
}
impl TwinRoster {
pub fn new(data_root: Option<&Path>) -> Self {
TwinRoster {
dir: data_root.map(|r| r.join(TWIN_ROSTER_DIR)),
heard: Mutex::new(BTreeMap::new()),
silent_after_ms: TWIN_SILENT_AFTER_MS,
stalled_after_ms: TWIN_STALLED_AFTER_MS,
}
}
/// Shorter stall bound (tests).
pub fn with_stalled_after(mut self, ms: u64) -> Self {
self.stalled_after_ms = ms;
self
}
/// Shorter silence bound (tests).
pub fn with_silent_after(mut self, ms: u64) -> Self {
self.silent_after_ms = ms;
self
}
/// Every enrolled twin: the roster directory plus anything heard.
pub fn members(&self) -> Result<Vec<String>, String> {
let mut out: Vec<String> = self.heard.lock_or_recover().keys().cloned().collect();
if let Some(dir) = &self.dir {
match std::fs::read_dir(dir) {
Ok(entries) => {
for e in entries.flatten() {
// A dot-name is the roster's own staging file, never a twin.
if let Some(n) = e.file_name().to_str().filter(|n| !n.starts_with('.')) {
out.push(n.to_owned());
}
}
}
Err(e) if e.kind() == std::io::ErrorKind::NotFound => {}
Err(e) => {
return Err(format!(
"the twin roster {} could not be read: {e}",
dir.display()
))
}
}
}
out.sort();
out.dedup();
Ok(out)
}
/// **A twin's word**: recorded, and enrolled the first time. The answer
/// is this box's position.
pub fn report(
&self,
req: &pb::TwinReportRequest,
hold: &Hold,
) -> Result<pb::TwinReportReply, String> {
let principal = req.principal.trim();
if principal.is_empty()
|| principal.len() > 128
|| principal.contains(['/', '\\', '\n'])
|| principal.starts_with('.')
{
return Err(format!(
"a twin report must name its principal as one plain word; {:?} is not one",
req.principal
));
}
if let Some(dir) = &self.dir {
let file = dir.join(principal);
if !file.exists() {
std::fs::create_dir_all(dir)
.and_then(|_| std::fs::write(&file, format!("{}\n", req.address)))
.map_err(|e| format!("enrol twin `{principal}` at {}: {e}", file.display()))?;
log::warn!("grow: twin `{principal}` ({}) enrolled — the drain now proves it holds the set before /data is unmounted", req.address);
}
}
let now = now_ms();
let mut heard = self.heard.lock_or_recover();
let prev = heard.get(principal);
let point = crate::follow_point::FollowPoint::heard(
prev.map(|h| &h.point),
req.applied_boot,
req.applied_seq,
hold.boot(),
hold.head(),
now,
);
let now_unix = (now.max(0) / 1000) as u64;
let day = now_unix / crate::visitors::WINDOW_SECS;
let today = prev
.filter(|h| h.day == day)
.map(|h| h.pulls.requests_today)
.unwrap_or(0);
let pulls = crate::peers::PeerCount {
last_pull_unix: Some(now_unix),
requests_today: today + 1,
..Default::default()
};
// ★ Applied SEQUENCE → a MOMENT (stinky SB #35): the latest answer whose
// head the twin has applied is the moment before which it holds every
// write — the retention guard's line.
let mut answered = prev.map(|h| h.answered.clone()).unwrap_or_default();
if answered.len() == ANSWERS_KEPT {
answered.pop_front();
}
answered.push_back((hold.boot(), hold.head(), now));
let before = prev.and_then(|h| h.followed_to_ms);
let reached = answered
.iter()
.rev()
.find(|(b, h, _)| {
*b == req.applied_boot && *h == req.applied_seq && req.applied_boot == hold.boot()
})
.map(|(_, _, ms)| *ms);
let followed_to_ms = match (before, reached) {
(Some(a), Some(b)) => Some(a.max(b)),
(a, b) => a.or(b),
};
if followed_to_ms != before {
if let (Some(dir), Some(ms)) = (&self.dir, followed_to_ms) {
let file = dir.join(principal);
let tmp = dir.join(format!(".{principal}.part"));
if let Err(e) =
std::fs::write(&tmp, format!("{}\n{FOLLOWED_TO} {ms}\n", req.address))
.and_then(|_| std::fs::rename(&tmp, &file))
{
log::warn!("grow: twin `{principal}` followed to {ms}, and the roster could not say so ({e}); a retention run keeps what it cannot prove followed");
}
}
}
heard.insert(
principal.to_owned(),
TwinHeard {
address: req.address.clone(),
point,
total: req.disk_total_bytes,
used: req.disk_used_bytes,
estate_root: req.estate_root.clone(),
estate_files: req.estate_files,
held: req
.held
.iter()
.take(crate::mirror_cycle::WRITTEN_KEPT)
.cloned()
.collect(),
pulls,
day,
answered,
followed_to_ms,
},
);
drop(heard);
Ok(pb::TwinReportReply {
boot: hold.boot(),
head_seq: hold.head(),
})
}
/// Every enrolled twin as `twin_verdict` reads it: a position from another
/// boot is no position (`applied 0`, never heard).
pub fn positions(&self, boot: u64, now: i64) -> Vec<TwinPosition> {
let heard = self.heard.lock_or_recover();
heard
.iter()
.map(|(p, h)| TwinPosition {
principal: p.clone(),
applied_seq: h.point.applied_in(boot),
last_seen_unix_ms: h.point.last_seen_ms,
seen_ago_ms: (now - h.point.last_seen_ms).max(0) as u64,
silent_after_ms: self.silent_after_ms,
})
.collect()
}
/// The rows monetize's twin sizing reads (`GrowService.Twins`).
pub fn rows(&self, hold: &Hold) -> Result<Vec<pb::TwinRow>, String> {
let now = now_ms();
let heard = self.heard.lock_or_recover().clone();
Ok(self
.members()?
.into_iter()
.map(|p| match heard.get(&p) {
None => pb::TwinRow {
principal: p,
state: "silent".into(),
..Default::default()
},
Some(h) => {
let j = h.point.judge(
hold.boot(),
hold.head(),
now,
self.silent_after_ms,
self.stalled_after_ms,
);
pb::TwinRow {
principal: p,
address: h.address.clone(),
state: j.state.into(),
last_seen_unix_ms: h.point.last_seen_ms,
lag_entries: j.lag,
disk_total_bytes: h.total,
disk_used_bytes: h.used,
fill_permille: if h.total == 0 {
0
} else {
((h.used as u128 * 1000) / h.total as u128) as u32
},
applied_seq: j.applied,
head_seq: hold.head(),
seen_ago_ms: j.seen_ago_ms,
estate_root: h.estate_root.clone(),
estate_files: h.estate_files,
held: h.held.clone(),
pulls_today: if h.day
== (now.max(0) / 1000) as u64 / crate::visitors::WINDOW_SECS
{
h.pulls.requests_today
} else {
0
},
last_pull_unix: h.pulls.last_pull_unix.unwrap_or(0) as i64,
}
}
})
.collect())
}
/// **Have every enrolled twin's positions reached `barrier`?** — the rule
/// is `appliance_grow::world::twin_verdict`'s, and nothing else.
pub fn verdict(&self, hold: &Hold, barrier: u64) -> Result<(u64, u64), FlushVerdict> {
let members = self.members().map_err(FlushVerdict::CouldNotLook)?;
twin_verdict(&members, &self.positions(hold.boot(), now_ms()), barrier)
}
}
// ── holger's world ──────────────────────────────────────────────────────────
/// **holger, as the growth sees it** — `appliance_grow::World` over the
/// [`Hold`], the [`TwinRoster`] and a volume layer (`MdVolumes` on the box, a
/// fake in the tests).
pub struct HolgerWorld {
hold: Arc<Hold>,
roster: Arc<TwinRoster>,
volumes: Arc<dyn Volumes>,
/// The box's posture, read LIVE: `BecomeTwin` switches a running box to a
/// twin, and a pending recharge makes it not serve (`TwinPosture::serving`).
posture: Arc<crate::twin::TwinPosture>,
data_root: PathBuf,
hotplug: Box<dyn Fn() -> bool + Send + Sync>,
close_wait: Duration,
}
impl HolgerWorld {
pub fn new(
hold: Arc<Hold>,
roster: Arc<TwinRoster>,
volumes: Arc<dyn Volumes>,
posture: Arc<crate::twin::TwinPosture>,
data_root: PathBuf,
) -> Self {
HolgerWorld {
hold,
roster,
volumes,
posture,
data_root,
hotplug: Box::new(|| appliance_grow::probe::pci_hotplug().present),
close_wait: CLOSE_WAIT,
}
}
/// Another answer to "does the running kernel have PCI hot-plug" (tests;
/// the box reads the kernel).
pub fn with_hotplug(mut self, f: impl Fn() -> bool + Send + Sync + 'static) -> Self {
self.hotplug = Box::new(f);
self
}
}
impl World for HolgerWorld {
fn engine_present(&self) -> bool {
self.volumes.engine_present()
}
fn hotplug(&self) -> bool {
(self.hotplug)()
}
fn journal_path(&self) -> PathBuf {
self.volumes.journal_path()
}
/// **Mounted AND open AND not awaiting a recharge** — the bit monetize
/// recharges on (`restore_decision`: a failed run that reads `serving: false`
/// is re-laid and pulled back from the twin). `TwinPosture::serving` is the
/// recharge lane's seam: false while this box waits to be filled.
fn serving(&self) -> bool {
self.volumes.is_mounted() && !self.hold.closed() && self.posture.serving()
}
fn mounted_bytes(&self) -> u64 {
if !self.volumes.engine_present() && self.serving() {
return crate::metering::statvfs(&self.data_root)
.map(|f| f.total_bytes)
.unwrap_or(0);
}
self.volumes.mounted_bytes()
}
fn pause_admission(&self) {
// On a TWIN this also holds its FOLLOWER (T21, gunnar's ruling: the twin
// grows exactly like main, never restarted): the follow cycle parks while
// paused (`crate::twin_follow`), and every put it still has in flight
// counts in `in_flight` like any upload, because it passes the same
// `Held` backends.
self.hold.pause();
}
fn unpause_admission(&self) {
self.hold.unpause();
}
/// **A TWIN HAS NO STANDBY OF ITS OWN** (gunnar b0922aa): its set is a copy
/// of its primary's, which is where it is pulled back from. A PRIMARY holds
/// every enrolled twin to the BARRIER — its head as it stands NOW, the gate
/// shut and nothing in flight (gunnar 364a4f5): a twin past it holds
/// everything it must.
fn flush_replication(&self, ceiling: Duration) -> FlushVerdict {
if self.posture.is_twin() {
return FlushVerdict::NoStandby("this box is a twin: its set is a copy of its primary's, which is where it is pulled back from".to_owned());
}
let barrier = self.hold.head();
let started = std::time::Instant::now();
match await_twins(ceiling, Duration::from_millis(250), || self.roster.verdict(&self.hold, barrier)) {
Ok((0, _)) => FlushVerdict::NoStandby(
"no twin is enrolled on this primary (none has ever reported a position to it), so there is no standby to prove".to_owned(),
),
Ok((n, _)) => FlushVerdict::CaughtUp { standbys: n, waited_ms: started.elapsed().as_millis() as u64 },
Err(v) => v,
}
}
fn in_flight(&self) -> u64 {
self.hold.in_flight()
}
/// holger keeps NO identity table on the stripe: its auth is its config
/// (bearer, OIDC, mTLS roles) and the passkey store is files read per
/// request, which the closed stores refuse meanwhile. Nothing to copy.
fn suspend_identity(&self, _why: &str) -> Result<IdentityFacts, String> {
Ok(IdentityFacts::default())
}
fn resume_identity(&self) -> Result<ResumeFacts, String> {
Ok(ResumeFacts::default())
}
fn close_stores(&self) -> Result<(), String> {
let released = self.hold.close(self.close_wait)?;
log::info!(
"grow: stores closed — released {}",
if released.is_empty() {
"nothing (nothing was registered)".to_owned()
} else {
released.join(", ")
}
);
Ok(())
}
fn reopen_stores(&self) -> Result<(), String> {
self.hold.open()
}
fn sync_and_unmount(&self) -> Result<(), String> {
self.volumes.sync_and_unmount()
}
fn stop_array(&self) -> Result<(), String> {
self.volumes.stop_array()
}
fn rescan(&self) -> Result<(), String> {
self.volumes.rescan()
}
fn survey(&self) -> SurveyFacts {
self.volumes.survey()
}
fn grow_and_mount(&self) -> Result<Grown, String> {
self.volumes.grow_and_mount()
}
fn epoch_for(&self, target_sectors: u64) -> u64 {
self.volumes.epoch_for(target_sectors)
}
fn note(&self, epoch: u64, target_sectors: u64, step: Step) -> Result<(), String> {
self.volumes.note(epoch, target_sectors, step)
}
fn remember_nonce(&self, nonce: &str) -> Result<bool, String> {
self.volumes.remember_nonce(nonce)
}
}
/// The volume layer this BUILD carries: korp-installer's engine with
/// `grow-runtime`, none without (status answers `engine_present: false`, start
/// is refused by name).
pub fn build_volumes() -> Arc<dyn Volumes> {
#[cfg(feature = "grow-runtime")]
{
Arc::new(appliance_grow::md::MdVolumes::new())
}
#[cfg(not(feature = "grow-runtime"))]
{
Arc::new(appliance_grow::world::NoEngine)
}
}
/// **The metered box's go-ahead policy, kept** (A7 row 8): monetize's key when it has one,
/// else AWAITING, so a key the installer feeds after the start is applied in place
/// ([`appliance_grow::MonetizePolicy::set_key`]).
#[cfg(feature = "metered")]
pub fn monetize_policy(monetize_pubkey: Option<[u8; 32]>) -> Arc<appliance_grow::MonetizePolicy> {
if let Some(k) = monetize_pubkey {
match appliance_grow::MonetizePolicy::new(&k) {
Ok(p) => return Arc::new(p),
Err(e) => log::error!(
"grow: {e}; a monetize go-ahead is refused until a valid key is configured"
),
}
}
Arc::new(appliance_grow::MonetizePolicy::awaiting())
}
/// Who may say go: monetize's key on a metered box that has one (the SAME key
/// its entitlement facts verify under), the operator on this admin door always.
pub fn go_ahead_policy(
monetize_pubkey: Option<[u8; 32]>,
) -> Arc<dyn appliance_grow::GoAheadPolicy> {
#[cfg(feature = "metered")]
{
if let Some(k) = monetize_pubkey {
match appliance_grow::MonetizePolicy::new(&k) {
Ok(p) => return Arc::new(p),
Err(e) => log::error!(
"grow: {e}; a monetize go-ahead is refused until a valid key is configured"
),
}
}
Arc::new(appliance_grow::MonetizePolicy::awaiting())
}
#[cfg(not(feature = "metered"))]
{
let _ = monetize_pubkey;
Arc::new(appliance_grow::OperatorPolicy::default())
}
}
// ── the plane the doors hold ────────────────────────────────────────────────
/// **The growth, as holger's doors hold it**: the machine, the hold, the roster.
pub struct GrowPlane {
pub machine: Arc<GrowMachine>,
pub hold: Arc<Hold>,
pub roster: Arc<TwinRoster>,
pub posture: Arc<crate::twin::TwinPosture>,
}
impl std::fmt::Debug for GrowPlane {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.debug_struct("GrowPlane")
.field("phase", &self.machine.phase())
.field("twin", &self.is_twin())
.finish()
}
}
impl GrowPlane {
pub fn new(world: HolgerWorld, policy: Arc<dyn appliance_grow::GoAheadPolicy>) -> Arc<Self> {
let (hold, roster, posture) = (
Arc::clone(&world.hold),
Arc::clone(&world.roster),
Arc::clone(&world.posture),
);
let machine = GrowMachine::with_policy(Arc::new(world), policy);
Arc::new(GrowPlane {
machine,
hold,
roster,
posture,
})
}
/// Is this box a twin — read live from the posture.
pub fn is_twin(&self) -> bool {
self.posture.is_twin()
}
/// The status, with the recharge lane's seam folded in: a box awaiting its
/// recharge (`TwinPosture::serving`) does not serve, whatever the set says.
pub fn status(&self) -> pb::GrowStatusReply {
self.fold(status_reply(self.machine.status(), self.is_twin()))
}
fn fold(&self, mut s: pb::GrowStatusReply) -> pb::GrowStatusReply {
if !self.posture.serving() {
s.serving = false;
s.total_bytes = 0;
}
s
}
pub fn start(&self, req: &pb::GrowStartRequest) -> Result<pb::GrowStatusReply, String> {
if let Some(from) = self.posture.awaiting_from() {
return Err(format!("this box is awaiting its recharge from `{from}` and serves nothing; a growth cannot start until the recharge is whole"));
}
let r = appliance_grow::StartRequest {
target_sectors_per_member: req.target_sectors_per_member,
go_ahead: req.go_ahead.clone(),
drain_timeout_secs: req.drain_timeout_secs,
wait_members_timeout_secs: req.wait_members_timeout_secs,
};
self.machine
.start(&r)
.map(|s| self.fold(status_reply(s, self.is_twin())))
}
pub fn resume(&self, give_up: bool) -> Result<pb::GrowStatusReply, String> {
self.machine
.resume(give_up)
.map(|s| self.fold(status_reply(s, self.is_twin())))
}
/// **The watermark** (T11, monetize's `flushing`): every enrolled twin at
/// this box's head as of NOW, waited for within `timeout` — the gate stays
/// OPEN (this is step 0, taken before the growth; the drain takes its own).
/// Blocking: the door runs it off the executor.
pub fn flush(&self, timeout: Duration) -> pb::GrowFlushReply {
let started = std::time::Instant::now();
let at = now_ms();
let catalog_repos = 0;
if self.is_twin() {
return pb::GrowFlushReply {
caught_up: true,
flushed_at_unix_ms: at,
verdict: "this box is a twin: it has no standby of its own; its set is a copy of its primary's".into(),
catalog_repos,
..Default::default()
};
}
let head = self.hold.head();
let verdict = await_twins(timeout, Duration::from_millis(250), || {
self.roster.verdict(&self.hold, head)
});
let waited_ms = started.elapsed().as_millis() as u64;
match verdict {
Ok((0, _)) => pb::GrowFlushReply {
caught_up: true,
flushed_at_unix_ms: at,
waited_ms,
verdict: "no twin is enrolled on this primary (none has ever reported to it): nothing to prove, and nothing is claimed".into(),
..Default::default()
},
Ok((n, _)) => pb::GrowFlushReply {
caught_up: true,
standbys: n,
flushed_at_unix_ms: at,
last_success_unix_ms: at,
waited_ms,
in_flight: self.hold.in_flight(),
verdict: format!("{n} twin(s) have applied this box's head {head}: everything written before the flush is on them"),
..Default::default()
},
Err(FlushVerdict::Behind { pending, why, .. }) => pb::GrowFlushReply {
caught_up: false,
pending,
flushed_at_unix_ms: at,
waited_ms,
in_flight: self.hold.in_flight(),
verdict: why,
..Default::default()
},
Err(FlushVerdict::CouldNotLook(why)) | Err(FlushVerdict::NoStandby(why)) => pb::GrowFlushReply {
caught_up: false,
flushed_at_unix_ms: at,
waited_ms,
verdict: format!("COULD NOT BE ESTABLISHED: {why}"),
..Default::default()
},
Err(FlushVerdict::CaughtUp { .. }) => unreachable!("twin_verdict answers CaughtUp as Ok"),
}
}
/// A twin's report, on the primary. A twin does not take reports.
pub fn twin_report(&self, req: &pb::TwinReportRequest) -> Result<pb::TwinReportReply, String> {
if self.is_twin() {
return Err(
"this box is a twin: a twin reports to its primary and takes no reports".into(),
);
}
self.roster.report(req, &self.hold)
}
pub fn twins(&self) -> Result<pb::TwinsReply, String> {
Ok(pb::TwinsReply {
twins: self.roster.rows(&self.hold)?,
})
}
}
/// The shared status, on holger's wire — field for field.
pub fn status_reply(s: GrowStatus, twin: bool) -> pb::GrowStatusReply {
pb::GrowStatusReply {
phase: s.phase,
why: s.why,
epoch: s.epoch,
target_sectors_per_member: s.target_sectors_per_member,
members: s
.members
.into_iter()
.map(|m| pb::GrowMember {
index: m.index,
device: m.device,
partition: m.partition,
disk_bytes: m.disk_bytes,
set_bytes: m.set_bytes,
offered_bytes: m.offered_bytes,
larger: m.larger,
})
.collect(),
members_verdict: s.members_verdict,
journal: s.journal.map(|j| pb::GrowState {
engine_present: true,
phase: j.phase_word().to_owned(),
epoch: j.epoch.unwrap_or(0),
target_sectors: j.target_sectors.unwrap_or(0),
steps: j.steps.clone(),
partial_members: j.partial_members.clone(),
detail: j.detail(),
journal_path: j.path.clone(),
}),
engine_present: s.engine_present,
hotplug: s.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: s.server_unix_ms,
detail: s.detail,
go_ahead_signer: s.go_ahead_signer,
set_uuid: s.set_uuid,
total_bytes: s.total_bytes,
twin,
}
}
// ── the capacity door's growth verdict ──────────────────────────────────────
/// `CapacityReply`'s growth half: `(SetGrowth wire number, reason, data set,
/// hotplug)` — edda's ladder over the facts measured off THIS machine.
pub fn capacity_growth(root: &Path) -> (i32, String, String, bool) {
let facts =
appliance_grow::capacity::GrowthFacts::measure(root, cfg!(feature = "grow-runtime"));
let (growth, reason) = appliance_grow::capacity::decide_growth(&facts, root, "holger");
let data_set = facts
.kind
.as_ref()
.map(|k| k.to_string())
.unwrap_or_default();
(growth.wire(), reason, data_set, facts.hotplug)
}