crates/gunnar-grpc/src/topology.rs
//! **`Admin.Topology`: the estate's boxes, as THIS gunnar knows them**
//! (owner, 2026-10-04).
//!
//! The console's Machine card draws a picture of the estate, and the owner's
//! rule for the console is that it is a thin layer: it calls and it renders,
//! and it decides nothing. So every decision behind the picture is here — which
//! boxes there are, what each one runs, and whether it is alive — and the
//! console draws the answer as it comes.
//!
//! # Evidence, never inference
//!
//! A box is on the list because this gunnar 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`.
//! * **`pairing`** — the installer's pairing feed (`joined.toml`): the front,
//! the primary and the twins it named. A peer nothing here can ask is
//! `unknown`, which is a different fact from `silent`.
//! * **`heartbeat`** — a twin's `Mirror.LogHead` polls (the primary half of
//! the dual heartbeat), or, on a twin, when its primary last answered. Its
//! state word (`ok`/`behind`/`silent`) is authoritative for that box.
//! * **`replication`** — a configured replication standby and its verdict.
//! * **`caller`** — a MACHINE principal that called this gunnar: a principal
//! the pairing feed says runs on the front (the console, monetize, the
//! guest, the certify key), or any machine that asked for this very picture
//! (a console drawing it is a console).
//! * **`record`** — the ESTATE RECORD the installer fed (owner, 2026-10-04:
//! "the installer has the full picture — at the end of the installation
//! gunnar-server should have all the info"). The kit lays the boxes, so it
//! feeds every box the estate as laid — each box's name, how it was laid
//! (front/main/twin), its public and utility addresses, its hostname and
//! title, its declared size, system disk and data disks — plus the variant
//! and the commits, as the shared feed's `estate.*` keys ([`Record`], one
//! encoding for both sides: [`Record::to_feed`] / [`Record::from_feed`]).
//! The pairing persists the feed (`joined.toml`) and every re-admission
//! renews it (upgrade, recharge, a twin laid again). With a record the
//! answer is EXACTLY the laid boxes, every other piece of evidence merged
//! onto them; without one (a solo gunnar, a box installed before the
//! record) it is what the evidence alone shows.
//!
//! A solo gunnar is one row; a front and a server two; with a twin three, and
//! one more per twin. Nothing here walks a network, resolves a name or dials a
//! box: every input is something the server already recorded.
use std::collections::BTreeMap;
use std::sync::Mutex;
use gunnar_proto::{PrimaryHeartbeat, StandbyStatus, TopologyBox, TopologyResponse, TwinHeartbeat};
pub const ROLE_PRIMARY: &str = "primary";
pub const ROLE_TWIN: &str = "twin";
pub const ROLE_FRONT: &str = "front";
pub const ROLE_CONSOLE: &str = "console";
pub const ROLE_MONETIZE: &str = "monetize";
pub const ROLE_STANDBY: &str = "standby";
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 SERVICE_GUNNAR: &str = "gunnar";
pub const SERVICE_CONSOLE: &str = "console";
pub const SERVICE_MONETIZE: &str = "monetize";
pub const SOURCE_SELF: &str = "self";
pub const SOURCE_PAIRING: &str = "pairing";
pub const SOURCE_HEARTBEAT: &str = "heartbeat";
pub const SOURCE_REPLICATION: &str = "replication";
pub const SOURCE_CALLER: &str = "caller";
pub const SOURCE_RECORD: &str = "record";
// ── the estate record's feed keys (the kit writes them, gunnar reads them) ──
/// The record's schema; a box fed a schema this build does not know keeps no
/// record (and says so in its answer: `recorded` false).
pub const FEED_RECORD: &str = "estate.record";
pub const RECORD_SCHEMA: &str = "1";
pub const FEED_VARIANT: &str = "estate.variant";
/// `<repository>=<commit>` pairs, comma-separated.
pub const FEED_VERSIONS: &str = "estate.versions";
/// The laid boxes' names, comma-separated, in lay order.
pub const FEED_BOXES: &str = "estate.boxes";
/// `estate.box.<name>.<field>`.
pub const FEED_BOX_PREFIX: &str = "estate.box.";
/// **One box as the installer laid it** — declared, never measured.
#[derive(Debug, Clone, Default, PartialEq, Eq)]
pub struct RecordBox {
/// The pairing name (`front`, `main`, `twin`).
pub name: String,
/// How the kit laid it: `front` | `main` | `twin`.
pub laid_as: String,
pub hostname: String,
pub title: String,
pub public: String,
pub utility: String,
/// The provider's size name (`STARTER-4xCPU-16GB`).
pub size: String,
pub vcpu: u32,
pub ram_mb: u64,
pub system_disk_gb: u64,
/// The data disks in words: `4 × 10 GB maxiops, md-raid0`.
pub data: String,
}
/// **The estate as the installer laid it.**
#[derive(Debug, Clone, Default, PartialEq, Eq)]
pub struct Record {
pub variant: String,
/// `(repository, commit)`, in the order fed.
pub versions: Vec<(String, String)>,
pub boxes: Vec<RecordBox>,
}
const BOX_FIELDS: [&str; 10] = [
"laid_as",
"hostname",
"title",
"public",
"utility",
"size",
"vcpu",
"ram_mb",
"system_disk_gb",
"data",
];
impl Record {
/// The record as feed keys — what the kit puts in the shared feed.
pub fn to_feed(&self) -> BTreeMap<String, String> {
let mut f = BTreeMap::new();
f.insert(FEED_RECORD.to_owned(), RECORD_SCHEMA.to_owned());
f.insert(FEED_VARIANT.to_owned(), self.variant.clone());
f.insert(
FEED_VERSIONS.to_owned(),
self.versions
.iter()
.map(|(r, c)| format!("{r}={c}"))
.collect::<Vec<_>>()
.join(","),
);
f.insert(
FEED_BOXES.to_owned(),
self.boxes
.iter()
.map(|b| b.name.clone())
.collect::<Vec<_>>()
.join(","),
);
for b in &self.boxes {
let values = [
b.laid_as.clone(),
b.hostname.clone(),
b.title.clone(),
b.public.clone(),
b.utility.clone(),
b.size.clone(),
b.vcpu.to_string(),
b.ram_mb.to_string(),
b.system_disk_gb.to_string(),
b.data.clone(),
];
for (field, value) in BOX_FIELDS.iter().zip(values) {
f.insert(format!("{FEED_BOX_PREFIX}{}.{field}", b.name), value);
}
}
f
}
/// **The record a feed carries**, or `None` — no `estate.record` key, or a
/// schema this build does not read. A box name the feed lists is a box
/// even when its fields are missing (they read as not declared).
pub fn from_feed(feed: &BTreeMap<String, String>) -> Option<Record> {
if feed.get(FEED_RECORD).map(String::as_str) != Some(RECORD_SCHEMA) {
return None;
}
let get = |k: &str| feed.get(k).cloned().unwrap_or_default();
let versions = get(FEED_VERSIONS)
.split(',')
.filter_map(|p| {
p.split_once('=')
.map(|(r, c)| (r.trim().to_owned(), c.trim().to_owned()))
})
.filter(|(r, c)| !r.is_empty() && !c.is_empty())
.collect();
let boxes = get(FEED_BOXES)
.split(',')
.map(str::trim)
.filter(|n| !n.is_empty())
.map(|name| {
let f = |field: &str| get(&format!("{FEED_BOX_PREFIX}{name}.{field}"));
RecordBox {
name: name.to_owned(),
laid_as: f("laid_as"),
hostname: f("hostname"),
title: f("title"),
public: f("public"),
utility: f("utility"),
size: f("size"),
vcpu: f("vcpu").parse().unwrap_or(0),
ram_mb: f("ram_mb").parse().unwrap_or(0),
system_disk_gb: f("system_disk_gb").parse().unwrap_or(0),
data: f("data"),
}
})
.collect();
Some(Record {
variant: get(FEED_VARIANT),
versions,
boxes,
})
}
}
/// **How long a caller's last call keeps its box `up`.** Longer than the
/// twin's poll bound on purpose: a twin polls every few seconds, a console
/// calls when a person does something, and ten quiet minutes on a console is
/// an idle evening, not an outage. The row's sentence always says how long ago.
pub const CALLER_SILENT_AFTER_MS: u64 = 10 * 60_000;
/// How many distinct machine principals the caller log keeps. Bounded, like
/// every roster on this listener: past it the oldest sighting goes.
pub const CALLERS_CAPACITY: usize = 64;
/// One machine principal's newest call.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct Sighting {
pub principal: String,
/// `ip:port` of the connection, empty when the transport gave none.
pub address: String,
pub last_unix_ms: i64,
/// It has asked for `Admin.Topology` — a console drawing the picture.
pub asked_topology: bool,
}
/// **The machine principals that call this listener**, newest call each.
/// Written by [`crate::ServiceContext::caller`] for every verified machine
/// caller, read by `Admin.Topology`. People are not recorded: a person's
/// laptop is not a box of the estate.
#[derive(Debug, Default)]
pub struct CallerLog {
seen: Mutex<BTreeMap<String, Sighting>>,
}
impl CallerLog {
/// Record one call. `topology` marks the caller as one that asked for the
/// picture (it stays marked).
pub fn note(&self, principal: &str, address: &str, now_ms: i64, topology: bool) {
let Ok(mut seen) = self.seen.lock() else {
return;
};
if !seen.contains_key(principal) && seen.len() >= CALLERS_CAPACITY {
if let Some(oldest) = seen
.values()
.min_by_key(|s| s.last_unix_ms)
.map(|s| s.principal.clone())
{
seen.remove(&oldest);
}
}
let row = seen
.entry(principal.to_owned())
.or_insert_with(|| Sighting {
principal: principal.to_owned(),
address: String::new(),
last_unix_ms: 0,
asked_topology: false,
});
if now_ms >= row.last_unix_ms {
row.last_unix_ms = now_ms;
if !address.is_empty() {
row.address = address.to_owned();
}
}
row.asked_topology |= topology;
}
/// Every sighting, in principal order.
pub fn sightings(&self) -> Vec<Sighting> {
self.seen
.lock()
.map(|s| s.values().cloned().collect())
.unwrap_or_default()
}
}
/// A peer the installer's pairing feed names.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct Peer {
pub name: String,
/// `front` | `primary` | `twin`.
pub role: String,
/// Every address the feed gives it (utility first, then the tunnel), bare
/// hosts — what a caller's or a heartbeat's address is matched against.
pub hosts: Vec<String>,
}
/// **What the control plane knows and the service does not**: the pairing
/// feed, which principals run on the front, the plugin registration, this
/// box's own data volume. [`crate::seam::ControlPlane::topology_facts`].
#[derive(Debug, Clone, Default, PartialEq, Eq)]
pub struct Facts {
/// The estate's name from the pairing feed; empty when unpaired.
pub estate: String,
/// This box's own pairing name; empty when unpaired.
pub me: String,
/// Where this box is: its utility (or tunnel) address, or its bind.
pub address: String,
pub peers: Vec<Peer>,
/// Principal id → the service it is on the front (`console`,
/// `monetize`, …): the principals the pairing feed's front offered.
pub front_principals: BTreeMap<String, String>,
/// The estate record the installer fed, when there is one.
pub record: Option<Record>,
/// This box's data volume `(total, used)`, when measured.
pub disk: Option<(u64, u64)>,
}
/// Everything [`assemble`] reads.
#[derive(Debug, Clone, Default)]
pub struct Input {
pub facts: Facts,
pub is_twin: bool,
pub twins: Vec<TwinHeartbeat>,
pub primary: Option<PrimaryHeartbeat>,
pub standbys: Vec<StandbyStatus>,
pub callers: Vec<Sighting>,
pub now_ms: i64,
/// The twin heartbeat's silence bound, reported on the answer.
pub silent_after_ms: u64,
}
/// `10.0.0.5:4242` → `10.0.0.5`; `[::1]:50051` → `::1`;
/// `https://host:8443/x` → `host`; a bare host stays itself.
pub fn host_of(address: &str) -> String {
let a = address.trim();
let a = a.split_once("://").map(|(_, rest)| rest).unwrap_or(a);
let a = a.split('/').next().unwrap_or(a);
let a = a.rsplit_once('@').map(|(_, h)| h).unwrap_or(a);
if let Some(rest) = a.strip_prefix('[') {
return rest.split(']').next().unwrap_or(rest).to_owned();
}
match a.rsplit_once(':') {
// One colon is host:port; more is a bare IPv6 address.
Some((host, port)) if !host.contains(':') && port.chars().all(|c| c.is_ascii_digit()) => {
host.to_owned()
}
_ => a.to_owned(),
}
}
/// One piece of evidence about a box.
#[derive(Debug, Clone)]
struct Evidence {
state: &'static str,
at_ms: i64,
why: String,
/// A heartbeat's word decides the box's state over everything else.
authoritative: bool,
/// The state in a few words, as this evidence says it ([`TopologyBox::headline`]).
headline: String,
}
#[derive(Debug, Clone)]
struct Draft {
row: TopologyBox,
hosts: Vec<String>,
evidence: Vec<Evidence>,
}
impl Draft {
fn new(name: &str, role: &str, address: &str, source: &str) -> Draft {
let host = host_of(address);
Draft {
row: TopologyBox {
name: name.to_owned(),
role: role.to_owned(),
address: address.to_owned(),
sources: vec![source.to_owned()],
..Default::default()
},
hosts: if host.is_empty() {
Vec::new()
} else {
vec![host]
},
evidence: Vec::new(),
}
}
fn has_host(&self, host: &str) -> bool {
!host.is_empty() && self.hosts.iter().any(|h| h == host)
}
fn source(&mut self, source: &str) {
if !self.row.sources.iter().any(|s| s == source) {
self.row.sources.push(source.to_owned());
}
}
fn service(&mut self, service: &str) {
if !service.is_empty() && !self.row.services.iter().any(|s| s == service) {
self.row.services.push(service.to_owned());
}
}
/// What the installer laid, onto this row.
fn record(&mut self, b: &RecordBox) {
self.source(SOURCE_RECORD);
let r = &mut self.row;
r.laid_as = b.laid_as.clone();
r.public_address = b.public.clone();
r.utility_address = b.utility.clone();
r.hostname = b.hostname.clone();
r.title = b.title.clone();
r.provider_size = b.size.clone();
r.declared_vcpu = b.vcpu;
r.declared_ram_mb = b.ram_mb;
r.declared_system_disk_gb = b.system_disk_gb;
r.declared_data = b.data.clone();
for a in [&b.utility, &b.public] {
if !a.is_empty() {
self.at(a);
}
}
}
fn at(&mut self, address: &str) {
let host = host_of(address);
if self.row.address.is_empty() {
self.row.address = address.to_owned();
}
if !host.is_empty() && !self.has_host(&host) {
self.hosts.push(host);
}
}
}
/// **How long, coarsely** — whole minutes once past one, whole hours past
/// two — so a headline changes with the state and not with the clock.
fn coarse(ms: u64) -> String {
let s = ms / 1000;
if s < 60 {
"under a minute".to_owned()
} else if s < 7200 {
format!("{} min", s / 60)
} else {
format!("{} h", s / 3600)
}
}
/// A heartbeat's headline: the state word, the lag, and the twin's OWN word
/// when it is not plain following.
fn heartbeat_headline(
state: &str,
last_seen_ms: i64,
ago_ms: u64,
lag: u64,
own: &str,
ok_word: &str,
) -> String {
let head = match state {
STATE_UP => ok_word.to_owned(),
STATE_BEHIND => format!("behind {lag} entries"),
_ if last_seen_ms == 0 => "never heard".to_owned(),
_ => format!("silent for {}", coarse(ago_ms)),
};
let own = own.trim();
if state != STATE_SILENT && !own.is_empty() && own != "following" && own != "idle" {
format!("{head} · {own}")
} else {
head
}
}
fn ago_words(ms: u64) -> String {
let s = ms / 1000;
if s < 120 {
format!("{s} s ago")
} else if s < 7200 {
format!("{} min ago", s / 60)
} else {
format!("{} h ago", s / 3600)
}
}
/// A standby's verdict word as a box state.
fn standby_state(verdict: &str, never: bool) -> &'static str {
match verdict {
"NEVER" | "STALE" => STATE_SILENT,
"BEHIND" => STATE_BEHIND,
"syncing" | "idle" => STATE_UP,
_ if never => STATE_SILENT,
_ => STATE_UNKNOWN,
}
}
fn role_rank(role: &str) -> u8 {
match role {
ROLE_PRIMARY => 0,
ROLE_TWIN => 1,
ROLE_STANDBY => 2,
ROLE_FRONT => 3,
ROLE_CONSOLE => 4,
ROLE_MONETIZE => 5,
_ => 6,
}
}
/// **The answer.** Pure: the same input always gives the same rows.
pub fn assemble(input: &Input) -> TopologyResponse {
let facts = &input.facts;
let now = input.now_ms;
let mut boxes: Vec<Draft> = Vec::new();
// ── 1. this box ────────────────────────────────────────────────────────
let mut me = Draft::new(
if facts.me.is_empty() {
"this box"
} else {
&facts.me
},
if input.is_twin {
ROLE_TWIN
} else {
ROLE_PRIMARY
},
&facts.address,
SOURCE_SELF,
);
me.row.this_box = true;
me.service(SERVICE_GUNNAR);
if let Some((total, used)) = facts.disk {
me.row.disk_measured = true;
me.row.disk_total_bytes = total;
me.row.disk_used_bytes = used;
}
me.evidence.push(Evidence {
state: STATE_UP,
at_ms: now,
why: "this box answered this call".into(),
authoritative: true,
headline: "answering".into(),
});
boxes.push(me);
// ── 1b. the estate record: every box the installer laid ────────────────
if let Some(rec) = &facts.record {
for b in &rec.boxes {
let i = if !facts.me.is_empty() && b.name == facts.me {
0
} else {
// How it was laid; the pairing's election (step 2) has the
// last word on primary and twin.
let role = match b.laid_as.as_str() {
"front" => ROLE_FRONT,
"twin" => ROLE_TWIN,
_ => ROLE_PRIMARY,
};
boxes.push(Draft::new(&b.name, role, "", SOURCE_RECORD));
boxes.len() - 1
};
let d = &mut boxes[i];
d.record(b);
if b.laid_as != ROLE_FRONT {
d.service(SERVICE_GUNNAR);
}
}
}
// ── 2. the pairing feed's peers ────────────────────────────────────────
for p in &facts.peers {
if !facts.me.is_empty() && p.name == facts.me {
continue;
}
// A box the record named is the same box: the pairing's elected role
// and its addresses merge onto it.
let i = match boxes
.iter()
.position(|b| !b.row.this_box && b.row.name == p.name)
{
Some(i) => {
boxes[i].row.role = p.role.clone();
boxes[i].source(SOURCE_PAIRING);
i
}
None => {
boxes.push(Draft::new(
&p.name,
&p.role,
p.hosts.first().map(String::as_str).unwrap_or(""),
SOURCE_PAIRING,
));
boxes.len() - 1
}
};
let d = &mut boxes[i];
for h in &p.hosts {
d.at(h);
}
if p.role == ROLE_PRIMARY || p.role == ROLE_TWIN {
d.service(SERVICE_GUNNAR);
}
}
// ── 3. the twins this primary hears ────────────────────────────────────
for t in &input.twins {
let host = host_of(&t.address);
let found = boxes
.iter()
.position(|b| !b.row.this_box && b.row.role == ROLE_TWIN && b.has_host(&host))
.or_else(|| {
// A twin never heard from has no address. If the feed names
// exactly one twin nothing else has claimed, it is that one.
if !host.is_empty() {
return None;
}
let open: Vec<usize> = boxes
.iter()
.enumerate()
.filter(|(_, b)| {
!b.row.this_box
&& b.row.role == ROLE_TWIN
&& !b.row.sources.iter().any(|s| s == SOURCE_HEARTBEAT)
})
.map(|(i, _)| i)
.collect();
(open.len() == 1).then(|| open[0])
});
let i = match found {
Some(i) => i,
None => {
boxes.push(Draft::new(
&t.principal,
ROLE_TWIN,
&t.address,
SOURCE_HEARTBEAT,
));
boxes.len() - 1
}
};
let d = &mut boxes[i];
d.source(SOURCE_HEARTBEAT);
d.service(SERVICE_GUNNAR);
d.at(&t.address);
if t.disk_total_bytes > 0 {
d.row.disk_measured = true;
d.row.disk_total_bytes = t.disk_total_bytes;
d.row.disk_used_bytes = t.disk_used_bytes;
}
let state = match t.state.as_str() {
"ok" => STATE_UP,
"behind" => STATE_BEHIND,
_ => STATE_SILENT,
};
let why = if t.last_seen_unix_ms == 0 {
format!(
"twin `{}` has not polled this primary since it started: {}",
t.principal, t.why
)
} else {
format!(
"twin `{}` polled {} (lag {} entries): {}",
t.principal,
ago_words(t.seen_ago_ms),
t.lag_entries,
t.why
)
};
let headline = heartbeat_headline(
state,
t.last_seen_unix_ms,
t.seen_ago_ms,
t.lag_entries,
&t.twin_state,
"following",
);
d.evidence.push(Evidence {
state,
at_ms: t.last_seen_unix_ms,
why,
authoritative: true,
headline,
});
}
// ── 4. on a twin: its primary ──────────────────────────────────────────
if let Some(p) = &input.primary {
let host = host_of(&p.address);
let i = match boxes.iter().position(|b| {
!b.row.this_box && b.row.role == ROLE_PRIMARY && (b.has_host(&host) || host.is_empty())
}) {
Some(i) => i,
None => {
boxes.push(Draft::new(
if p.address.is_empty() {
"primary"
} else {
&host
},
ROLE_PRIMARY,
&p.address,
SOURCE_HEARTBEAT,
));
boxes.len() - 1
}
};
let d = &mut boxes[i];
d.source(SOURCE_HEARTBEAT);
d.service(SERVICE_GUNNAR);
d.at(&p.address);
let state = if p.state == "ok" {
STATE_UP
} else {
STATE_SILENT
};
let why = if p.last_answer_unix_ms == 0 {
format!(
"the primary has not answered this twin since it started: {}",
p.why
)
} else {
format!(
"the primary answered {}: {}",
ago_words(p.answer_ago_ms),
p.why
)
};
let headline = heartbeat_headline(
state,
p.last_answer_unix_ms,
p.answer_ago_ms,
0,
"",
"answering",
);
d.evidence.push(Evidence {
state,
at_ms: p.last_answer_unix_ms,
why,
authoritative: true,
headline,
});
}
// ── 5. replication standbys ────────────────────────────────────────────
for s in &input.standbys {
let host = host_of(&s.remote);
let i = match boxes
.iter()
.position(|b| !b.row.this_box && b.has_host(&host))
{
Some(i) => i,
None => {
boxes.push(Draft::new(
&s.name,
ROLE_STANDBY,
&s.remote,
SOURCE_REPLICATION,
));
boxes.len() - 1
}
};
let d = &mut boxes[i];
d.source(SOURCE_REPLICATION);
d.service(SERVICE_GUNNAR);
let state = standby_state(&s.verdict, s.last_success_unix_ms == 0);
let why = format!(
"replication standby `{}`: {}{}",
s.name,
if s.verdict.is_empty() {
"no verdict"
} else {
&s.verdict
},
if s.last_error.is_empty() {
String::new()
} else {
format!(" — {}", s.last_error)
}
);
let headline = match s.verdict.as_str() {
"NEVER" => "never synced".to_owned(),
"" => "no verdict".to_owned(),
v => v.to_lowercase(),
};
d.evidence.push(Evidence {
state,
at_ms: s.last_success_unix_ms,
why,
authoritative: false,
headline,
});
}
// ── 6. the machines that call this gunnar ──────────────────────────────
let front = boxes
.iter()
.position(|b| !b.row.this_box && b.row.role == ROLE_FRONT);
for c in &input.callers {
let service = match facts.front_principals.get(&c.principal) {
Some(service) => service.clone(),
None if c.asked_topology => SERVICE_CONSOLE.to_owned(),
None => continue,
};
let host = host_of(&c.address);
let on_front = facts.front_principals.contains_key(&c.principal);
let i = match (on_front, front) {
(true, Some(i)) => i,
_ => match boxes.iter().position(|b| {
!b.row.this_box
&& matches!(
b.row.role.as_str(),
ROLE_FRONT | ROLE_CONSOLE | ROLE_MONETIZE
)
&& b.has_host(&host)
}) {
Some(i) => i,
None => {
let role = if service == SERVICE_MONETIZE {
ROLE_MONETIZE
} else {
ROLE_CONSOLE
};
boxes.push(Draft::new(&c.principal, role, &c.address, SOURCE_CALLER));
boxes.len() - 1
}
},
};
let d = &mut boxes[i];
d.source(SOURCE_CALLER);
if service != ROLE_FRONT {
d.service(&service);
}
d.at(&c.address);
let ago = now.saturating_sub(c.last_unix_ms).max(0) as u64;
let state = if ago <= CALLER_SILENT_AFTER_MS {
STATE_UP
} else {
STATE_SILENT
};
let from = if c.address.is_empty() {
String::new()
} else {
format!(" from {}", c.address)
};
d.evidence.push(Evidence {
state,
at_ms: c.last_unix_ms,
why: format!(
"`{}` called this gunnar {}{from}",
c.principal,
ago_words(ago)
),
authoritative: false,
headline: if state == STATE_UP {
"calling".to_owned()
} else {
format!("silent for {}", coarse(ago))
},
});
}
// ── the verdicts ───────────────────────────────────────────────────────
let mut rows: Vec<TopologyBox> = boxes
.into_iter()
.map(|mut d| {
let authoritative: Vec<&Evidence> = d.evidence.iter().filter(|e| e.authoritative).collect();
let state = if let Some(e) = authoritative.first() {
e.state
} else if d.evidence.iter().any(|e| e.state == STATE_UP) {
STATE_UP
} else if d.evidence.iter().any(|e| e.state == STATE_BEHIND) {
STATE_BEHIND
} else if d.evidence.iter().any(|e| e.state == STATE_SILENT) {
STATE_SILENT
} else {
STATE_UNKNOWN
};
d.row.state = state.to_owned();
// The words of the evidence that decided the state (the first
// authoritative one, else the first saying that state).
d.row.headline = authoritative
.first()
.map(|e| e.headline.clone())
.or_else(|| d.evidence.iter().find(|e| e.state == state).map(|e| e.headline.clone()))
.unwrap_or_else(|| "not asked".to_owned());
let last = d.evidence.iter().map(|e| e.at_ms).max().unwrap_or(0);
d.row.last_seen_unix_ms = last;
d.row.seen_ago_ms = if last > 0 { now.saturating_sub(last).max(0) as u64 } else { 0 };
d.row.why = if d.evidence.is_empty() {
"laid by the installer (its record and pairing feed); nothing on this box can ask it whether it is up".to_owned()
} else {
d.evidence.iter().map(|e| e.why.as_str()).collect::<Vec<_>>().join("; ")
};
d.row
})
.collect();
let this = rows.remove(0);
rows.sort_by(|a, b| (role_rank(&a.role), &a.name).cmp(&(role_rank(&b.role), &b.name)));
rows.insert(0, this);
let (recorded, variant, versions) = match &facts.record {
Some(r) => (
true,
r.variant.clone(),
r.versions
.iter()
.map(|(repo, commit)| format!("{repo} {commit}"))
.collect(),
),
None => (false, String::new(), Vec::new()),
};
TopologyResponse {
boxes: rows,
server_unix_ms: now,
silent_after_ms: input.silent_after_ms,
estate: facts.estate.clone(),
recorded,
variant,
versions,
}
}
#[cfg(test)]
mod tests {
use super::*;
const NOW: i64 = 1_800_000_000_000;
fn twin(principal: &str, address: &str, state: &str, seen_ago_ms: u64) -> TwinHeartbeat {
TwinHeartbeat {
principal: principal.into(),
address: address.into(),
last_seen_unix_ms: if address.is_empty() {
0
} else {
NOW - seen_ago_ms as i64
},
seen_ago_ms,
state: state.into(),
why: format!("{state} (test)"),
disk_total_bytes: 20 << 30,
disk_used_bytes: 1 << 30,
..Default::default()
}
}
fn caller(principal: &str, address: &str, ago_ms: i64, topology: bool) -> Sighting {
Sighting {
principal: principal.into(),
address: address.into(),
last_unix_ms: NOW - ago_ms,
asked_topology: topology,
}
}
fn input() -> Input {
Input {
now_ms: NOW,
silent_after_ms: 60_000,
..Default::default()
}
}
fn shape(r: &TopologyResponse) -> Vec<(String, String, String)> {
r.boxes
.iter()
.map(|b| (b.name.clone(), b.role.clone(), b.state.clone()))
.collect()
}
/// ★ **A solo gunnar is ONE box** — itself, up, the primary, first.
#[test]
fn a_solo_gunnar_is_one_box_and_it_is_up() {
let mut i = input();
i.facts.address = "127.0.0.1:50051".into();
i.facts.disk = Some((100, 40));
let r = assemble(&i);
assert_eq!(
shape(&r),
[("this box".into(), "primary".into(), "up".into())]
);
let b = &r.boxes[0];
assert!(
b.this_box && b.disk_measured && b.disk_total_bytes == 100 && b.disk_used_bytes == 40
);
assert_eq!(b.services, ["gunnar"]);
assert_eq!(b.sources, ["self"]);
assert_eq!(r.server_unix_ms, NOW);
}
/// ★ **A console that asks for the picture is a box of its own** — a front
/// and a server are TWO boxes. A machine that never asked (a builder's
/// key, a twin's principal) is not drawn as a console, and a person is
/// never in the caller log at all.
#[test]
fn a_console_that_asks_is_the_second_box_and_other_machines_are_not() {
let mut i = input();
i.callers = vec![
caller("uitest-console", "127.0.0.1:41000", 0, true),
caller("dwarves-oden", "10.0.0.9:5000", 1000, false),
];
let r = assemble(&i);
assert_eq!(
shape(&r),
[
("this box".into(), "primary".into(), "up".into()),
("uitest-console".into(), "console".into(), "up".into())
]
);
assert_eq!(r.boxes[1].services, ["console"]);
assert_eq!(r.boxes[1].address, "127.0.0.1:41000");
assert!(
r.boxes[1].why.contains("uitest-console") && r.boxes[1].why.contains("127.0.0.1:41000"),
"{}",
r.boxes[1].why
);
}
/// ★ **With a twin it is three boxes**, and one more per twin. The twin's
/// own heartbeat word decides its state: `behind` stays `behind`, a twin
/// never heard from is `silent`, never `up`.
#[test]
fn each_twin_is_one_more_box_with_its_heartbeat_state() {
let mut i = input();
i.callers = vec![caller("gunnar-ui", "10.13.8.10:40000", 5_000, true)];
i.twins = vec![twin("gunnar-twin", "10.13.8.122:5222", "ok", 2_000)];
assert_eq!(assemble(&i).boxes.len(), 3, "front + server + twin");
i.twins
.push(twin("gunnar-twin-2", "10.13.8.123:5222", "behind", 2_000));
i.twins.push(twin("gunnar-twin-3", "", "silent", 0));
let r = assemble(&i);
assert_eq!(r.boxes.len(), 5);
let state = |name: &str| {
r.boxes
.iter()
.find(|b| b.name == name)
.map(|b| b.state.clone())
.unwrap()
};
assert_eq!(state("gunnar-twin"), "up");
assert_eq!(state("gunnar-twin-2"), "behind");
assert_eq!(state("gunnar-twin-3"), "silent");
let t = r.boxes.iter().find(|b| b.name == "gunnar-twin").unwrap();
assert!(
t.disk_measured && t.disk_total_bytes == 20 << 30,
"the twin's own statvfs travels"
);
assert_eq!(
r.boxes.iter().map(|b| b.role.as_str()).collect::<Vec<_>>(),
["primary", "twin", "twin", "twin", "console"],
"this box first, then by role"
);
}
/// ★ **The estate as the installer paired it**: the front's principals
/// (console, monetize) and the plugin registration land ON the front
/// peer — one front box, not three — and a twin's heartbeat lands on the
/// twin peer the feed named, matched by address.
#[test]
fn the_paired_estate_is_front_main_twin_and_the_fronts_services_merge() {
let mut i = input();
i.facts.estate = "test2.gunnar.rs".into();
i.facts.me = "vm2".into();
i.facts.address = "10.13.8.121".into();
i.facts.peers = vec![
Peer {
name: "front".into(),
role: "front".into(),
hosts: vec!["10.99.0.1".into()],
},
Peer {
name: "vm2".into(),
role: "primary".into(),
hosts: vec!["10.13.8.121".into()],
},
Peer {
name: "vm3".into(),
role: "twin".into(),
hosts: vec!["10.13.8.122".into(), "10.99.0.3".into()],
},
];
i.facts.front_principals = [
("gunnar-ui".to_string(), "console".to_string()),
("monetize".to_string(), "monetize".to_string()),
]
.into_iter()
.collect();
i.callers = vec![
caller("gunnar-ui", "10.13.8.120:40000", 1_000, true),
caller("monetize", "10.13.8.120:40001", 20_000, false),
];
i.twins = vec![twin("gunnar-twin", "10.13.8.122:5222", "ok", 2_000)];
let r = assemble(&i);
assert_eq!(r.estate, "test2.gunnar.rs");
assert_eq!(
shape(&r),
[
("vm2".into(), "primary".into(), "up".into()),
("vm3".into(), "twin".into(), "up".into()),
("front".into(), "front".into(), "up".into()),
]
);
let front = &r.boxes[2];
assert_eq!(front.services, ["console", "monetize"]);
for s in ["pairing", "caller"] {
assert!(
front.sources.iter().any(|x| x == s),
"{s} in {:?}",
front.sources
);
}
assert!(
r.boxes[1].sources.iter().any(|s| s == "heartbeat"),
"the twin's poll landed on vm3"
);
}
/// ★ **A peer nothing here can ask is `unknown`, not `silent` and not
/// `up`**; a console that has not called for longer than the bound is
/// `silent`, with how long ago in its sentence.
#[test]
fn unknown_is_not_silent_and_an_old_call_is_silent() {
let mut i = input();
i.facts.peers = vec![Peer {
name: "front".into(),
role: "front".into(),
hosts: vec![],
}];
assert_eq!(assemble(&i).boxes[1].state, "unknown");
let mut i = input();
i.callers = vec![caller(
"old-console",
"10.0.0.7:1",
(CALLER_SILENT_AFTER_MS + 60_000) as i64,
true,
)];
let r = assemble(&i);
assert_eq!(r.boxes[1].state, "silent");
assert!(r.boxes[1].why.contains("min ago"), "{}", r.boxes[1].why);
}
/// ★ **On a twin: its primary is a box**, from when it last answered.
#[test]
fn a_twin_draws_its_primary_from_the_primarys_answers() {
let mut i = input();
i.is_twin = true;
i.primary = Some(PrimaryHeartbeat {
address: "10.13.8.121:50051".into(),
last_answer_unix_ms: NOW - 3_000,
answer_ago_ms: 3_000,
state: "ok".into(),
why: "ok".into(),
..Default::default()
});
let r = assemble(&i);
assert_eq!(
shape(&r),
[
("this box".into(), "twin".into(), "up".into()),
("10.13.8.121".into(), "primary".into(), "up".into())
]
);
i.primary.as_mut().unwrap().state = "silent".into();
assert_eq!(assemble(&i).boxes[1].state, "silent");
}
/// A replication standby with no other evidence is a box of its own; a
/// standby that has never taken a byte is `silent`.
#[test]
fn a_standby_is_a_box_and_never_is_silent() {
let mut i = input();
i.standbys = vec![StandbyStatus {
name: "dr".into(),
remote: "ssh://git@10.0.0.40:2222".into(),
verdict: "NEVER".into(),
..Default::default()
}];
let r = assemble(&i);
assert_eq!(
shape(&r)[1],
("dr".into(), "standby".into(), "silent".into())
);
}
#[test]
fn hosts_are_read_out_of_every_address_shape() {
assert_eq!(host_of("10.0.0.5:4242"), "10.0.0.5");
assert_eq!(host_of("[::1]:50051"), "::1");
assert_eq!(host_of("https://host.example:8443/x"), "host.example");
assert_eq!(host_of("ssh://git@10.0.0.40:2222"), "10.0.0.40");
assert_eq!(host_of("10.0.0.5"), "10.0.0.5");
assert_eq!(host_of("fe80::1"), "fe80::1");
assert_eq!(host_of(""), "");
}
/// The caller log keeps the NEWEST call per principal, keeps the
/// topology mark once set, and is bounded.
#[test]
fn the_caller_log_is_newest_per_principal_and_bounded() {
let log = CallerLog::default();
log.note("c", "1.1.1.1:1", 10, true);
log.note("c", "1.1.1.1:2", 20, false);
log.note("c", "1.1.1.1:3", 15, false);
let s = log.sightings();
assert_eq!(s.len(), 1);
assert_eq!(
(
s[0].address.as_str(),
s[0].last_unix_ms,
s[0].asked_topology
),
("1.1.1.1:2", 20, true)
);
for n in 0..(CALLERS_CAPACITY + 5) {
log.note(&format!("m{n}"), "", 100 + n as i64, false);
}
let s = log.sightings();
assert_eq!(s.len(), CALLERS_CAPACITY);
assert!(
!s.iter().any(|x| x.principal == "c"),
"the oldest went first"
);
}
fn laid() -> Record {
let b = |name: &str,
laid_as: &str,
public: &str,
utility: &str,
size: &str,
vcpu: u32,
ram_mb: u64,
data: &str| RecordBox {
name: name.into(),
laid_as: laid_as.into(),
hostname: format!("gunnar-{name}"),
title: format!("the {name}"),
public: public.into(),
utility: utility.into(),
size: size.into(),
vcpu,
ram_mb,
system_disk_gb: 40,
data: data.into(),
};
Record {
variant: "gunnar-twin-one".into(),
versions: vec![
("gunnar".into(), "abc123".into()),
("gunnar-ui".into(), "def456".into()),
],
boxes: vec![
b(
"front",
"front",
"203.0.113.9",
"10.13.8.99",
"STARTER-4xCPU-8GB",
4,
8192,
"",
),
b(
"main",
"main",
"203.0.113.10",
"10.13.8.101",
"STARTER-4xCPU-16GB",
4,
16384,
"4 × 10 GB maxiops, md-raid0",
),
b(
"twin",
"twin",
"203.0.113.11",
"10.13.8.122",
"STARTER-1xCPU-2GB",
1,
2048,
"1 × 20 GB hdd, xfs",
),
],
}
}
/// ★ **The record is one encoding for both sides**: what the kit feeds is
/// what gunnar reads, field for field; a feed with no record, or a schema
/// this build does not read, is no record (never a half one).
#[test]
fn the_estate_record_round_trips_through_the_feed() {
let r = laid();
let feed = r.to_feed();
assert_eq!(
feed.get(FEED_RECORD).map(String::as_str),
Some(RECORD_SCHEMA)
);
assert_eq!(
feed.get("estate.box.main.data").map(String::as_str),
Some("4 × 10 GB maxiops, md-raid0")
);
assert!(
feed.keys().all(|k| k.starts_with("estate.")),
"the record stays in its own namespace of the shared feed"
);
assert_eq!(Record::from_feed(&feed), Some(r));
assert_eq!(Record::from_feed(&BTreeMap::new()), None);
let mut newer = feed.clone();
newer.insert(FEED_RECORD.into(), "2".into());
assert_eq!(
Record::from_feed(&newer),
None,
"a schema this build does not read is no record"
);
}
/// ★ **With the installer's record, the topology is EXACTLY the laid
/// boxes** (owner, 2026-10-04): the front, main and the twin — with what
/// the kit declared (addresses, size, disks), the pairing's elected roles,
/// and the live evidence merged on: the console and the plugin
/// registration on the front, the twin's poll on the twin. No extra box.
#[test]
fn with_the_record_the_topology_is_exactly_the_laid_boxes() {
let mut i = input();
i.facts.estate = "test2.gunnar.rs".into();
i.facts.me = "main".into();
i.facts.address = "10.13.8.101".into();
i.facts.record = Some(laid());
i.facts.peers = vec![
Peer {
name: "front".into(),
role: "front".into(),
hosts: vec!["10.13.8.99".into(), "10.99.0.1".into()],
},
Peer {
name: "twin".into(),
role: "twin".into(),
hosts: vec!["10.13.8.122".into(), "10.99.0.3".into()],
},
];
i.facts.front_principals = [
("gunnar-ui".to_string(), "console".to_string()),
("monetize".to_string(), "monetize".to_string()),
]
.into_iter()
.collect();
i.callers = vec![caller("gunnar-ui", "10.13.8.99:40000", 1_000, true)];
i.twins = vec![twin("gunnar-twin", "10.13.8.122:5222", "ok", 2_000)];
let r = assemble(&i);
assert!(r.recorded && r.variant == "gunnar-twin-one");
assert_eq!(r.versions, ["gunnar abc123", "gunnar-ui def456"]);
assert_eq!(
shape(&r),
[
("main".into(), "primary".into(), "up".into()),
("twin".into(), "twin".into(), "up".into()),
("front".into(), "front".into(), "up".into()),
],
"exactly the laid boxes"
);
let main = &r.boxes[0];
assert!(
main.this_box
&& main.laid_as == "main"
&& main.declared_vcpu == 4
&& main.declared_ram_mb == 16384
);
assert_eq!(
(main.public_address.as_str(), main.utility_address.as_str()),
("203.0.113.10", "10.13.8.101")
);
assert_eq!(main.declared_data, "4 × 10 GB maxiops, md-raid0");
let twin = &r.boxes[1];
assert!(
twin.sources.iter().any(|s| s == "record")
&& twin.sources.iter().any(|s| s == "heartbeat"),
"{:?}",
twin.sources
);
assert_eq!(twin.provider_size, "STARTER-1xCPU-2GB");
let front = &r.boxes[2];
// The front's monetize is seen when it CALLS this box (the register fact that also
// named it went with self_register, v2.0 step 1); here only gunnar-ui called.
assert_eq!(front.services, ["console"]);
assert_eq!(front.hostname, "gunnar-front");
}
/// The pairing's ELECTION decides primary and twin; how the kit laid a
/// box stays on the row as `laid_as`. A laid box no evidence reaches is
/// still drawn — `unknown`, never dropped.
#[test]
fn the_election_names_the_role_and_a_silent_laid_box_stays_on_the_picture() {
let mut i = input();
i.facts.me = "twin".into();
i.is_twin = false; // this box was ELECTED primary although laid as twin
i.facts.record = Some(laid());
i.facts.peers = vec![Peer {
name: "main".into(),
role: "twin".into(),
hosts: vec!["10.13.8.101".into()],
}];
let r = assemble(&i);
let row = |n: &str| r.boxes.iter().find(|b| b.name == n).unwrap().clone();
assert_eq!(
(row("twin").role.as_str(), row("twin").laid_as.as_str()),
("primary", "twin")
);
assert_eq!(
(row("main").role.as_str(), row("main").laid_as.as_str()),
("twin", "main")
);
assert_eq!(
row("front").state,
"unknown",
"laid, and nothing here can ask it"
);
assert_eq!(r.boxes.len(), 3);
}
fn headline(r: &TopologyResponse, name: &str) -> String {
r.boxes
.iter()
.find(|b| b.name == name)
.map(|b| b.headline.clone())
.unwrap_or_else(|| panic!("no box {name}"))
}
/// ★ **gunnar says the state in a few words** — the words a live picture
/// draws, so the console composes none: this box answering, a following
/// twin, a twin behind by N entries, a twin silent for whole minutes (coarse,
/// so the answer changes when the state does), one never heard.
#[test]
fn every_box_carries_gunnars_headline() {
let mut i = input();
i.twins = vec![
twin("t-ok", "10.0.0.1:1", "ok", 2_000),
TwinHeartbeat {
lag_entries: 120,
..twin("t-behind", "10.0.0.2:1", "behind", 2_000)
},
twin("t-dead", "10.0.0.3:1", "silent", 4 * 60_000 + 30_000),
twin("t-fresh", "10.0.0.4:1", "silent", 45_000),
twin("t-never", "", "silent", 0),
];
i.callers = vec![caller("console", "10.0.0.9:1", 1_000, true)];
let r = assemble(&i);
assert_eq!(r.boxes[0].headline, "answering");
assert_eq!(headline(&r, "t-ok"), "following");
assert_eq!(headline(&r, "t-behind"), "behind 120 entries");
assert_eq!(headline(&r, "t-dead"), "silent for 4 min");
assert_eq!(headline(&r, "t-fresh"), "silent for under a minute");
assert_eq!(headline(&r, "t-never"), "never heard");
assert_eq!(headline(&r, "console"), "calling");
// Ten seconds later a dead twin's words are the same: the answer
// changes with the state, not with the clock's seconds.
let mut later = i.clone();
later.now_ms += 10_000;
later.twins[2].seen_ago_ms += 10_000;
assert_eq!(headline(&assemble(&later), "t-dead"), "silent for 4 min");
}
/// The twin's OWN word rides the headline when it is not plain following —
/// a twin being refilled after a recharge says so (`snapshot-needed`), as
/// gunnar heard it on the poll; nothing is invented for it.
#[test]
fn the_twins_own_word_rides_the_headline() {
let mut i = input();
i.twins = vec![TwinHeartbeat {
twin_state: "snapshot-needed".into(),
lag_entries: 900,
..twin("t", "10.0.0.1:1", "behind", 1_000)
}];
assert_eq!(
headline(&assemble(&i), "t"),
"behind 900 entries · snapshot-needed"
);
i.twins[0].state = "ok".into();
assert_eq!(headline(&assemble(&i), "t"), "following · snapshot-needed");
}
/// ★ **After a grow the measured disk wins** (no re-feed needed): the twin's
/// own statvfs on its poll is the row's disk, whatever the installer's
/// record declared at the lay; the declaration stays what it was, as laid.
#[test]
fn after_a_grow_the_measured_disk_wins_over_the_record() {
let mut i = input();
i.facts.me = "main".into();
i.facts.record = Some(laid());
i.facts.peers = vec![Peer {
name: "twin".into(),
role: "twin".into(),
hosts: vec!["10.13.8.122".into()],
}];
i.twins = vec![TwinHeartbeat {
disk_total_bytes: 40 << 30,
disk_used_bytes: 2 << 30,
..twin("gunnar-twin", "10.13.8.122:5222", "ok", 1_000)
}];
let r = assemble(&i);
let t = r.boxes.iter().find(|b| b.name == "twin").unwrap();
assert!(
t.disk_measured && t.disk_total_bytes == 40 << 30,
"the grown volume is the row's disk: {t:?}"
);
assert_eq!(
t.declared_data, "1 × 20 GB hdd, xfs",
"the record stays what was laid"
);
}
}