server/lib/src/grow.rs
//! **holger's data set grows WITHOUT a restart** — the door monetize's growth
//! drives (`holger.grow.v1.GrowService`), over edda's `appliance-grow`: gunnar's
//! phase machine, moved AS IS (gunnar 18a44e3 + 364a4f5 + b0922aa + dd672ce),
//! so both products grow the same way and monetize reads the same facts.
//!
//! ```text
//! monetize (vm1) holger (vm2 main, vm3 twin)
//! ────────────── ──────────────────────────
//! Capacity: total / used / growth ◀──── AdminService.Capacity (+ SetGrowth)
//! twin-sizing: Twins (fill) ◀──── GrowService.Twins (primary's roster)
//! flushing: Flush ────▶ every enrolled twin ≥ head, or not caught up
//! starting: Start{target, go_ahead}────▶ GrowMachine::start: draining (writes HELD)
//! twins held to the BARRIER; stores RELEASED;
//! umount; md stop — only THEN `unmounted`
//! cloud: detach → resize → attach ◀──── Status (no credential): unmounted | waiting-members
//! resuming: Resume ────▶ look again → growing → mounting → done
//! (twin lane: the same four on vm3) a twin grows too: its drain holds its
//! FOLLOWER and proves no standby (NoStandby)
//! ```
//!
//! The [`Hold`], the releases and the twin's follow position are in every build; the
//! growth plane itself (edda's machine, [`TwinRoster`], [`HolgerWorld`], the
//! [`GrowPlane`]) is the estate's (`estate`, `grow/plane.rs`), and a build without it
//! answers `GrowService` from `grow/plane_off.rs`: nothing to grow, refused by name.
//!
//! # What is shared and what is holger's
//!
//! The machine, the go-ahead, the journal, the twin verdict, the capacity ladder
//! and the volume layer (`appliance_grow::md::MdVolumes`, feature `grow-runtime`)
//! are edda's. What is holger's is here: the [`Hold`] (holger has no admission
//! gate, so the growth brought one: every write — every door AND every slow path,
//! because every repository backend is wrapped once at wiring in [`Held`] —
//! counts in and out of it), the [`Releasable`] registry (the descriptors holger
//! keeps open under its data root, given up before the unmount and taken back
//! after), the [`TwinRoster`] (holger has no transaction log: its sequence is
//! the [`Hold`]'s write counter, and a twin reports how far into it it has
//! followed), and [`HolgerWorld`], which is `appliance_grow::World` over them.
//!
//! # `serving` is the truth monetize recharges on
//!
//! `GrowStatus.serving` is true only while the volumes are mounted AND the
//! stores are open. monetize's restore decision (`restore_decision`) orders a
//! recharge from the twin on a FAILED run that reads `serving: false`, so this
//! bit is never a guess: [`HolgerWorld::serving`] reads the mount and the hold.
//!
//! # What only a VM proves
//!
//! Every volume verb is a real syscall on a real block device (umount2, md
//! `array_state`, `/sys/bus/pci/rescan`, XFS_IOC_FSGROWFSDATA), and the provider's
//! detach → resize → attach is the cloud's. The in-process tests drive the REAL
//! machine and holger's REAL hold, releases and roster against a fake volume
//! layer; the mock-upcloud functest proves the rest.
use crate::lock::{LockOrRecover as _, RwLockOrRecover as _};
use std::path::Path;
use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
use std::sync::{Arc, Mutex, RwLock};
use std::time::Duration;
use traits::RepositoryBackendTrait;
use crate::grpc::holger_grow_proto as pb;
/// The sentence every held upload answers with — one for every door.
pub const HELD: &str = "holger is growing its storage to make room for new members. Uploads are paused — please try again in a few minutes.";
/// Seconds a held HTTP upload is told to wait (`Retry-After`).
pub const RETRY_AFTER_SECS: u32 = 120;
/// A twin report older than this says nothing about now: three follow cycles
/// at the default minute ([`crate::twin::DEFAULT_POLL_SECONDS`]).
pub const TWIN_SILENT_AFTER_MS: u64 = 3 * crate::twin::DEFAULT_POLL_SECONDS * 1000;
/// **A twin that keeps reporting and does not move is `stalled`** after this
/// long: its applied position unchanged at every report while this primary's
/// head was ahead of it (lag above zero) — gunnar's heartbeat rule for a stuck
/// applied head (`gunnar-grpc` heartbeat, `BEHIND_POLLS` = 5 polls), in follow
/// cycles: five at the default minute ([`crate::twin::DEFAULT_POLL_SECONDS`]).
/// A twin that moves, however slowly, is `behind`; `stalled` is the one the
/// owner is woken for (the ops kit's `estate watch`: `twin.<p>.stalled`).
pub const TWIN_STALLED_AFTER_MS: u64 = 5 * crate::twin::DEFAULT_POLL_SECONDS * 1000;
/// Where a primary keeps the names of the twins that have ever reported to it
/// (`<data_root>/holger-twins/<principal>`) — enrolment, so a twin that has
/// not reported since a restart is UNKNOWN, never absent.
pub const TWIN_ROSTER_DIR: &str = "holger-twins";
/// How long the close waits for the reads already in flight to finish.
pub const CLOSE_WAIT: Duration = Duration::from_secs(30);
/// Unix ms on this box's clock (the same reading as edda's `appliance_grow::now_unix_ms`,
/// which the growth plane's own tests use; the hold is in every build).
fn now_ms() -> i64 {
std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.map(|d| d.as_millis() as i64)
.unwrap_or(0)
}
// ── the hold ────────────────────────────────────────────────────────────────
/// **holger's ONE admission gate for the growth.** Per process, shared by the
/// gRPC write gates, the HTTP door and every [`Held`] backend, so they all say
/// the same sentence at the same moment.
///
/// It is also holger's SEQUENCE: [`Hold::head`] counts every write admitted
/// since this boot (from 1, so a head of 0 never happens — `twin_verdict`
/// refuses 0 as "proves nothing"), and [`Hold::boot`] names the boot, so a
/// twin's position from before a restart is never read as one after it.
pub struct Hold {
paused: AtomicBool,
closed: AtomicBool,
in_flight: AtomicU64,
reading: AtomicU64,
head: AtomicU64,
boot: u64,
releases: Mutex<Vec<Arc<dyn Releasable>>>,
}
impl std::fmt::Debug for Hold {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.debug_struct("Hold")
.field("paused", &self.paused())
.field("closed", &self.closed())
.field("in_flight", &self.in_flight())
.field("head", &self.head())
.finish()
}
}
impl Default for Hold {
fn default() -> Self {
Self::new()
}
}
/// One admitted write; dropping it counts it out and advances the head.
pub struct WriteTicket<'a>(&'a Hold);
impl Drop for WriteTicket<'_> {
fn drop(&mut self) {
self.0.head.fetch_add(1, Ordering::SeqCst);
self.0.in_flight.fetch_sub(1, Ordering::SeqCst);
}
}
/// One admitted read; the close waits for these.
pub struct ReadTicket<'a>(&'a Hold);
impl Drop for ReadTicket<'_> {
fn drop(&mut self) {
self.0.reading.fetch_sub(1, Ordering::SeqCst);
}
}
impl Hold {
pub fn new() -> Self {
let boot = std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.map(|d| d.as_nanos() as u64)
.unwrap_or(1)
.max(1);
Hold {
paused: AtomicBool::new(false),
closed: AtomicBool::new(false),
in_flight: AtomicU64::new(0),
reading: AtomicU64::new(0),
head: AtomicU64::new(1),
boot,
releases: Mutex::new(Vec::new()),
}
}
pub fn paused(&self) -> bool {
self.paused.load(Ordering::SeqCst)
}
pub fn closed(&self) -> bool {
self.closed.load(Ordering::SeqCst)
}
pub fn in_flight(&self) -> u64 {
self.in_flight.load(Ordering::SeqCst)
}
/// Writes admitted since this boot, from 1.
pub fn head(&self) -> u64 {
self.head.load(Ordering::SeqCst)
}
/// This boot's name.
pub fn boot(&self) -> u64 {
self.boot
}
pub fn pause(&self) {
self.paused.store(true, Ordering::SeqCst);
}
pub fn unpause(&self) {
self.paused.store(false, Ordering::SeqCst);
}
/// **Admit one write, or refuse it by name.** Counted IN before the pause
/// is read, so a writer racing the pause is either refused or seen by the
/// drain — never neither.
pub fn admit_write(&self) -> Result<WriteTicket<'_>, String> {
self.in_flight.fetch_add(1, Ordering::SeqCst);
if self.paused() || self.closed() {
self.in_flight.fetch_sub(1, Ordering::SeqCst);
return Err(holger_errcode::GROW_WRITES_HELD.tag(HELD));
}
Ok(WriteTicket(self))
}
/// Admit one read of `repo`, or refuse it while the stores are closed.
pub fn admit_read(&self, repo: &str) -> Result<ReadTicket<'_>, String> {
self.reading.fetch_add(1, Ordering::SeqCst);
if self.closed() {
self.reading.fetch_sub(1, Ordering::SeqCst);
return Err(closed_refusal(repo));
}
Ok(ReadTicket(self))
}
/// The gRPC / HTTP gates' question: `Some(sentence)` while writes are held.
pub fn write_refusal(&self) -> Option<String> {
(self.paused() || self.closed()).then(|| HELD.to_owned())
}
/// Register something that holds a descriptor under the data root.
pub fn register(&self, r: Arc<dyn Releasable>) {
self.releases.lock_or_recover().push(r);
}
/// What is registered, for the status and the tests.
pub fn registered(&self) -> Vec<String> {
self.releases
.lock_or_recover()
.iter()
.map(|r| r.what())
.collect()
}
/// **Close the stores**: reads are refused from here, the reads already in
/// flight are waited for (≤ `wait`), then every registered holder gives up
/// its descriptor. Any refusal puts back what was released and reopens —
/// the growth then aborts with nothing changed (EBUSY would say the same,
/// later and less clearly).
pub fn close(&self, wait: Duration) -> Result<Vec<String>, String> {
self.closed.store(true, Ordering::SeqCst);
let deadline = std::time::Instant::now() + wait;
while self.reading.load(Ordering::SeqCst) > 0 {
if std::time::Instant::now() >= deadline {
let n = self.reading.load(Ordering::SeqCst);
self.closed.store(false, Ordering::SeqCst);
return Err(format!(
"{n} read(s) still in flight after {} s; the stores were not closed",
wait.as_secs()
));
}
std::thread::sleep(Duration::from_millis(20));
}
let all: Vec<Arc<dyn Releasable>> = self.releases.lock_or_recover().clone();
let mut done: Vec<Arc<dyn Releasable>> = Vec::new();
for r in &all {
if let Err(e) = r.release() {
for back in done.iter().rev() {
if let Err(e2) = back.reacquire() {
log::error!(
"grow: {} could not be taken back after an aborted close: {e2}",
back.what()
);
}
}
self.closed.store(false, Ordering::SeqCst);
return Err(format!("{} would not let go: {e}", r.what()));
}
done.push(Arc::clone(r));
}
Ok(done.iter().map(|r| r.what()).collect())
}
/// **Reopen**: every holder takes its descriptor back, then reads are let
/// in. Every failure is named; the rest still reopen.
pub fn open(&self) -> Result<(), String> {
let all: Vec<Arc<dyn Releasable>> = self.releases.lock_or_recover().clone();
let failed: Vec<String> = all
.iter()
.filter_map(|r| r.reacquire().err().map(|e| format!("{}: {e}", r.what())))
.collect();
self.closed.store(false, Ordering::SeqCst);
if failed.is_empty() {
Ok(())
} else {
Err(failed.join("; "))
}
}
}
fn closed_refusal(repo: &str) -> String {
holger_errcode::GROW_STORES_CLOSED.tag(format!(
"repository `{repo}` is closed: holger's data set is unmounted while its volumes are grown — it is back in a few minutes"
))
}
// ── what holds a descriptor under the data root ─────────────────────────────
/// **Something that keeps a file open under the data root for the process's
/// life** — MEASURED 2026-09-27: the audit segment, the promotion job ledger,
/// the job queue's ledger, and every znippy archive a repository serves from.
/// Any one of them makes the unmount answer EBUSY.
pub trait Releasable: Send + Sync {
/// What it is, for the refusal and the log.
fn what(&self) -> String;
/// Close the descriptor.
fn release(&self) -> Result<(), String>;
/// Open it again.
fn reacquire(&self) -> Result<(), String>;
}
/// The audit sink's segment ([`crate::audit::AuditLog::release_files`]).
pub struct AuditFiles(pub Arc<dyn crate::audit::AuditLog>);
impl Releasable for AuditFiles {
fn what(&self) -> String {
"the audit segment".into()
}
fn release(&self) -> Result<(), String> {
self.0.release_files().map_err(|e| e.to_string())
}
fn reacquire(&self) -> Result<(), String> {
self.0.reacquire_files().map_err(|e| e.to_string())
}
}
/// The promotion job ledger (`holger-jobs/jobs.redb`).
#[cfg(feature = "intake")]
pub struct PromotionLedger(pub Arc<crate::promote::JobLedger>);
#[cfg(feature = "intake")]
impl Releasable for PromotionLedger {
fn what(&self) -> String {
"the promotion job ledger".into()
}
fn release(&self) -> Result<(), String> {
self.0.release()
}
fn reacquire(&self) -> Result<(), String> {
self.0.reacquire()
}
}
/// The durable job queue's ledger (`jobs/jobs.redb`, feature `jobs` — which the
/// appliance variants carry): the pool is wound down and the file dropped, then
/// reopened and the pool restarted ([`crate::jobs::JobQueue::release`]).
#[cfg(feature = "jobs")]
pub struct JobQueueLedger(pub Arc<crate::jobs::JobQueue>);
#[cfg(feature = "jobs")]
impl Releasable for JobQueueLedger {
fn what(&self) -> String {
"the job queue ledger".into()
}
fn release(&self) -> Result<(), String> {
self.0.release()
}
fn reacquire(&self) -> Result<(), String> {
self.0.reacquire()
}
}
/// **A holder that cannot let go, said by name** — for a descriptor holger has no
/// way to close while it runs: registered, it refuses the growth at the close
/// with nothing changed, rather than letting the unmount meet EBUSY later.
pub struct CannotRelease(pub String);
impl Releasable for CannotRelease {
fn what(&self) -> String {
self.0.clone()
}
fn release(&self) -> Result<(), String> {
Err(format!("{} is held open by its workers for the life of the process and cannot be released while holger runs — build the appliance without it, or grow at a restart", self.0))
}
fn reacquire(&self) -> Result<(), String> {
Ok(())
}
}
// ── every repository backend, held ──────────────────────────────────────────
/// How a released backend is built again: the repository's own config, through
/// the same `backend_from_config_with` the boot used.
pub type Rebuild = Box<dyn Fn() -> anyhow::Result<Arc<dyn RepositoryBackendTrait>> + Send + Sync>;
/// **A repository's backend under the growth's hold** — wrapped ONCE at wiring
/// (`wire_holger`), before any route, proxy, mirror, replication pair or job
/// takes a clone, so every write that exists passes [`Hold::admit_write`] and
/// every read [`Hold::admit_read`]: the doors, the mirror, a twin's follow, the
/// proxy's cache fill, `Holger::put`.
///
/// It is also the backend's [`Releasable`]: the archive it serves from is
/// DROPPED at the close (nothing else holds the inner `Arc`) and REBUILT from the
/// config after the reopen. The static facts (`name`, `format`, …) are read once
/// at wrap time, so they answer while the archive is closed.
///
/// It is a [`traits::Decorator`]: every trait method reaches the inner backend
/// through [`around`](traits::Decorator::around), which admits the read or the
/// write first, so no method can bypass the hold.
pub struct Held {
name: String,
hold: Arc<Hold>,
inner: RwLock<Option<Arc<dyn RepositoryBackendTrait>>>,
rebuild: Option<Rebuild>,
format: traits::ArtifactFormat,
writable: bool,
caps: traits::Caps,
}
impl Held {
/// Wrap `inner`; `rebuild` is how it is built again after a release
/// (`None`: it cannot be released, and the close says so).
pub fn wrap(
inner: Arc<dyn RepositoryBackendTrait>,
name: impl Into<String>,
hold: Arc<Hold>,
rebuild: Option<Rebuild>,
) -> Arc<Held> {
Arc::new(Held {
name: name.into(),
format: inner.format(),
writable: inner.is_writable(),
caps: inner.caps(),
hold,
inner: RwLock::new(Some(inner)),
rebuild,
})
}
fn inner(&self) -> anyhow::Result<Arc<dyn RepositoryBackendTrait>> {
self.inner
.read_or_recover()
.clone()
.ok_or_else(|| anyhow::anyhow!(closed_refusal(&self.name)))
}
}
impl Releasable for Held {
fn what(&self) -> String {
format!("repository `{}`", self.name)
}
fn release(&self) -> Result<(), String> {
if self.rebuild.is_none() {
return Err(format!("repository `{}` has no way to be rebuilt after a release, so its archive stays open", self.name));
}
let taken = self.inner.write_or_recover().take();
if let Some(inner) = taken {
// Somebody still holds the backend — a download stream's chunk
// iterator, a proxy mid-fill. Its file stays open whatever this
// does, so the release is refused and the backend put back.
if Arc::strong_count(&inner) > 1 {
*self.inner.write_or_recover() = Some(inner);
return Err(format!(
"repository `{}` is still in use by a stream",
self.name
));
}
}
Ok(())
}
fn reacquire(&self) -> Result<(), String> {
let mut slot = self.inner.write_or_recover();
if slot.is_none() {
let build = self
.rebuild
.as_ref()
.ok_or_else(|| format!("repository `{}` cannot be rebuilt", self.name))?;
*slot =
Some(build().map_err(|e| {
format!("repository `{}` could not be reopened: {e:#}", self.name)
})?);
}
Ok(())
}
}
impl traits::Decorator for Held {
fn name(&self) -> &str {
&self.name
}
fn format(&self) -> traits::ArtifactFormat {
self.format.clone()
}
fn is_writable(&self) -> bool {
self.writable
}
fn caps(&self) -> traits::Caps {
self.caps
}
fn around<T>(
&self,
access: traits::Access,
call: impl FnOnce(&dyn RepositoryBackendTrait) -> anyhow::Result<T>,
) -> anyhow::Result<T> {
match access {
traits::Access::Read => {
let _t = self
.hold
.admit_read(&self.name)
.map_err(anyhow::Error::msg)?;
call(self.inner()?.as_ref())
}
traits::Access::Write => {
let _t = self.hold.admit_write().map_err(anyhow::Error::msg)?;
call(self.inner()?.as_ref())
}
}
}
/// Pure path logic, asked before any read: not admitted through the hold.
fn coordinate_for_path(&self, suburl: &str) -> anyhow::Result<Option<traits::ArtifactId>> {
self.inner()?.coordinate_for_path(suburl)
}
/// A hold changes no answer, so a read in pieces is the inner backend's, admitted
/// like every read (lane HOL-RSS).
fn stream_get(
&self,
suburl: &str,
piece_bytes: usize,
) -> anyhow::Result<Option<traits::StreamedGet>> {
self.around(traits::Access::Read, |b| b.stream_get(suburl, piece_bytes))
}
}
/// A held backend, as the release registry takes it.
pub struct HeldRelease(pub Arc<Held>);
impl Releasable for HeldRelease {
fn what(&self) -> String {
self.0.what()
}
fn release(&self) -> Result<(), String> {
self.0.release()
}
fn reacquire(&self) -> Result<(), String> {
self.0.reacquire()
}
}
// ── the primary's twins ─────────────────────────────────────────────────────
/// The roster file's line that carries a twin's `followed_to_ms` (the roster's `TwinHeard`).
const FOLLOWED_TO: &str = "followed_to_ms";
/// **What every enrolled twin has followed to** (stinky SB #35), read off the
/// roster under `data_root` — the retention executor's guard reads it in its own
/// process. `None` for a twin that has not completed a follow this primary has
/// heard of. An unreadable roster is an `Err`, never an empty one: an empty one
/// would let a reclaim drop what a twin has not followed.
pub fn followed_to(data_root: &Path) -> Result<Vec<(String, Option<i64>)>, String> {
let dir = data_root.join(TWIN_ROSTER_DIR);
let entries = match std::fs::read_dir(&dir) {
Ok(e) => e,
Err(e) if e.kind() == std::io::ErrorKind::NotFound => return Ok(Vec::new()),
Err(e) => {
return Err(format!(
"the twin roster {} could not be read: {e}",
dir.display()
))
}
};
let mut out = Vec::new();
for e in entries.flatten() {
let Some(name) = e.file_name().to_str().map(str::to_owned) else {
continue;
};
if name.starts_with('.') {
continue;
}
let text = std::fs::read_to_string(e.path())
.map_err(|err| format!("the twin roster entry {}: {err}", e.path().display()))?;
let at = text
.lines()
.find_map(|l| l.strip_prefix(FOLLOWED_TO))
.and_then(|v| v.trim().parse::<i64>().ok());
out.push((name, at));
}
out.sort();
Ok(out)
}
// ── the growth plane: edda's machine over holger's world (`estate`) ─────────
#[cfg(feature = "estate")]
mod plane;
#[cfg(feature = "estate")]
pub use plane::*;
#[cfg(not(feature = "estate"))]
mod plane_off;
#[cfg(not(feature = "estate"))]
pub use plane_off::*;
// ── the twin's half: report where it has followed to ────────────────────────
/// **Why a twin backfills before a cycle** ([`FollowPosition::gap`]).
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum FollowGap {
/// This process has not completed a cycle yet: its change maps were written
/// by an earlier one, and nothing proves the store still holds what they say.
NeverFollowedHere,
/// The primary restarted: what it admitted between this twin's last whole
/// cycle and its death is counted under a boot that is gone.
PrimaryRestarted { applied_boot: u64, boot: u64 },
/// The primary's head is BEHIND what this twin applied in the same boot —
/// a sequence that went backwards counts nothing it says.
Rewound { applied_seq: u64, head: u64 },
}
impl std::fmt::Display for FollowGap {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
match self {
FollowGap::NeverFollowedHere => write!(f, "no cycle completed in this process yet"),
FollowGap::PrimaryRestarted { applied_boot, boot } => {
write!(f, "the primary restarted (boot {applied_boot} → {boot})")
}
FollowGap::Rewound { applied_seq, head } => write!(
f,
"the primary's head {head} is behind the applied {applied_seq}"
),
}
}
}
/// **What a twin remembers between follow cycles**: the primary's position at
/// the start of its last COMPLETE cycle — what it has applied.
#[derive(Debug, Default)]
pub struct FollowPosition {
applied: Mutex<(u64, u64)>,
/// The twin's own point against the primary's `(boot, head)` its last report's
/// answer named — judged by the one follow-point rule for the twin's words.
primary: Mutex<Option<(crate::follow_point::FollowPoint, u64, u64)>>,
}
impl FollowPosition {
pub fn applied(&self) -> (u64, u64) {
*self.applied.lock_or_recover()
}
/// **A gap between what this twin applied and where its primary stands
/// now** (stinky SB #35), judged at the START of a cycle against the head
/// the primary just answered (`at`). Holger has no transaction log to
/// replay, so a gap is never filled by sequence — it is filled by a
/// BACKFILL: the change maps are held against what the store really holds
/// ([`crate::mirror_cycle::backfill_pair`]) and the cycle fetches the rest.
pub fn gap(&self, at: (u64, u64)) -> Option<FollowGap> {
let (applied_boot, applied_seq) = self.applied();
let (boot, head) = at;
if (applied_boot, applied_seq) == (0, 0) {
Some(FollowGap::NeverFollowedHere)
} else if applied_boot != boot {
Some(FollowGap::PrimaryRestarted { applied_boot, boot })
} else if head < applied_seq {
Some(FollowGap::Rewound { applied_seq, head })
} else {
None
}
}
/// A cycle that started at `(boot, head)` completed.
pub fn completed(&self, at: (u64, u64)) {
*self.applied.lock_or_recover() = at;
}
/// **The primary answered a report**: it stands at `(boot, head)` now (`now_ms`).
pub fn heard_primary(&self, boot: u64, head: u64, now_ms: i64) {
let (applied_boot, applied_seq) = self.applied();
let mut p = self.primary.lock_or_recover();
let point = crate::follow_point::FollowPoint::heard(
p.as_ref().map(|(fp, _, _)| fp),
applied_boot,
applied_seq,
boot,
head,
now_ms,
);
*p = Some((point, boot, head));
}
/// **Where this twin stands**, as the one rule judges its last point against the
/// primary's last answered head: `None` before the primary ever answered.
pub fn look(&self, now_ms: i64) -> Option<crate::follow_point::Judged> {
let p = self.primary.lock_or_recover();
p.as_ref().map(|(fp, boot, head)| {
fp.judge(
*boot,
*head,
now_ms,
TWIN_SILENT_AFTER_MS,
TWIN_STALLED_AFTER_MS,
)
})
}
}
/// The twin's own disk, for its report: `(total, used)` over its data root.
pub fn own_fill(root: &Path) -> (u64, u64) {
crate::metering::statvfs(root)
.map(|f| {
(
f.total_bytes,
f.total_bytes.saturating_sub(f.servable_bytes),
)
})
.unwrap_or((0, 0))
}
/// The twin's principal on its primary's roster: `HOLGER_TWIN_PRINCIPAL`, else the
/// principal vm1 certified it as ([`crate::twin_tls::TWIN_PRINCIPAL`]) — never its
/// host name, which on an appliance is the DHCP-derived `holger-<n>` nobody else
/// knows it by (MEASURED on the dwarf mock 2026-09-29: the roster named
/// `holger-123456` and the functest, reading `holger-twin`, found no twin).
pub fn twin_principal() -> String {
twin_principal_from(std::env::var("HOLGER_TWIN_PRINCIPAL").ok().as_deref())
}
/// Pure: [`twin_principal`] over the operator's override.
pub fn twin_principal_from(declared: Option<&str>) -> String {
declared
.map(str::trim)
.filter(|s| !s.is_empty())
.map(str::to_owned)
.unwrap_or_else(|| crate::twin_tls::TWIN_PRINCIPAL.to_owned())
}
/// **One report to the primary** over `holger.grow.v1.TwinReport`, on a channel the
/// follow already dials (`twin_follow::Follower::report`): the twin's last complete
/// position and fill; the answer is where the primary is NOW, which the next
/// cycle starts from.
pub async fn report_over(
channel: tonic::transport::Channel,
bearer: Option<&str>,
req: pb::TwinReportRequest,
) -> Result<(u64, u64), String> {
use pb::grow_service_client::GrowServiceClient;
let mut r = tonic::Request::new(req);
if let Some(b) = bearer {
let v = format!("Bearer {b}")
.parse()
.map_err(|e| format!("the follow bearer is not a header value: {e}"))?;
r.metadata_mut().insert("authorization", v);
}
let reply = GrowServiceClient::new(channel)
.twin_report(r)
.await
.map_err(|s| format!("TwinReport: {s}"))?
.into_inner();
Ok((reply.boot, reply.head_seq))
}
// ── the gRPC door ───────────────────────────────────────────────────────────
fn refused(why: String) -> tonic::Status {
crate::grpc::coded(
&holger_errcode::GROW_REFUSED,
tonic::Code::FailedPrecondition,
why,
)
}
#[cfg(feature = "estate")]
fn no_plane() -> tonic::Status {
refused("this holger was started without the growth plane (no data root), so it has nothing to grow".into())
}
/// A build without `estate` has no growth plane at all (H2's estate cut).
#[cfg(not(feature = "estate"))]
fn no_plane() -> tonic::Status {
refused("this holger was built without the estate (feature `estate`), so it has no growth plane and nothing to grow".into())
}
#[tonic::async_trait]
impl pb::grow_service_server::GrowService for Arc<crate::grpc::HolgerGrpc> {
async fn start(
&self,
request: tonic::Request<pb::GrowStartRequest>,
) -> Result<tonic::Response<pb::GrowStatusReply>, tonic::Status> {
self.authorize_grow(&request, "start a data-set growth")
.await?;
let plane = self.grow.clone().ok_or_else(no_plane)?;
let req = request.into_inner();
let out = tokio::task::spawn_blocking(move || plane.start(&req))
.await
.map_err(|e| crate::grpc::internal_err(format!("grow start task: {e}")))?;
out.map(tonic::Response::new).map_err(refused)
}
/// NO credential, by design (gunnar's T6): the actor outside must read the
/// phase while the stores are closed.
async fn status(
&self,
_request: tonic::Request<pb::GrowStatusRequest>,
) -> Result<tonic::Response<pb::GrowStatusReply>, tonic::Status> {
let plane = self.grow.clone().ok_or_else(no_plane)?;
let s = tokio::task::spawn_blocking(move || plane.status())
.await
.map_err(|e| crate::grpc::internal_err(format!("grow status task: {e}")))?;
Ok(tonic::Response::new(s))
}
async fn resume(
&self,
request: tonic::Request<pb::GrowResumeRequest>,
) -> Result<tonic::Response<pb::GrowStatusReply>, tonic::Status> {
self.authorize_grow(&request, "resume a data-set growth")
.await?;
let plane = self.grow.clone().ok_or_else(no_plane)?;
let give_up = request.get_ref().give_up;
let out = tokio::task::spawn_blocking(move || plane.resume(give_up))
.await
.map_err(|e| crate::grpc::internal_err(format!("grow resume task: {e}")))?;
out.map(tonic::Response::new).map_err(refused)
}
async fn flush(
&self,
request: tonic::Request<pb::GrowFlushRequest>,
) -> Result<tonic::Response<pb::GrowFlushReply>, tonic::Status> {
self.authorize_grow(&request, "flush to the twins").await?;
let plane = self.grow.clone().ok_or_else(no_plane)?;
let timeout = Duration::from_secs(u64::from(request.get_ref().timeout_secs.clamp(1, 3600)));
let r = tokio::task::spawn_blocking(move || plane.flush(timeout))
.await
.map_err(|e| crate::grpc::internal_err(format!("grow flush task: {e}")))?;
Ok(tonic::Response::new(r))
}
async fn twin_report(
&self,
request: tonic::Request<pb::TwinReportRequest>,
) -> Result<tonic::Response<pb::TwinReportReply>, tonic::Status> {
self.authorize_admin_read(&request, "").await?;
let plane = self.grow.clone().ok_or_else(no_plane)?;
plane
.twin_report(request.get_ref())
.map(tonic::Response::new)
.map_err(refused)
}
async fn twins(
&self,
request: tonic::Request<pb::TwinsRequest>,
) -> Result<tonic::Response<pb::TwinsReply>, tonic::Status> {
self.authorize_admin_read(&request, "").await?;
let plane = self.grow.clone().ok_or_else(no_plane)?;
plane.twins().map(tonic::Response::new).map_err(refused)
}
/// A paired DOWNSTREAM's sync point. NO role check: the paired client
/// certificate is the credential (`crate::peers::PeerBoard::report` names
/// the caller by its pin, or refuses it by name). Needs no growth plane —
/// the position is this box's write sequence, which every build counts.
///
/// ★ A SYNC downstream (the RON's or the console's, `via: Front` or `via: Wireguard`;
/// lane OX-WGDS2) is no paired one: its report carries the proof its sync key signed
/// (`x-holger-sync-report`), and the sync hub counts it by the downstream's enrolled
/// name ([`crate::sync::SyncHub::heard`]), the name the front put on a WireGuard caller
/// (`x-holger-downstream`) checked against it.
async fn downstream_report(
&self,
request: tonic::Request<pb::DownstreamReportRequest>,
) -> Result<tonic::Response<pb::TwinReportReply>, tonic::Status> {
let leaf = request
.peer_certs()
.and_then(|c| c.first().map(|l| l.as_ref().to_vec()));
let header = |name: &str| {
request
.metadata()
.get(name)
.and_then(|v| v.to_str().ok())
.map(str::to_string)
};
let (proof, named) = (
header(crate::mirror_cycle::REPORT_HEADER),
header("x-holger-downstream"),
);
let (peers, hold) = (self.peers.clone(), self.grow_hold.clone());
let req = request.into_inner();
#[cfg(feature = "sync")]
let hub = self.intake.as_ref().and_then(|c| c.sync());
let out = tokio::task::spawn_blocking(move || {
let paired = peers.report(leaf.as_deref(), &req, hold.boot(), hold.head(), now_ms());
#[cfg(feature = "sync")]
if let (Err(crate::peers::ReportRefused::NotPaired(_)), Some(proof), Some(hub)) =
(&paired, proof.as_deref(), hub)
{
let (boot, head) = (hold.boot(), hold.head());
return hub
.heard(proof, &req, named.as_deref(), boot, head, now_ms())
.map(|_| pb::TwinReportReply {
boot,
head_seq: head,
})
.map_err(|e| {
let code = match e.status {
400 => tonic::Code::InvalidArgument,
401 | 403 => tonic::Code::PermissionDenied,
404 => tonic::Code::NotFound,
409 => tonic::Code::FailedPrecondition,
_ => tonic::Code::Internal,
};
crate::grpc::coded(e.code, code, &e.message)
});
}
let _ = (&proof, &named);
paired.map_err(crate::peers::ReportRefused::status)
})
.await
.map_err(|e| crate::grpc::internal_err(format!("downstream report task: {e}")))?;
out.map(tonic::Response::new)
}
async fn downstreams(
&self,
request: tonic::Request<pb::DownstreamsRequest>,
) -> Result<tonic::Response<pb::DownstreamsReply>, tonic::Status> {
self.authorize_admin_read(&request, "").await?;
let (boot, head) = (self.grow_hold.boot(), self.grow_hold.head());
Ok(tonic::Response::new(pb::DownstreamsReply {
downstreams: self.peers.rows(boot, head, now_ms()),
boot,
head_seq: head,
}))
}
}
// The growth plane's tests drive edda's machine (`estate`); the hold's own are there too.
#[cfg(all(test, feature = "estate"))]
mod tests;