server/lib/src/twin_follow.rs
//! **The twin follows its primary over the network** (owner plan 2b,
//! 2026-09-27): a dumb replica with no UI that holds what the primary holds and
//! can recharge it.
//!
//! Two halves, both on engines that already exist, neither a second copy:
//!
//! * **the artifacts** — every repository this box declares (its config is the
//! primary's byte for byte, `[estate-twin-drift]`) mirrored from the SAME
//! repository on the primary, by [`crate::mirror_cycle::run_pair`] (its change
//! maps) once per follow cycle: a mirror source built here, not written in
//! anybody's config;
//! * **the estate** — the config and the side stores, pulled through
//! `holger.estate.v1` ([`crate::estate_wire::pull_over`]) on the twin's own
//! cadence (`poll_seconds`).
//!
//! **One cycle, then a word to the primary** ([`Follower::cycle`]): the twin says
//! how far into the primary's sequence it has followed (`holger.grow.v1.TwinReport`),
//! which is what the primary's growth drain holds it to (`crate::grow`), and its
//! cycle parks while the twin grows itself.
//!
//! **The wire is the primary's replication door, mutual TLS 1.3** (stinky 14
//! §P3, [`crate::twin_tls`]): once the front has certified this twin, both
//! halves dial the first recorded door that answers, with the pinned issuer and
//! the twin's own certificate — read per cycle, so a certification (or a
//! renewal) installed while the follow runs is taken on the next tick. Before
//! that, a LOOPBACK primary is followed in the clear (+ `HOLGER_PRIMARY_BEARER`)
//! — the in-process tests' road, and the only one; a remote primary is refused
//! by name every cycle (`holger-twin-13`) and never followed in the clear.
//!
//! **The follow starts live.** `BecomeTwin` records the election, switches the
//! posture and calls [`Follower::start`] — no restart (stinky 14 §2
//! correction 3). A box that boots already a twin starts it through [`spawn`].
//!
//! The twin's doors refuse every write by name ([`crate::twin`]); these timers
//! write through the backends and the data root, never through a door. Nothing
//! here writes to the primary.
//!
//! **Recharge** is the same pull the other way round: a new main pulls the
//! twin's estate and every repository — over the TWIN's replication door with
//! the `recharge` credential the front certified onto it
//! ([`Follower::recharge`], the `Recharge` RPC), or, at boot, from
//! `recharge_from` ([`recharge`]).
use crate::lock::LockOrRecover as _;
use std::path::{Path, PathBuf};
use std::sync::{Arc, Mutex};
use std::time::Duration;
use traits::RepositoryBackendTrait;
use crate::mirror_cycle::{MirrorSource, SourcePlan};
use crate::twin_tls::{self, Purpose};
/// The mirror source's name for the twin's own follow.
pub const SOURCE_NAME: &str = "twin-of-primary";
/// The env key of the admin-read bearer the primary's estate door accepts —
/// the LOOPBACK road only (an uncertified twin of an in-process primary); a
/// certified twin presents its certificate and no bearer.
pub const BEARER_ENV: &str = "HOLGER_PRIMARY_BEARER";
/// Pure: the primary's endpoint as a URL (`host:port` → `http://host:port`).
pub fn endpoint_of(primary: &str) -> String {
if primary.starts_with("http://") || primary.starts_with("https://") {
primary.to_string()
} else {
format!("http://{primary}")
}
}
/// **The mirror source that follows every repository**, anonymous — pure over
/// the repository names and the cadence. RON so the config's own defaults
/// (page size, in-flight, retry, timeouts) apply, not a second opinion.
pub fn source(
primary: &str,
repositories: &[String],
every_secs: u64,
) -> Result<MirrorSource, String> {
source_with(primary, repositories, every_secs, None)
}
/// [`source`] over this box's replication credential in `cred_dir`
/// (`MirrorCredential::Twin`): mutual TLS once installed, the clear to a
/// loopback primary before.
pub fn source_with(
primary: &str,
repositories: &[String],
every_secs: u64,
cred_dir: Option<&Path>,
) -> Result<MirrorSource, String> {
let pairs: Vec<String> = repositories
.iter()
.map(|r| format!("(ron_from: {r:?}, ron_to: {r:?})"))
.collect();
let cred = match cred_dir {
Some(d) => format!(
"ron_credential: Twin(ron_dir: {:?}), ",
d.display().to_string()
),
None => String::new(),
};
let text = format!(
"(ron_name: {SOURCE_NAME:?}, ron_kind: Holger, ron_endpoint: {:?}, {cred}ron_every_secs: {every_secs}, ron_repositories: [{}])",
endpoint_of(primary),
pairs.join(", ")
);
ron::from_str(&text).map_err(|e| format!("the twin's mirror source does not parse: {e}"))
}
/// One estate pull over whatever this box may dial: the replication door when
/// the `purpose` credential is installed, the clear to a loopback source
/// before, a named refusal otherwise. `Ok((pulled, how))` when the source was
/// REACHED (the pull may still have failed items); `Err` when it was not.
pub async fn estate_pull(
root: &Path,
endpoint: &str,
purpose: Purpose,
) -> Result<(crate::estate_wire::Pulled, String), String> {
match twin_tls::load(root, purpose)? {
Some(c) => {
let (ch, at) = twin_tls::channel(&c).await?;
let how = format!("mtls {at}");
Ok((
crate::estate_wire::pull_over(ch, &format!("replication door {at}"), None, root)
.await?,
how,
))
}
None if twin_tls::is_loopback(endpoint) => {
let bearer = std::env::var(BEARER_ENV)
.ok()
.filter(|b| !b.trim().is_empty());
Ok((
crate::estate_wire::pull_remote(endpoint, bearer.as_deref(), root).await?,
format!("plain {endpoint}"),
))
}
None => Err(twin_tls::not_certified(endpoint)),
}
}
/// [`estate_pull`] said as one line: `Ok((line, how))` only when whole.
pub async fn estate_once(
root: &Path,
endpoint: &str,
purpose: Purpose,
) -> Result<(String, String), String> {
let (r, how) = estate_pull(root, endpoint, purpose).await?;
if r.failed.is_empty() {
Ok((
format!(
"twin: estate over {how}: {} fetched ({} bytes), {} already held",
r.fetched, r.fetched_bytes, r.already
),
how,
))
} else {
Err(format!(
"twin: estate over {how}: {} fetched, {} FAILED: {}",
r.fetched,
r.failed.len(),
r.failed.join("; ")
))
}
}
/// **What a follow (or a recharge) needs of this box**, captured once at start
/// so `BecomeTwin` can start the follow LIVE without a `Holger` in hand: the
/// repositories and their backends, the data root, and the posture it reports
/// to. The started tasks are kept here, so a live follow lives as long as the
/// server.
#[derive(Clone)]
pub struct Follower {
/// Every repository this box follows, by name: the boot's, and every repository a sale
/// made on the primary that the estate brought since ([`Follower::adopt_sold`]).
#[allow(clippy::type_complexity)]
repos: Arc<Mutex<Vec<(String, Arc<dyn RepositoryBackendTrait>)>>>,
/// What makes (and serves) a repository at run time — a sold one the estate names.
live: Option<Arc<crate::live_routes::LiveRoutes>>,
root: PathBuf,
posture: Arc<crate::twin::TwinPosture>,
tasks: Arc<Mutex<Vec<tokio::task::JoinHandle<()>>>>,
/// The configured audit directory's name under the data root — the box's
/// own history of what it refused while it waited, which a recharge may
/// land beside ([`Follower::recharge_blocker`]).
audit_dir: Option<String>,
/// Where the running config keeps repository bytes ([`crate::Holger::history_stores`]):
/// history for [`Follower::recharge_blocker`] wherever the config on disk is.
stores: Vec<PathBuf>,
/// This box's control plane as a client dials it — the first exposed
/// endpoint's CONFIGURED gRPC address, an unspecified host named by the
/// replication door's first advertised address (or its bound one).
control: String,
/// **This box's OWN growth hold** (`crate::grow::Hold`): while the twin grows,
/// the follow parks — the follower is its only writer under /data.
hold: Arc<crate::grow::Hold>,
/// The primary's position at the start of the last COMPLETE cycle —
/// what this twin has applied, as its reports say.
position: Arc<crate::grow::FollowPosition>,
/// Who this twin is on its primary's roster.
principal: String,
/// What this twin can say it holds, sent with every report ([`Evidence`]).
evidence: Arc<Mutex<Evidence>>,
/// Whether this process has said its first WHOLE follow over mutual TLS
/// (stinky SB #33/#46): said once, at WARN, so the KVM gate reads it in the
/// twin's own log — the proof that the box-to-box follow ran over mTLS 1.3
/// on real boxes, not only in-process.
said_mtls: Arc<std::sync::atomic::AtomicBool>,
/// **The runtime the follow runs on**, captured where the Follower is built (the
/// server's start, inside Tokio). `BecomeTwin` reaches [`Follower::start`] from
/// the pairing loop's own thread too, where `tokio::spawn` has no reactor —
/// MEASURED on the mock 2026-09-28: `thread 'box-pairing' panicked at
/// twin_follow.rs: there is no reactor running`, and the certified twin never
/// followed. `None` only when built outside a runtime (tests).
rt: Option<tokio::runtime::Handle>,
}
impl Follower {
pub fn of(holger: &crate::Holger, posture: Arc<crate::twin::TwinPosture>) -> Follower {
let repos = holger
.repositories
.iter()
.filter_map(|r| {
r.backend_repository
.clone()
.map(|b| (r.ron_name.clone(), b))
})
.collect();
let audit_dir = holger
.resolved_audit_dir()
.file_name()
.map(|n| n.to_string_lossy().to_string());
Follower {
repos: Arc::new(Mutex::new(repos)),
live: holger.live_routes_cell.get().cloned(),
root: holger.peers_root(),
posture,
tasks: Arc::default(),
audit_dir,
stores: holger.history_stores(),
control: control_of(holger),
hold: holger.grow_hold(),
position: Arc::default(),
principal: crate::grow::twin_principal(),
evidence: Arc::default(),
said_mtls: Arc::default(),
rt: tokio::runtime::Handle::try_current().ok(),
}
}
fn names(&self) -> Vec<String> {
self.repos
.lock_or_recover()
.iter()
.map(|(n, _)| n.clone())
.collect()
}
fn repos(&self) -> Vec<(String, Arc<dyn RepositoryBackendTrait>)> {
self.repos.lock_or_recover().clone()
}
/// **The repositories a sale made on the primary, taken on here** (lane P4, A8 lane 10).
/// The estate brings `fleet/sold-repositories.json`; each repository it names that this
/// box does not follow yet gets its backend (served at once where it can be,
/// [`crate::live_routes::LiveRoutes::backend_for`]) and is followed from now on — so a
/// twin holds a buyer's artifacts whether the sale came before its start or during its
/// follow, and a recharged main gets them back. Answers the names taken on; what could
/// not be taken on is said, and asked again on the next cycle.
pub fn adopt_sold(&self) -> Vec<String> {
let path = self.root.join(crate::sold::SOLD_REPOS_REL);
let Ok(bytes) = std::fs::read(&path) else {
return Vec::new();
};
let sold: std::collections::BTreeMap<String, crate::sold::SoldRepository> =
match serde_json::from_slice(&bytes) {
Ok(s) => s,
Err(e) => {
log::error!(
"{}: the signed-up repositories cannot be read ({e}); none is followed",
path.display()
);
return Vec::new();
}
};
let mut taken = Vec::new();
for (name, s) in sold {
if self.repos.lock_or_recover().iter().any(|(n, _)| *n == name) {
continue;
}
let Some(live) = &self.live else {
log::warn!("twin: '{name}' (a signed-up repository) is not followed: this box makes no repository at run time, so it is followed from its next start");
continue;
};
// A repository made at run time is its owner's here too (lane P1).
live.own_entry(&name, crate::sold::gate_entry(&self.root, &s));
match live.backend_for(&name, &s.template, s.private) {
Ok(b) => {
self.repos.lock_or_recover().push((name.clone(), b));
taken.push(name);
}
Err(why) => log::error!("twin: '{name}' (a signed-up repository, shaped like '{}') is not followed: {why}", s.template),
}
}
taken
}
/// [`Follower::adopt_sold`], and each repository taken on added to `plan`.
fn adopt_sold_into(&self, plan: &mut SourcePlan) {
for name in self.adopt_sold() {
let Some(backend) = self
.repos
.lock_or_recover()
.iter()
.find(|(n, _)| *n == name)
.map(|(_, b)| b.clone())
else {
continue;
};
let pair = crate::mirror_cycle::MirrorPair {
ron_from: name.clone(),
ron_to: name.clone(),
ron_unproven: Default::default(),
};
plan.source.ron_repositories.push(pair.clone());
plan.pairs.push((pair, backend));
log::warn!("twin: '{name}' (signed up on the primary) is followed from now on");
}
}
fn plan(&self, src: MirrorSource, state: &str) -> Result<SourcePlan, String> {
let mut pairs = Vec::new();
for pair in &src.ron_repositories {
let backend = self
.repos
.lock_or_recover()
.iter()
.find(|(n, _)| *n == pair.ron_to)
.map(|(_, b)| b.clone())
.ok_or_else(|| {
format!(
"repository `{}` has no backend in this process",
pair.ron_to
)
})?;
pairs.push((pair.clone(), backend));
}
let state_dir = self.root.join(state);
std::fs::create_dir_all(&state_dir).map_err(|e| format!("{}: {e}", state_dir.display()))?;
Ok(SourcePlan {
source: src,
pairs,
state_dir,
pinned: None,
report_as: None,
summary: None,
summary_as: None,
console: None,
})
}
/// **Start the follow of `role`** — ONE loop running [`Follower::cycle`]
/// every `poll_seconds`. Nothing on a primary. The handles are returned (and
/// nothing is kept) — see [`Follower::start`] for the live form.
///
/// It was two timers — the mirror's supervised timer for the artifacts and a
/// loop for the estate — and neither knew when the other had finished, so
/// nothing on the twin could say "I hold everything the primary had when I
/// started". The growth needs exactly that sentence (`crate::grow`: the
/// primary's drain holds every enrolled twin to its BARRIER, gunnar 364a4f5),
/// so the halves are one cycle now, on the same engines
/// (`mirror_cycle::run_pair` with its change maps, `estate_pull`).
pub fn spawn(
&self,
role: &crate::twin::NodeRole,
) -> Result<Vec<tokio::task::JoinHandle<()>>, String> {
let crate::twin::NodeRole::Twin {
primary,
poll_seconds,
} = role
else {
return Ok(Vec::new());
};
// An estate twin follows its primary over wg0 only (lane E1): a primary named by
// its utility address (a BecomeTwin from the ONE box's order) is followed at its
// overlay address.
let primary = &crate::tunnel_addr::estate_route("the primary", primary)?;
let every = (*poll_seconds).max(1);
let src = source_with(
primary,
&self.names(),
every,
Some(&Purpose::Twin.dir(&self.root)),
)?;
let endpoint = endpoint_of(primary);
let mut plan = self.plan(src, "holger-twin-mirror")?;
log::warn!("twin: following {endpoint} — {} repositories mirrored and the estate pulled every {every}s over its replication door once certified; this box serves reads only", plan.pairs.len());
let me = self.clone();
let spawn_on = match (&self.rt, tokio::runtime::Handle::try_current()) {
(_, Ok(here)) => here,
(Some(h), Err(_)) => h.clone(),
(None, Err(_)) => return Err("the follow has no runtime to run on: the Follower was built outside Tokio and is started from a thread without one".into()),
};
// A32-14, stays a task: the twin's follow loop; its handles go back to the server's main, which keeps them for its life (`_twin_tasks`).
Ok(vec![spawn_on.spawn(async move {
let mut said = String::new();
loop {
let line = me.cycle(&mut plan, &endpoint).await;
// Said when it changes, not every cycle.
let gist: String = line.split(':').take(2).collect();
if gist != said || line.contains("FAILED") {
log::info!("{line}");
said = gist;
}
me.posture
.rest_between_follows(Duration::from_secs(every))
.await;
}
})])
}
/// **One follow cycle**: say where this twin stands (and learn where the
/// primary is), take every repository and the estate, and — only when ALL of
/// it landed — record the primary's position from the START of the cycle as
/// applied, and say so at once, so the primary's drain need not wait a whole
/// cycle for the word. The estate first (the config and the side stores the
/// artifacts are served under), then every repository.
///
/// While this twin is ITSELF growing (its [`crate::grow::Hold`] paused or
/// closed) the cycle PARKS: the follower is the twin's only writer under
/// /data, and its growth drains it the way main drains its uploads (gunnar's
/// T21 ruling: the twin grows exactly like main, never restarted).
pub async fn cycle(&self, plan: &mut SourcePlan, endpoint: &str) -> String {
if self.hold.paused() || self.hold.closed() {
return "twin: follow held: this twin is growing its data set; the follower resumes when /data is back".into();
}
let now = i64::try_from(crate::clock::now_unix_secs()).unwrap_or(0);
let mut report_failed = None;
let at = match self.report(endpoint).await {
Ok(at) => Some(at),
Err(e) => {
log::warn!("twin: position not reported to {endpoint}: {e}");
report_failed = Some(format!("{e:#}"));
None
}
};
let mut failed: Vec<String> = Vec::new();
let mut moved = 0usize;
// Whole, partial (reached and something failed — a LOSS if it stays),
// or unreachable (a dead primary — not a loss).
let now_ms = crate::clock::now_unix_ms();
let estate = match estate_pull(&self.root, endpoint, Purpose::Twin).await {
// What the primary listed, whole or not: it ANSWERS (the fence gate's hearing).
Ok((r, how)) if r.failed.is_empty() => {
if let Some(line) = first_whole_over_mtls(&how, &self.said_mtls) {
log::warn!("{line}");
}
self.posture
.record_reached(now_ms, (r.fetched + r.already + r.failed.len()) as u64);
self.posture.record_follow(how.clone());
self.posture.record_follow_whole(now);
self.evidence.lock_or_recover().estate(&r);
format!(
"estate over {how}: {} fetched ({} bytes), {} already held",
r.fetched, r.fetched_bytes, r.already
)
}
Ok((r, how)) => {
self.posture
.record_reached(now_ms, (r.fetched + r.already + r.failed.len()) as u64);
let why = format!(
"estate over {how}: {} fetched, {} FAILED: {}",
r.fetched,
r.failed.len(),
r.failed.join("; ")
);
self.posture.record_follow(how);
self.posture.record_follow_partial(now, why.clone());
failed.push(why.clone());
why
}
Err(e) => {
self.posture.record_unreached(now_ms);
failed.push(e.clone());
e
}
};
// A repository the estate says a sale made is followed from this cycle on.
self.adopt_sold_into(plan);
let plan = &*plan;
// ★ A GAP (stinky SB #35): a first cycle in this process, a primary that
// restarted, a head that went backwards — the change maps are held
// against what the store really holds before the mirror trusts their
// "unchanged", so what this twin lost is fetched again this cycle.
let backfill = match at.and_then(|at| self.position.gap(at)) {
Some(gap) => self.backfill(plan, gap).await,
None => String::new(),
};
// The main's PURGES, applied here: the mirror below never deletes, so a
// repository the main emptied would otherwise live on in this twin.
let purges = self.apply_purges(plan).await;
failed.extend(purges.failures.iter().cloned());
match crate::mirror_cycle::connect_upstream(&plan.source).await {
Err(e) => failed.push(format!("{endpoint}: {e:#}")),
Ok(upstream) => {
for (pair, local) in &plan.pairs {
match crate::mirror_cycle::run_pair(
upstream.as_ref(),
&plan.source.ron_name,
pair,
local,
&plan.state_dir,
plan.source.ron_page_size,
plan.source.ron_in_flight,
crate::mirror_cycle::now_nanos(),
)
.await
{
Ok(r) => {
moved += usize::from(!r.moved_nothing());
self.evidence
.lock_or_recover()
.wrote(&pair.ron_to, &r.written);
}
Err(e) => failed.push(format!("{}: {e:#}", pair.ron_to)),
}
}
}
}
if !failed.is_empty() {
// ★ Refused for certificate DATES: this twin's clock or the primary's is wrong, and
// the front is the judge (Gunnar f2e602dd: a re-laid twin's clock stayed two hours
// off while every poll was refused, because only refusals THIS box made asked).
if crate::clock::saw_refusals(failed.iter().chain(&report_failed).map(String::as_str)) {
log::warn!("twin: the follow of {endpoint} was refused for certificate dates — the front's clock is asked");
}
return format!(
"twin: follow of {endpoint} FAILED ({moved} repositories moved): {}",
failed.join("; ")
);
}
if let Some(at) = at {
self.position.completed(at);
if let Err(e) = self.report(endpoint).await {
log::warn!("twin: position not reported to {endpoint}: {e}");
}
}
let purged = if purges.applied.is_empty() {
String::new()
} else {
format!(
"; purges applied: {}",
purges
.applied
.iter()
.map(|(rec, repo, deleted, kept)| format!(
"{rec} ({repo}: {deleted} deleted, {kept} newer kept)"
))
.collect::<Vec<_>>()
.join(", ")
)
};
format!("twin: followed {endpoint}: {moved} repositories moved; {estate}{purged}{backfill}; applied {:?}", self.position.applied())
}
/// **Backfill every followed pair** after a [`crate::grow::FollowGap`], off the
/// reactor: each change-map entry whose bytes this store does not hold is
/// forgotten, so this cycle's mirror fetches it. Answers the summary's words.
async fn backfill(&self, plan: &SourcePlan, gap: crate::grow::FollowGap) -> String {
let (source, state_dir, pairs) = (
plan.source.ron_name.clone(),
plan.state_dir.clone(),
plan.pairs.clone(),
);
let done = tokio::task::spawn_blocking(move || {
pairs
.iter()
.map(|(pair, local)| {
(
pair.ron_to.clone(),
crate::mirror_cycle::backfill_pair(&source, pair, local, &state_dir),
)
})
.collect::<Vec<_>>()
})
.await;
let mut forgotten = 0usize;
let mut words = Vec::new();
match done {
Ok(rows) => {
for (repo, r) in rows {
match r {
Ok(0) => {}
Ok(n) => {
forgotten += n;
words.push(format!("{repo}: {n}"));
}
Err(e) => log::warn!("twin: backfill of `{repo}` after a gap ({gap}) could not read its change map: {e:#}"),
}
}
}
Err(e) => log::warn!("twin: the backfill task after a gap ({gap}) failed: {e}"),
}
if forgotten > 0 {
log::warn!("twin: a gap ({gap}): {forgotten} artifact(s) the change maps claimed are not held here and are fetched again ({})", words.join(", "));
format!(
"; backfilled after a gap ({gap}): {forgotten} re-fetched ({})",
words.join(", ")
)
} else {
String::new()
}
}
/// **Apply every purge record the estate brought that this box has not**
/// ([`crate::purge::apply_records`]: stored at or before the purge ⇒ deleted,
/// after ⇒ kept, each record once), off the reactor. A repository emptied here
/// has its mirror change map forgotten, so a coordinate the main holds again —
/// even byte-identical — is fetched again rather than believed held.
async fn apply_purges(&self, plan: &SourcePlan) -> crate::purge::Applied {
let (root, repos) = (self.root.clone(), self.repos());
let applied =
match tokio::task::spawn_blocking(move || crate::purge::apply_records(&root, &repos))
.await
{
Ok(a) => a,
Err(e) => crate::purge::Applied {
failures: vec![format!("purge records: the apply task failed: {e}")],
..Default::default()
},
};
for repo in applied.repositories() {
for (pair, _) in plan.pairs.iter().filter(|(p, _)| p.ron_to == repo) {
let map = crate::mirror::ChangeMap::path_in(
&plan.state_dir,
&plan.source.ron_name,
&pair.ron_from,
);
if let Err(e) = std::fs::remove_file(&map) {
if e.kind() != std::io::ErrorKind::NotFound {
log::warn!(
"twin: purge of `{repo}` applied, its change map {} not forgotten: {e}",
map.display()
);
}
}
}
}
for (rec, repo, deleted, kept) in &applied.applied {
log::warn!("twin: applied the main's purge {rec}: `{repo}` — {deleted} artifact(s) deleted here, {kept} stored after it kept");
}
applied
}
/// **This twin's word to its primary** (`holger.grow.v1.TwinReport`) over the
/// SAME road the follow takes: the replication door with the twin
/// credential once certified, the clear (+ `HOLGER_PRIMARY_BEARER`) to a
/// loopback primary before, refused by name otherwise.
async fn report(&self, endpoint: &str) -> Result<(u64, u64), String> {
let (total, used) = crate::grow::own_fill(&self.root);
let (boot, seq) = self.position.applied();
let ev = self.evidence.lock_or_recover().clone();
let req = crate::grpc::holger_grow_proto::TwinReportRequest {
principal: self.principal.clone(),
address: self.control.clone(),
applied_boot: boot,
applied_seq: seq,
disk_total_bytes: total,
disk_used_bytes: used,
estate_root: ev.estate_root,
estate_files: ev.estate_files,
held: ev.held.into_iter().collect(),
};
let answered = self.report_with(endpoint, req).await;
if let Ok((boot, head)) = answered {
self.position
.heard_primary(boot, head, crate::clock::now_unix_ms());
}
answered
}
async fn report_with(
&self,
endpoint: &str,
req: crate::grpc::holger_grow_proto::TwinReportRequest,
) -> Result<(u64, u64), String> {
match twin_tls::load(&self.root, Purpose::Twin)? {
Some(c) => {
let (ch, _at) = twin_tls::channel(&c).await?;
crate::grow::report_over(ch, None, req).await
}
None if twin_tls::is_loopback(endpoint) => {
let bearer = std::env::var(BEARER_ENV)
.ok()
.filter(|b| !b.trim().is_empty());
crate::egress::check("twin follow", endpoint).map_err(|e| e.to_string())?;
let ch = tonic::transport::Channel::from_shared(endpoint.to_string())
.map_err(|e| format!("`{endpoint}` is not an endpoint: {e}"))?
.connect()
.await
.map_err(|e| format!("{endpoint}: {e}"))?;
crate::grow::report_over(ch, bearer.as_deref(), req).await
}
None => Err(twin_tls::not_certified(endpoint)),
}
}
/// **Where this twin stands against its primary now** (its words and review; the one
/// follow-point rule): `None` before the primary ever answered a report.
pub fn look(&self) -> Option<crate::follow_point::Judged> {
self.position.look(crate::clock::now_unix_ms())
}
/// Where this twin has followed to (the growth's tests read it).
pub fn applied(&self) -> (u64, u64) {
self.position.applied()
}
/// What this twin would say it holds in its next report.
pub fn evidence(&self) -> Evidence {
self.evidence.lock_or_recover().clone()
}
/// **Start the follow LIVE and keep it** — what `BecomeTwin` calls after the
/// election is recorded. Once: a second call while it runs starts nothing.
pub fn start(&self, role: &crate::twin::NodeRole) -> Result<usize, String> {
let mut tasks = self.tasks.lock_or_recover();
if !tasks.is_empty() {
return Ok(0);
}
// A32-14, stays a task: the follow's own spawn (above); its handles are the caller's.
let h = self.spawn(role)?;
let n = h.len();
tasks.extend(h);
Ok(n)
}
/// **Pull the estate and every repository from `from`, once** — over the
/// `recharge` credential's door when one is installed, in the clear to a
/// loopback `from` otherwise. No marker is written here; [`recharge`] keeps
/// the boot road once, and the `Recharge` RPC answers what happened.
pub async fn recharge(&self, from: &str) -> Result<Recharged, String> {
// An estate main is filled from its twin over wg0 only (lane E1): a twin the
// ONE box names by its utility address is dialled at its overlay address.
let from = crate::tunnel_addr::estate_route("the twin", from)?;
let from = from.as_str();
let endpoint = endpoint_of(from);
let mut out = Recharged::default();
match estate_once(&self.root, &endpoint, Purpose::Recharge).await {
Ok((l, _)) => out.estate = l,
Err(e) => out.failed.push(e),
}
// The repositories the estate says sales made come back with their artifacts.
for name in self.adopt_sold() {
log::warn!(
"recharge: '{name}' (a signed-up repository) is taken back with its artifacts"
);
}
let src = source_with(
from,
&self.names(),
60,
Some(&Purpose::Recharge.dir(&self.root)),
)?;
let plan = self.plan(src, "holger-recharge-mirror")?;
// The records came with the estate: applied (by the same time rule — nothing
// stored here is older than them) and remembered, before the artifacts land.
let purges = self.apply_purges(&plan).await;
out.failed.extend(purges.failures);
match crate::mirror_cycle::connect_upstream(&plan.source).await {
Err(e) => out.failed.push(format!("{endpoint}: {e:#}")),
Ok(upstream) => {
for (pair, local) in &plan.pairs {
match crate::mirror_cycle::run_pair(
upstream.as_ref(),
&plan.source.ron_name,
pair,
local,
&plan.state_dir,
plan.source.ron_page_size,
plan.source.ron_in_flight,
crate::mirror_cycle::now_nanos(),
)
.await
{
Ok(r) => {
out.failed
.extend(not_restored(&pair.ron_to, &r.failed_coords));
out.artifacts.push(format!("{}: {r:?}", pair.ron_to));
}
Err(e) => out.failed.push(format!("{}: {e:#}", pair.ron_to)),
}
}
}
}
Ok(out)
}
/// The data root this follower writes.
pub fn root(&self) -> &Path {
&self.root
}
/// This box's control plane as a client dials it.
pub fn control(&self) -> &str {
&self.control
}
/// The posture it reports to.
pub fn posture(&self) -> &Arc<crate::twin::TwinPosture> {
&self.posture
}
/// **What stops a recharge from starting here** — the first path of state
/// this box holds of its own, not counting its own audit (what it refused
/// while it waited) and its recharge markers. `None` = fresh for a recharge.
pub fn recharge_blocker(&self) -> Option<PathBuf> {
let mut except: Vec<&str> = vec!["holger-audit", RECHARGED, RECHARGING];
if let Some(a) = &self.audit_dir {
except.push(a);
}
crate::twin::held_state_with(&self.root, &except, &self.stores)
}
/// Every artifact this box holds, over every repository.
pub fn artifact_count(&self) -> u64 {
self.repos()
.iter()
.map(|(_, b)| {
b.list(None, usize::MAX)
.map(|v| v.len() as u64)
.unwrap_or(0)
})
.sum()
}
}
/// **The line the KVM gate reads** (stinky SB #33/#46): the first whole estate
/// pull of this process over the replication door's mutual TLS 1.3 (`how` is
/// `mtls <door>`) says so ONCE — `twin: following over mutual TLS 1.3 (<door>)`;
/// a pull in the clear (the in-process loopback road) never says it.
pub const OVER_MTLS_SAID: &str = "twin: following over mutual TLS 1.3";
fn first_whole_over_mtls(how: &str, said: &std::sync::atomic::AtomicBool) -> Option<String> {
let door = how.strip_prefix("mtls ")?;
(!said.swap(true, std::sync::atomic::Ordering::SeqCst)).then(|| format!("{OVER_MTLS_SAID} ({door}): the estate pulled whole from the primary's replication door"))
}
/// **What a twin can say it holds** (2026-09-28): the root of the primary's
/// estate manifest from its last WHOLE pull (`estate_wire::manifest_root`) and
/// the newest artifacts it wrote with the sha256 of the bytes it wrote — sent
/// with every `TwinReport`, kept by the primary and read back through
/// `GrowService.Twins`. It is gunnar's pattern: a twin publishes no door outside
/// the utility network, so the estate reads it through what its PRIMARY heard
/// (gunnar's functest reads `Admin.Health` through vm1's console).
#[derive(Debug, Clone, Default)]
pub struct Evidence {
pub estate_root: String,
pub estate_files: u64,
pub held: std::collections::VecDeque<crate::grpc::holger_grow_proto::HeldArtifact>,
}
impl Evidence {
/// A whole estate pull: its root replaces the last one. A pull that was
/// not whole says nothing and changes nothing.
pub fn estate(&mut self, r: &crate::estate_wire::Pulled) {
if !r.root.is_empty() {
self.estate_root = r.root.clone();
self.estate_files = r.files as u64;
}
}
/// What a mirror cycle of `repository` wrote, into the ring (the newest
/// [`crate::mirror_cycle::WRITTEN_KEPT`] kept).
pub fn wrote(&mut self, repository: &str, written: &[(String, String, u64)]) {
for (coordinate, sha256, size) in written {
if self.held.len() == crate::mirror_cycle::WRITTEN_KEPT {
self.held.pop_front();
}
self.held
.push_back(crate::grpc::holger_grow_proto::HeldArtifact {
repository: repository.to_string(),
coordinate: coordinate.clone(),
sha256: sha256.clone(),
size: *size,
});
}
}
}
/// **The running server's follower, for an in-process caller that is not a
/// door** — the pairing loop of an installer-built box (`holger-server`'s
/// `paired`), which must make its box a twin LIVE and read how it follows. Set
/// by [`crate::Holger::start`]; the FIRST start in a process wins (an appliance
/// runs one holger; an in-process test that starts several reads its own
/// posture through the doors instead).
static LIVE: std::sync::OnceLock<Follower> = std::sync::OnceLock::new();
/// Publish this process's follower (first call wins).
pub fn set_live(f: Follower) {
let _ = LIVE.set(f);
}
/// This process's follower, once a server has started.
pub fn live() -> Option<&'static Follower> {
LIVE.get()
}
/// **Start the twin's follow** (a twin only; nothing on a primary): one cycle
/// loop, artifacts then estate ([`Follower::cycle`]). Returns the handle to keep.
pub fn spawn(holger: &crate::Holger) -> Result<Vec<tokio::task::JoinHandle<()>>, String> {
// The STARTED server's follower when there is one: its posture is the one
// the doors read (`TwinStatus.complete`, `/-/twin`). A follow reporting to a
// posture of its own would be a twin that follows and says it never did.
match holger.live_follower.get() {
// A32-14, stays a task: the follow's own spawn (above); its handles are the caller's.
Some(f) => f.spawn(&holger.role),
// A32-14, stays a task: the follow's own spawn (above); its handles are the caller's.
None => Follower::of(
holger,
Arc::new(crate::twin::TwinPosture::new(holger.role.clone())),
)
.spawn(&holger.role),
}
}
/// Pure: **a repository's artifacts a recharge did not put** — a failure of the
/// recharge, never a footnote of a whole one. A pair's run that answered while some
/// of its puts failed was counted whole, and the new main served without them
/// (MEASURED by the private-holger-ops estate-box gate, 2026-10-04, R1: `RECHARGED
/// … 7 fetched` while three artifacts of a lapsed repository were refused).
pub fn not_restored(repository: &str, failed: &[String]) -> Option<String> {
(!failed.is_empty()).then(|| {
format!(
"{repository}: {} artifact(s) not restored: {}",
failed.len(),
failed
.iter()
.take(5)
.cloned()
.collect::<Vec<_>>()
.join(", ")
)
})
}
/// The marker that a box was recharged — once per data root.
pub const RECHARGED: &str = "holger-recharged";
/// The marker that a recharge BEGAN and is not whole: the box serves nothing
/// (`holger-twin-15`) and the next `Recharge` resumes — gunnar's RESTORING.
pub const RECHARGING: &str = "holger-recharging";
/// **The boot hold** (Gunnar's `repair/hold`, Holger's names): the start found this box DEAD
/// — a main, fresh, with a twin to be filled from — and held it. The mark names the twin. While
/// it stands the box serves nothing (`holger-twin-15`), across every restart, whether or not
/// its config still names the twin; the recharge (the `Recharge` a pulled
/// `recharge_from_twin` order runs) releases it once the box is whole. A box that holds data
/// of its own again is released by itself. To start empty on purpose an admin removes it.
pub const RECHARGE_HOLD: &str = "holger-recharge-hold";
/// **Hold this box** for a recharge from `twin`.
pub fn hold(root: &Path, twin: &str) -> Result<(), String> {
let at = root.join(RECHARGE_HOLD);
std::fs::write(&at, format!("{}\n", twin.trim())).map_err(|e| format!("{}: {e}", at.display()))
}
/// **The twin this box is held for**, if it is held.
pub fn held(root: &Path) -> Option<String> {
let twin = std::fs::read_to_string(root.join(RECHARGE_HOLD))
.ok()?
.trim()
.to_string();
(!twin.is_empty()).then_some(twin)
}
/// **Release the hold**: the recharge is whole (or the box holds data of its own again).
pub fn release(root: &Path) {
if std::fs::remove_file(root.join(RECHARGE_HOLD)).is_ok() {
log::warn!("the boot hold is released: this box holds its data set again and serves");
}
}
/// Pure: **the hold at boot**, Gunnar's rule. `dead` = a main that is fresh (or mid-recharge);
/// `named` = the twin its config or feed names (`recharge_from`); `held` = the twin a hold on
/// its disk names. Answers the twin it is held for (`None` = it serves), and whether the mark
/// is to be written (`Some(true)`) or removed (`Some(false)`).
pub fn boot_hold(
dead: bool,
named: Option<&str>,
held: Option<&str>,
) -> (Option<String>, Option<bool>) {
let named = named.map(str::trim).filter(|t| !t.is_empty());
match (dead, held, named) {
(true, Some(h), _) => (Some(h.to_string()), None),
(true, None, Some(n)) => (Some(n.to_string()), Some(true)),
(true, None, None) => (None, None),
(false, Some(_), _) => (None, Some(false)),
(false, None, _) => (None, None),
}
}
/// What a recharge did.
#[derive(Debug, Clone, Default)]
pub struct Recharged {
pub estate: String,
pub artifacts: Vec<String>,
pub failed: Vec<String>,
}
/// Write the once-marker when `r` is whole.
pub fn mark_recharged(root: &Path, from: &str, r: &Recharged) -> Result<(), String> {
if r.failed.is_empty() {
std::fs::write(root.join(RECHARGED), format!("recharged from {from}\n"))
.map_err(|e| format!("{}: {e}", root.join(RECHARGED).display()))?;
}
Ok(())
}
/// **Recharge this box from `recharge_from`** (a twin or a paired downstream) —
/// the estate and every repository this box declares, ONCE, before it serves.
/// `None` when there is nothing to do (no `recharge_from`, or already done).
/// A partial recharge is said and the marker is NOT written, so the next start
/// tries again rather than serving half a box as if it were whole.
pub async fn recharge(holger: &crate::Holger) -> Result<Option<Recharged>, String> {
let Some(from) = holger
.recharge_from
.as_deref()
.map(str::trim)
.filter(|s| !s.is_empty())
else {
return Ok(None);
};
let root = holger.peers_root();
if root.join(RECHARGED).exists() {
return Ok(None);
}
// A REMOTE twin with no `recharge` credential yet cannot be read at boot:
// the box starts, serving nothing (`holger-twin-15`, `Holger::start`), and
// the recharge lane certifies it and runs `Recharge` — the estate's road.
if twin_tls::load(&root, Purpose::Recharge)?.is_none()
&& !twin_tls::is_loopback(&endpoint_of(from))
{
log::warn!("recharge: `{from}` is not on loopback and no recharge credential is installed — this box serves nothing until the recharge lane has run");
return Ok(None);
}
// The recharge is in flight: the paid limit does not police its writes
// (`metered::unpoliced`), as on the lane's own road (`Recharge`).
std::fs::write(root.join(RECHARGING), format!("recharging from {from}\n"))
.map_err(|e| format!("{}: {e}", root.join(RECHARGING).display()))?;
let out = Follower::of(holger, Arc::default()).recharge(from).await?;
mark_recharged(&root, &endpoint_of(from), &out)?;
if out.failed.is_empty() {
let _ = std::fs::remove_file(root.join(RECHARGING));
release(&root);
}
Ok(Some(out))
}
/// Pure over the config: this box's control plane as a client dials it.
fn control_of(holger: &crate::Holger) -> String {
let Some(first) = holger.exposed_endpoints.first() else {
return String::new();
};
let Ok(at) = first.ron_url.parse::<std::net::SocketAddr>() else {
return first.ron_url.clone();
};
if !at.ip().is_unspecified() && !(at.ip().is_loopback() && holger.role.is_twin()) {
return at.to_string();
}
// Unspecified (or a twin's loopback rewrite): the address the replication
// door is reached at is the address this box is reached at.
let named = holger
.replication_door
.as_ref()
.and_then(|d| d.ron_advertise.first().cloned())
.or_else(|| {
holger
.replication_bind
.as_deref()
.and_then(|b| b.parse::<std::net::SocketAddr>().ok())
.filter(|b| !b.ip().is_unspecified())
.map(|b| b.ip().to_string())
});
match named {
Some(host) => format!("{host}:{}", at.port()),
None => at.to_string(),
}
}
#[cfg(test)]
mod tests {
use super::*;
/// ★ A pair whose puts failed makes the recharge NOT whole (the box keeps serving
/// nothing and the next `Recharge` resumes); a pair with none says nothing.
/// ★ The boot hold is Gunnar's rule: a dead main with a twin named is held (and the mark
/// written); a mark on its disk holds it with or without the name; a box that is not dead
/// is never held, and a mark it carries is removed.
#[test]
fn the_boot_hold_holds_a_dead_main_and_lifts_once_it_is_not_dead() {
assert_eq!(
boot_hold(true, Some(" 10.0.0.2:50051 "), None),
(Some("10.0.0.2:50051".into()), Some(true))
);
assert_eq!(
boot_hold(true, None, Some("10.0.0.2:50051")),
(Some("10.0.0.2:50051".into()), None),
"the mark holds without the config"
);
assert_eq!(
boot_hold(true, Some("other:1"), Some("10.0.0.2:50051")),
(Some("10.0.0.2:50051".into()), None),
"the mark names the twin it was held for"
);
assert_eq!(
boot_hold(true, Some(" "), None),
(None, None),
"no twin, no hold: nothing to be filled from"
);
assert_eq!(
boot_hold(false, Some("10.0.0.2:50051"), None),
(None, None),
"a box with data of its own is not held"
);
assert_eq!(
boot_hold(false, None, Some("10.0.0.2:50051")),
(None, Some(false)),
"it lifts by itself"
);
let d = tempfile::tempdir().unwrap();
assert_eq!(held(d.path()), None);
hold(d.path(), "10.0.0.2:50051").unwrap();
assert_eq!(held(d.path()).as_deref(), Some("10.0.0.2:50051"));
release(d.path());
assert_eq!(held(d.path()), None);
}
/// ★ The twin says its first whole follow over mutual TLS ONCE, at WARN — the
/// line `gate holger` reads in the twin's log (stinky SB #33/#46); the clear
/// road never says it.
#[test]
fn the_first_whole_follow_over_mtls_is_said_once() {
let said = std::sync::atomic::AtomicBool::new(false);
assert_eq!(
first_whole_over_mtls("plain http://127.0.0.1:1", &said),
None,
"the clear road says nothing"
);
let line = first_whole_over_mtls("mtls 10.13.8.2:50052", &said)
.expect("the first whole pull over mTLS is said");
assert!(
line.starts_with(OVER_MTLS_SAID) && line.contains("10.13.8.2:50052"),
"{line}"
);
assert_eq!(
first_whole_over_mtls("mtls 10.13.8.2:50052", &said),
None,
"once"
);
}
#[test]
fn an_artifact_a_recharge_did_not_put_is_a_failure_of_the_recharge() {
let f = not_restored(
"migtest",
&[
"/functest/1/seed.bin".to_string(),
"/functest/2/seed.bin".to_string(),
],
)
.unwrap();
assert!(
f.starts_with("migtest: 2 artifact(s) not restored")
&& f.contains("/functest/1/seed.bin"),
"{f}"
);
assert_eq!(not_restored("bundles", &[]), None);
}
/// `BecomeTwin` starts the follow from the pairing loop's own thread, which
/// has no Tokio reactor (mock 2026-09-28: `thread 'box-pairing' panicked …
/// there is no reactor running`). The Follower runs it on the runtime it was
/// built in.
#[test]
fn the_live_follow_starts_from_a_thread_without_a_runtime() {
let rt = tokio::runtime::Builder::new_multi_thread()
.worker_threads(1)
.enable_all()
.build()
.unwrap();
let root = std::env::temp_dir().join(format!("holger-follow-rt-{}", std::process::id()));
std::fs::create_dir_all(&root).unwrap();
let f = rt.block_on(async {
Follower {
repos: Arc::default(),
live: None,
root: root.clone(),
posture: Arc::default(),
tasks: Arc::default(),
audit_dir: None,
stores: Vec::new(),
control: "127.0.0.1:1".into(),
hold: Arc::default(),
position: Arc::default(),
principal: "holger-twin".into(),
evidence: Arc::default(),
said_mtls: Arc::default(),
rt: tokio::runtime::Handle::try_current().ok(),
}
});
let role = crate::twin::NodeRole::Twin {
primary: "127.0.0.1:9".into(),
poll_seconds: 3600,
};
let started = std::thread::spawn(move || f.start(&role))
.join()
.expect("the pairing thread does not panic");
assert_eq!(started, Ok(1), "one follow loop, on the server's runtime");
rt.shutdown_background();
let _ = std::fs::remove_dir_all(&root);
}
#[test]
fn the_twin_mirrors_every_repository_from_the_same_one_on_the_primary() {
let s = source("10.13.8.2:50051", &["crates".into(), "bundles".into()], 60).unwrap();
assert_eq!(s.ron_name, SOURCE_NAME);
assert_eq!(s.ron_endpoint, "http://10.13.8.2:50051");
assert_eq!(s.ron_every_secs, 60);
let pairs: Vec<(String, String)> = s
.ron_repositories
.iter()
.map(|p| (p.ron_from.clone(), p.ron_to.clone()))
.collect();
assert_eq!(
pairs,
[
("crates".into(), "crates".into()),
("bundles".into(), "bundles".into())
]
);
assert_eq!(
endpoint_of("https://grpc.holger.rs:50051"),
"https://grpc.holger.rs:50051"
);
let t = source_with(
"10.13.8.2:50051",
&["crates".into()],
60,
Some(Path::new("/var/lib/holger/holger-twin-tls")),
)
.unwrap();
assert_eq!(
t.ron_credential,
crate::mirror_cycle::MirrorCredential::Twin {
ron_dir: "/var/lib/holger/holger-twin-tls".into()
}
);
}
#[test]
fn the_evidence_keeps_the_last_whole_root_and_the_newest_writes() {
let mut ev = Evidence::default();
let whole = crate::estate_wire::Pulled {
root: "r1".into(),
files: 3,
..Default::default()
};
ev.estate(&whole);
ev.estate(&crate::estate_wire::Pulled {
failed: vec!["x".into()],
..Default::default()
});
assert_eq!(
(ev.estate_root.as_str(), ev.estate_files),
("r1", 3),
"a partial pull changes nothing"
);
let n = crate::mirror_cycle::WRITTEN_KEPT;
let many: Vec<(String, String, u64)> = (0..n + 5)
.map(|i| (format!("/a{i}@"), format!("{i:064x}"), i as u64))
.collect();
ev.wrote("bundles", &many);
assert_eq!(ev.held.len(), n);
assert_eq!(
ev.held.front().unwrap().coordinate,
"/a5@",
"the oldest fall out"
);
assert_eq!(
(
ev.held.back().unwrap().repository.as_str(),
ev.held.back().unwrap().size
),
("bundles", (n + 4) as u64)
);
}
}