nordisk/holger
server/lib/src/flow.rs
//! **The artifact flow, drawn from the configuration that produces it** — and the
//! estate it runs in, with the counters on every box and arrow.
//!
//! Restored by lane HOL-SVG-RON (owner 2026-10-30) from the module the lean cut deleted
//! (holger 9521bf06, `server/lib/src/flow.rs` of 2026-09-28): the estate board that
//! replaced it (`crate::topology`) drew front | main | twin and nothing of the RON. The
//! RON is the source of truth, so the picture is generated from it again, and the
//! estate's boxes and the live counters are written onto that shape.
//!
//! ONE data model ([`FlowGraph`], also served as JSON so another renderer — facett — can
//! draw the same thing) and ONE renderer ([`FlowGraph::render_svg`]). The server draws;
//! a console shows the drawing as an image (`AdminService.FlowSvg`) and decides nothing.
//! Both doors call [`live`], so `GET /-/flow.svg` and `AdminService.FlowSvg` can never
//! draw two different pictures.
//!
//! # What is drawn, and where it comes from
//!
//! Every relation in the RON that has a shape is drawn, and nothing the RON does not
//! declare is drawn as if it did:
//!
//! | RON | drawn as |
//! |---|---|
//! | `sources:` | a SOURCE box per entry, by its name: `upload` and the pull kinds feed `incoming`, `crates` is the remote repository it lowers to, `holger` the mirror source it lowers to |
//! | no `sources:`, no remote, no mirror | one box that says **no sources configured** — never a sample |
//! | `repositories` | a box per repository (`repository` / `group` / `upstream`) |
//! | `ron_upstreams` | a **proxy** arrow — dashed, lazy, "on miss" |
//! | `ron_members` | a group-member arrow |
//! | `promotion.stages` (`steps:`) | the ladder, one arrow per consecutive rung |
//! | `promotion.intake` (`steps:`) | the intake chain: a STEP box per required scanner, the malware inspection, the approvals; a GATE box per rung gate and for the floor (`rules:`) |
//! | `storage_endpoints` | a box per endpoint, in its own band under the flow |
//! | `ron_in` / `ron_out` → `ron_storage_endpoint` | an arrow from the repository down into that storage box |
//! | `exposed_endpoints` | a door box |
//! | `ron_out` / `ron_in` → `ron_exposed_endpoint` | repository → door ("served via"), door → repository ("ingest via") |
//! | `mirror.ron_sources` | an off-box source box, with its schedule, and a **mirror** arrow per pair |
//! | `sync.downstreams` | a DOWNSTREAM box per entry, by name |
//!
//! And what is not configuration but RECORDED, each from its own evidence and drawn in
//! its own band so it never reads as declared:
//!
//! * the ESTATE — the console and front, this main, its twin — from the pairing feed and
//! the twin's reports (`crate::topology`'s rows, the same words the estate board said);
//! * the downstreams that paired or enrolled (the peer records, the sync board), each by
//! name with how it reaches this holger (`via front` / `via wireguard`), its last sync,
//! its lag and the bytes sent to it. **A downstream is never drawn or worded as an
//! upstream**: it is its own kind, below the main, and the arrow runs from the main to it;
//! * the observed pullers — ONE box, never one per address (see below).
//!
//! # Three levels of one model (HOL-FLOW-SMART)
//!
//! Every node says which machine it runs on ([`FlowNode::machine`], `box` in the JSON):
//! [`MAIN_ID`] for the parts of this server's RON, a machine for itself, a twin or a
//! downstream for its own parts (a downstream's come from the RON summary its sync report
//! carried, [`RonSummary`]), and none for what runs on no machine of the estate (an upstream
//! source, the diode drop area, a one-way inbox). Every arrow that crosses a machine's border
//! names the port it leaves or enters by. [`FlowGraph::zoom`] folds every part of a machine
//! that is not open into the machine, so a line ends at the same port at every level:
//!
//! * level 1 (`?open=` empty, the Overview): the machines and the upstream sources;
//! * level 2 (`?open=main,lab-1`, the Flow): those machines unfolded in place;
//! * level 3 (`?open=all`): the whole model.
//!
//! Nothing that is not there is drawn: no console without a console box or a recent sign-in,
//! no front, twin or downstream that does not exist, no placeholder box. A downstream's line
//! says where it is known from (declared, paired, seen) and looks it (live, silent, unknown).
//! A machine with parts carries the relative link that toggles it (`flow?open=…`), and
//! [`FlowGraph::render_with_links`] says where, for a console that shows the drawing as an
//! image.
//!
//! # A mirror is not a proxy
//!
//! `ron_upstreams` is proxy fall-through: lazy, on demand, cached. `mirror:` is scheduled
//! and eager. They are different edge kinds with different strokes and labels, and the
//! page's legend names the difference in words.
//!
//! # Facts and measurements are different rows
//!
//! A box carries declared FACTS (from the RON) above a rule and MEASURED rows below it.
//! A fact the config does not state reads `not declared`; a counter with no source reads
//! `not recorded`, never `0` — a zero says nobody pulled, and the truth may be that
//! nothing wrote it down. Where every counter comes from:
//!
//! * repository `artifacts`, `bytes stored` — the backend's bounded listing (the
//! `/metrics` census); `quota` — the paid repository's disk cap (metered builds);
//! * repository `fetches` / `bytes sent` (24 h and total), promotion arrows, the observed
//! pullers — the audit log (successful downloads and promotions only);
//! * a proxy's `hits` / `misses` and each proxy arrow's fetches, last fetch and errors —
//! the proxy walk's record ([`crate::upstreams`], since this boot);
//! * a mirror source's last run, next run, items synced, lag, failures — the mirror's
//! health ([`crate::metrics::MirrorSourceHealth`], since this boot);
//! * the intake steps and gates — the intake record (every artifact's state and the
//! evidence each transition carried; durable) and the scanners' findings by severity
//! ([`crate::metrics::Metrics::scan_tally`], since this boot);
//! * doors — the visitor tally (requests today) and the door's 5xx answers (since boot);
//! * storage endpoints — one `statvfs` of the declared path;
//! * the estate — `crate::topology`'s rows (the twin's follow, the downstreams' reports).
//!
//! # An observed puller is ONE box
//!
//! Client addresses the audit log saw collapse into a single `observed pullers` box whose
//! rows list the busiest few, by a per-day keyed hash ([`crate::visitors`]) — never an
//! address. Loopback is the box talking to itself and is counted on its own row.
//!
//! # Airgapped: the page fetches nothing
//!
//! Hand-written SVG with an inline `<style>`: no `<script>`, no `<link>`, no `src=`, no
//! `@import`, no `url(` that is not a same-document fragment.
use std::collections::{BTreeMap, BTreeSet};
use std::sync::Arc;
use serde::{Deserialize, Serialize};
/// The HTML page: the drawing plus its legend.
pub const FLOW_PATH: &str = "/-/flow";
/// The bare drawing, `image/svg+xml`.
pub const FLOW_SVG_PATH: &str = "/-/flow.svg";
/// The data model the drawing is rendered from, JSON (for another renderer).
pub const FLOW_JSON_PATH: &str = "/-/flow.json";
/// The single box every observed puller collapses into.
pub const PULLERS_NODE: &str = "observed pullers";
/// The box a RON with no source at all is drawn with.
pub const NO_SOURCES_NODE: &str = "no sources configured";
/// **This server**: the machine every part of its own RON runs on. The id is fixed, so
/// `?open=main` names it whatever the estate calls the box.
pub const MAIN_ID: &str = "estate:main";
/// The console (holger-ui), drawn only when one signed in or called.
pub const CONSOLE_ID: &str = "estate:console";
/// The diode drop area (`export: (push_to: …)`): where signed bundles leave, one way.
pub const DIODE_ID: &str = "diode:drop";
/// A one-way downstream's inbox (`boundary: OneWay`): where bundles come in, one way.
pub const INBOX_ID: &str = "diode:inbox";
/// How long after its last sign-in or call a console is still drawn.
pub const CONSOLE_SEEN_WITHIN_MS: u64 = 24 * 3600 * 1000;
/// **The `flow:` block of the RON config — on by default.**
#[derive(Serialize, Deserialize, Clone, Debug, PartialEq, Eq)]
#[serde(deny_unknown_fields)]
pub struct FlowSettings {
/// Mount the `/-/flow` doors and `AdminService.FlowSvg`. Default `true`.
#[serde(default = "default_true")]
pub enabled: bool,
/// Heading on the drawing. Default `artifact flow`.
#[serde(default)]
pub title: Option<String>,
/// `<meta http-equiv="refresh">` interval of the page, seconds. `0` ⇒ none.
#[serde(default = "default_refresh_secs")]
pub refresh_secs: u32,
/// How many observed pullers to list, biggest by bytes first (a cap, not a filter).
#[serde(default = "default_max_consumers")]
pub max_consumers: usize,
}
fn default_true() -> bool {
true
}
fn default_refresh_secs() -> u32 {
30
}
fn default_max_consumers() -> usize {
8
}
impl Default for FlowSettings {
fn default() -> Self {
Self {
enabled: true,
title: None,
refresh_secs: default_refresh_secs(),
max_consumers: default_max_consumers(),
}
}
}
/// What a box in the drawing is.
#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum NodeKind {
/// A repository that holds bytes on this server.
Repository,
/// A virtual `group` repository.
Group,
/// A repository whose bytes live on another server (`*-remote`).
Remote,
/// A `storage_endpoints` entry.
Storage,
/// An `exposed_endpoints` entry.
Door,
/// A `mirror.ron_sources` entry.
MirrorSource,
/// A `sources:` entry that is not a remote repository or a mirror source (`upload`,
/// the pull-through kinds).
Source,
/// The statement that the RON names no source.
NoSources,
/// An intake step (a required scanner, the malware inspection, the approvals).
Step,
/// A gate (a rung's `gates`, the floor `rules:`).
Gate,
/// The console (holger-ui), served by the front.
Console,
/// The estate's front.
Front,
/// This holger server (or the main a twin follows).
Main,
/// A twin.
Twin,
/// A downstream holger: declared (`sync.downstreams`), paired or enrolled.
Downstream,
/// A twin's part: its mirror of the main, with its lag.
Mirror,
/// The diode's drop area: the folder the upstream pushes signed bundles into.
DiodeDrop,
/// A one-way downstream's inbox: the only way bytes come in.
Inbox,
/// The collapsed observed pullers.
Consumer,
/// A name the front serves, and where it is routed (the front's own report).
Host,
}
impl NodeKind {
/// The CSS class.
pub fn css(self) -> &'static str {
match self {
NodeKind::Repository => "repo",
NodeKind::Group => "group",
NodeKind::Remote => "remote",
NodeKind::Storage => "storage",
NodeKind::Door => "door",
NodeKind::MirrorSource => "mirrorsrc",
NodeKind::Source => "source",
NodeKind::NoSources => "none",
NodeKind::Step => "step",
NodeKind::Gate => "gate",
NodeKind::Console => "console",
NodeKind::Front => "front",
NodeKind::Main => "main",
NodeKind::Twin => "twin",
NodeKind::Downstream => "downstream",
NodeKind::Mirror => "mirror",
NodeKind::DiodeDrop => "diode",
NodeKind::Inbox => "inbox",
NodeKind::Consumer => "consumer",
NodeKind::Host => "host",
}
}
/// The word printed in the box's corner.
pub fn word(self) -> &'static str {
match self {
NodeKind::Repository => "repository",
NodeKind::Group => "group",
NodeKind::Remote => "upstream registry",
NodeKind::Storage => "storage endpoint",
NodeKind::Door => "exposed endpoint",
NodeKind::MirrorSource => "upstream holger · mirror",
NodeKind::Source => "source",
NodeKind::NoSources => "sources",
NodeKind::Step => "intake step",
NodeKind::Gate => "gate",
NodeKind::Console => "console",
NodeKind::Front => "front",
NodeKind::Main => "holger server",
NodeKind::Twin => "twin",
NodeKind::Downstream => "downstream holger",
NodeKind::Mirror => "the mirror",
NodeKind::DiodeDrop => "diode drop area · one way",
NodeKind::Inbox => "inbox · one way in",
NodeKind::Consumer => "observed pullers",
NodeKind::Host => "served name · route",
}
}
/// Is this a repository that carries the per-repository counters?
pub fn is_measured_repository(self) -> bool {
matches!(
self,
NodeKind::Repository | NodeKind::Group | NodeKind::Remote
)
}
/// Is this a box that brings bytes INTO this holger? A downstream never is.
pub fn is_upstream(self) -> bool {
matches!(
self,
NodeKind::Remote | NodeKind::MirrorSource | NodeKind::Source | NodeKind::Inbox
)
}
/// Is this a machine of the estate (a box the others run on)?
pub fn is_machine(self) -> bool {
matches!(
self,
NodeKind::Console
| NodeKind::Front
| NodeKind::Main
| NodeKind::Twin
| NodeKind::Downstream
)
}
}
/// Which relation an arrow is.
#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum EdgeKind {
/// A rung of the promotion ladder.
Promotion,
/// `ron_upstreams`: proxy fall-through, lazy.
Pull,
/// `mirror.ron_sources[].ron_repositories`: scheduled, eager.
Mirror,
/// `ron_members`.
Member,
/// The bytes rest in that storage endpoint.
Stores,
/// Served through that door.
Serves,
/// Written through that door.
Ingests,
/// A `sources:` entry feeding the repository it lands in.
Feeds,
/// An intake step or gate checking the repository hop it guards.
Checks,
/// The console reaches holger through the front.
Console,
/// The front serves the main.
Fronts,
/// A twin follows the main.
Follows,
/// A downstream syncs from this holger (the arrow runs main → downstream).
Syncs,
/// Measured: bytes sent to the observed pullers.
Served,
/// Signed bundles pushed into the diode's folder: one way, no live state.
Export,
/// Bundles carried into a one-way inbox (diode or USB): one way, no live state.
Carries,
/// A front's door takes a name it serves (the front's own report).
Routes,
}
impl EdgeKind {
/// Every kind, in legend order.
pub const ALL: [EdgeKind; 17] = [
EdgeKind::Feeds,
EdgeKind::Pull,
EdgeKind::Mirror,
EdgeKind::Promotion,
EdgeKind::Checks,
EdgeKind::Member,
EdgeKind::Stores,
EdgeKind::Serves,
EdgeKind::Ingests,
EdgeKind::Console,
EdgeKind::Routes,
EdgeKind::Fronts,
EdgeKind::Follows,
EdgeKind::Syncs,
EdgeKind::Export,
EdgeKind::Carries,
EdgeKind::Served,
];
/// The CSS class.
pub fn css(self) -> &'static str {
match self {
EdgeKind::Promotion => "promote",
EdgeKind::Pull => "pull",
EdgeKind::Mirror => "mirror",
EdgeKind::Member => "member",
EdgeKind::Stores => "stores",
EdgeKind::Serves => "serves",
EdgeKind::Ingests => "ingests",
EdgeKind::Feeds => "feeds",
EdgeKind::Checks => "checks",
EdgeKind::Console => "consoleedge",
EdgeKind::Fronts => "fronts",
EdgeKind::Follows => "follows",
EdgeKind::Syncs => "syncs",
EdgeKind::Served => "served",
EdgeKind::Export => "export",
EdgeKind::Carries => "carries",
EdgeKind::Routes => "routes",
}
}
/// The words in the legend.
pub fn legend(self) -> &'static str {
match self {
EdgeKind::Promotion => "promotion rung",
EdgeKind::Pull => "proxy upstream, on miss",
EdgeKind::Mirror => "mirror, scheduled",
EdgeKind::Member => "group member",
EdgeKind::Stores => "bytes rest in",
EdgeKind::Serves => "served through",
EdgeKind::Ingests => "written through",
EdgeKind::Feeds => "source feeds",
EdgeKind::Checks => "step / gate checks",
EdgeKind::Console => "console via front",
EdgeKind::Fronts => "front serves main",
EdgeKind::Follows => "twin follows",
EdgeKind::Syncs => "downstream syncs from here",
EdgeKind::Served => "measured: sent to pullers",
EdgeKind::Export => "signed bundles out, one way",
EdgeKind::Carries => "carried in, one way (diode, USB)",
EdgeKind::Routes => "front door takes a name",
}
}
}
/// One number, or one declared fact. `0` means zero, and nothing else ever means `0`.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum Cell {
/// A plain count.
Count(u64),
/// A byte total, printed SI-humanised.
Bytes(u64),
/// Free text.
Text(String),
/// The backend could not enumerate. Not zero.
Unlisted,
/// There is no source for this number. Not zero.
NotRecorded,
/// The config does not state this.
NotDeclared,
/// Measured, and zero distinct clients.
NoConsumer,
}
impl Cell {
/// How the cell prints.
pub fn render(&self) -> String {
match self {
Cell::Count(n) => group_thousands(*n),
Cell::Bytes(n) => humanise_bytes(*n),
Cell::Text(s) => s.clone(),
Cell::Unlisted => "unlisted".to_string(),
Cell::NotRecorded => "not recorded".to_string(),
Cell::NotDeclared => "not declared".to_string(),
Cell::NoConsumer => "no consumer".to_string(),
}
}
/// A real measurement or a stated fact?
pub fn is_known(&self) -> bool {
!matches!(self, Cell::Unlisted | Cell::NotRecorded | Cell::NotDeclared)
}
/// The machine word for the JSON model.
fn kind_word(&self) -> &'static str {
match self {
Cell::Count(_) => "count",
Cell::Bytes(_) => "bytes",
Cell::Text(_) => "text",
Cell::Unlisted => "unlisted",
Cell::NotRecorded => "not_recorded",
Cell::NotDeclared => "not_declared",
Cell::NoConsumer => "no_consumer",
}
}
}
/// The JSON form: `{"kind": "count", "value": 3, "text": "3"}` — the number for a
/// renderer that computes, the words for one that prints.
impl Serialize for Cell {
fn serialize<S: serde::Serializer>(&self, s: S) -> Result<S::Ok, S::Error> {
use serde::ser::SerializeStruct;
let mut st = s.serialize_struct("Cell", 3)?;
st.serialize_field("kind", self.kind_word())?;
match self {
Cell::Count(n) | Cell::Bytes(n) => st.serialize_field("value", n)?,
_ => st.serialize_field("value", &Option::<u64>::None)?,
}
st.serialize_field("text", &self.render())?;
st.end()
}
}
/// One line inside a box.
#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
pub struct Row {
pub label: String,
pub cell: Cell,
}
impl Row {
pub fn new(label: impl Into<String>, cell: Cell) -> Self {
Self {
label: label.into(),
cell,
}
}
}
/// A box in the drawing.
#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
pub struct FlowNode {
/// Identity: a repository's bare `ron_name`; every other kind namespaced
/// (`storage:`, `door:`, `mirror:`, `source:`, `step:`, `gate:`, `estate:`,
/// `downstream:`).
pub id: String,
/// What the box prints: the name as written in the RON (or the evidence).
pub label: String,
pub kind: NodeKind,
/// The second line.
pub detail: Option<String>,
/// The promotion rung this repository is at.
pub stage: Option<String>,
/// Does arriving at this rung run gates?
pub gated: bool,
/// DECLARED rows.
pub facts: Vec<Row>,
/// MEASURED rows.
pub rows: Vec<Row>,
/// Which band (see [`Band`]).
pub band: Band,
/// Layout column, left to right.
pub column: usize,
/// Position within the column.
pub slot: usize,
/// A status the box is drawn in, bound to a real record: `none` (grey), `warn`,
/// `bad`, `good`.
pub state: Option<&'static str>,
/// **The machine this runs on** (`box` in the JSON): [`MAIN_ID`] for the parts of this
/// server's RON, a machine's own id for a machine, the twin or downstream for its own
/// parts, and `None` for what runs on no machine of the estate (an upstream source).
#[serde(rename = "box")]
pub machine: Option<String>,
/// **That server's own flow, at its console** (HOL-FLOW-FRONT, Henrik's Q4): set only on a
/// box that is another Holger server whose console this server knows (a downstream that
/// reported its console's address). `None`: there is nowhere to go, and nothing to click.
pub link: Option<String>,
}
/// An arrow.
#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
pub struct FlowEdge {
pub from: String,
pub to: String,
pub kind: EdgeKind,
/// The number or words on the arrow.
pub cell: Cell,
/// Where the arrow leaves its machine's border (`<machine>/<port>`), when it crosses one.
pub from_port: Option<String>,
/// Where the arrow enters its machine's border, when it crosses one.
pub to_port: Option<String>,
/// How a live link looks: `live` (solid: declared or paired, and seen), `silent`
/// (dashed: declared or paired, not seen lately), `unknown` (grey: seen, never
/// declared or paired). `None` for what is no live link.
pub look: Option<&'static str>,
/// Which sources know this link: `declared`, `paired`, `seen`.
pub known_from: Vec<&'static str>,
}
impl FlowEdge {
/// An arrow that crosses no border yet ([`FlowGraph::finish`] names its ports).
pub fn new(from: impl Into<String>, to: impl Into<String>, kind: EdgeKind, cell: Cell) -> Self {
Self {
from: from.into(),
to: to.into(),
kind,
cell,
from_port: None,
to_port: None,
look: None,
known_from: Vec::new(),
}
}
/// `from->to`, as the drawing tags it.
pub fn id(&self) -> String {
format!("{}->{}", self.from, self.to)
}
}
/// **Which machines are unfolded** (`?open=`): none (level 1, the Overview), some (level 2),
/// or all (level 3, the full RON). A machine is named by its id (`estate:main`) or its short
/// name (`main`, `twin`, a downstream's name).
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum Open {
Machines(BTreeSet<String>),
All,
}
impl Open {
/// Nothing unfolded: level 1.
pub fn none() -> Self {
Open::Machines(BTreeSet::new())
}
/// `main,lab-1` (order and spaces do not matter); `all` unfolds everything.
pub fn parse(s: &str) -> Self {
let words: BTreeSet<String> = s
.split(',')
.map(str::trim)
.filter(|w| !w.is_empty())
.map(str::to_string)
.collect();
if words.contains("all") {
Open::All
} else {
Open::Machines(words)
}
}
fn names(&self, machine: &str) -> bool {
match self {
Open::All => true,
Open::Machines(w) => w.contains(machine) || w.contains(short_name(machine)),
}
}
}
/// A machine's short name: `estate:main` ⇒ `main`, `downstream:lab-1` ⇒ `lab-1`.
pub fn short_name(id: &str) -> &str {
id.strip_prefix("estate:")
.or_else(|| id.strip_prefix("downstream:"))
.unwrap_or(id)
}
/// The port an arrow of `kind` uses on `machine`'s border, going out or coming in.
fn port_name(kind: EdgeKind, out: bool, machine: NodeKind) -> &'static str {
match (kind, out, machine) {
(EdgeKind::Feeds | EdgeKind::Pull | EdgeKind::Mirror, false, _) => "intake",
(EdgeKind::Fronts, false, _) => "doors",
(EdgeKind::Fronts, true, _) => "main",
(EdgeKind::Console, false, NodeKind::Front) => "console",
(EdgeKind::Console, false, _) => "doors",
(EdgeKind::Console, true, _) => "console",
(EdgeKind::Syncs, true, _) => "export",
(EdgeKind::Syncs, false, _) => "fetch",
(EdgeKind::Follows, true, _) => "twin",
(EdgeKind::Follows, false, _) => "follow",
(EdgeKind::Export, true, _) => "export",
(EdgeKind::Carries, false, _) => "inbox",
(_, true, _) => "out",
(_, false, _) => "in",
}
}
/// The bands, top to bottom.
#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Hash, Serialize)]
#[serde(rename_all = "snake_case")]
pub enum Band {
/// Recorded: the console, the front, the main, its twins and the downstreams.
Estate,
/// Declared: sources, repositories, the ladder, the doors, the pullers.
Flow,
/// Declared: the intake chain's steps and gates.
Chain,
/// Declared: where the bytes rest.
Storage,
}
impl Band {
const ALL: [Band; 4] = [Band::Estate, Band::Flow, Band::Chain, Band::Storage];
fn caption(self) -> &'static str {
match self {
Band::Estate => {
"THE ESTATE · recorded: pairing, the twin's reports, the downstreams' syncs"
}
Band::Flow => {
"THE FLOW · declared in the RON: sources, repositories, the ladder, the doors"
}
Band::Chain => {
"THE INTAKE CHAIN · declared steps and gates (steps: / promotion.intake)"
}
Band::Storage => "WHERE THE BYTES REST · storage_endpoints",
}
}
}
/// **The whole picture**: nodes, edges, and the caption under them. Serialised as the
/// JSON model (`GET /-/flow.json`, `AdminService.FlowSvg`'s `graph_json`).
#[derive(Debug, Clone, PartialEq, Eq, Default, Serialize)]
pub struct FlowGraph {
/// The model's schema word.
pub schema: &'static str,
pub title: String,
/// The configured promotion ladder's name.
pub ladder: Option<String>,
pub nodes: Vec<FlowNode>,
pub edges: Vec<FlowEdge>,
/// The process-wide counters, one caption line.
pub summary: Vec<Row>,
pub refresh_secs: u32,
pub max_consumers: usize,
/// Does this server keep an audit log the counters can be read from?
pub audit_recorded: bool,
/// The zoom: 1 the machines only, 2 some unfolded, 3 everything.
pub level: u8,
/// The machines that have parts of their own (so a click can unfold them).
pub openable: Vec<String>,
/// The machines unfolded in this picture.
pub open: Vec<String>,
/// The repository the downstreams and an export take (the intake chain's production),
/// when the RON has one.
#[serde(skip)]
pub served_from: Option<String>,
}
/// The JSON model's schema.
pub const SCHEMA: &str = "holger.flow/v2";
pub fn storage_node_id(name: &str) -> String {
format!("storage:{name}")
}
pub fn door_node_id(name: &str) -> String {
format!("door:{name}")
}
pub fn mirror_node_id(name: &str) -> String {
format!("mirror:{name}")
}
pub fn source_node_id(name: &str) -> String {
format!("source:{name}")
}
pub fn step_node_id(name: &str) -> String {
format!("step:{name}")
}
pub fn gate_node_id(name: &str) -> String {
format!("gate:{name}")
}
pub fn estate_node_id(name: &str) -> String {
format!("estate:{name}")
}
pub fn downstream_node_id(name: &str) -> String {
format!("downstream:{name}")
}
fn is_remote_type(repo_type: &str) -> bool {
repo_type.ends_with("-remote")
}
/// One intake step as the RON declares it.
#[cfg(feature = "intake")]
#[derive(Debug, Clone, PartialEq, Eq)]
struct StepDecl {
/// The id suffix and label (`scan · vulnerability`, `malware`, `approval`).
label: String,
/// What it checks, as the evidence names it (`scan`, `malware`, `license`, `provenance`, `approval`).
evidence: &'static str,
/// The scanner kind's wire label, for the findings tally; empty for none.
scanner: String,
/// The repository whose hop it guards.
guards: String,
facts: Vec<Row>,
}
impl FlowGraph {
/// **The graph, read out of the configuration** — the shape plus the declared
/// facts, built once at boot. Measurements are added per request by [`measure`].
pub fn from_config(cfg: &crate::Holger) -> Self {
let mut nodes: Vec<FlowNode> = Vec::new();
let mut edges: Vec<FlowEdge> = Vec::new();
let node = |id: String,
label: String,
kind: NodeKind,
detail: Option<String>,
band: Band,
column: usize| FlowNode {
id,
label,
kind,
detail,
stage: None,
gated: false,
facts: Vec::new(),
rows: Vec::new(),
band,
column,
slot: 0,
state: None,
// The RON's parts run on this server; an upstream runs on none of the estate's.
machine: (!kind.is_upstream()).then(|| MAIN_ID.to_string()),
link: None,
};
// The ladder: rung index per repository.
#[allow(unused_mut)]
let mut rung: BTreeMap<String, (usize, String, bool)> = BTreeMap::new();
#[allow(unused_mut)]
let mut ladder: Option<String> = None;
#[allow(unused_mut)]
let mut ladder_len = 0usize;
#[cfg(feature = "intake")]
if let Some(p) = cfg.promotion.as_ref() {
for (i, stage) in p.stages.iter().enumerate() {
rung.insert(
stage.repo.clone(),
(
i,
stage.name.clone(),
stage.gated || !stage.gates.is_empty(),
),
);
}
ladder = Some(p.ladder_name.clone());
ladder_len = p.stages.len();
}
let mut is_upstream_of_somebody: BTreeSet<&str> = BTreeSet::new();
for repo in &cfg.repositories {
for up in &repo.ron_upstreams {
is_upstream_of_somebody.insert(up.as_str());
}
}
// Columns: 0 the sources, 1..=n the ladder, then the repositories on no rung, then
// groups, then doors.
let plain_col = ladder_len + 1;
let group_col = ladder_len + 2;
let door_col = ladder_len + 3;
// The `sources:` entries by name: which kind each is, so the box it lowered to
// says so.
let declared_sources: BTreeMap<String, String> = cfg
.sources
.iter()
.map(|s| (s.name.clone(), s.kind.name().to_string()))
.collect();
for repo in &cfg.repositories {
let name = repo.ron_name.as_str();
let kind = if repo.ron_repo_type == "group" {
NodeKind::Group
} else if is_remote_type(&repo.ron_repo_type) {
NodeKind::Remote
} else {
NodeKind::Repository
};
let column = match rung.get(name) {
Some((i, _, _)) => i + 1,
None if kind == NodeKind::Remote => 0,
None if repo.ron_upstreams.is_empty() && is_upstream_of_somebody.contains(name) => {
0
}
None if kind == NodeKind::Group => group_col,
None if ladder_len == 0 => 1,
None => plain_col,
};
let detail = match kind {
NodeKind::Remote => repo
.ron_base_url
.clone()
.map(|u| match &repo.ron_upstream_repo {
Some(r) => format!("{u} · {r}"),
None => u,
})
.or_else(|| Some(repo.ron_repo_type.clone())),
_ => Some(repo.ron_repo_type.clone()),
};
let mut facts = Vec::new();
if let Some(k) = declared_sources.get(name) {
facts.push(Row::new("source", Cell::Text(format!("sources: {k}"))));
}
if kind != NodeKind::Group {
let bytes_at = if let Some(a) = &repo.ron_writable_archive {
Cell::Text(format!("archive rw {a}"))
} else if let Some(a) = &repo.ron_archive_path {
Cell::Text(format!("archive ro {a}"))
} else if let Some(d) = &repo.ron_data_dir {
Cell::Text(format!("dir {d}"))
} else {
Cell::NotDeclared
};
facts.push(Row::new("bytes at", bytes_at));
}
if kind == NodeKind::Group {
facts.push(Row::new(
"members",
Cell::Count(repo.ron_members.len() as u64),
));
}
let mut n = node(
name.to_string(),
name.to_string(),
kind,
detail,
Band::Flow,
column,
);
n.stage = rung.get(name).map(|(_, s, _)| s.clone());
n.gated = rung.get(name).map(|(_, _, g)| *g).unwrap_or(false);
n.facts = facts;
nodes.push(n);
for up in &repo.ron_upstreams {
edges.push(FlowEdge {
from: up.clone(),
to: name.to_string(),
kind: EdgeKind::Pull,
cell: Cell::Text("on miss".into()),
from_port: None,
to_port: None,
look: None,
known_from: Vec::new(),
});
}
for member in &repo.ron_members {
edges.push(FlowEdge {
from: member.clone(),
to: name.to_string(),
kind: EdgeKind::Member,
cell: Cell::Text(String::new()),
from_port: None,
to_port: None,
look: None,
known_from: Vec::new(),
});
}
for io in [repo.ron_out.as_ref(), repo.ron_in.as_ref()]
.into_iter()
.flatten()
{
edges.push(FlowEdge {
from: name.to_string(),
to: storage_node_id(&io.ron_storage_endpoint),
kind: EdgeKind::Stores,
cell: Cell::Text(String::new()),
from_port: None,
to_port: None,
look: None,
known_from: Vec::new(),
});
}
if let Some(io) = repo.ron_out.as_ref() {
edges.push(FlowEdge {
from: name.to_string(),
to: door_node_id(&io.ron_exposed_endpoint),
kind: EdgeKind::Serves,
cell: Cell::Text(String::new()),
from_port: None,
to_port: None,
look: None,
known_from: Vec::new(),
});
}
if let Some(io) = repo.ron_in.as_ref() {
edges.push(FlowEdge {
from: door_node_id(&io.ron_exposed_endpoint),
to: name.to_string(),
kind: EdgeKind::Ingests,
cell: Cell::Text(String::new()),
from_port: None,
to_port: None,
look: None,
known_from: Vec::new(),
});
}
}
// The promotion ladder.
#[cfg(feature = "intake")]
if let Some(p) = cfg.promotion.as_ref() {
for pair in p.stages.windows(2) {
edges.push(FlowEdge {
from: pair[0].repo.clone(),
to: pair[1].repo.clone(),
kind: EdgeKind::Promotion,
cell: Cell::NotRecorded,
from_port: None,
to_port: None,
look: None,
known_from: Vec::new(),
});
}
}
// The doors.
for (i, ep) in cfg.exposed_endpoints.iter().enumerate() {
let mut n = node(
door_node_id(&ep.ron_name),
ep.ron_name.clone(),
NodeKind::Door,
Some(format!("grpc {}", ep.ron_url)),
Band::Flow,
door_col,
);
n.slot = i;
n.facts.push(Row::new(
"http",
match &ep.ron_http_url {
Some(u) => Cell::Text(u.clone()),
None => Cell::NotDeclared,
},
));
n.facts.push(Row::new(
"tls",
Cell::Text(
if ep.ron_tls.is_some() {
"configured"
} else {
"cleartext"
}
.into(),
),
));
nodes.push(n);
}
// The storage endpoints.
for (i, se) in cfg.storage_endpoints.iter().enumerate() {
let mut n = node(
storage_node_id(&se.ron_name),
se.ron_name.clone(),
NodeKind::Storage,
Some(se.ron_storage_type.clone()),
Band::Storage,
i,
);
n.facts
.push(Row::new("path", Cell::Text(se.ron_path.clone())));
nodes.push(n);
}
// The mirror: scheduled and eager.
if let Some(m) = cfg.mirror.as_ref() {
for src in &m.ron_sources {
let mut n = node(
mirror_node_id(&src.ron_name),
src.ron_name.clone(),
NodeKind::MirrorSource,
Some(src.ron_endpoint.clone()),
Band::Flow,
0,
);
if let Some(k) = declared_sources.get(&src.ron_name) {
n.facts
.push(Row::new("source", Cell::Text(format!("sources: {k}"))));
}
n.facts.push(Row::new(
"speaks",
Cell::Text(format!("{:?}", src.ron_kind)),
));
n.facts.push(Row::new(
"every",
Cell::Text(humanise_secs(src.ron_every_secs)),
));
n.facts.push(Row::new(
"credential",
Cell::Text(credential_words(&src.ron_credential)),
));
nodes.push(n);
for pair in &src.ron_repositories {
let mut label = format!("every {}", humanise_secs(src.ron_every_secs));
if pair.ron_from != pair.ron_to {
label.push_str(&format!(" · from {}", pair.ron_from));
}
if matches!(
pair.ron_unproven,
crate::mirror_cycle::RonUnprovenPolicy::TrustSize
) {
label.push_str(" · trust size");
}
edges.push(FlowEdge {
from: mirror_node_id(&src.ron_name),
to: pair.ron_to.clone(),
kind: EdgeKind::Mirror,
cell: Cell::Text(label),
from_port: None,
to_port: None,
look: None,
known_from: Vec::new(),
});
}
}
}
// The `sources:` entries that lowered to no repository and no mirror source:
// the upload doors and the pull-through kinds. Each feeds `incoming`.
let incoming = intake_incoming(cfg);
for s in &cfg.sources {
if nodes.iter().any(|n| {
n.label == s.name
&& (n.kind == NodeKind::Remote || n.kind == NodeKind::MirrorSource)
}) {
continue;
}
let detail = match &s.url {
Some(u) => format!("{} · {u}", s.kind.name()),
None => s.kind.name().to_string(),
};
let mut n = node(
source_node_id(&s.name),
s.name.clone(),
NodeKind::Source,
Some(detail),
Band::Flow,
0,
);
if !s.importers.is_empty() {
n.facts
.push(Row::new("importers", Cell::Text(s.importers.join(", "))));
}
if !s.serves.is_empty() {
n.facts
.push(Row::new("serves", Cell::Text(s.serves.join(", "))));
}
#[cfg(feature = "intake")]
if let Some(g) = &s.gates {
n.facts.push(Row::new("gates", Cell::Text(gates_words(g))));
}
nodes.push(n);
// What it feeds: the repositories it serves, else the chain's incoming, else
// this server as a whole (no chain names where an upload lands).
let into: Vec<String> = if !s.serves.is_empty() {
s.serves.clone()
} else if let Some(i) = &incoming {
vec![i.clone()]
} else {
vec![MAIN_ID.to_string()]
};
for to in into {
edges.push(FlowEdge {
from: source_node_id(&s.name),
to,
kind: EdgeKind::Feeds,
cell: Cell::Text(String::new()),
from_port: None,
to_port: None,
look: None,
known_from: Vec::new(),
});
}
}
// A RON with no source at all says so — never a sample.
if !nodes.iter().any(|n| n.kind.is_upstream()) {
let mut n = node(
NO_SOURCES_NODE.into(),
NO_SOURCES_NODE.into(),
NodeKind::NoSources,
None,
Band::Flow,
0,
);
n.detail = Some(match &incoming {
Some(i) => format!("uploads into {i} through the doors only"),
None => "the RON names no sources:, no remote, no mirror".into(),
});
n.state = Some("none");
nodes.push(n);
}
// The intake chain: its steps and gates.
#[cfg(feature = "intake")]
{
let (steps, gates, floor_adds) = chain_of(cfg);
if let Some((repo, words)) = floor_adds {
if let Some(n) = nodes.iter_mut().find(|n| n.id == repo) {
n.facts.push(Row::new("floor scans", Cell::Text(words)));
}
}
for s in &steps {
let mut n = node(
step_node_id(&s.label),
s.label.clone(),
NodeKind::Step,
Some(format!("checks {}", s.evidence)),
Band::Chain,
0,
);
n.facts = s.facts.clone();
n.column = rung.get(&s.guards).map(|(r, _, _)| r + 1).unwrap_or(1);
nodes.push(n);
edges.push(FlowEdge {
from: step_node_id(&s.label),
to: s.guards.clone(),
kind: EdgeKind::Checks,
cell: Cell::Text(String::new()),
from_port: None,
to_port: None,
look: None,
known_from: Vec::new(),
});
}
for g in gates {
let mut n = node(
gate_node_id(&g.0),
g.0.clone(),
NodeKind::Gate,
Some(g.1.clone()),
Band::Chain,
0,
);
n.column = rung.get(&g.2).map(|(r, _, _)| r + 1).unwrap_or(1);
n.facts = g.3.clone();
nodes.push(n);
edges.push(FlowEdge {
from: gate_node_id(&g.0),
to: g.2.clone(),
kind: EdgeKind::Checks,
cell: Cell::Text(String::new()),
from_port: None,
to_port: None,
look: None,
known_from: Vec::new(),
});
}
}
// The declared downstreams (`sync.downstreams`), each by name.
#[cfg(feature = "sync")]
if let Some(sync) = &cfg.sync {
for d in &sync.downstreams {
let mut n = node(
downstream_node_id(&d.name),
d.name.clone(),
NodeKind::Downstream,
Some("declared: sync.downstreams".into()),
Band::Estate,
4,
);
n.machine = Some(n.id.clone());
n.facts.push(Row::new(
"via",
Cell::Text(match d.via {
holger_downstream_rules::via::Via::Front => "front".into(),
holger_downstream_rules::via::Via::Wireguard { .. } => "wireguard".into(),
}),
));
nodes.push(n);
}
}
// The diode, one way out: `export: (push_to: …)` pushes signed bundles of the
// production into a folder; nothing comes back.
let landing = intake_production(cfg)
.filter(|p| nodes.iter().any(|n| &n.id == p))
.unwrap_or_else(|| MAIN_ID.to_string());
if let Some((x, dir)) = cfg
.export
.as_ref()
.and_then(|x| x.push_to.as_ref().map(|d| (x, d)))
{
let mut n = node(
DIODE_ID.into(),
"diode drop area".into(),
NodeKind::DiodeDrop,
Some(dir.clone()),
Band::Flow,
0,
);
n.machine = None;
n.facts.push(Row::new("every", Cell::Text(x.every.clone())));
n.facts.push(Row::new(
"bundles",
Cell::Text(format!("{} · {}", x.vendor, x.product)),
));
nodes.push(n);
edges.push(FlowEdge::new(
landing.clone(),
DIODE_ID,
EdgeKind::Export,
Cell::Text(format!("signed bundles · every {}", x.every)),
));
}
// The diode, one way in: a `boundary: OneWay` Holger takes bundles through its
// inbox only, and sends nothing back.
if cfg.boundary == Some(crate::boundary::Boundary::OneWay) {
let dir = cfg.data_root().join(INBOX_DIR);
let mut n = node(
INBOX_ID.into(),
"inbox".into(),
NodeKind::Inbox,
Some(dir.display().to_string()),
Band::Flow,
0,
);
n.facts.push(Row::new(
"first import",
Cell::Text(format!("{:?}", cfg.trust.first_import).to_lowercase()),
));
if let Some(d) = &cfg.domain {
n.facts.push(Row::new("domain", Cell::Text(d.clone())));
}
nodes.push(n);
edges.push(FlowEdge::new(
INBOX_ID,
landing,
EdgeKind::Carries,
Cell::Text("one way · diode or USB".into()),
));
}
// A RON whose only way in is the inbox has a source after all.
if nodes.iter().any(|n| n.kind == NodeKind::Inbox) {
nodes.retain(|n| n.kind != NodeKind::NoSources);
}
// This server: always there, it is the box answering ([`measure`] names it).
let mut main = node(
MAIN_ID.into(),
"this holger".into(),
NodeKind::Main,
Some("holger-server".into()),
Band::Estate,
2,
);
main.machine = Some(MAIN_ID.into());
nodes.push(main);
let known: BTreeSet<String> = nodes.iter().map(|n| n.id.clone()).collect();
edges.retain(|e| known.contains(&e.from) && known.contains(&e.to));
let mut seen: BTreeSet<(String, String)> = BTreeSet::new();
edges.retain(|e| seen.insert((e.from.clone(), e.to.clone())));
let served_from = intake_production(cfg).filter(|p| known.contains(p));
let mut graph = Self {
schema: SCHEMA,
title: cfg
.flow
.title
.clone()
.unwrap_or_else(|| "artifact flow".to_string()),
ladder,
nodes,
edges,
summary: Vec::new(),
refresh_secs: cfg.flow.refresh_secs,
max_consumers: cfg.flow.max_consumers,
audit_recorded: false,
level: 3,
openable: Vec::new(),
open: Vec::new(),
served_from,
};
graph.tidy();
graph.finish();
graph
}
/// The machines, in the order the drawing places them: console, front, main, twin,
/// then each downstream.
fn machine_ids(&self) -> Vec<String> {
let mut m: Vec<&FlowNode> = self
.nodes
.iter()
.filter(|n| n.machine.as_deref() == Some(n.id.as_str()))
.collect();
m.sort_by_key(|n| (machine_rank(n.kind), n.slot, n.id.clone()));
m.into_iter().map(|n| n.id.clone()).collect()
}
/// The machine `id` runs on, as far as this graph knows (`None`: outside every machine).
fn machine_of(&self, id: &str) -> Option<&str> {
self.node(id).and_then(|n| n.machine.as_deref())
}
/// **Name the ports and the zoom of the whole model**: every arrow that crosses a
/// machine's border gets the port it leaves or enters by. Run after the model changes
/// shape; the result is level 3 (nothing folded).
fn finish(&mut self) {
let kinds: BTreeMap<String, NodeKind> =
self.nodes.iter().map(|n| (n.id.clone(), n.kind)).collect();
let machines: BTreeMap<String, Option<String>> = self
.nodes
.iter()
.map(|n| (n.id.clone(), n.machine.clone()))
.collect();
for e in &mut self.edges {
let mf = machines.get(&e.from).cloned().flatten();
let mt = machines.get(&e.to).cloned().flatten();
if mf == mt {
e.from_port = None;
e.to_port = None;
continue;
}
e.from_port = mf.map(|m| {
let k = kinds.get(&m).copied().unwrap_or(NodeKind::Main);
format!("{m}/{}", port_name(e.kind, true, k))
});
e.to_port = mt.map(|m| {
let k = kinds.get(&m).copied().unwrap_or(NodeKind::Main);
format!("{m}/{}", port_name(e.kind, false, k))
});
}
let with_parts: BTreeSet<&str> = self
.nodes
.iter()
.filter_map(|n| n.machine.as_deref().filter(|m| *m != n.id))
.collect();
self.openable = self
.machine_ids()
.into_iter()
.filter(|m| with_parts.contains(m.as_str()))
.collect();
self.open = self.openable.clone();
self.level = 3;
}
/// **The picture at one zoom**: every part of a machine that is not unfolded is folded
/// into that machine, its arrows end at the machine's border (the same port), and an
/// arrow folded inside one machine is not drawn. `Open::All` is the whole model.
pub fn zoom(&self, open: &Open) -> FlowGraph {
let unfolded: Vec<String> = self
.openable
.iter()
.filter(|m| open.names(m))
.cloned()
.collect();
let shown = |id: &str| -> String {
match self.machine_of(id) {
Some(m) if m != id && !unfolded.iter().any(|u| u == m) => m.to_string(),
_ => id.to_string(),
}
};
let mut g = self.clone();
g.nodes.retain(|n| shown(&n.id) == n.id);
let mut seen: BTreeSet<(String, String, EdgeKind)> = BTreeSet::new();
g.edges = self
.edges
.iter()
.filter_map(|e| {
let (from, to) = (shown(&e.from), shown(&e.to));
if from == to || !seen.insert((from.clone(), to.clone(), e.kind)) {
return None;
}
Some(FlowEdge {
from,
to,
..e.clone()
})
})
.collect();
// The level is what was asked: everything (3), some machines by name (2), none (1).
g.level = match open {
Open::All => 3,
_ if unfolded.is_empty() => 1,
_ => 2,
};
g.open = unfolded;
g
}
/// The link that toggles `machine` in this picture (`flow?open=…`; `flow?open=` for none).
fn toggle_href(&self, machine: &str) -> String {
let mut open: Vec<&str> = self.open.iter().map(String::as_str).collect();
if let Some(i) = open.iter().position(|m| *m == machine) {
open.remove(i);
} else {
open.push(machine);
}
let order = self.openable.clone();
open.sort_by_key(|m| order.iter().position(|o| o == m));
open_href(open.into_iter().map(short_name))
}
/// Close the columns a band does not use (order kept), then stack each column's
/// boxes in declaration order.
fn tidy(&mut self) {
// The chain's steps stand under the rung they guard, so the flow and the chain
// share one column space; the storage has its own. The estate's columns are fixed
// (console, front, main, twin, downstreams), so a downstream never moves left of
// the main.
let groups: [&[Band]; 2] = [&[Band::Flow, Band::Chain], &[Band::Storage]];
for bands in groups {
let used: BTreeSet<usize> = self
.nodes
.iter()
.filter(|n| bands.contains(&n.band))
.map(|n| n.column)
.collect();
let remap: BTreeMap<usize, usize> =
used.into_iter().enumerate().map(|(i, c)| (c, i)).collect();
for node in self.nodes.iter_mut().filter(|n| bands.contains(&n.band)) {
node.column = remap[&node.column];
}
}
let mut next: BTreeMap<(Band, usize), usize> = BTreeMap::new();
let mut order: Vec<usize> = (0..self.nodes.len()).collect();
order.sort_by_key(|&i| {
(
self.nodes[i].band,
self.nodes[i].column,
self.nodes[i].slot,
i,
)
});
for i in order {
let n = &mut self.nodes[i];
let slot = next.entry((n.band, n.column)).or_insert(0);
n.slot = *slot;
*slot += 1;
}
}
/// One node by id.
pub fn node(&self, id: &str) -> Option<&FlowNode> {
self.nodes.iter().find(|n| n.id == id)
}
fn node_mut(&mut self, id: &str) -> Option<&mut FlowNode> {
self.nodes.iter_mut().find(|n| n.id == id)
}
/// Every node id, sorted.
pub fn node_ids(&self) -> BTreeSet<String> {
self.nodes.iter().map(|n| n.id.clone()).collect()
}
/// Every edge id (`from->to`), sorted.
pub fn edge_ids(&self) -> BTreeSet<String> {
self.edges.iter().map(|e| e.id()).collect()
}
/// The JSON model.
pub fn to_json(&self) -> String {
serde_json::to_string(self)
.unwrap_or_else(|e| format!("{{\"error\":\"the flow model did not serialise: {e}\"}}"))
}
fn legend_kinds(&self) -> Vec<EdgeKind> {
let set: BTreeSet<EdgeKind> = self.edges.iter().map(|e| e.kind).collect();
EdgeKind::ALL
.into_iter()
.filter(|k| set.contains(k))
.collect()
}
}
/// The one-way inbox's folder under the data root: the import door's own
/// (`import_door::BUNDLE_INBOX` is this constant, kept where a build without the import
/// door names it too).
use crate::boundary::INBOX_DIR;
/// Where the order of the machines comes from: left to right, then the downstreams.
fn machine_rank(kind: NodeKind) -> u8 {
match kind {
NodeKind::Console => 0,
NodeKind::Front => 1,
NodeKind::Main => 2,
NodeKind::Twin => 3,
_ => 4,
}
}
/// `flow?open=a,b` (`flow?open=` when nothing is open): relative, so the page, the bare
/// drawing and the console's Flow page each open their own Flow.
fn open_href<'a>(names: impl Iterator<Item = &'a str>) -> String {
let words: Vec<String> = names
.map(|n| {
n.bytes()
.map(|b| {
if b.is_ascii_alphanumeric() || b"-._~".contains(&b) {
(b as char).to_string()
} else {
format!("%{b:02X}")
}
})
.collect()
})
.collect();
if words.is_empty() {
// Nothing open, said: a console's Flow page tells it from its own default.
"flow?open=".into()
} else {
format!("flow?open={}", words.join(","))
}
}
/// The intake chain's `production` repository, when the RON declares one.
fn intake_production(cfg: &crate::Holger) -> Option<String> {
#[cfg(feature = "intake")]
if let Some(i) = cfg.promotion.as_ref().and_then(|p| p.intake.as_ref()) {
return Some(i.production.clone());
}
let _ = cfg;
None
}
/// The intake chain's `incoming` repository, when the RON declares one.
fn intake_incoming(cfg: &crate::Holger) -> Option<String> {
#[cfg(feature = "intake")]
if let Some(i) = cfg.promotion.as_ref().and_then(|p| p.intake.as_ref()) {
return Some(i.incoming.clone());
}
let _ = cfg;
None
}
/// A mirror credential's KIND, never the credential.
fn credential_words(c: &crate::mirror_cycle::MirrorCredential) -> String {
use crate::mirror_cycle::MirrorCredential as C;
match c {
C::Anonymous => "anonymous".into(),
C::Tls { .. } => "TLS (server verified, none presented)".into(),
C::Mtls {
ron_client_cert, ..
} => format!("mTLS client cert {ron_client_cert}"),
C::Vault { ron_entry, .. } => format!("vault {ron_entry}"),
#[allow(unreachable_patterns)]
_ => "declared".into(),
}
}
/// A source's gates, in words.
#[cfg(feature = "intake")]
fn gates_words(g: &crate::sources::Gates) -> String {
let mut w: Vec<String> = Vec::new();
if let Some(p) = &g.provenance {
w.push(format!("provenance {p:?}").to_lowercase());
}
if let Some(c) = &g.cve {
w.push(format!("cve max {:?}", c.max).to_lowercase());
}
if let Some(l) = &g.license {
if !l.deny.is_empty() {
w.push(format!("deny {}", l.deny.join(" ")));
}
}
if g.sbom.is_some() {
w.push("sbom".into());
}
if g.malware.is_some() {
w.push("malware".into());
}
if g.secrets.is_some() {
w.push("secrets".into());
}
if w.is_empty() {
"the floor".into()
} else {
w.join(", ")
}
}
/// The intake chain's steps and gates, as the RON declares them:
/// `(steps, [(gate name, detail, guarded repository, facts)])`.
#[cfg(feature = "intake")]
#[allow(clippy::type_complexity)]
fn chain_of(
cfg: &crate::Holger,
) -> (
Vec<StepDecl>,
Vec<(String, String, String, Vec<Row>)>,
Option<(String, String)>,
) {
let mut steps: Vec<StepDecl> = Vec::new();
let mut gates: Vec<(String, String, String, Vec<Row>)> = Vec::new();
let mut floor_adds: Option<(String, String)> = None;
let Some(p) = cfg.promotion.as_ref() else {
return (steps, gates, floor_adds);
};
if let Some(i) = p.intake.as_ref() {
// The scan, the inspection and the provenance act at SCANNED / POLICY_CHECKED:
// the hop into quarantine. The approvals act at APPROVED.
let scanned_at = i.quarantine.clone();
// The scanners as the RON names them: `steps.scan` when the RON says `steps:`, else
// `promotion.intake.scan.required` as written. What the floor adds beyond them is a
// fact on the repository the scan guards, never a box the RON did not write.
let required: Vec<String> = i
.scan
.as_ref()
.map(|sc| sc.required.iter().map(|k| k.as_str().to_string()).collect())
.unwrap_or_default();
let named: Vec<String> = match &cfg.steps_declared {
Some(st) => st.scan.iter().map(|k| k.as_str().to_string()).collect(),
None => required.clone(),
};
for word in &named {
let evidence = match word.as_str() {
"malware" => "malware",
"license" => "license",
"provenance" => "provenance",
_ => "scan",
};
steps.push(StepDecl {
label: format!("scan · {word}"),
evidence,
scanner: word.clone(),
guards: scanned_at.clone(),
facts: vec![Row::new("required", Cell::Text("yes".into()))],
});
}
let added: Vec<String> = required
.iter()
.filter(|k| !named.contains(k))
.cloned()
.collect();
if !added.is_empty() {
floor_adds = Some((scanned_at.clone(), added.join(", ")));
}
#[cfg(feature = "scanners")]
if let Some(c) = &i.inspect.clamav {
let _ = c;
if !steps.iter().any(|s| s.scanner == "malware") {
steps.push(StepDecl {
label: "malware · clamav".into(),
evidence: "malware",
scanner: "malware".into(),
guards: scanned_at.clone(),
facts: vec![Row::new("engine", Cell::Text("clamav".into()))],
});
}
}
if i.approvals > 0 {
steps.push(StepDecl {
label: "approval".into(),
evidence: "approval",
scanner: String::new(),
guards: i.approved.clone(),
facts: vec![Row::new("approvers", Cell::Count(u64::from(i.approvals)))],
});
}
if let Some(r) = &cfg.rules {
let mut facts = vec![
Row::new(
"signed",
Cell::Text(
if r.require_signed {
"required"
} else {
"not required"
}
.into(),
),
),
Row::new(
"cve max",
Cell::Text(format!("{:?}", r.cve_max).to_lowercase()),
),
];
if !r.deny_licenses.is_empty() {
facts.push(Row::new("deny", Cell::Text(r.deny_licenses.join(" "))));
}
gates.push((
"rules · the floor".into(),
"every source; a gate may only tighten it".into(),
scanned_at.clone(),
facts,
));
}
}
for stage in &p.stages {
for g in &stage.gates {
gates.push((
format!("{g} → {}", stage.repo),
format!("gate on the rung {}", stage.name),
stage.repo.clone(),
Vec::new(),
));
}
}
(steps, gates, floor_adds)
}
// ── the counters ────────────────────────────────────────────────────────────
/// What one repository's audit rows say.
#[derive(Clone, Debug, Default, PartialEq, Eq)]
pub struct RepoTally {
pub fetches: u64,
pub bytes: u64,
pub fetches_24h: u64,
pub bytes_24h: u64,
}
/// What one client's audit rows say.
#[derive(Clone, Debug, Default, PartialEq, Eq)]
pub struct ClientTally {
pub fetches: u64,
pub bytes: u64,
pub last_nanos: i64,
}
/// **The audit log, folded** — successful downloads per repository and per client, and
/// successful promotions per target repository. Pure.
#[derive(Clone, Debug, Default, PartialEq, Eq)]
pub struct AuditTally {
pub repos: BTreeMap<String, RepoTally>,
/// Keyed by client address (port dropped), loopback and paired peers apart.
pub clients: BTreeMap<String, ClientTally>,
pub loopback: ClientTally,
/// Paired peers by principal (`holger-peer:<name>`), bytes.
pub peers: BTreeMap<String, ClientTally>,
/// Target repository → (count, last nanos).
pub promotions: BTreeMap<String, (u64, i64)>,
}
/// The client address of an audit row, without its port.
fn client_of(source_ip: &str) -> String {
crate::audit::client_host(source_ip)
}
/// Is this client address the box talking to itself?
pub fn is_loopback_ident(ident: &str) -> bool {
let s = ident.trim().trim_start_matches('[').trim_end_matches(']');
s == "::1" || s == "localhost" || s == "0:0:0:0:0:0:0:1" || s.starts_with("127.")
}
impl AuditTally {
/// **The tally from the audit log's aggregates** (lane HOL-AUDIT): three
/// aggregates over ONE read of the Arrow index — successful downloads per
/// repository, actor and client (all time and the last 24 h) and successful
/// promotions per target repository. Equal to [`AuditTally::fold`] over the
/// same records (`the_tally_from_the_aggregates_is_the_fold`).
pub fn from_table(
table: &crate::audit::arrow_ipc::AuditTable,
now_nanos: i64,
) -> Result<Self, crate::audit::AuditError> {
use crate::audit::aggregate::{aggregate_table, AggregateQuery, GroupKey};
use crate::audit::query::AuditFilter;
let ok = |action: &str, since: Option<i64>| AuditFilter {
action: Some(action.into()),
status_class: Some(2),
since_nanos: since,
..Default::default()
};
let day_ago = now_nanos - 86_400 * 1_000_000_000;
let all = aggregate_table(
table,
&AggregateQuery {
group_by: vec![GroupKey::Repo, GroupKey::Ident, GroupKey::Client],
filter: ok("download", None),
max_groups: 0,
},
)?;
let day = aggregate_table(
table,
&AggregateQuery {
group_by: vec![GroupKey::Repo],
filter: ok("download", Some(day_ago)),
max_groups: 0,
},
)?;
let promo = aggregate_table(
table,
&AggregateQuery {
group_by: vec![GroupKey::Repo],
filter: ok("promote", None),
max_groups: 0,
},
)?;
let mut t = AuditTally::default();
for g in &all.rows {
let r = t.repos.entry(g.repo.clone()).or_default();
r.fetches += g.count;
r.bytes = r.bytes.saturating_add(g.bytes);
let slot = if let Some(peer) = g.ident.strip_prefix(crate::peers::PEER_PRINCIPAL_PREFIX)
{
t.peers.entry(peer.to_string()).or_default()
} else if g.client.is_empty() || is_loopback_ident(&g.client) {
&mut t.loopback
} else {
t.clients.entry(g.client.clone()).or_default()
};
slot.fetches += g.count;
slot.bytes = slot.bytes.saturating_add(g.bytes);
slot.last_nanos = slot.last_nanos.max(g.last_ts_nanos);
}
for g in &day.rows {
let r = t.repos.entry(g.repo.clone()).or_default();
r.fetches_24h += g.count;
r.bytes_24h = r.bytes_24h.saturating_add(g.bytes);
}
for g in &promo.rows {
let p = t.promotions.entry(g.repo.clone()).or_insert((0, i64::MIN));
p.0 += g.count;
p.1 = p.1.max(g.last_ts_nanos);
}
Ok(t)
}
/// Fold `events` as of `now_nanos`.
pub fn fold(events: &[crate::audit::AuditEvent], now_nanos: i64) -> Self {
use crate::audit::AuditAction;
let day_ago = now_nanos - 86_400 * 1_000_000_000;
let mut t = AuditTally::default();
for e in events {
if !(200..300).contains(&e.status) {
continue;
}
match e.action {
AuditAction::Download => {
let r = t.repos.entry(e.repo.clone()).or_default();
r.fetches += 1;
r.bytes = r.bytes.saturating_add(e.bytes);
if e.ts_nanos >= day_ago {
r.fetches_24h += 1;
r.bytes_24h = r.bytes_24h.saturating_add(e.bytes);
}
let slot = if e.ident.starts_with(crate::peers::PEER_PRINCIPAL_PREFIX) {
t.peers
.entry(
e.ident
.trim_start_matches(crate::peers::PEER_PRINCIPAL_PREFIX)
.to_string(),
)
.or_default()
} else {
let c = client_of(&e.source_ip);
if c.is_empty() || is_loopback_ident(&c) {
&mut t.loopback
} else {
t.clients.entry(c).or_default()
}
};
slot.fetches += 1;
slot.bytes = slot.bytes.saturating_add(e.bytes);
slot.last_nanos = slot.last_nanos.max(e.ts_nanos);
}
AuditAction::Promote => {
let p = t.promotions.entry(e.repo.clone()).or_insert((0, i64::MIN));
p.0 += 1;
p.1 = p.1.max(e.ts_nanos);
}
_ => {}
}
}
t
}
}
/// **The intake record, folded** — durable: every artifact's state now, and the evidence
/// verdicts its transitions carried.
#[derive(Clone, Debug, Default, PartialEq, Eq)]
pub struct IntakeTally {
/// Artifacts per state name (`INGESTED`, `QUARANTINED`, …).
pub states: BTreeMap<String, u64>,
/// Quarantined artifacts per the state they were held at.
pub held_from: BTreeMap<String, u64>,
/// Evidence kind (`scan`, `malware`, `license`, `provenance`, `approval`) → (pass, warn, fail).
pub verdicts: BTreeMap<String, (u64, u64, u64)>,
/// Hops between intake repositories: `(from repo, to repo)` → (count, last at, unix s).
pub hops: BTreeMap<(String, String), (u64, i64)>,
}
#[cfg(feature = "intake")]
impl IntakeTally {
/// Fold a chain's record (bounded: the first `cap` artifacts).
pub fn of(chain: &crate::promote::intake::IntakeChain, cap: usize) -> Self {
use crate::promote::intake::EvidenceVerdict as V;
let settings = chain.settings();
let repo_of = |s: crate::promote::intake::ArtifactState| -> Option<String> {
use crate::promote::intake::IntakeRepo as R;
s.repository().map(|r| match r {
R::Incoming => settings.incoming.clone(),
R::Quarantine => settings.quarantine.clone(),
R::Approved => settings.approved.clone(),
R::Production => settings.production.clone(),
})
};
let mut t = IntakeTally::default();
for (id, status) in chain.list().into_iter().take(cap) {
*t.states.entry(status.state.name().to_string()).or_insert(0) += 1;
if let Some(h) = status.held_from {
*t.held_from.entry(h.name().to_string()).or_insert(0) += 1;
}
for entry in chain.history(&id) {
for ev in &entry.evidence {
let v = t
.verdicts
.entry(ev.kind.name().to_string())
.or_insert((0, 0, 0));
match ev.verdict {
V::Pass => v.0 += 1,
V::Warn => v.1 += 1,
V::Fail => v.2 += 1,
}
}
if let (Some(a), Some(b)) = (repo_of(entry.from), repo_of(entry.to)) {
if a != b {
let h = t.hops.entry((a, b)).or_insert((0, 0));
h.0 += 1;
h.1 = h.1.max(entry.at);
}
}
}
}
t
}
}
/// One downstream as the estate knows it now.
#[derive(Clone, Debug, Default, PartialEq, Eq)]
pub struct DownstreamView {
pub name: String,
/// `front` | `wireguard` | empty (not said).
pub via: String,
/// The server's state word (`up`, `behind`, `silent`, `unknown`) and headline.
pub state: String,
pub headline: String,
pub last_seen_unix_ms: i64,
pub seen_ago_ms: u64,
pub lag_entries: u64,
/// Bytes sent to it, when its peer record counts them.
pub bytes_total: Option<u64>,
pub fetched_total: Option<u64>,
/// Which repositories it mirrors (its pairing record).
pub repos: Vec<String>,
/// Named in the RON's `sync.downstreams`.
pub declared: bool,
/// Paired: enrolled from a signed invite (the sync hub's record), or mTLS-paired.
pub paired: bool,
/// What it said of its own RON in its last report.
pub summary: Option<RonSummary>,
}
/// **What a downstream says of its own RON** (its sync report's `summary`): names, words
/// and folders, never a key. The flow unfolds a downstream to it.
#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize)]
pub struct RonSummary {
/// `OneWay` or `Fetch`.
pub boundary: String,
/// Its repositories, by name.
pub repositories: Vec<String>,
/// Its one-way inbox folder; empty when it fetches.
pub inbox: String,
/// Who allows a vendor's first import (`admin`).
pub first_import: String,
/// Its security domain; empty when it declares none.
pub domain: String,
/// How many downstreams it declares itself.
pub downstreams: u32,
/// Its console's address, as its own front reported it (HOL-FLOW-FRONT); empty when it
/// knows none, or its report did not prove it.
#[serde(default)]
pub console_url: String,
}
impl From<&crate::grpc::holger_grow_proto::DownstreamSummary> for RonSummary {
fn from(s: &crate::grpc::holger_grow_proto::DownstreamSummary) -> Self {
Self {
boundary: s.boundary.clone(),
repositories: s.repositories.clone(),
inbox: s.inbox.clone(),
first_import: s.first_import.clone(),
domain: s.domain.clone(),
downstreams: s.downstreams,
console_url: s.console_url.clone(),
}
}
}
impl From<RonSummary> for crate::grpc::holger_grow_proto::DownstreamSummary {
fn from(s: RonSummary) -> Self {
Self {
boundary: s.boundary,
repositories: s.repositories,
inbox: s.inbox,
first_import: s.first_import,
domain: s.domain,
downstreams: s.downstreams,
console_url: s.console_url,
}
}
}
/// **This Holger's own RON, summarised** for its upstream (the summary its sync reports
/// carry): its repositories, its boundary, its inbox when it is one-way, who allows a first
/// import, its domain, how many downstreams it declares. Nothing secret is read.
pub fn ron_summary(cfg: &crate::Holger) -> RonSummary {
let boundary = cfg.resolved_boundary();
#[cfg(feature = "sync")]
let downstreams = cfg
.sync
.as_ref()
.map(|s| s.downstreams.len() as u32)
.unwrap_or(0);
#[cfg(not(feature = "sync"))]
let downstreams = 0;
RonSummary {
boundary: boundary.name().to_string(),
repositories: cfg
.repositories
.iter()
.map(|r| r.ron_name.clone())
.collect(),
inbox: if boundary == crate::boundary::Boundary::OneWay {
cfg.data_root().join(INBOX_DIR).display().to_string()
} else {
String::new()
},
first_import: format!("{:?}", cfg.trust.first_import).to_lowercase(),
domain: cfg.domain.clone().unwrap_or_default(),
downstreams,
// Filled when a report is sent, from what this box's front last reported.
console_url: String::new(),
}
}
/// One estate box (front, main, twin) as the estate knows it now.
#[derive(Clone, Debug, Default, PartialEq, Eq)]
pub struct EstateBox {
pub name: String,
/// `front` | `primary` | `twin`.
pub role: String,
pub this_box: bool,
pub state: String,
pub headline: String,
pub address: String,
pub twin_state: String,
pub lag_entries: u64,
pub seen_ago_ms: u64,
pub last_seen_unix_ms: i64,
pub disk: Option<(u64, u64)>,
}
/// One of the front's doors, as it reported it (HOL-FLOW-FRONT).
#[derive(Clone, Debug, Default, PartialEq, Eq)]
pub struct FrontDoorView {
/// `https`, `http`, `downstreams`.
pub name: String,
/// `tcp/443`.
pub listen: String,
pub what: String,
}
/// One name the front serves and where it goes, as it reported it (HOL-FLOW-FRONT).
#[derive(Clone, Debug, Default, PartialEq, Eq)]
pub struct FrontRouteView {
pub host: String,
/// The door it comes in by.
pub door: String,
/// `main`, `console`, `monetize`.
pub to: String,
pub what: String,
}
/// **What the front last reported of itself** (HOL-FLOW-FRONT): its doors and routes, proven
/// (`crate::front_report`). The flow unfolds the front to it; nothing else of the front is drawn.
#[derive(Clone, Debug, Default, PartialEq, Eq)]
pub struct FrontView {
/// The front's pairing name, as it signed it.
pub front: String,
pub last_seen_unix_ms: i64,
pub seen_ago_ms: u64,
pub doors: Vec<FrontDoorView>,
pub routes: Vec<FrontRouteView>,
/// Its console's address; empty when none.
pub console_url: String,
}
/// The last time a console (holger-ui) signed in or called this server.
#[derive(Clone, Debug, Default, PartialEq, Eq)]
pub struct ConsoleView {
pub last_seen_unix_ms: i64,
pub seen_ago_ms: u64,
}
/// ★ **When a console (holger-ui) last called this server** (HOL-SYNC-SKEW): ONE per
/// server, shared by its HTTP and gRPC doors like the visitor tally, so a console of one
/// server in a process never draws on another. Marked only by holger-ui's own calls: those
/// that carry its proof (`x-holger-proxy-authorization`, its own administrator's bearer, on
/// the passkey door and the settings door). A person's call, a CLI or a script is no console.
#[derive(Clone, Debug, Default)]
pub struct ConsoleSeen(Arc<std::sync::atomic::AtomicI64>);
impl ConsoleSeen {
/// **The console called, now.**
pub fn mark(&self) {
self.mark_at(crate::audit::now_nanos() / 1_000_000);
}
/// The console called at `now_ms` (Unix ms); an earlier time never moves it back.
pub fn mark_at(&self, now_ms: i64) {
self.0
.fetch_max(now_ms, std::sync::atomic::Ordering::Relaxed);
}
/// When the console was last seen, as of `now_ms`; `None` when never.
pub fn view(&self, now_ms: i64) -> Option<ConsoleView> {
let at = self.0.load(std::sync::atomic::Ordering::Relaxed);
(at > 0).then(|| ConsoleView {
last_seen_unix_ms: at,
seen_ago_ms: now_ms.saturating_sub(at).max(0) as u64,
})
}
}
/// The export push's last bundle, from the export audit: its sequence and when.
#[derive(Clone, Debug, Default, PartialEq, Eq)]
pub struct ExportView {
pub sequence: u64,
pub last_unix: Option<u64>,
}
/// The one-way inbox's record, from the import audit: the bundles imported and refused.
#[derive(Clone, Debug, Default, PartialEq, Eq)]
pub struct InboxView {
pub imported: u64,
pub refused: u64,
pub last_name: String,
/// The last imported bundle's sequence.
pub last_sequence: Option<u64>,
pub last_unix: Option<u64>,
}
/// One scanner kind's tally since boot.
pub use crate::metrics::ScanTally;
/// **Everything [`measure`] reads**, gathered at one instant (a live server through
/// [`Sources::gather`], a test by hand).
#[derive(Clone, Debug, Default)]
pub struct Measured {
pub now_unix: u64,
/// The bounded listing per repository: `(count, bytes stored)`; `None` = could not list.
pub listing: BTreeMap<String, Option<(u64, u64)>>,
/// The paid disk cap per repository, where one is set.
pub quota: BTreeMap<String, u64>,
/// The audit log folded; `None` = this server keeps none.
pub audit: Option<AuditTally>,
/// The visitor tally now.
pub visitors: Option<crate::visitors::Snapshot>,
/// What a puller row is called (the keyed hash), per client address.
pub puller_labels: BTreeMap<String, String>,
/// Per door: 5xx answers since boot.
pub door_errors: BTreeMap<String, u64>,
/// The proxy walks' and mirrors' record.
pub upstreams: Vec<crate::upstreams::UpstreamFact>,
/// Mirror source → (last success unix, cycles ok, cycles failed, fetched, consecutive failures, stalled secs).
pub mirrors: BTreeMap<String, MirrorView>,
/// The intake record; `None` = no intake chain.
pub intake: Option<IntakeTally>,
/// Scanner kind → tally since boot.
pub scans: BTreeMap<String, ScanTally>,
/// Storage endpoint path → (total, used); absent = not measured.
pub storage: BTreeMap<String, (u64, u64)>,
/// The estate's boxes; empty = no estate evidence.
pub estate: Vec<EstateBox>,
/// The estate's name.
pub estate_name: String,
/// The downstreams.
pub downstreams: Vec<DownstreamView>,
/// The console's last sign-in or call; `None` = no console seen.
pub console: Option<ConsoleView>,
/// The front's last report of its doors and routes; `None` = it never reported.
pub front: Option<FrontView>,
/// A link not heard from for longer than this is silent; `0` = the estate's default.
pub silent_after_ms: u64,
/// The export push's record; `None` = none kept.
pub export: Option<ExportView>,
/// The one-way inbox's record; `None` = none kept.
pub inbox: Option<InboxView>,
/// The consumers the door admission counts (people, machines, anonymous today, window days).
pub consumers: Option<(u64, u64, u64, u32)>,
/// The process-wide `/metrics` counters.
pub summary: Vec<Row>,
}
/// One mirror source's health now.
#[derive(Clone, Debug, Default, PartialEq, Eq)]
pub struct MirrorView {
pub last_success_unix: Option<u64>,
pub cycles_ok: u64,
pub cycles_failed: u64,
pub fetched: u64,
pub consecutive_failures: u64,
pub link: String,
}
/// "3m ago" for a unix time, or `never`.
fn ago(then: Option<u64>, now_unix: u64) -> String {
match then {
Some(t) => format!("{} ago", humanise_secs(now_unix.saturating_sub(t))),
None => "never".to_string(),
}
}
/// A count, or `not recorded` when `known` is false.
fn count_or(known: bool, n: u64) -> Cell {
if known {
Cell::Count(n)
} else {
Cell::NotRecorded
}
}
/// **Put the measurements on the shape** — pure. Every counter the owner listed
/// (2026-10-30), each from its source, `not recorded` where there is none.
pub fn measure(shape: &FlowGraph, m: &Measured) -> FlowGraph {
let mut g = shape.clone();
g.summary = m.summary.clone();
g.audit_recorded = m.audit.is_some();
let now = m.now_unix;
// ── repositories ──
let repo_ids: Vec<String> = g
.nodes
.iter()
.filter(|n| n.kind.is_measured_repository())
.map(|n| n.id.clone())
.collect();
for id in &repo_ids {
let mut rows = Vec::new();
let (artifacts, stored) = match m.listing.get(id) {
Some(Some((n, b))) => (Cell::Count(*n), Cell::Bytes(*b)),
_ => (Cell::Unlisted, Cell::Unlisted),
};
rows.push(Row::new("artifacts", artifacts));
rows.push(Row::new("bytes stored", stored));
// A paid repository's disk cap: only a metered build reads one (the open build
// speaks no money, tests/open_build_speaks_no_money.rs).
#[cfg(feature = "metered")]
if let Some(q) = m.quota.get(id) {
rows.push(quota_row(*q));
}
match &m.audit {
Some(a) => {
let t = a.repos.get(id).cloned().unwrap_or_default();
rows.push(Row::new("fetches 24 h", Cell::Count(t.fetches_24h)));
rows.push(Row::new("fetches total", Cell::Count(t.fetches)));
rows.push(Row::new("sent 24 h", Cell::Bytes(t.bytes_24h)));
rows.push(Row::new("sent total", Cell::Bytes(t.bytes)));
}
None => {
for l in ["fetches 24 h", "fetches total", "sent 24 h", "sent total"] {
rows.push(Row::new(l, Cell::NotRecorded));
}
}
}
// A proxy: its upstreams' walks.
let ups: Vec<&FlowEdge> = g
.edges
.iter()
.filter(|e| e.kind == EdgeKind::Pull && &e.to == id)
.collect();
if !ups.is_empty() {
let mut hits = 0u64;
let mut misses = 0u64;
let mut any = false;
for e in &ups {
if let Some(u) = m.upstreams.iter().find(|u| u.name == e.from) {
any = true;
hits += u.fetched;
misses += u.answered.saturating_sub(u.fetched);
}
}
rows.push(Row::new("proxy hits", count_or(any, hits)));
rows.push(Row::new("proxy misses", count_or(any, misses)));
}
if let Some(v) = &m.visitors {
let c = v.repos.get(id).copied().unwrap_or_default();
rows.push(Row::new("requests today", Cell::Count(c.requests)));
rows.push(Row::new("consumers today", consumers_cell(&c)));
}
if let Some(n) = g.node_mut(id) {
n.rows = rows;
}
}
// ── proxy arrows: fetches on miss, the last fetch, errors ──
for e in g.edges.iter_mut().filter(|e| e.kind == EdgeKind::Pull) {
e.cell = match m.upstreams.iter().find(|u| u.name == e.from) {
Some(u) => {
let last = if u.last_ok_ms > 0 {
ago(Some((u.last_ok_ms / 1000) as u64), now)
} else {
"never".into()
};
Cell::Text(format!(
"on miss · {} fetched · last {last} · {} errors",
u.fetched, u.failed
))
}
None => Cell::Text("on miss · not recorded".into()),
};
}
// ── mirror sources: last run, next run, items synced, lag, failures ──
let mirror_ids: Vec<(String, String)> = g
.nodes
.iter()
.filter(|n| n.kind == NodeKind::MirrorSource)
.map(|n| (n.id.clone(), n.label.clone()))
.collect();
for (id, name) in mirror_ids {
let every = g
.node(&id)
.and_then(|n| n.facts.iter().find(|r| r.label == "every"))
.and_then(|r| parse_every(&r.cell.render()));
let rows = match m.mirrors.get(&name) {
Some(h) => {
let next = match (h.last_success_unix, every) {
(Some(t), Some(p)) => {
let at = t + p;
if at <= now {
"due now".to_string()
} else {
format!("in {}", humanise_secs(at - now))
}
}
_ => "at the next cycle".to_string(),
};
let lag = match (h.last_success_unix, every) {
(Some(t), Some(p)) => {
let late = now.saturating_sub(t).saturating_sub(p);
if late == 0 {
"on time".to_string()
} else {
format!("{} late", humanise_secs(late))
}
}
(None, _) => "no good cycle yet".to_string(),
(Some(_), None) => "not declared".to_string(),
};
let mut r = vec![
Row::new("last run", Cell::Text(ago(h.last_success_unix, now))),
Row::new("next run", Cell::Text(next)),
Row::new("items synced", Cell::Count(h.fetched)),
Row::new("lag", Cell::Text(lag)),
Row::new("failures", Cell::Count(h.cycles_failed)),
];
if !h.link.is_empty() {
r.push(Row::new("pin", Cell::Text(h.link.clone())));
}
r
}
None => ["last run", "next run", "items synced", "lag", "failures"]
.into_iter()
.map(|l| Row::new(l, Cell::NotRecorded))
.collect(),
};
let state = m.mirrors.get(&name).and_then(|h| match h.link.as_str() {
"key-mismatch" => Some("bad"),
"unreachable" => Some("warn"),
_ if h.consecutive_failures > 0 => Some("warn"),
_ => None,
});
if let Some(n) = g.node_mut(&id) {
n.rows = rows;
n.state = state;
}
}
// ── sources (upload / pull kinds): what came in through them ──
let source_ids: Vec<(String, String)> = g
.nodes
.iter()
.filter(|n| n.kind == NodeKind::Source)
.map(|n| (n.id.clone(), n.label.clone()))
.collect();
for (id, name) in source_ids {
let rows = match m.upstreams.iter().find(|u| u.name == name) {
Some(u) => vec![
Row::new("fetched", Cell::Count(u.fetched)),
Row::new("errors", Cell::Count(u.failed)),
Row::new(
"last",
Cell::Text(if u.last_ok_ms > 0 {
ago(Some((u.last_ok_ms / 1000) as u64), now)
} else {
"never".into()
}),
),
],
None => vec![Row::new("fetched", Cell::NotRecorded)],
};
if let Some(n) = g.node_mut(&id) {
n.rows = rows;
}
}
// ── the intake chain: steps and gates ──
let chain_ids: Vec<(String, NodeKind)> = g
.nodes
.iter()
.filter(|n| matches!(n.kind, NodeKind::Step | NodeKind::Gate))
.map(|n| (n.id.clone(), n.kind))
.collect();
for (id, kind) in chain_ids {
let detail = g
.node(&id)
.and_then(|n| n.detail.clone())
.unwrap_or_default();
let evidence = detail.strip_prefix("checks ").unwrap_or("").to_string();
let scanner = g.node(&id).map(|n| n.label.clone()).unwrap_or_default();
let scanner = scanner
.strip_prefix("scan · ")
.map(str::to_string)
.unwrap_or_else(|| {
if evidence == "malware" {
"malware".into()
} else {
String::new()
}
});
let mut rows = Vec::new();
match &m.intake {
Some(t) => {
let st = |s: &str| t.states.get(s).copied().unwrap_or(0);
if kind == NodeKind::Step && evidence == "approval" {
rows.push(Row::new(
"waiting for approval",
Cell::Count(st("POLICY_CHECKED")),
));
let (p, _, f) = t.verdicts.get("approval").copied().unwrap_or_default();
rows.push(Row::new("approvals cast", Cell::Count(p)));
rows.push(Row::new("refused", Cell::Count(f)));
rows.push(Row::new(
"quarantined",
Cell::Count(t.held_from.get("POLICY_CHECKED").copied().unwrap_or(0)),
));
} else if kind == NodeKind::Step {
rows.push(Row::new(
"in queue",
Cell::Count(st("INGESTED") + st("SCANNED")),
));
let (p, w, f) = t.verdicts.get(&evidence).copied().unwrap_or_default();
if evidence == "provenance" {
rows.push(Row::new("verified", Cell::Count(p + w)));
rows.push(Row::new("failed", Cell::Count(f)));
} else {
rows.push(Row::new("passed", Cell::Count(p + w)));
rows.push(Row::new("rejected", Cell::Count(f)));
}
rows.push(Row::new(
"quarantined",
Cell::Count(
t.held_from.get("INGESTED").copied().unwrap_or(0)
+ t.held_from.get("SCANNED").copied().unwrap_or(0),
),
));
rows.push(Row::new("blocked", Cell::Count(st("BLOCKED"))));
} else {
// A gate: what stands at its rung, and what is held or blocked.
rows.push(Row::new("quarantined", Cell::Count(st("QUARANTINED"))));
rows.push(Row::new("rejected", Cell::Count(st("BLOCKED"))));
rows.push(Row::new(
"waiting for approval",
Cell::Count(st("POLICY_CHECKED")),
));
}
}
None => {
for l in ["in queue", "passed", "quarantined", "rejected"] {
rows.push(Row::new(l, Cell::NotRecorded));
}
}
}
if kind == NodeKind::Step && !scanner.is_empty() {
match m.scans.get(&scanner) {
Some(s) => {
rows.push(Row::new("scanned since boot", Cell::Count(s.scanned)));
for (sev, n) in &s.findings {
rows.push(Row::new(format!("findings {sev}"), Cell::Count(*n)));
}
if s.findings.is_empty() {
rows.push(Row::new("findings", Cell::Count(0)));
}
}
None => rows.push(Row::new("findings", Cell::NotRecorded)),
}
}
if let Some(n) = g.node_mut(&id) {
n.rows = rows;
}
}
// ── promotion arrows: how many, and the last one ──
for e in g.edges.iter_mut().filter(|e| e.kind == EdgeKind::Promotion) {
let from_intake = m
.intake
.as_ref()
.and_then(|t| t.hops.get(&(e.from.clone(), e.to.clone())).copied());
let from_audit = m
.audit
.as_ref()
.and_then(|a| a.promotions.get(&e.to).copied());
e.cell = match (from_intake, from_audit, &m.audit, &m.intake) {
(Some((n, at)), _, _, _) => {
Cell::Text(format!("{n} · last {}", ago(Some(at.max(0) as u64), now)))
}
(None, Some((n, at)), _, _) => Cell::Text(format!(
"{n} · last {}",
ago(Some((at / 1_000_000_000).max(0) as u64), now)
)),
(None, None, Some(_), _) | (None, None, _, Some(_)) => Cell::Count(0),
_ => Cell::NotRecorded,
};
}
// ── doors: requests served, errors ──
let door_ids: Vec<(String, String)> = g
.nodes
.iter()
.filter(|n| n.kind == NodeKind::Door)
.map(|n| (n.id.clone(), n.label.clone()))
.collect();
for (id, name) in door_ids {
let mut rows = Vec::new();
match &m.visitors {
Some(v) => {
let c = v.doors.get(&name).copied().unwrap_or_default();
rows.push(Row::new("requests today", Cell::Count(c.requests)));
rows.push(Row::new("consumers today", consumers_cell(&c)));
}
None => rows.push(Row::new("requests today", Cell::NotRecorded)),
}
rows.push(Row::new(
"errors since boot",
Cell::Count(m.door_errors.get(&name).copied().unwrap_or(0)),
));
if let Some(n) = g.node_mut(&id) {
n.rows = rows;
}
}
// ── storage endpoints: used / free ──
let storage_ids: Vec<(String, String)> = g
.nodes
.iter()
.filter(|n| n.kind == NodeKind::Storage)
.map(|n| {
(
n.id.clone(),
n.facts
.iter()
.find(|r| r.label == "path")
.map(|r| r.cell.render())
.unwrap_or_default(),
)
})
.collect();
for (id, path) in storage_ids {
let rows = match m.storage.get(&path) {
Some((total, used)) => vec![
Row::new("used", Cell::Bytes(*used)),
Row::new("free", Cell::Bytes(total.saturating_sub(*used))),
],
None => vec![
Row::new("used", Cell::NotRecorded),
Row::new("free", Cell::NotRecorded),
],
};
if let Some(n) = g.node_mut(&id) {
n.rows = rows;
}
}
// ── the diode: the last bundle out, what came in ──
apply_diode(&mut g, m);
// ── the estate: console, front, main, twins ──
apply_estate(&mut g, m);
// ── the downstreams ──
apply_downstreams(&mut g, m);
// ── the observed pullers ──
apply_pullers(&mut g, m);
g.tidy();
g.finish();
g
}
/// The paid repository's disk cap as a row (metered builds only).
#[cfg(feature = "metered")]
fn quota_row(bytes: u64) -> Row {
Row::new("quota", Cell::Bytes(bytes))
}
/// A distinct-client count as the picture prints it.
pub fn consumers_cell(c: &crate::visitors::Count) -> Cell {
if c.saturated {
Cell::Text(format!("≥ {}", group_thousands(c.consumers)))
} else if c.consumers == 0 {
Cell::NoConsumer
} else {
Cell::Count(c.consumers)
}
}
fn state_class(state: &str) -> Option<&'static str> {
match state {
"up" => Some("good"),
"behind" => Some("warn"),
"silent" => Some("bad"),
_ => Some("none"),
}
}
fn seen_words(last_seen_unix_ms: i64, seen_ago_ms: u64) -> String {
if last_seen_unix_ms <= 0 {
"never".into()
} else {
format!("{} ago", humanise_secs(seen_ago_ms / 1000))
}
}
use holger_downstream_rules::front::{TO_CONSOLE, TO_MAIN};
/// The console, the front, the main and its twins, from the estate's rows.
fn apply_estate(g: &mut FlowGraph, m: &Measured) {
g.nodes.retain(|n| {
!matches!(
n.kind,
NodeKind::Console
| NodeKind::Front
| NodeKind::Main
| NodeKind::Twin
| NodeKind::Mirror
)
});
g.edges.retain(|e| {
!matches!(
e.kind,
EdgeKind::Console | EdgeKind::Fronts | EdgeKind::Follows
)
});
let base =
|id: String, label: String, kind: NodeKind, detail: String, column: usize| FlowNode {
machine: Some(id.clone()),
link: None,
id,
label,
kind,
detail: Some(detail),
stage: None,
gated: false,
facts: Vec::new(),
rows: Vec::new(),
band: Band::Estate,
column,
slot: 0,
state: None,
};
let main = m
.estate
.iter()
.find(|b| b.role == crate::topology::ROLE_PRIMARY);
// This server: always drawn — it is the box answering.
{
let mut n = base(
MAIN_ID.to_string(),
main.map(|b| b.name.clone())
.unwrap_or_else(|| "this holger".into()),
NodeKind::Main,
if m.estate_name.is_empty() {
"holger-server".into()
} else {
format!("holger-server · estate {}", m.estate_name)
},
2,
);
if let Some(b) = main {
n.facts.push(Row::new(
"address",
if b.address.is_empty() {
Cell::NotDeclared
} else {
Cell::Text(b.address.clone())
},
));
n.rows.push(Row::new(
"state",
Cell::Text(if b.headline.is_empty() {
b.state.clone()
} else {
b.headline.clone()
}),
));
if b.this_box {
n.rows.push(Row::new("this box", Cell::Text("yes".into())));
}
if let Some((total, used)) = b.disk {
n.rows.push(Row::new("disk used", Cell::Bytes(used)));
n.rows.push(Row::new(
"disk free",
Cell::Bytes(total.saturating_sub(used)),
));
}
n.state = state_class(&b.state);
} else {
n.rows
.push(Row::new("state", Cell::Text("answering".into())));
}
// What an operator needs first, on the box itself (level 1).
let listed: Vec<u64> = g
.nodes
.iter()
.filter(|r| r.kind.is_measured_repository() && r.machine.as_deref() == Some(MAIN_ID))
.filter_map(|r| m.listing.get(&r.id).copied().flatten().map(|(c, _)| c))
.collect();
n.rows.push(Row::new(
"artifacts",
if listed.is_empty() {
Cell::NotRecorded
} else {
Cell::Count(listed.iter().sum())
},
));
let state = |s: &str| match &m.intake {
Some(t) => Cell::Count(t.states.get(s).copied().unwrap_or(0)),
None => Cell::NotRecorded,
};
n.rows
.push(Row::new("waiting for approval", state("POLICY_CHECKED")));
n.rows.push(Row::new("quarantined", state("QUARANTINED")));
g.nodes.push(n);
}
// Where the front and the console reach this server: its doors, or the box itself.
let doors: Vec<String> = g
.nodes
.iter()
.filter(|n| n.kind == NodeKind::Door && n.machine.as_deref() == Some(MAIN_ID))
.map(|n| n.id.clone())
.collect();
let into_main: Vec<String> = if doors.is_empty() {
vec![MAIN_ID.to_string()]
} else {
doors
};
// Each front, and — when it reported them (HOL-FLOW-FRONT) — its doors and the names it
// serves inside it: a front that never reported is one closed box, and its line runs from
// the box itself. `(front, the part the console comes in at)`.
let front_boxes: Vec<&EstateBox> = m
.estate
.iter()
.filter(|b| b.role == crate::topology::ROLE_FRONT)
.collect();
let mut fronts: Vec<(String, Option<String>)> = Vec::new();
for b in &front_boxes {
let id = estate_node_id(&b.name);
let mut n = base(
id.clone(),
b.name.clone(),
NodeKind::Front,
"estate front · TLS · plugin door".into(),
1,
);
n.facts.push(Row::new(
"address",
if b.address.is_empty() {
Cell::NotDeclared
} else {
Cell::Text(b.address.clone())
},
));
// Its report: signed by the front as itself; the only front of the estate may sign
// under another spelling of its name.
let report = m
.front
.as_ref()
.filter(|f| f.front == b.name || front_boxes.len() == 1);
match report {
Some(f) => {
n.facts.push(Row::new(
"reported",
Cell::Text(seen_words(f.last_seen_unix_ms, f.seen_ago_ms)),
));
if !f.console_url.is_empty() {
n.facts
.push(Row::new("console", Cell::Text(f.console_url.clone())));
}
}
// An older front sends none: say so, never nothing.
None => n
.facts
.push(Row::new("report", Cell::Text("no report".into()))),
}
// The front's own counters live on the front: nothing on this main records them.
n.rows.push(Row::new("requests", Cell::NotRecorded));
n.rows.push(Row::new("peers", Cell::NotRecorded));
n.rows.push(Row::new(
"state",
Cell::Text(if b.headline.is_empty() {
b.state.clone()
} else {
b.headline.clone()
}),
));
n.state = state_class(&b.state);
g.nodes.push(n);
// What reaches main through it, and where the console comes in.
let mut carries: Vec<String> = Vec::new();
let mut console_at: Option<String> = None;
if let Some(f) = report {
let part =
|pid: String, label: String, kind: NodeKind, detail: String, column: usize| {
let mut p = base(pid, label, kind, detail, column);
p.machine = Some(id.clone());
p.band = Band::Flow;
p
};
for d in &f.doors {
let did = format!("{id}:door:{}", d.name);
if g.node(&did).is_some() {
continue;
}
let mut p = part(did, d.name.clone(), NodeKind::Door, d.what.clone(), 0);
p.facts
.push(Row::new("listens", Cell::Text(d.listen.clone())));
g.nodes.push(p);
}
for r in &f.routes {
let hid = format!("{id}:host:{}", r.host);
if g.node(&hid).is_some() {
continue;
}
let mut p = part(
hid.clone(),
r.host.clone(),
NodeKind::Host,
r.what.clone(),
1,
);
p.facts.push(Row::new("door", Cell::Text(r.door.clone())));
p.facts.push(Row::new("to", Cell::Text(r.to.clone())));
g.nodes.push(p);
let did = format!("{id}:door:{}", r.door);
if g.node(&did).is_some() {
g.edges.push(FlowEdge::new(
did,
hid.clone(),
EdgeKind::Routes,
Cell::Text(String::new()),
));
}
if r.to == TO_MAIN || r.to == TO_CONSOLE {
carries.push(hid.clone());
}
if r.to == TO_CONSOLE && console_at.is_none() {
console_at = Some(hid);
}
}
}
if carries.is_empty() {
carries.push(id.clone());
}
for from in &carries {
for to in &into_main {
g.edges.push(FlowEdge::new(
from.clone(),
to.clone(),
EdgeKind::Fronts,
Cell::Text(String::new()),
));
}
}
fronts.push((id, console_at));
}
// The console: only when this installation has one — a console box in the estate, or a
// console that signed in or called recently. A front alone is no console.
let console_box = m
.estate
.iter()
.any(|b| b.role == crate::topology::ROLE_CONSOLE);
let seen = m
.console
.as_ref()
.filter(|c| c.seen_ago_ms <= CONSOLE_SEEN_WITHIN_MS);
if console_box || seen.is_some() {
let mut c = base(
CONSOLE_ID.into(),
"console".into(),
NodeKind::Console,
if fronts.is_empty() {
"holger-ui".into()
} else {
"holger-ui, served by the front".into()
},
0,
);
c.rows.push(Row::new(
"last seen",
match seen {
Some(s) => Cell::Text(seen_words(s.last_seen_unix_ms, s.seen_ago_ms)),
None => Cell::NotRecorded,
},
));
g.nodes.push(c);
// Through the front: at the name it serves the console at, when it reported one.
let to: Vec<String> = match fronts.first() {
None => into_main.clone(),
Some((f, at)) => vec![at.clone().unwrap_or_else(|| f.clone())],
};
for t in &to {
g.edges.push(FlowEdge::new(
CONSOLE_ID,
t.clone(),
EdgeKind::Console,
Cell::Text(String::new()),
));
}
}
for b in m
.estate
.iter()
.filter(|b| b.role == crate::topology::ROLE_TWIN)
{
let id = estate_node_id(&b.name);
let mut n = base(
id.clone(),
b.name.clone(),
NodeKind::Twin,
"holger-server · the follower".into(),
3,
);
let following = matches!(b.twin_state.as_str(), "ok" | "behind" | "stalled");
let behind = if b.twin_state.is_empty() {
Cell::NotRecorded
} else {
Cell::Count(b.lag_entries)
};
let last = Cell::Text(seen_words(b.last_seen_unix_ms, b.seen_ago_ms));
n.rows.push(Row::new(
"following",
Cell::Text(if following { "yes" } else { "no" }.into()),
));
n.rows.push(Row::new("entries behind", behind.clone()));
n.rows.push(Row::new("last sync", last.clone()));
if let Some((total, used)) = b.disk {
n.rows.push(Row::new(
"disk free",
Cell::Bytes(total.saturating_sub(used)),
));
}
n.state = state_class(&b.state);
g.nodes.push(n);
// Its part: the mirror of this main, and its lag.
let mirror = format!("{id}:mirror");
let mut p = base(
mirror.clone(),
"mirror".into(),
NodeKind::Mirror,
"every repository of the main, followed".into(),
0,
);
p.machine = Some(id.clone());
p.band = Band::Flow;
p.facts.push(Row::new(
"of",
Cell::Text(
main.map(|b| b.name.clone())
.unwrap_or_else(|| "this holger".into()),
),
));
p.rows.push(Row::new("entries behind", behind));
p.rows.push(Row::new("last sync", last));
p.state = state_class(&b.state);
g.nodes.push(p);
g.edges.push(FlowEdge::new(
MAIN_ID,
mirror,
EdgeKind::Follows,
Cell::Text(if b.headline.is_empty() {
b.state.clone()
} else {
b.headline.clone()
}),
));
}
}
/// The diode drop area's last bundle and the inbox's imports, each from its own record.
fn apply_diode(g: &mut FlowGraph, m: &Measured) {
let now = m.now_unix;
if let Some(n) = g.node_mut(DIODE_ID) {
n.rows = vec![Row::new(
"last bundle",
match &m.export {
Some(x) => Cell::Text(match x.last_unix {
Some(t) => format!("{} · {}", x.sequence, ago(Some(t), now)),
None => x.sequence.to_string(),
}),
None => Cell::NotRecorded,
},
)];
}
if let Some(n) = g.node_mut(INBOX_ID) {
n.rows = match &m.inbox {
Some(i) => {
let mut r = vec![
Row::new("imported", Cell::Count(i.imported)),
Row::new("refused", Cell::Count(i.refused)),
Row::new("last import", Cell::Text(ago(i.last_unix, now))),
];
if !i.last_name.is_empty() {
r.push(Row::new(
"last bundle",
Cell::Text(match i.last_sequence {
Some(n) => format!("{} · seq {n}", i.last_name),
None => i.last_name.clone(),
}),
));
}
r
}
None => vec![
Row::new("imported", Cell::NotRecorded),
Row::new("refused", Cell::NotRecorded),
],
};
}
}
/// **What a downstream's line is known from, and how it looks**: `(look, known_from)`.
/// Declared or paired AND seen within `silent_after_ms` is live; declared or paired but not
/// seen is silent; neither declared nor paired but seen, lately or long ago (`ever_seen`), is
/// unknown (the spec's grey "?", never the silent dash of one that is declared).
fn link_look(
declared: bool,
paired: bool,
seen: bool,
ever_seen: bool,
) -> (&'static str, Vec<&'static str>) {
let known: Vec<&'static str> = [(declared, "declared"), (paired, "paired"), (seen, "seen")]
.into_iter()
.filter_map(|(on, w)| on.then_some(w))
.collect();
let look = match (declared || paired, seen) {
(true, true) => "live",
(true, false) => "silent",
(false, _) if seen || ever_seen => "unknown",
// Nothing knows it, and it was never seen: no source says anything of it.
(false, _) => "silent",
};
(look, known)
}
/// The words on a downstream's line: how it reaches this holger, its lag, its last
/// contact — or that it is silent, or that nothing declares or pairs it.
fn sync_words(d: &DownstreamView, look: &str) -> String {
let mut w = Vec::new();
if look == "unknown" {
w.push("? seen, not declared or paired".to_string());
}
if !d.via.is_empty() {
w.push(format!("via {}", d.via));
}
if look == "silent" && d.lag_entries > 0 {
w.push(format!("lag {}", d.lag_entries));
}
match look {
"silent" if d.last_seen_unix_ms <= 0 => w.push("never seen".into()),
"silent" => w.push(format!(
"not seen since {}",
seen_words(d.last_seen_unix_ms, d.seen_ago_ms)
)),
_ => {
w.push(format!("lag {}", d.lag_entries));
w.push(format!(
"seen {}",
seen_words(d.last_seen_unix_ms, d.seen_ago_ms)
));
}
}
w.join(" · ")
}
/// Every downstream by name — declared, paired or enrolled — below the main, the arrow
/// from what it takes on the main to it. Never an upstream.
fn apply_downstreams(g: &mut FlowGraph, m: &Measured) {
g.edges.retain(|e| e.kind != EdgeKind::Syncs);
let silent_after = if m.silent_after_ms == 0 {
crate::topology::SILENT_AFTER_MS
} else {
m.silent_after_ms
};
// The ones the RON declares are in the shape already.
let in_the_ron: BTreeSet<String> = g
.nodes
.iter()
.filter(|n| n.kind == NodeKind::Downstream)
.map(|n| n.id.clone())
.collect();
for d in &m.downstreams {
let id = downstream_node_id(&d.name);
if g.node(&id).is_none() {
g.nodes.push(FlowNode {
id: id.clone(),
label: d.name.clone(),
kind: NodeKind::Downstream,
detail: Some("paired or enrolled".into()),
stage: None,
gated: false,
facts: Vec::new(),
rows: Vec::new(),
band: Band::Estate,
column: 4,
slot: 0,
state: None,
machine: Some(id.clone()),
link: None,
});
}
let n = g.node_mut(&id).expect("pushed above");
n.rows = vec![
Row::new(
"via",
if d.via.is_empty() {
Cell::NotRecorded
} else {
Cell::Text(d.via.clone())
},
),
Row::new(
"status",
Cell::Text(if d.headline.is_empty() {
d.state.clone()
} else {
d.headline.clone()
}),
),
Row::new(
"last sync",
Cell::Text(seen_words(d.last_seen_unix_ms, d.seen_ago_ms)),
),
Row::new("entries behind", Cell::Count(d.lag_entries)),
Row::new(
"bytes sent",
d.bytes_total.map(Cell::Bytes).unwrap_or(Cell::NotRecorded),
),
];
if let Some(f) = d.fetched_total {
n.rows.push(Row::new("packages fetched", Cell::Count(f)));
}
n.state = state_class(&d.state);
}
// A declared downstream nothing has measured yet.
for n in g
.nodes
.iter_mut()
.filter(|n| n.kind == NodeKind::Downstream && n.rows.is_empty())
{
n.rows = vec![
Row::new("status", Cell::Text("not measured".into())),
Row::new("last sync", Cell::NotRecorded),
Row::new("entries behind", Cell::NotRecorded),
Row::new("bytes sent", Cell::NotRecorded),
];
n.state = Some("none");
}
let downs: Vec<String> = g
.nodes
.iter()
.filter(|n| n.kind == NodeKind::Downstream)
.map(|n| n.id.clone())
.collect();
let on_main = |id: &str, g: &FlowGraph| {
g.node(id)
.is_some_and(|n| n.machine.as_deref() == Some(MAIN_ID) && n.id != MAIN_ID)
};
for id in downs {
let view = m
.downstreams
.iter()
.find(|d| downstream_node_id(&d.name) == id);
// What it takes on the main: the repositories its pairing names, else the
// production the chain serves, else the main as a whole.
let mut takes: Vec<String> = view
.map(|d| d.repos.iter().filter(|r| on_main(r, g)).cloned().collect())
.unwrap_or_default();
if takes.is_empty() {
takes = g
.served_from
.clone()
.filter(|p| on_main(p, g))
.into_iter()
.collect();
}
if takes.is_empty() {
takes.push(MAIN_ID.into());
}
let declared = in_the_ron.contains(&id) || view.is_some_and(|d| d.declared);
let paired = view.is_some_and(|d| d.paired);
let seen = view.is_some_and(|d| d.last_seen_unix_ms > 0 && d.seen_ago_ms <= silent_after);
let ever_seen = view.is_some_and(|d| d.last_seen_unix_ms > 0);
let (look, known) = link_look(declared, paired, seen, ever_seen);
let words = match view {
Some(d) => sync_words(d, look),
None => "declared · never seen".into(),
};
if let Some(n) = g.node_mut(&id) {
n.facts.retain(|r| r.label != "known from");
n.facts
.push(Row::new("known from", Cell::Text(known.join(" · "))));
}
// Another Holger server whose console it reported (HOL-FLOW-FRONT, Henrik's Q4): its
// box leads to THAT server's own flow; with no console known there is nowhere to go.
if let Some(n) = g.node_mut(&id) {
n.facts.retain(|r| r.label != OWN_FLOW_ROW);
n.link = view
.and_then(|d| d.summary.as_ref())
.and_then(|s| own_flow_href(&s.console_url));
if let Some(href) = n.link.clone() {
let shown = href.split_once("://").map(|(_, r)| r).unwrap_or(&href);
n.facts
.push(Row::new(OWN_FLOW_ROW, Cell::Text(format!("↗ {shown}"))));
}
}
// What it said of its own RON: its box's facts, and its parts inside it.
let fills = match view.and_then(|d| d.summary.as_ref()) {
Some(s) => unfold_downstream(g, &id, s),
None => {
// An older downstream sends none (HOL-SYNC-SKEW): say so, never nothing.
if let Some(n) = g.node_mut(&id) {
n.facts.retain(|r| r.label != "summary");
n.facts
.push(Row::new("summary", Cell::Text("no summary".into())));
}
vec![id.clone()]
}
};
for from in takes {
for to in &fills {
let mut e = FlowEdge::new(
from.clone(),
to.clone(),
EdgeKind::Syncs,
Cell::Text(words.clone()),
);
e.look = Some(look);
e.known_from = known.clone();
g.edges.push(e);
}
}
}
}
/// The fact row a box that leads to another server's own flow carries; the link lies over it.
pub const OWN_FLOW_ROW: &str = "its own flow";
/// Pure: **where a server's own flow is**, from its console's address: `<console>/flow`, when
/// the address is a plain `https://` URL (or `http://` on a loopback host,
/// `holger_downstream_rules::front::console_url`); `None` otherwise.
pub fn own_flow_href(console_url: &str) -> Option<String> {
holger_downstream_rules::front::console_url(console_url).map(|u| format!("{u}/flow"))
}
/// **A downstream's own RON, inside its border**: its boundary, domain and how many it is
/// upstream of on its box, a part per repository (and its inbox when it is one-way).
/// Returns what the line from this holger fills (its repositories, or its inbox).
fn unfold_downstream(g: &mut FlowGraph, id: &str, s: &RonSummary) -> Vec<String> {
let name = g
.node(id)
.map(|n| n.label.clone())
.unwrap_or_else(|| short_name(id).to_string());
if let Some(n) = g.node_mut(id) {
n.facts
.push(Row::new("boundary", Cell::Text(s.boundary.clone())));
n.facts.push(Row::new(
"domain",
if s.domain.is_empty() {
Cell::NotDeclared
} else {
Cell::Text(s.domain.clone())
},
));
if s.downstreams > 0 {
n.facts.push(Row::new(
"upstream of",
Cell::Count(u64::from(s.downstreams)),
));
}
}
let part =
|pid: String, label: String, kind: NodeKind, detail: String, column: usize| FlowNode {
machine: Some(id.to_string()),
link: None,
id: pid,
label,
kind,
detail: Some(detail),
stage: None,
gated: false,
facts: Vec::new(),
rows: Vec::new(),
band: Band::Flow,
column,
slot: 0,
state: None,
};
let repos: Vec<String> = s.repositories.iter().map(|r| format!("{id}:{r}")).collect();
for (r, pid) in s.repositories.iter().zip(&repos) {
let mut p = part(
pid.clone(),
r.clone(),
NodeKind::Repository,
format!("on {name}"),
1,
);
p.facts
.push(Row::new("first import", Cell::Text(s.first_import.clone())));
g.nodes.push(p);
}
if s.inbox.is_empty() {
return repos;
}
let inbox = format!("{id}:inbox");
g.nodes.push(part(
inbox.clone(),
"inbox".into(),
NodeKind::Inbox,
s.inbox.clone(),
0,
));
for r in &repos {
g.edges.push(FlowEdge::new(
inbox.clone(),
r.clone(),
EdgeKind::Carries,
Cell::Text(String::new()),
));
}
vec![inbox]
}
/// The observed pullers, one box: count and the busiest by bytes.
fn apply_pullers(g: &mut FlowGraph, m: &Measured) {
g.nodes.retain(|n| n.kind != NodeKind::Consumer);
g.edges.retain(|e| e.kind != EdgeKind::Served);
let column = g
.nodes
.iter()
.filter(|n| n.band == Band::Flow)
.map(|n| n.column)
.max()
.map(|c| c + 1)
.unwrap_or(0);
let mut rows = Vec::new();
let detail;
match &m.audit {
None => {
detail = "audit is off · nothing recorded".to_string();
rows.push(Row::new("pullers", Cell::NotRecorded));
rows.push(Row::new("fetches", Cell::NotRecorded));
rows.push(Row::new("bytes sent", Cell::NotRecorded));
}
Some(a) => {
detail = "audit log · loopback excluded".to_string();
let mut list: Vec<(&String, &ClientTally)> = a.clients.iter().collect();
list.sort_by(|x, y| y.1.bytes.cmp(&x.1.bytes).then(x.0.cmp(y.0)));
rows.push(Row::new("pullers", Cell::Count(list.len() as u64)));
rows.push(Row::new(
"fetches",
Cell::Count(list.iter().map(|c| c.1.fetches).sum()),
));
rows.push(Row::new(
"bytes sent",
Cell::Bytes(list.iter().map(|c| c.1.bytes).sum()),
));
rows.push(Row::new("loopback", Cell::Count(a.loopback.fetches)));
for (addr, c) in list.into_iter().take(g.max_consumers) {
let label = m
.puller_labels
.get(addr)
.cloned()
.unwrap_or_else(|| "client ?".into());
rows.push(Row::new(label, Cell::Bytes(c.bytes)));
}
}
}
if let Some((people, machines, anonymous, days)) = m.consumers {
rows.push(Row::new(format!("people ({days} d)"), Cell::Count(people)));
rows.push(Row::new(
format!("machines ({days} d)"),
Cell::Count(machines),
));
rows.push(Row::new("anonymous today", Cell::Count(anonymous)));
}
g.nodes.push(FlowNode {
id: PULLERS_NODE.into(),
label: PULLERS_NODE.into(),
kind: NodeKind::Consumer,
detail: Some(detail),
stage: None,
gated: false,
facts: Vec::new(),
rows,
band: Band::Flow,
column,
slot: 0,
state: None,
// What this server's audit saw: drawn inside it.
machine: Some(MAIN_ID.into()),
link: None,
});
if let Some(a) = &m.audit {
for (repo, t) in &a.repos {
if t.fetches > 0 && g.nodes.iter().any(|n| &n.id == repo) {
g.edges.push(FlowEdge {
from: repo.clone(),
to: PULLERS_NODE.into(),
kind: EdgeKind::Served,
cell: Cell::Count(t.fetches),
from_port: None,
to_port: None,
look: None,
known_from: Vec::new(),
});
}
}
}
}
/// Seconds out of a humanised period (`5m`, `24h`, `1h 30m`).
fn parse_every(s: &str) -> Option<u64> {
let s = s.trim().trim_start_matches("every").trim();
let mut total = 0u64;
let mut any = false;
for part in s.split_whitespace() {
let (num, unit) = part.split_at(part.find(|c: char| !c.is_ascii_digit())?);
let n: u64 = num.parse().ok()?;
total += n * match unit {
"s" => 1,
"m" => 60,
"h" => 3600,
"d" => 86_400,
_ => return None,
};
any = true;
}
any.then_some(total)
}
// ── the live server ─────────────────────────────────────────────────────────
/// A paid repository's disk cap, when one is set (metered builds).
pub type QuotaOf = Arc<dyn Fn(&str) -> Option<u64> + Send + Sync>;
/// **What a door holds of the flow**: the shape (read once from the RON) and the quota
/// reader. The HTTP gateway and the gRPC service each hold one, built by
/// [`crate::Holger::flow_door`] from the same RON.
#[derive(Clone)]
pub struct Door {
pub shape: Arc<FlowGraph>,
pub quota: Option<QuotaOf>,
}
/// **The live server's evidence**, one handle both doors build the same way.
#[derive(Clone)]
pub struct Sources {
/// The shape, read once at boot ([`FlowGraph::from_config`]).
pub shape: Arc<FlowGraph>,
pub repos: Vec<(String, Arc<dyn traits::RepositoryBackendTrait>)>,
pub audit: Arc<dyn crate::audit::AuditLog>,
pub visitors: Arc<crate::visitors::Visitors>,
/// When this server's console last called ([`ConsoleSeen`]).
pub console: ConsoleSeen,
/// The estate board's evidence (`None` on a bare gateway).
pub board: Option<crate::topology::Sources>,
#[cfg(feature = "intake")]
pub intake: Option<Arc<crate::promote::intake::IntakeChain>>,
pub quota: Option<QuotaOf>,
}
/// How many artifacts the intake fold reads at most.
#[cfg(feature = "intake")]
const INTAKE_CAP: usize = 100_000;
impl Sources {
/// Read every input now. BLOCKS (listings, the audit log, statvfs): call it off the
/// executor.
pub fn gather(&self) -> Measured {
let now_unix = crate::visitors::now_unix();
let mut m = Measured {
now_unix,
..Default::default()
};
for (name, backend) in &self.repos {
let listed = backend
.list(None, crate::exposed::http::METRICS_LIST_CAP)
.ok()
.map(|v| {
let bytes: u64 = v.iter().map(|e| e.size_bytes.max(0) as u64).sum();
(v.len() as u64, bytes)
});
m.listing.insert(name.clone(), listed);
if let Some(q) = self.quota.as_ref().and_then(|q| q(name)) {
m.quota.insert(name.clone(), q);
}
}
// The audit counters are the audit log's aggregates (its Arrow index); a
// server without a readable log, or an index that cannot be read, leaves
// `m.audit` empty and every such counter reads "not recorded".
if let Some(table) = self.audit.query_source().and_then(|s| s.table().ok()) {
if let Ok(tally) = AuditTally::from_table(&table, crate::audit::now_nanos()) {
for addr in tally.clients.keys() {
m.puller_labels
.insert(addr.clone(), self.visitors.puller_label(addr, now_unix));
}
m.audit = Some(tally);
}
// The diode's counters are the export and import audit's (HOL-SYNC-SKEW), read
// only where the RON draws the diode drop area or the inbox.
let (diode, inbox) = (
self.shape.node(DIODE_ID).is_some(),
self.shape.node(INBOX_ID).is_some(),
);
if diode || inbox {
let (export, imports) = airgap_of_audit(&table);
m.export = export.filter(|_| diode);
m.inbox = inbox.then_some(imports);
}
}
m.visitors = Some(self.visitors.snapshot(now_unix));
let metrics = crate::metrics::global();
m.door_errors = metrics.door_errors();
m.scans = metrics.scan_tally();
m.upstreams = crate::upstreams::snapshot();
for h in metrics.mirror_sources() {
let (ok, failed, fetched, _, _) = h.counts();
m.mirrors.insert(
h.name().to_string(),
MirrorView {
last_success_unix: h.last_success_unix(),
cycles_ok: ok,
cycles_failed: failed,
fetched,
consecutive_failures: h.consecutive_failures(),
link: h.link_state().word().to_string(),
},
);
}
#[cfg(feature = "intake")]
if let Some(chain) = &self.intake {
m.intake = Some(IntakeTally::of(chain, INTAKE_CAP));
}
for n in self
.shape
.nodes
.iter()
.filter(|n| n.kind == NodeKind::Storage)
{
if let Some(path) = n
.facts
.iter()
.find(|r| r.label == "path")
.map(|r| r.cell.render())
{
if let Ok(f) = crate::metering::statvfs(std::path::Path::new(&path)) {
m.storage.insert(path, (f.total_bytes, f.used_bytes));
}
}
}
m.console = self.console.view(now_unix as i64 * 1000);
// The front's own report of its doors and routes (HOL-FLOW-FRONT).
if let Some(root) = self.board.as_ref().and_then(|b| b.data_root.as_deref()) {
m.front = crate::front_report::kept(root)
.map(|k| crate::front_report::view(&k, now_unix as i64 * 1000));
}
if let Some(board) = &self.board {
let facts = board.gather();
let reply = crate::topology::board(&facts);
let peers = self.audit_peers(&m);
estate_of(&mut m, &facts, &reply, peers);
}
m.summary = vec![
Row::new(
"fetches",
Cell::Count(metrics.request_count(crate::metrics::Verb::Get)),
),
Row::new(
"uploads",
Cell::Count(metrics.request_count(crate::metrics::Verb::Put)),
),
Row::new(
"promotions",
Cell::Count(metrics.request_count(crate::metrics::Verb::Promote)),
),
Row::new("bytes served", Cell::Bytes(metrics.bytes_served())),
Row::new("errors", Cell::Count(metrics.errors())),
];
m
}
fn audit_peers(&self, m: &Measured) -> BTreeMap<String, u64> {
m.audit
.as_ref()
.map(|a| a.peers.iter().map(|(k, v)| (k.clone(), v.bytes)).collect())
.unwrap_or_default()
}
}
/// The export push's actor in the audit (`export_push`).
const EXPORT_PUSH_ACTOR: &str = "export-push";
/// The number after `key` in an audit detail (`seq=3`, `seq 7`).
fn number_after(detail: &str, key: &str) -> Option<u64> {
let rest = &detail[detail.find(key)? + key.len()..];
let digits: String = rest.chars().take_while(char::is_ascii_digit).collect();
digits.parse().ok()
}
/// The word after `key` in an audit detail (`bundle_id=acme-pkgs-3`).
fn word_after<'a>(detail: &'a str, key: &str) -> Option<&'a str> {
let rest = &detail[detail.find(key)? + key.len()..];
rest.split_whitespace().next()
}
/// Pure: **the diode's counters, from the audit** (HOL-SYNC-SKEW), never a file's mtime or a
/// folder's listing:
///
/// * the export push's last bundle: its `airgap export <id> seq <n>` row (actor
/// `export-push`, `export_door`), its sequence and time; `None` when it pushed none;
/// * the inbox: one `bundle-import: bundle_id=<id> seq=<n>` row per imported bundle (the
/// per-payload rows are not bundles), one `bundle import refused:` row per refused one
/// (`import_door`), and the last import's bundle, sequence and time.
pub fn airgap_of_audit(
table: &crate::audit::arrow_ipc::AuditTable,
) -> (Option<ExportView>, InboxView) {
let (mut export, mut inbox) = (None::<(i64, ExportView)>, InboxView::default());
let mut last_import = i64::MIN;
let _ = table.for_each_row(|r| {
let ok = (200..300).contains(&r.status);
let secs = u64::try_from(r.ts_nanos / 1_000_000_000).ok();
if ok && r.ident == EXPORT_PUSH_ACTOR && r.detail.starts_with("airgap export ") {
if export.as_ref().is_none_or(|(t, _)| r.ts_nanos >= *t) {
export = Some((
r.ts_nanos,
ExportView {
sequence: number_after(r.detail, " seq ").unwrap_or(0),
last_unix: secs,
},
));
}
} else if ok && r.detail.starts_with("bundle-import: bundle_id=") {
inbox.imported += 1;
if r.ts_nanos >= last_import {
last_import = r.ts_nanos;
inbox.last_name = word_after(r.detail, "bundle_id=")
.unwrap_or_default()
.to_string();
inbox.last_sequence = number_after(r.detail, " seq=");
inbox.last_unix = secs;
}
} else if !ok && r.detail.starts_with("bundle import refused:") {
inbox.refused += 1;
}
});
(export.map(|(_, x)| x), inbox)
}
/// The estate's boxes and downstreams, out of the estate board's facts and rows.
pub fn estate_of(
m: &mut Measured,
facts: &crate::topology::Facts,
reply: &crate::grpc::holger_proto::TopologyReply,
audit_peers: BTreeMap<String, u64>,
) {
use crate::topology::{ROLE_CONSOLE, ROLE_DOWNSTREAM, ROLE_FRONT, ROLE_PRIMARY, ROLE_TWIN};
m.estate_name = reply.estate.clone();
m.silent_after_ms = reply.silent_after_ms;
for b in &reply.boxes {
match b.role.as_str() {
ROLE_CONSOLE | ROLE_FRONT | ROLE_PRIMARY | ROLE_TWIN => m.estate.push(EstateBox {
name: b.name.clone(),
role: b.role.clone(),
this_box: b.this_box,
state: b.state.clone(),
headline: b.headline.clone(),
address: b.address.clone(),
twin_state: b.twin_state.clone(),
lag_entries: b.lag_entries,
seen_ago_ms: b.seen_ago_ms,
last_seen_unix_ms: b.last_seen_unix_ms,
disk: b
.disk_measured
.then_some((b.disk_total_bytes, b.disk_used_bytes)),
}),
ROLE_DOWNSTREAM => {
let peer = facts.downstreams.iter().find(|d| d.name == b.name);
// Declared in the RON, or recorded by the hub (a downstream is one or the
// other, never both), or mTLS-paired.
let declared = facts.declared.downstreams.contains(&b.name);
let enrolled = !declared && facts.sync.iter().any(|s| s.name == b.name);
let bytes = peer
.map(|p| p.bytes_total)
.or_else(|| audit_peers.get(&b.name).copied());
m.downstreams.push(DownstreamView {
name: b.name.clone(),
via: b.via.clone(),
state: b.state.clone(),
headline: b.headline.clone(),
last_seen_unix_ms: b.last_seen_unix_ms,
seen_ago_ms: b.seen_ago_ms,
lag_entries: b.lag_entries,
bytes_total: bytes,
fetched_total: peer.map(|p| p.fetched_total),
repos: peer
.map(|p| {
p.repositories
.iter()
.map(|r| r.repository.clone())
.collect()
})
.unwrap_or_default(),
declared,
paired: peer.is_some() || enrolled,
summary: peer
.and_then(|p| p.summary.as_ref().map(RonSummary::from))
.or_else(|| {
facts
.sync
.iter()
.find(|s| s.name == b.name)
.and_then(|s| s.summary.clone())
}),
});
}
_ => {}
}
}
if let Some(c) = &reply.consumers {
if c.counted {
m.consumers = Some((c.principals, c.peers, c.anonymous_today, c.window_days));
}
}
}
/// **The live picture** — the ONE function `GET /-/flow.svg`, `GET /-/flow.json` and
/// `AdminService.FlowSvg` call. Blocking.
pub fn live(sources: &Sources) -> FlowGraph {
measure(&sources.shape, &sources.gather())
}
// ── geometry ────────────────────────────────────────────────────────────────
const NODE_W: f64 = 244.0;
const COL_GAP: f64 = 104.0;
const ROW_GAP: f64 = 30.0;
const HEAD_H: f64 = 52.0;
const ROW_H: f64 = 18.0;
const SEP_H: f64 = 9.0;
const PAD_B: f64 = 13.0;
const MARGIN: f64 = 32.0;
const TOP: f64 = 158.0;
const BAND_GAP: f64 = 96.0;
fn node_height(facts: usize, rows: usize) -> f64 {
let sep = if facts > 0 && rows > 0 { SEP_H } else { 0.0 };
HEAD_H + (facts + rows) as f64 * ROW_H + sep + PAD_B
}
#[derive(Debug, Clone, Copy)]
struct Box2 {
x: f64,
y: f64,
w: f64,
h: f64,
band: Band,
/// The column it stands in, among the boxes it is laid out with.
col: usize,
}
impl Box2 {
fn mid_x(&self) -> f64 {
self.x + self.w / 2.0
}
fn mid_y(&self) -> f64 {
self.y + self.h / 2.0
}
fn right(&self) -> f64 {
self.x + self.w
}
fn bottom(&self) -> f64 {
self.y + self.h
}
}
/// Which side of a machine a port sits on.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum Side {
Top,
Bottom,
Left,
Right,
}
/// The side a port named `name` sits on: what comes in from above enters at the top, what
/// leaves for the downstreams leaves at the bottom, and the estate's own links (the front,
/// the console, the twin) run sideways.
fn port_side(name: &str, out: bool) -> Side {
match (name, out) {
("export", true) => Side::Bottom,
(_, true) => Side::Right,
("doors" | "follow" | "console", false) => Side::Left,
(_, false) => Side::Top,
}
}
/// The inner padding of an unfolded machine.
const PAD: f64 = 28.0;
/// Room above an unfolded machine's first band, for its caption.
const INNER_TOP: f64 = 48.0;
/// A side port's height under the machine's top edge.
const SIDE_PORT_Y: f64 = 26.0;
/// How many of an unfolded machine's own rows stand in one column of its head.
const HEAD_ROWS: usize = 4;
const HEAD_COL_W: f64 = 232.0;
/// The gap between the rows of machines.
const ROW_GAP_ESTATE: f64 = 92.0;
struct Placed {
boxes: BTreeMap<String, Box2>,
width: f64,
height: f64,
/// Each band caption: the band, where it starts (x, y).
band_y: Captions,
/// The machines drawn unfolded, and the height of each one's head.
open: BTreeMap<String, f64>,
/// Each port: where it is and on which side.
ports: BTreeMap<String, (f64, f64, Side)>,
}
/// Each band caption: the band, where it starts (x, y).
type Captions = Vec<(Band, f64, f64)>;
/// The boxes of one set of nodes in their bands (flow, chain, storage), top-left at
/// `(x0, y0)`; their columns closed up among themselves. `(boxes, right, bottom, captions)`.
fn layout_bands(
nodes: &[&FlowNode],
x0: f64,
y0: f64,
) -> (BTreeMap<String, Box2>, f64, f64, Captions) {
let mut ordered: Vec<&FlowNode> = nodes.to_vec();
ordered.sort_by_key(|n| (n.band, n.column, n.slot));
// The chain's steps stand under the rung they guard: flow and chain share columns.
let mut remap: BTreeMap<(bool, usize), usize> = BTreeMap::new();
for storage in [false, true] {
let used: BTreeSet<usize> = ordered
.iter()
.filter(|n| (n.band == Band::Storage) == storage)
.map(|n| n.column)
.collect();
for (i, c) in used.into_iter().enumerate() {
remap.insert((storage, c), i);
}
}
let mut boxes = BTreeMap::new();
let mut right = x0;
let mut band_origin = y0;
let mut bottom = y0;
let mut captions = Vec::new();
for band in Band::ALL {
let in_band: Vec<&&FlowNode> = ordered.iter().filter(|n| n.band == band).collect();
if in_band.is_empty() {
continue;
}
captions.push((band, x0, band_origin));
let mut col_y: BTreeMap<usize, f64> = BTreeMap::new();
let mut band_bottom = band_origin;
for node in in_band {
let col = remap[&(band == Band::Storage, node.column)];
let h = node_height(node.facts.len(), node.rows.len());
let x = x0 + col as f64 * (NODE_W + COL_GAP);
let y = *col_y.get(&col).unwrap_or(&band_origin);
col_y.insert(col, y + h + ROW_GAP);
boxes.insert(
node.id.clone(),
Box2 {
x,
y,
w: NODE_W,
h,
band,
col,
},
);
right = right.max(x + NODE_W);
band_bottom = band_bottom.max(y + h);
}
bottom = band_bottom;
band_origin = band_bottom + BAND_GAP;
}
(boxes, right, bottom, captions)
}
/// The height of an unfolded machine's head: its name, then its own facts and rows in
/// columns of [`HEAD_ROWS`].
fn open_head_height(lines: usize) -> f64 {
HEAD_H + lines.min(HEAD_ROWS) as f64 * ROW_H + PAD_B
}
/// **Where everything stands.** The machines in rows — the upstream sources on top, then
/// console, front, main and twin, then the downstreams — and an unfolded machine as a frame
/// holding its own parts laid out in their bands.
fn layout(graph: &FlowGraph) -> Placed {
let ids: BTreeSet<&str> = graph.nodes.iter().map(|n| n.id.as_str()).collect();
let machines: Vec<String> = graph.machine_ids();
let parts_of = |m: &str| -> Vec<&FlowNode> {
graph
.nodes
.iter()
.filter(|n| n.machine.as_deref() == Some(m) && n.id != m)
.collect()
};
// Outside every machine: no box, or a box this picture does not hold.
let outside: Vec<&FlowNode> = graph
.nodes
.iter()
.filter(|n| match n.machine.as_deref() {
None => true,
Some(m) => !ids.contains(m),
})
.collect();
// Each machine's size, and its parts' boxes relative to its top-left corner.
let mut size: BTreeMap<String, (f64, f64)> = BTreeMap::new();
let mut inner: BTreeMap<String, (BTreeMap<String, Box2>, Captions)> = BTreeMap::new();
let mut open: BTreeMap<String, f64> = BTreeMap::new();
for m in &machines {
let node = graph.node(m).expect("a machine is a node");
let parts = parts_of(m);
if parts.is_empty() {
size.insert(
The file goes on; the console shows all of it.