server/lib/src/topology.rs
//! **`AdminService.Topology`: the estate's boxes, as THIS holger knows them**
//! (lane E3, 2026-10-19; Gunnar's `Admin.Topology` pattern, Holger's own code).
//!
//! The console draws one board of the estate — front | main | twin, and every
//! downstream holger that pulls from main hanging below — and the console is a thin
//! layer: it calls and it renders, and it decides nothing. So every decision behind
//! the board is here: which boxes there are, and whether each one is alive.
//!
//! # Evidence, never inference
//!
//! A box is on the board because this holger holds EVIDENCE of it, and each row
//! says which (`sources`) and in one sentence how (`why`):
//!
//! * **`self`** — this box. It answered the call, so it is `up`, unless it is HELD
//! (a main awaiting its recharge from its twin, M3's boot hold; or a fenced
//! primary): then it serves nothing and is drawn dark with the word `held`.
//! * **`pairing`** — the installer's pairing feed (`joined.toml`, set by the
//! `paired` lane through [`set_pairing`]): the front, the main and the twin, with
//! their WireGuard overlay and utility addresses. A peer nothing here can ask (the
//! front) is `unknown`, which is a different fact from `silent`.
//! * **`report`** — on a primary, the twins' `GrowService.TwinReport`s (the roster's
//! rows: L25's TwinRow, M2's `twin_state` and `lag_entries`); on a twin, its own
//! follow of the primary (`twin_follow::Follower::look`). The follow-point word is
//! authoritative for that box.
//! * **`downstream`** — a paired downstream holger's sync reports (`PeerBoard`),
//! and the packages delivered to it (counted at each delivery, kept in its
//! sync record across a restart; lane A13).
//! * **`config` + `proxy` | `mirror`** — an UPSTREAM (lane A13): the config
//! declares it (a repository another one names in `ron_upstreams`, or a
//! `mirror.ron_sources` entry), and its state is what the proxy's walks or the
//! mirror's cycles recorded ([`crate::upstreams`]); `pairing` too when the
//! mirrored holger is paired. Drawn on a row above the main box.
//! * **the consumers** — the door admission's usage book counts every consumer
//! (named in the window, anonymous today); drawn under the title.
//!
//! Nothing here walks a network, resolves a name or dials a box: every input is
//! something the server already recorded. [`Sources::gather`] reads them (it
//! blocks: the roster, the credential files, one statvfs); [`board`] is pure and
//! decides every word; [`json`] is pure and is the same answer for the HTTP door.
//!
//! # The board is data (Holger v2.0 H2, owner 2026-10-25)
//!
//! The server draws nothing. `AdminService.Topology` and `GET /-/estate.json` carry the
//! rows and the links; holger-ui draws them (its own renderer). `TopologyReply.svg` is
//! retired and always empty, so no server build links a renderer.
use crate::lock::RwLockOrRecover as _;
use std::net::Ipv4Addr;
use std::path::PathBuf;
use std::sync::{Arc, RwLock};
use crate::follow_point::Judged;
use crate::grpc::holger_grow_proto as gpb;
use crate::grpc::holger_proto as pb;
pub const ROLE_FRONT: &str = "front";
pub const ROLE_PRIMARY: &str = "primary";
pub const ROLE_TWIN: &str = "twin";
pub const ROLE_DOWNSTREAM: &str = "downstream";
pub const ROLE_UPSTREAM: &str = "upstream";
/// A console (holger-ui) as a box of the estate.
pub const ROLE_CONSOLE: &str = "console";
pub const STATE_UP: &str = "up";
pub const STATE_BEHIND: &str = "behind";
pub const STATE_SILENT: &str = "silent";
pub const STATE_UNKNOWN: &str = "unknown";
pub const SOURCE_SELF: &str = "self";
pub const SOURCE_PAIRING: &str = "pairing";
pub const SOURCE_REPORT: &str = "report";
pub const SOURCE_DOWNSTREAM: &str = "downstream";
pub const SOURCE_CONFIG: &str = "config";
pub const SOURCE_PROXY: &str = "proxy";
pub const SOURCE_MIRROR: &str = "mirror";
pub const LINK_SERVES: &str = "serves";
pub const LINK_FOLLOWS: &str = "follows";
pub const LINK_PULLS: &str = "pulls";
pub const LINK_FEEDS: &str = "feeds";
/// **Every consumer this box serves** (lane A13), as the door admission's book
/// counts them ([`crate::admission::consumers`]).
#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
pub struct ConsumerCount {
/// False: this build keeps no book, and nothing below is a count.
pub counted: bool,
pub principals: u64,
pub peers: u64,
pub anonymous_today: u64,
pub window_days: u32,
}
/// The word a held box carries in its headline.
pub const HELD: &str = "held";
/// The bound every state below is judged against: the twin's (three follow cycles).
pub const SILENT_AFTER_MS: u64 = crate::grow::TWIN_SILENT_AFTER_MS;
// ── the pairing feed, as the `paired` lane hands it over ────────────────────
/// **One box the pairing feed names.**
#[derive(Clone, Debug, Default, PartialEq, Eq)]
pub struct PairedBox {
/// The pairing name (`front`, `main`, `twin`, …).
pub name: String,
/// `front` | `primary` | `twin` (nordisk-pairing's role words).
pub role: String,
/// wg0's address on the estate overlay.
pub overlay: Option<Ipv4Addr>,
/// The utility network's address, when fed.
pub utility: Option<Ipv4Addr>,
}
/// **The estate as the pairing feed names it.**
#[derive(Clone, Debug, Default, PartialEq, Eq)]
pub struct Pairing {
pub estate: String,
/// This box's own pairing name.
pub me: String,
/// Every box the feed names, this one included.
pub boxes: Vec<PairedBox>,
/// When the pairing's key expires (Unix s); 0 = not known (lane E2).
pub not_after_unix: i64,
}
static PAIRING: RwLock<Option<Pairing>> = RwLock::new(None);
/// **Set the pairing view** (the `paired` lane, every time a feed is kept).
pub fn set_pairing(p: Option<Pairing>) {
*PAIRING.write_or_recover() = p;
}
/// The pairing view now; `None` on a holger nobody paired.
pub fn pairing() -> Option<Pairing> {
#[cfg_attr(not(feature = "estate"), allow(unused_mut))]
let mut p = PAIRING.read_or_recover().clone()?;
// The pairing certificate's end, read live (lane E2): the front renews it at two
// thirds of its life, so the end moves on without a new feed. Only an estate build
// reads certificates (`estate`); a box without it was never paired.
#[cfg(feature = "estate")]
if let Some(end) = crate::cert_health::pairing_not_after() {
p.not_after_unix = end;
}
Some(p)
}
// ── what the board is drawn from ────────────────────────────────────────────
/// **Everything [`board`] reads**, gathered at one instant. Built by
/// [`Sources::gather`] on a live server and by hand in a test.
#[derive(Clone, Debug, Default)]
pub struct Facts {
pub now_ms: i64,
/// This box's hostname, the name of last resort.
pub host: String,
pub is_twin: bool,
/// On a twin: the primary it follows (`host:port`).
pub primary_at: String,
/// `Some(source)` while this box is held for its recharge.
pub held_from: Option<String>,
/// `Some(epoch)` once this primary was fenced.
pub fenced: Option<u64>,
/// On a twin: where it stands against its primary; `None` = never answered.
pub look: Option<Judged>,
/// This box's data volume `(total, used)`, measured; `None` = not measured.
pub own_disk: Option<(u64, u64)>,
/// This box's wg0 address.
pub own_overlay: Option<Ipv4Addr>,
/// When this box's mutual-TLS credential expires (Unix s); 0 = none.
pub cert_not_after_unix: i64,
/// On a primary: its twins, as their reports say.
pub twins: Vec<gpb::TwinRow>,
/// Its paired downstreams, as their reports say.
pub downstreams: Vec<gpb::DownstreamRow>,
pub pairing: Option<Pairing>,
/// Where its bytes come from: the declared upstreams with their evidence.
pub upstreams: Vec<crate::upstreams::UpstreamFact>,
/// Its consumers; `None` = this build keeps no book.
pub consumers: Option<ConsumerCount>,
/// What the RON declares: every source and every downstream connector (HBOARD-1).
pub declared: Declared,
/// The declared downstreams' sync lag, as the sync board measures it now.
pub sync: Vec<SyncDownstream>,
}
/// **The RON's own graph** (HBOARD-1, the owner's board of 2026-10-23): every input source
/// and upstream the declaration names, and every downstream connector, read once from the
/// config. The board draws each one, measured or not.
#[derive(Clone, Debug, Default, PartialEq, Eq)]
pub struct Declared {
pub upstreams: Vec<DeclaredUpstream>,
/// The `sync.downstreams` names.
pub downstreams: Vec<String>,
}
/// One declared source or upstream.
#[derive(Clone, Debug, Default, PartialEq, Eq)]
pub struct DeclaredUpstream {
pub name: String,
/// [`crate::upstreams::KIND_PROXY`] | `KIND_MIRROR` | `KIND_PULL` | `KIND_UPLOAD`.
pub kind: String,
/// The source kind or repository type as declared (`crates`, `npm`, `rust-remote`, …).
pub speaks: String,
pub address: String,
/// The local repositories it feeds.
pub feeds: Vec<String>,
}
/// One declared downstream's sync, as the sync board (`GET /-/sync/board`) measures it.
#[derive(Clone, Debug, Default, PartialEq, Eq)]
pub struct SyncDownstream {
pub name: String,
/// How it reaches this holger: `front` or `wireguard` (lane HOL-WGDS).
pub via: String,
/// `current`, `behind`, `never`, `alarm`.
pub state: String,
pub headline: String,
pub last_generation_applied: u64,
pub lag_generations: u64,
pub lag_secs: i64,
/// When it last acknowledged a generation (Unix seconds); 0 = never.
pub last_ack_at: i64,
/// When its mirror last reported its sync (Unix seconds; 0 = never; lane OX-WGDS2).
pub last_report_at: i64,
/// How many of this holger's writes its mirror had not applied at that report.
pub entries_behind: u64,
/// What its mirror's last report said of its own RON (HOL-FLOW-SMART).
pub summary: Option<crate::flow::RonSummary>,
}
/// The live sync rows, read at each gathering (a closure, so the board names no feature).
pub type SyncRows = Arc<dyn Fn() -> Vec<SyncDownstream> + Send + Sync>;
/// The intake chain's door every upload lands in (`promotion.intake`'s first repository).
pub const UPLOAD_DOOR: &str = "incoming";
/// The row an estate with an intake chain and no upload source is given: its uploads.
pub const UPLOADS: &str = "uploads";
/// A downstream measured by the sync board (a row's `sources`).
pub const SOURCE_SYNC: &str = "sync";
/// Pure: **what the declaration names** (HBOARD-1): every `sources:` entry (lowered or
/// not), every repository's `ron_upstreams`, every `mirror.ron_sources` entry, a downstream
/// box's `sync.upstreams`; the uploads into `incoming` when no upload source is named; and
/// every `sync.downstreams` connector. One row per name, its feeds merged; by name.
pub fn declared(h: &crate::Holger) -> Declared {
use crate::sources::SourceKind;
use crate::upstreams::{KIND_MIRROR, KIND_PROXY, KIND_PULL, KIND_UPLOAD};
let mut ups: Vec<DeclaredUpstream> = Vec::new();
fn merge(ups: &mut Vec<DeclaredUpstream>, u: DeclaredUpstream) {
match ups.iter_mut().find(|x| x.name == u.name) {
Some(x) => {
for f in u.feeds {
if !x.feeds.contains(&f) {
x.feeds.push(f);
}
}
if x.address.is_empty() {
x.address = u.address;
}
}
None => ups.push(u),
}
}
let mut add = |u: DeclaredUpstream| merge(&mut ups, u);
let up = |name: &str, kind: &str, speaks: &str, address: String, feeds: Vec<String>| {
DeclaredUpstream {
name: name.to_string(),
kind: kind.to_string(),
speaks: speaks.to_string(),
address,
feeds,
}
};
for s in &h.sources {
let url = s.url.clone().unwrap_or_default();
let speaks = s.kind.name();
add(if s.kind == SourceKind::Holger {
up(
&s.name,
KIND_MIRROR,
speaks,
url,
s.repositories.iter().map(|p| p.ron_to.clone()).collect(),
)
} else if s.skips_the_chain() {
up(
&s.name,
KIND_PROXY,
speaks,
if url.is_empty() {
"https://index.crates.io".into()
} else {
url
},
s.serves.clone(),
)
} else if s.pulls_into_the_chain() {
up(&s.name, KIND_PULL, speaks, url, vec![UPLOAD_DOOR.into()])
} else {
up(
&s.name,
KIND_UPLOAD,
speaks,
String::new(),
vec![UPLOAD_DOOR.into()],
)
});
}
for r in &h.repositories {
for name in &r.ron_upstreams {
let (speaks, address) = match h.repositories.iter().find(|x| &x.ron_name == name) {
Some(u) => (
u.ron_repo_type.clone(),
match (&u.ron_base_url, u.ron_repo_type.as_str()) {
(Some(b), _) => b.clone(),
(None, "rust-remote") => "https://index.crates.io".to_string(),
(None, _) => String::new(),
},
),
None => (String::new(), String::new()),
};
add(up(
name,
KIND_PROXY,
&speaks,
address,
vec![r.ron_name.clone()],
));
}
}
if let Some(m) = &h.mirror {
for src in &m.ron_sources {
add(up(
&src.ron_name,
KIND_MIRROR,
&format!("{:?}", src.ron_kind),
src.ron_endpoint.clone(),
src.ron_repositories
.iter()
.map(|p| p.ron_to.clone())
.collect(),
));
}
}
#[cfg(feature = "sync")]
if let Some(sync) = &h.sync {
for p in &sync.upstreams {
add(up(
&p.name,
crate::upstreams::KIND_SYNC,
"holger",
p.url.clone().unwrap_or_default(),
Vec::new(),
));
}
}
if !ups.iter().any(|u| u.kind == KIND_UPLOAD)
&& h.repositories.iter().any(|r| r.ron_name == UPLOAD_DOOR)
{
merge(
&mut ups,
up(
UPLOADS,
KIND_UPLOAD,
"upload",
String::new(),
vec![UPLOAD_DOOR.into()],
),
);
}
ups.sort_by(|a, b| a.name.cmp(&b.name));
#[cfg(feature = "sync")]
let downstreams = h
.sync
.as_ref()
.map(|s| s.downstreams.iter().map(|d| d.name.clone()).collect())
.unwrap_or_default();
#[cfg(not(feature = "sync"))]
let downstreams = Vec::new();
Declared {
upstreams: ups,
downstreams,
}
}
/// **The declared downstreams' sync, live**: the sync board of the chain's hub at each
/// gathering (the alarms due raised first, as `GET /-/sync/board` raises them). `None` on a
/// box with no intake chain; empty rows while the chain has no `sync:` block.
#[cfg(feature = "sync")]
pub fn sync_rows(chain: Option<&Arc<crate::promote::intake::IntakeChain>>) -> Option<SyncRows> {
let chain = Arc::clone(chain?);
Some(Arc::new(move || {
let Some(hub) = chain.sync() else {
return Vec::new();
};
// What each one's last report said of its own RON.
let said: std::collections::BTreeMap<String, crate::flow::RonSummary> = hub
.reports()
.unwrap_or_default()
.into_iter()
.filter_map(|h| h.summary.map(|s| (h.name, s)))
.collect();
match hub.board(crate::sync::protocol::now_secs()) {
Ok(b) => b
.downstreams
.into_iter()
.map(|d| SyncDownstream {
summary: said.get(&d.name).cloned(),
name: d.name,
via: d.via,
state: d.state,
headline: d.headline,
last_generation_applied: d.last_generation_applied,
lag_generations: d.lag_generations,
lag_secs: d.lag_secs,
last_ack_at: d.last_ack_at,
last_report_at: d.last_report_at,
entries_behind: d.entries_behind,
})
.collect(),
Err(e) => {
log::warn!("the estate board could not read the sync board: {e}");
Vec::new()
}
}
}))
}
/// **The live server's evidence**, one handle the gRPC door and the HTTP door share.
#[derive(Clone)]
pub struct Sources {
pub twin: Arc<crate::twin::TwinPosture>,
pub grow: Option<Arc<crate::grow::GrowPlane>>,
pub follower: Option<crate::twin_follow::Follower>,
pub peers: Arc<crate::peers::PeerBoard>,
pub hold: Arc<crate::grow::Hold>,
pub data_root: Option<PathBuf>,
/// The declaration's graph, read once at boot ([`declared`]).
pub declared: Arc<Declared>,
/// The declared downstreams' sync, read at each gathering ([`sync_rows`]).
pub sync: Option<SyncRows>,
}
impl Sources {
/// Read every input now. BLOCKS (the twin roster, the credential files, one
/// statvfs): call it off the executor.
pub fn gather(&self) -> Facts {
let now_ms = crate::clock::now_unix_ms();
let role = self.twin.role();
let is_twin = role.is_twin();
let twins = match (&self.grow, is_twin) {
(Some(g), false) => g.twins().map(|r| r.twins).unwrap_or_default(),
_ => Vec::new(),
};
let cert_not_after_unix = self
.data_root
.as_deref()
.map(|root| {
[
crate::twin_tls::Purpose::Twin,
crate::twin_tls::Purpose::Recharge,
]
.into_iter()
.map(|p| crate::admin_view::credential(root, p, now_ms.div_euclid(1000)))
.find(|c| c.installed && c.not_after_unix > 0)
.map(|c| c.not_after_unix)
.unwrap_or(0)
})
.unwrap_or(0);
Facts {
now_ms,
host: crate::admin_view::host(),
is_twin,
primary_at: role.primary().unwrap_or_default().to_string(),
held_from: self.twin.awaiting_from(),
fenced: self.twin.fenced(),
look: if is_twin {
self.follower.as_ref().and_then(|f| f.look())
} else {
None
},
own_disk: self
.data_root
.as_deref()
.and_then(|r| crate::metering::statvfs(r).ok())
.map(|f| (f.total_bytes, f.used_bytes)),
own_overlay: crate::tunnel_addr::estate().and_then(|e| e.overlay),
cert_not_after_unix,
twins,
downstreams: self.peers.rows(self.hold.boot(), self.hold.head(), now_ms),
pairing: pairing(),
upstreams: crate::upstreams::snapshot(),
consumers: crate::admission::consumers(),
declared: (*self.declared).clone(),
sync: self.sync.as_ref().map(|rows| rows()).unwrap_or_default(),
}
}
}
// ── the words ───────────────────────────────────────────────────────────────
/// Pure: a span in coarse words — whole minutes once past one, whole hours past two.
pub fn ago(ms: u64) -> String {
let s = ms / 1000;
match s {
0..=59 => "under a minute".into(),
60..=7199 => format!("{} min", s / 60),
_ => format!("{} h", s / 3600),
}
}
/// Pure: a follower's word (`ok | behind | stalled | silent`) as the board's state
/// and headline.
pub fn follower_words(
word: &str,
lag: u64,
seen_ago_ms: u64,
ever_seen: bool,
) -> (&'static str, String) {
match word {
"ok" => (STATE_UP, "following".into()),
"behind" => (STATE_BEHIND, format!("behind {lag} {}", entries(lag))),
"stalled" => (
STATE_BEHIND,
format!("stalled, behind {lag} {}", entries(lag)),
),
_ if !ever_seen => (STATE_SILENT, "never heard".into()),
_ => (STATE_SILENT, format!("silent for {}", ago(seen_ago_ms))),
}
}
fn entries(n: u64) -> &'static str {
if n == 1 {
"entry"
} else {
"entries"
}
}
fn host_ip(at: &str) -> Option<Ipv4Addr> {
let rest = at
.trim()
.split_once("://")
.map(|(_, r)| r)
.unwrap_or(at.trim());
let hostport = rest.split('/').next().unwrap_or(rest);
hostport
.parse::<std::net::SocketAddrV4>()
.map(|s| *s.ip())
.ok()
.or_else(|| hostport.parse().ok())
}
fn paired_is(b: &PairedBox, at: &str) -> bool {
match host_ip(at) {
Some(ip) => b.overlay == Some(ip) || b.utility == Some(ip),
None => false,
}
}
fn ip(o: Option<Ipv4Addr>) -> String {
o.map(|i| i.to_string()).unwrap_or_default()
}
fn rank(role: &str) -> u8 {
match role {
ROLE_FRONT => 0,
ROLE_PRIMARY => 1,
ROLE_TWIN => 2,
ROLE_UPSTREAM => 3,
_ => 4,
}
}
/// Pure: **an upstream's state, headline and why** (lane A13), from what the
/// proxy's walks or the mirror's cycles recorded — never from a dial.
pub fn upstream_words(
u: &crate::upstreams::UpstreamFact,
now_ms: i64,
) -> (&'static str, String, String) {
let since = |t: i64| ago((now_ms - t).max(0) as u64);
let feeds = if u.feeds.is_empty() {
"no repository".to_string()
} else {
u.feeds.join(", ")
};
if u.kind == crate::upstreams::KIND_MIRROR {
let every = if u.every_secs == 0 {
300_000
} else {
u.every_secs as i64 * 1000
};
if u.link == crate::peers::LinkState::KeyMismatch.word() {
return (STATE_SILENT, "key mismatch".into(), format!("the paired holger `{}` presented another key than the one pinned at pairing: this box refuses to pull from it", u.paired_as));
}
if u.last_ok_ms == 0 {
return if u.failed > 0 {
(STATE_SILENT, "never synced".into(), format!("the mirror into {feeds} has not completed a cycle since this boot; {} failed", u.failed))
} else {
(
STATE_UNKNOWN,
"no cycle yet".into(),
format!("the config mirrors it into {feeds}; no cycle has run since this boot"),
)
};
}
if now_ms - u.last_ok_ms > 3 * every {
return (
STATE_SILENT,
format!("silent for {}", since(u.last_ok_ms)),
format!(
"its last completed cycle was {} ago, more than three of its {} s cycles",
since(u.last_ok_ms),
every / 1000
),
);
}
if u.consecutive_failures > 0 {
return (
STATE_BEHIND,
format!("{} failed cycles", u.consecutive_failures),
format!(
"its last cycle completed {} ago; {} failed since",
since(u.last_ok_ms),
u.consecutive_failures
),
);
}
return (STATE_UP, "syncing".into(), format!("its last cycle into {feeds} completed {} ago; {} artifacts fetched since this boot", since(u.last_ok_ms), u.fetched));
}
if u.kind == crate::upstreams::KIND_UPLOAD {
return (
STATE_UP,
"takes uploads".into(),
format!("an upload door into {feeds}: what is uploaded walks the intake steps"),
);
}
if u.kind == crate::upstreams::KIND_SYNC {
return (STATE_UNKNOWN, "syncs by signed manifest".into(), "the declaration trusts this upstream's sync manifests (`sync.upstreams`); the sync record says what was applied".into());
}
if u.last_ok_ms == 0 && u.last_err_ms == 0 {
return (STATE_UNKNOWN, "not asked since boot".into(), format!("the config names it as the upstream of {feeds}; no read has walked to it since this boot"));
}
if u.last_err_ms > u.last_ok_ms {
let last = if u.last_ok_ms == 0 {
"it has not answered since this boot".to_string()
} else {
format!("its last answer was {} ago", since(u.last_ok_ms))
};
return (
STATE_SILENT,
format!("failing for {}", since(u.last_err_ms)),
format!(
"the last read that walked to it ({} ago) failed: {}; {last}",
since(u.last_err_ms),
u.last_error
),
);
}
(STATE_UP, "answering".into(), format!("it answered the last read that walked to it, {} ago; since this boot it served {} and failed {}", since(u.last_ok_ms), u.fetched, u.failed))
}
/// Pure: **the board** — every row, every link, and the drawing.
pub fn board(f: &Facts) -> pb::TopologyReply {
let pairing = f.pairing.clone().unwrap_or_default();
let me_paired = pairing.boxes.iter().find(|b| b.name == pairing.me).cloned();
let mut rows: Vec<pb::TopologyBox> = Vec::new();
// ── this box ──
let mut me = pb::TopologyBox {
name: [pairing.me.as_str(), f.host.as_str(), "this box"]
.into_iter()
.find(|n| !n.is_empty())
.unwrap_or("this box")
.to_string(),
role: if f.is_twin { ROLE_TWIN } else { ROLE_PRIMARY }.into(),
this_box: true,
last_seen_unix_ms: f.now_ms,
sources: vec![SOURCE_SELF.into()],
overlay_address: ip(f.own_overlay.or(me_paired.as_ref().and_then(|b| b.overlay))),
utility_address: ip(me_paired.as_ref().and_then(|b| b.utility)),
cert_not_after_unix: f.cert_not_after_unix,
pairing_not_after_unix: pairing.not_after_unix,
..Default::default()
};
if me_paired.is_some() {
me.sources.push(SOURCE_PAIRING.into());
}
me.address = me.overlay_address.clone();
if let Some((total, used)) = f.own_disk {
(me.disk_measured, me.disk_total_bytes, me.disk_used_bytes) = (true, total, used);
}
if let Some(from) = &f.held_from {
me.state = STATE_SILENT.into();
me.held = true;
me.headline = format!("{HELD} for its recharge from {from}");
me.why = format!("this box answered, and it is held for its recharge from `{from}`: it serves nothing until the recharge is whole");
} else if let Some(epoch) = f.fenced {
me.state = STATE_SILENT.into();
me.held = true;
me.headline = format!("{HELD}: fenced at epoch {epoch}");
me.why = format!("this box answered, and it is fenced at epoch {epoch}: it takes no writes until it is recharged or replaced");
} else if f.is_twin {
match &f.look {
Some(j) => {
let (state, headline) = follower_words(j.state, j.lag, j.seen_ago_ms, true);
// Its own primary silent is not this twin's fault: it answers.
me.state = if state == STATE_SILENT {
STATE_UP.into()
} else {
state.into()
};
me.headline = if state == STATE_SILENT {
"keeping the estate".into()
} else {
headline
};
me.twin_state = j.state.into();
me.lag_entries = j.lag;
me.why = format!(
"this box answered; it follows its primary ({}, {} entries behind)",
j.state, j.lag
);
}
None => {
me.state = STATE_UP.into();
me.headline = "answering".into();
me.why =
"this box answered; its primary has not answered a follow report yet".into();
}
}
} else {
me.state = STATE_UP.into();
me.headline = "answering".into();
me.why = "this box answered this call".into();
}
rows.push(me);
// ── the boxes the pairing feed names ──
let mut matched_twins = vec![false; f.twins.len()];
for b in pairing.boxes.iter().filter(|b| b.name != pairing.me) {
let mut r = pb::TopologyBox {
name: b.name.clone(),
role: b.role.clone(),
overlay_address: ip(b.overlay),
utility_address: ip(b.utility),
address: ip(b.overlay.or(b.utility)),
sources: vec![SOURCE_PAIRING.into()],
pairing_not_after_unix: pairing.not_after_unix,
..Default::default()
};
match b.role.as_str() {
ROLE_PRIMARY if f.is_twin => primary_from_look(&mut r, f),
ROLE_TWIN if !f.is_twin => {
match f.twins.iter().position(|t| paired_is(b, &t.address)) {
Some(i) => {
matched_twins[i] = true;
twin_from_report(&mut r, &f.twins[i]);
}
None => {
r.state = STATE_SILENT.into();
r.headline = "never heard".into();
r.why = "the pairing feed names this twin, and it has never reported to this primary".into();
}
}
}
_ => {
r.state = STATE_UNKNOWN.into();
r.headline = "not asked".into();
r.why = "the pairing feed names this box; nothing on this holger asks it".into();
}
}
rows.push(r);
}
// ── a twin's primary that no feed names ──
if f.is_twin && !rows.iter().any(|r| r.role == ROLE_PRIMARY) && !f.primary_at.is_empty() {
let mut r = pb::TopologyBox {
name: "main".into(),
role: ROLE_PRIMARY.into(),
address: f.primary_at.clone(),
..Default::default()
};
primary_from_look(&mut r, f);
rows.push(r);
}
// ── twins that report and that no feed names ──
for (t, _) in f.twins.iter().zip(&matched_twins).filter(|(_, m)| !**m) {
let mut r = pb::TopologyBox {
name: if t.principal.is_empty() {
t.address.clone()
} else {
t.principal.clone()
},
role: ROLE_TWIN.into(),
address: t.address.clone(),
..Default::default()
};
twin_from_report(&mut r, t);
rows.push(r);
}
// ── downstream holgers ──
for d in &f.downstreams {
let (state, headline) = follower_words(
&d.state,
d.lag_entries,
d.seen_ago_ms,
d.last_seen_unix_ms > 0,
);
rows.push(pb::TopologyBox {
via: f
.sync
.iter()
.find(|s| s.name == d.name)
.map(|s| s.via.clone())
.filter(|v| !v.is_empty())
.unwrap_or_else(|| "front".into()),
name: d.name.clone(),
role: ROLE_DOWNSTREAM.into(),
state: state.into(),
headline,
why: format!(
"a paired downstream: its sync reports say {} ({} entries behind at most)",
d.state, d.lag_entries
),
last_seen_unix_ms: d.last_seen_unix_ms,
seen_ago_ms: d.seen_ago_ms,
sources: vec![SOURCE_DOWNSTREAM.into()],
twin_state: d.state.clone(),
lag_entries: d.lag_entries,
fetched_total: d.fetched_total,
fetched_today: d.fetched_today,
..Default::default()
});
}
// ── the downstream connectors the declaration names (`sync.downstreams`), and the
// ones the console enrolled (the sync board's live rows, lane HOL-WGDS) ──
let mut sync_names: Vec<String> = f.declared.downstreams.clone();
for s in &f.sync {
if !sync_names.contains(&s.name) {
sync_names.push(s.name.clone());
}
}
for name in &sync_names {
if rows
.iter()
.any(|r| r.role == ROLE_DOWNSTREAM && &r.name == name)
{
continue;
}
let mut r = pb::TopologyBox {
name: name.clone(),
role: ROLE_DOWNSTREAM.into(),
sources: vec![SOURCE_CONFIG.into(), SOURCE_SYNC.into()],
..Default::default()
};
match f.sync.iter().find(|s| &s.name == name) {
Some(s) => {
r.via = s.via.clone();
r.state = match s.state.as_str() {
"current" => STATE_UP,
"behind" => STATE_BEHIND,
"alarm" => STATE_SILENT,
_ => STATE_UNKNOWN,
}
.into();
r.headline = s.headline.clone();
r.twin_state = s.state.clone();
// Its lag: the generations it has not acknowledged, else the writes its
// mirror reported it had not applied (lane OX-WGDS2).
r.lag_entries = if s.lag_generations > 0 {
s.lag_generations
} else {
s.entries_behind
};
// Last seen: its last acknowledgement or its mirror's last report, the newer.
let seen = s.last_ack_at.max(s.last_report_at);
if seen > 0 {
r.last_seen_unix_ms = seen * 1000;
r.seen_ago_ms = (f.now_ms - r.last_seen_unix_ms).max(0) as u64;
}
r.why = format!(
"a declared sync downstream: generation {} applied, {} behind, the oldest unacknowledged issued {} ago",
s.last_generation_applied,
s.lag_generations,
ago(s.lag_secs.max(0) as u64 * 1000)
);
}
None => {
r.state = STATE_UNKNOWN.into();
r.headline = "not measured".into();
r.why = "the declaration names this sync downstream; nothing on this holger has measured its sync".into();
}
}
rows.push(r);
}
// ── where the bytes come from: every declared source, with what the walks learnt ──
let mut upstreams: Vec<crate::upstreams::UpstreamFact> = f
.declared
.upstreams
.iter()
.map(|d| match f.upstreams.iter().find(|u| u.name == d.name) {
Some(u) => {
let mut u = u.clone();
for feed in &d.feeds {
if !u.feeds.contains(feed) {
u.feeds.push(feed.clone());
}
}
if u.address.is_empty() {
u.address = d.address.clone();
}
u
}
None => crate::upstreams::UpstreamFact {
name: d.name.clone(),
kind: d.kind.clone(),
speaks: d.speaks.clone(),
address: d.address.clone(),
feeds: d.feeds.clone(),
..Default::default()
},
})
.collect();
for u in &f.upstreams {
if !upstreams.iter().any(|x| x.name == u.name) {
upstreams.push(u.clone());
}
}
for u in &upstreams {
let (state, headline, why) = upstream_words(u, f.now_ms);
let mut sources = vec![SOURCE_CONFIG.to_string(), u.kind.clone()];
if !u.paired_as.is_empty() {
sources.push(SOURCE_PAIRING.into());
}
let last = u.last_ok_ms.max(u.last_err_ms);
rows.push(pb::TopologyBox {
name: u.name.clone(),
role: ROLE_UPSTREAM.into(),
address: u.address.clone(),
state: state.into(),
headline,
why,
last_seen_unix_ms: last,
seen_ago_ms: if last > 0 {
(f.now_ms - last).max(0) as u64
} else {
0
},
sources,
fetched_total: u.fetched,
upstream_kind: u.kind.clone(),
feeds: u.feeds.clone(),
..Default::default()
});
}
// This box first, then front, primary, twins, downstreams, each by name.
let mine = rows.remove(0);
rows.sort_by(|a, b| (rank(&a.role), &a.name).cmp(&(rank(&b.role), &b.name)));
rows.insert(0, mine);
let links = links_of(&rows);
let mut reply = pb::TopologyReply {
boxes: rows,
server_unix_ms: f.now_ms,
silent_after_ms: SILENT_AFTER_MS,
estate: pairing.estate.clone(),
links,
svg: String::new(),
summary: String::new(),
consumers: f.consumers.map(|c| pb::TopologyConsumers {
counted: c.counted,
principals: c.principals,
peers: c.peers,
anonymous_today: c.anonymous_today,
window_days: c.window_days,
}),
};
reply.summary = subtitle(&reply, &f.sync);
reply
}
/// On a twin: its primary, from its own follow. A primary the twin cannot hear is
/// HELD — the twin keeps every byte and nothing is promoted (owner, 2026-10-12): the
/// dead main is rebuilt from the twin.
fn primary_from_look(r: &mut pb::TopologyBox, f: &Facts) {
r.sources.push(SOURCE_REPORT.into());
match &f.look {
Some(j) if !j.silent => {
r.state = STATE_UP.into();
r.headline = "answering".into();
r.last_seen_unix_ms = f.now_ms - j.seen_ago_ms as i64;
r.seen_ago_ms = j.seen_ago_ms;
r.why = format!(
"it answered this twin's follow report {} ago",
ago(j.seen_ago_ms)
);
}
Some(j) => {
r.state = STATE_SILENT.into();
r.held = true;
r.headline = format!("{HELD}: silent for {}", ago(j.seen_ago_ms));
r.last_seen_unix_ms = f.now_ms - j.seen_ago_ms as i64;
r.seen_ago_ms = j.seen_ago_ms;
r.why = format!("it has not answered this twin for {}; the twin keeps the estate until the main is recharged from it", ago(j.seen_ago_ms));
}
None => {
r.state = STATE_SILENT.into();
r.held = true;
r.headline = format!("{HELD}: never heard");
r.why = "it has never answered this twin's follow report".into();
}
}
}
/// On a primary: a twin, from its reports (L25 TwinRow, M2's twin_state / lag).
fn twin_from_report(r: &mut pb::TopologyBox, t: &gpb::TwinRow) {
let (state, headline) = follower_words(
&t.state,
t.lag_entries,
t.seen_ago_ms,
t.last_seen_unix_ms > 0,
);
r.state = state.into();
r.headline = headline;
r.twin_state = t.state.clone();
r.lag_entries = t.lag_entries;
r.last_seen_unix_ms = t.last_seen_unix_ms;
r.seen_ago_ms = t.seen_ago_ms;
if t.disk_total_bytes > 0 {
(r.disk_measured, r.disk_total_bytes, r.disk_used_bytes) =
(true, t.disk_total_bytes, t.disk_used_bytes);
}
if r.address.is_empty() {
r.address = t.address.clone();
}
r.sources.push(SOURCE_REPORT.into());
r.why = format!(
"its follow reports say {} — applied {} of this primary's {} (last report {} ago)",
t.state,
t.applied_seq,
t.head_seq,
ago(t.seen_ago_ms)
);
}
/// Pure: front → primary, primary → each twin, primary → each downstream, each
/// upstream → primary; each link carries the state of the box it leads into —
/// an upstream's link carries the UPSTREAM's state (the primary answered this
/// call; what the link says is whether the bytes still come).
fn links_of(rows: &[pb::TopologyBox]) -> Vec<pb::TopologyLink> {
let Some(main) = rows.iter().find(|r| r.role == ROLE_PRIMARY) else {
return Vec::new();
};
let mut out = Vec::new();
for r in rows {
let (from, to, kind) = match r.role.as_str() {
ROLE_FRONT => (r.name.clone(), main.name.clone(), LINK_SERVES),
ROLE_TWIN => (main.name.clone(), r.name.clone(), LINK_FOLLOWS),
ROLE_DOWNSTREAM => (main.name.clone(), r.name.clone(), LINK_PULLS),
ROLE_UPSTREAM => (r.name.clone(), main.name.clone(), LINK_FEEDS),
_ => continue,
};
let state = if kind == LINK_SERVES {
main.state.clone()
} else {
r.state.clone()
};
out.push(pb::TopologyLink {
from,
to,
kind: kind.into(),
state,
});
}
out
}
// ── the board as JSON ───────────────────────────────────────────────────────
/// Pure: **the board as JSON** — `GET /-/estate.json`, the same rows and words as
/// `AdminService.Topology` (the proto's field names), for any client that is not gRPC.
pub fn json(reply: &pb::TopologyReply) -> serde_json::Value {
use serde_json::json;
let boxes: Vec<serde_json::Value> = reply
.boxes
.iter()
.map(|b| {
json!({
"name": b.name,
"role": b.role,
"this_box": b.this_box,
"state": b.state,
"headline": b.headline,
"why": b.why,
"sources": b.sources,
"address": b.address,
"overlay_address": b.overlay_address,
"utility_address": b.utility_address,
"held": b.held,
"twin_state": b.twin_state,
"lag_entries": b.lag_entries,
"last_seen_unix_ms": b.last_seen_unix_ms,
"seen_ago_ms": b.seen_ago_ms,
"disk_measured": b.disk_measured,
"disk_total_bytes": b.disk_total_bytes,
"disk_used_bytes": b.disk_used_bytes,
"cert_not_after_unix": b.cert_not_after_unix,
"pairing_not_after_unix": b.pairing_not_after_unix,
"fetched_total": b.fetched_total,
"fetched_today": b.fetched_today,
"upstream_kind": b.upstream_kind,
"feeds": b.feeds,
})
})
.collect();
let links: Vec<serde_json::Value> = reply
.links
.iter()
.map(|l| json!({ "from": l.from, "to": l.to, "kind": l.kind, "state": l.state }))
.collect();
json!({
"estate": reply.estate,
"summary": reply.summary,
"server_unix_ms": reply.server_unix_ms,
"silent_after_ms": reply.silent_after_ms,
"boxes": boxes,
"links": links,
"consumers": reply.consumers.as_ref().map(|c| json!({
"counted": c.counted,
"principals": c.principals,
"peers": c.peers,
"anonymous_today": c.anonymous_today,
"window_days": c.window_days,
})),
})
}
/// Pure: **the line under the title** (OX-HOLF, the owner on holger.rs's board, 2026-10-27),
/// from the SAME rows the board draws, so the line and the drawing never disagree: the
/// estate's boxes by name (front, main, twins: an upstream or a downstream is no box of this
/// estate), where the bytes come from (the uploads alone are no upstream), every downstream
/// with its last sync or its lag (a declared sync downstream's lag is the sync board's,
/// `sync`), and who used it in the window.
pub fn subtitle(reply: &pb::TopologyReply, sync: &[SyncDownstream]) -> String {
let rows = |role: &'static str| reply.boxes.iter().filter(move |b| b.role == role);
let plural =
|n: usize, one: &str, many: &str| format!("{n} {}", if n == 1 { one } else { many });
let mut parts: Vec<String> = Vec::new();
// The estate's boxes.
let (fronts, mains, twins) = (
rows(ROLE_FRONT).count(),
rows(ROLE_PRIMARY).count(),
rows(ROLE_TWIN).count(),
);
let mut names: Vec<String> = Vec::new();
for (n, one, many) in [
(fronts, "front", "fronts"),
(mains, "main", "mains"),
(twins, "twin", "twins"),
] {
match n {
0 => {}
1 => names.push(one.into()),
n => names.push(format!("{n} {many}")),
}
}
let boxes = fronts + mains + twins;
parts.push(if names.is_empty() {
"no box known".into()
} else {
format!("{}: {}", plural(boxes, "box", "boxes"), names.join(", "))
});
// Where the bytes come from.
let uploads = rows(ROLE_UPSTREAM).any(|b| b.upstream_kind == crate::upstreams::KIND_UPLOAD);
let ups: Vec<&str> = rows(ROLE_UPSTREAM)
.filter(|b| b.upstream_kind != crate::upstreams::KIND_UPLOAD)
.map(|b| b.name.as_str())
.collect();
parts.push(match (ups.len(), uploads) {
(0, true) => "no upstream (uploads only)".into(),
(0, false) => "no upstream".into(),
(n, uploads) => format!(
"{}: {}{}",
plural(n, "upstream", "upstreams"),
ups.join(", "),
if uploads { ", and uploads" } else { "" }
),
});
// Every downstream, with its last sync or its lag.
let downs: Vec<String> = rows(ROLE_DOWNSTREAM)
.map(|d| downstream_words(d, sync.iter().find(|s| s.name == d.name)))
.collect();
parts.push(if downs.is_empty() {
"no downstream".into()
} else {
format!(
"{}: {}",
plural(downs.len(), "downstream", "downstreams"),
downs.join(", ")
)
});
// Who used it.
parts.push(match &reply.consumers {
Some(c) if c.counted => {
let (people, machines) = (c.principals as usize, c.peers as usize);
let who = match (people, machines) {
(0, 0) => "no one".to_string(),
(p, 0) => plural(p, "person", "people"),
(0, m) => plural(m, "machine", "machines"),
(p, m) => format!(
"{} and {}",
plural(p, "person", "people"),
plural(m, "machine", "machines")
),
};
let anonymous = if c.anonymous_today > 0 {
format!(", {} anonymous today", c.anonymous_today)
} else {
String::new()
};
format!("{who} used it in {} days{anonymous}", c.window_days)
}
_ => "consumers not counted in this build".into(),
});
parts.join(" · ")
}
/// Pure: one downstream in the line: behind (a sync downstream's lag as a time, a paired one's
/// in entries), else when it last synced, else never.
fn downstream_words(d: &pb::TopologyBox, sync: Option<&SyncDownstream>) -> String {
// How it reaches this holger (lane HOL-WGDS): `front` unless the row says otherwise.
let name = &format!(
"{} via {}",
d.name,
if d.via.is_empty() {
"front"
} else {
d.via.as_str()
}
);
if let Some(s) = sync.filter(|s| s.lag_generations > 0) {
return format!("{name} {} behind", span(s.lag_secs.max(0) as u64 * 1000));
}
// A sync downstream heard by its mirror's report (lane OX-WGDS2): the writes it is behind.
if let Some(s) = sync.filter(|s| s.entries_behind > 0) {
return format!(
"{name} {} {} behind",
s.entries_behind,
entries(s.entries_behind)
);
}
if sync.is_none() && d.lag_entries > 0 {
return format!("{name} {} {} behind", d.lag_entries, entries(d.lag_entries));
}
if d.last_seen_unix_ms > 0 {
return format!("{name} synced {} ago", span(d.seen_ago_ms));
}
format!("{name} never synced")
}
/// Pure: a span in plain words, days past two days ([`ago`] stops at hours).
pub fn span(ms: u64) -> String {
let h = ms / 3_600_000;
if h >= 48 {
format!("{} days", h / 24)
} else {
ago(ms)
}
}
#[cfg(test)]
mod tests;