src/sync_board.rs
//! **The sync board, drawn** (Holger v2.0 H6, the coordinator's B6): the upstream's
//! `GET /-/sync/board` (H4's `holger.sync-board/v1`), as the Admin → Sync pane shows it.
//!
//! Per downstream: the last sync it applied, its lag, whether each invalidation reached it,
//! and the ALARM when an invalidation was not acknowledged in time. The SERVER decides every
//! one of those (holger-sync's ledger: the state word `current` | `behind` | `never` |
//! `alarm`, the headline, `alarm`, each pending invalidation's `overdue`); this module only
//! says the server's numbers in words and picks a badge tone for its word. No logic.
#![cfg_attr(not(target_arch = "wasm32"), allow(dead_code))]
use serde::Deserialize;
use crate::util::Tone;
/// The board door.
pub const BOARD_PATH: &str = "/-/sync/board";
/// The schema the console reads.
pub const SCHEMA: &str = "holger.sync-board/v1";
/// `holger.sync-board/v1`, as the server serves it (holger-sync `SyncBoard`).
#[derive(Clone, Debug, Default, Deserialize, PartialEq)]
#[serde(default)]
pub struct Board {
pub schema: String,
pub upstream: String,
pub generation: u64,
pub generated_at: i64,
pub now: i64,
pub ack_deadline_secs: u64,
pub alarm: bool,
pub downstreams: Vec<Downstream>,
pub invalidations: Vec<Invalidation>,
}
#[derive(Clone, Debug, Default, Deserialize, PartialEq)]
#[serde(default)]
pub struct Downstream {
pub name: String,
/// How it reaches the upstream: `front` or `wireguard` (lane HOL-WGDS; absent = `front`).
pub via: String,
pub state: String,
pub headline: String,
pub last_generation_applied: u64,
pub lag_generations: u64,
pub lag_secs: i64,
pub last_ack_at: i64,
/// When its mirror last reported its sync (Unix seconds; 0 = never; holger lane OX-WGDS2).
pub last_report_at: i64,
/// How many of the upstream's writes its mirror had not applied at that report.
pub entries_behind: u64,
pub acks: u64,
pub packages_added: u64,
pub carrier: String,
pub invalidations_reached: usize,
pub invalidations_total: usize,
pub pending: Vec<Pending>,
pub alarm: bool,
}
#[derive(Clone, Debug, Default, Deserialize, PartialEq)]
#[serde(default)]
pub struct Pending {
pub id: String,
pub coordinate: String,
pub reason: String,
pub issued_at: i64,
pub waiting_secs: i64,
pub overdue: bool,
}
#[derive(Clone, Debug, Default, Deserialize, PartialEq)]
#[serde(default)]
pub struct Invalidation {
pub id: String,
pub coordinate: String,
pub reason: String,
pub detail: String,
pub generation: u64,
pub issued_at: i64,
pub reached: Vec<String>,
pub pending: Vec<String>,
}
/// One downstream's line on the pane.
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct Row {
pub name: String,
/// `front` or `wireguard`, as the server says (lane HOL-WGDS).
pub via: String,
/// The server's word, as it came.
pub state: String,
pub tone: Tone,
pub headline: String,
pub last_applied: String,
pub lag: String,
pub last_ack: String,
/// When it last synced: its mirror's last report or its last acknowledgement, the newer.
pub last_sync: String,
/// Holger signs its invite: a `front` one's direct (its sync door, HOL-PRODUCTS-7), a
/// `wireguard` one's through the front.
pub invitable: bool,
pub invalidations: String,
/// Each invalidation it has not confirmed, in words; `OVERDUE` where the server says so.
pub pending: Vec<String>,
pub alarm: bool,
}
/// One invalidation's line: where it reached and where it waits.
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct InvalidationLine {
pub coordinate: String,
pub why: String,
pub reached: String,
pub waiting: String,
}
/// The pane: the alarm banner (when the server raised one), a header and the lines.
#[derive(Clone, Debug, Default, PartialEq, Eq)]
pub struct View {
pub alarm: Option<String>,
pub header: String,
pub rows: Vec<Row>,
pub invalidations: Vec<InvalidationLine>,
}
/// Pure: the board door's answer read. `Err` in words for anything the console cannot draw.
pub fn read(status: u16, body: &str) -> Result<Board, String> {
match status {
200 => {}
404 => return Err("this Holger runs no sync (no `sync:` block): there is no board to draw".into()),
s => return Err(format!("holger refused the sync board ({s}): {}", body.trim())),
}
let b: Board = serde_json::from_str(body).map_err(|e| format!("holger answered something unreadable: {e}"))?;
if b.schema != SCHEMA {
return Err(format!("holger serves the board as {:?}; this console draws {SCHEMA}", b.schema));
}
Ok(b)
}
/// The console's revocation of downstream `name` (holger `POST /-/sync/downstreams/revoke`,
/// Repository Admin, audited there): from then on it takes no manifest and gives no ack,
/// and a WireGuard one is dropped by the front (lane HOL-WGDS).
pub fn revoke_path(name: &str) -> String {
let safe: String = name.chars().map(|c| if c.is_ascii_alphanumeric() || matches!(c, '.' | '_' | '-') { c.to_string() } else { format!("%{:02X}", c as u32 & 0xff) }).collect();
format!("/-/sync/downstreams/revoke?name={safe}")
}
fn query_safe(v: &str) -> String {
v.chars().map(|c| if c.is_ascii_alphanumeric() || matches!(c, '.' | '_' | '-') { c.to_string() } else { c.to_string().bytes().map(|b| format!("%{b:02X}")).collect() }).collect()
}
/// The console's enrolment of a downstream (holger `POST /-/sync/downstreams/enrol`,
/// Repository Admin, audited there; holger checks every field and refuses by name): its
/// name, its ack key (`holger sync key` on it), `front` or `wireguard`, and for a
/// `wireguard` one its WireGuard key (`holger sync wg-key` on it).
pub fn enrol_path(name: &str, key: &str, via: &str, wireguard_key: &str) -> String {
let mut p = format!("/-/sync/downstreams/enrol?name={}&key={}&via={}", query_safe(name.trim()), query_safe(key.trim()), query_safe(via.trim()));
if via.trim() == "wireguard" {
p.push_str(&format!("&wireguard_key={}", query_safe(wireguard_key.trim())));
}
p
}
/// The console's invite of a downstream (holger `POST /-/sync/downstreams/invite`, Repository
/// Admin): holger signs it with its sync key — a `front` one's direct, its door holger's sync
/// door (HOL-PRODUCTS-7); a `wireguard` one's with the front's interface as its offer was kept.
pub fn invite_path(name: &str) -> String {
format!("/-/sync/downstreams/invite?name={}", query_safe(name))
}
/// Pure: holger's answer to an enrolment, in words.
pub fn enrolled(status: u16, body: &str) -> Result<String, String> {
let v: serde_json::Value = match status {
200 => serde_json::from_str(body).map_err(|e| format!("holger answered something unreadable: {e}"))?,
s => return Err(format!("holger refused the enrolment ({s}): {}", body.trim())),
};
Ok(format!("{} enrolled, via {}", v["name"].as_str().unwrap_or("?"), v["via"].as_str().unwrap_or("front")))
}
/// What holger signed for a downstream: the invite file's content, and what it says.
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct Invited {
pub name: String,
/// The signed invite, as the downstream keeps it (its `sync.upstreams[].invite` file).
pub invite: String,
/// A `wireguard` downstream's address on the front's downstream interface; empty for a
/// direct one.
pub address: String,
/// The door the downstream reaches holger at.
pub door: String,
/// `holger.downstream-invite/v1` (direct) or `/v2` (WireGuard, the link's pre-shared key).
pub schema: String,
/// Holger's sync name and key: what the downstream's RON trusts.
pub upstream: String,
pub key: String,
}
impl Invited {
/// The entry the downstream's RON names its upstream with, the invite in `<name>.invite.json`.
pub fn upstream_line(&self) -> String {
format!(r#"(name: "{}", key: "{}", invite: Some("{}.invite.json"))"#, self.upstream, self.key, self.name)
}
/// The page's line about it.
pub fn said(&self) -> String {
if self.address.is_empty() {
format!(" for {}, door {}: put it on the downstream as its upstream's `invite` file.", self.name, self.door)
} else {
format!(" for {} at {}: put it on the downstream as its upstream's `invite` file.", self.name, self.address)
}
}
}
/// Pure: holger's answer to an invite.
pub fn invited(name: &str, status: u16, body: &str) -> Result<Invited, String> {
let v: serde_json::Value = match status {
200 => serde_json::from_str(body).map_err(|e| format!("holger answered something unreadable: {e}"))?,
s => return Err(format!("holger refused the invite ({s}): {}", body.trim())),
};
let invite = v.get("invite").filter(|i| i.is_object()).ok_or("holger's answer carries no invite")?;
let word = |k: &str| v[k].as_str().unwrap_or("").to_string();
Ok(Invited {
name: name.to_string(),
invite: serde_json::to_string_pretty(invite).unwrap_or_default(),
address: word("address"),
door: word("door"),
schema: word("schema"),
upstream: word("upstream"),
key: word("key"),
})
}
/// Pure: what the page says once holger answered the revocation of `name`: `Ok` with
/// holger's record of it (who revoked it), or `Err` with holger's refusal verbatim and its
/// status. The console adds no judgement: a refusal is holger's sentence.
pub fn revoke_said(name: &str, status: u16, body: &str) -> Result<String, String> {
if status != 200 {
let why = if body.trim().is_empty() { "holger gave no reason" } else { body.trim() };
return Err(format!("holger refused the revocation of {name} ({status}): {why}"));
}
let v: serde_json::Value = serde_json::from_str(body).map_err(|e| format!("holger revoked {name}? its answer is unreadable: {e}"))?;
let r = &v["revoked"];
let (who, by) = (r["name"].as_str().unwrap_or(name), r["revoked_by"].as_str().unwrap_or_default());
Ok(if by.is_empty() { format!("{who} revoked") } else { format!("{who} revoked by {by}") })
}
/// Pure: the badge tone of the server's state word (a word the console does not know is
/// shown as it came, untoned).
pub fn state_tone(state: &str) -> Tone {
match state {
"current" => Tone::Good,
"behind" => Tone::Warn,
"alarm" => Tone::Bad,
_ => Tone::Neutral,
}
}
/// Pure: seconds in words (`45 s`, `12 min`, `25 h`, `3 days`).
pub fn span(secs: i64) -> String {
match secs.max(0) {
s if s < 120 => format!("{s} s"),
s if s < 3_600 => format!("{} min", s / 60),
s if s < 172_800 => format!("{} h", s / 3_600),
s => format!("{} days", s / 86_400),
}
}
fn list_or_none(names: &[String]) -> String {
if names.is_empty() {
"none".into()
} else {
names.join(", ")
}
}
/// Pure: **the board, in words**, every line in the server's order.
pub fn view(board: &Board) -> View {
let rows: Vec<Row> = board
.downstreams
.iter()
.map(|d| Row {
name: d.name.clone(),
via: if d.via.is_empty() { "front".into() } else { d.via.clone() },
state: d.state.clone(),
tone: state_tone(&d.state),
headline: d.headline.clone(),
last_applied: if d.last_generation_applied == 0 { "none yet".into() } else { format!("generation {}", d.last_generation_applied) },
lag: match (d.lag_generations, d.entries_behind) {
(0, 0) => "none".into(),
(0, n) => format!("{n} write(s)"),
(g, _) => format!("{g} generation(s), {}", span(d.lag_secs)),
},
last_ack: if d.last_ack_at <= 0 { "never".into() } else { format!("{} ago", span(board.now - d.last_ack_at)) },
last_sync: match d.last_ack_at.max(d.last_report_at) {
t if t <= 0 => "never".into(),
t => format!("{} ago", span(board.now - t)),
},
invitable: d.state != "revoked",
invalidations: format!("{} of {} reached", d.invalidations_reached, d.invalidations_total),
pending: d
.pending
.iter()
.map(|p| format!("{} ({}), waiting {}{}", p.coordinate, p.reason, span(p.waiting_secs), if p.overdue { " — OVERDUE" } else { "" }))
.collect(),
alarm: d.alarm,
})
.collect();
let alarm = board.alarm.then(|| {
let said: Vec<String> = rows.iter().filter(|r| r.alarm).map(|r| format!("{}: {}", r.name, r.headline)).collect();
format!("ALARM: {}", if said.is_empty() { "an invalidation is not acknowledged in time".to_string() } else { said.join("; ") })
});
View {
alarm,
header: format!("upstream {} · generation {} · acknowledgement deadline {}", board.upstream, board.generation, span(board.ack_deadline_secs as i64)),
rows,
invalidations: board
.invalidations
.iter()
.map(|i| InvalidationLine {
coordinate: i.coordinate.clone(),
why: format!("{}: {} (generation {})", i.reason, i.detail, i.generation),
reached: list_or_none(&i.reached),
waiting: list_or_none(&i.pending),
})
.collect(),
}
}
#[cfg(test)]
mod tests {
use super::*;
const ALARM: &str = include_str!("../tests/fixtures/sync-board/alarm.json");
/// The board holger's own end-to-end gate left (`cargo xtask e2e`: the invalidation of the
/// wheel, carried on the USB, acknowledged by `onprem`).
const REACHED: &str = include_str!("../tests/fixtures/sync-board/reached.json");
/// ★ Lane HOL-WGDS: each downstream says how it reaches the upstream (absent = front), and
/// the console's revocation names it on holger's door.
#[test]
fn each_downstream_says_its_via_and_the_revocation_names_it() {
let v = view(&read(200, ALARM).unwrap());
assert!(v.rows.iter().all(|r| r.via == "front"), "a board without `via` is the front's");
let mut b = read(200, ALARM).unwrap();
b.downstreams[1].via = "wireguard".into();
assert_eq!(view(&b).rows[1].via, "wireguard");
assert_eq!(revoke_path("lab-2"), "/-/sync/downstreams/revoke?name=lab-2");
assert_eq!(revoke_path("a b&c"), "/-/sync/downstreams/revoke?name=a%20b%26c");
}
/// ★ Holger lane OX-WGDS2: a downstream its upstream heard by its mirror's signed report
/// says when it last synced and how many writes it is behind, instead of "never".
#[test]
fn a_downstream_heard_by_its_mirrors_report_says_its_last_sync_and_its_lag() {
let mut b = read(200, ALARM).unwrap();
let lab = &mut b.downstreams[3];
(lab.state, lab.lag_generations, lab.lag_secs, lab.last_report_at, lab.entries_behind) = ("behind".into(), 0, 0, b.now - 120, 4);
let v = view(&b);
assert_eq!((v.rows[3].last_sync.as_str(), v.rows[3].lag.as_str(), v.rows[3].tone), ("2 min ago", "4 write(s)", Tone::Warn));
assert_eq!(v.rows[3].last_ack, "never", "its acknowledgements are still its own column");
assert_eq!(v.rows[0].last_sync, "24 h ago", "an acknowledgement is a sync too");
}
/// ★ Holger lane OX-WGDS2: the console enrols and invites through holger's doors; holger
/// checks and signs, the console says its answer.
#[test]
fn the_console_enrols_and_invites_through_holgers_doors() {
assert_eq!(enrol_path("lab-1", "ab12", "front", "ignored"), "/-/sync/downstreams/enrol?name=lab-1&key=ab12&via=front");
assert_eq!(enrol_path(" lab-2 ", "cd", "wireguard", "AAEC+/x="), "/-/sync/downstreams/enrol?name=lab-2&key=cd&via=wireguard&wireguard_key=AAEC%2B%2Fx%3D");
assert_eq!(invite_path("lab-2"), "/-/sync/downstreams/invite?name=lab-2");
assert_eq!(enrolled(200, r#"{"name":"lab-2","via":"wireguard"}"#), Ok("lab-2 enrolled, via wireguard".into()));
assert_eq!(enrolled(400, "[holger-sync-11] downstream 'x' is declared twice\n"), Err("holger refused the enrolment (400): [holger-sync-11] downstream 'x' is declared twice".into()));
let got = invited("lab-2", 200, r#"{"invite":{"schema":"holger.sync-signed/v1","kind":"invite","signer":"cloud"},"address":"10.45.0.2"}"#).unwrap();
assert_eq!((got.address.as_str(), got.name.as_str()), ("10.45.0.2", "lab-2"));
assert!(got.invite.contains("\"kind\": \"invite\""), "{}", got.invite);
assert!(invited("lab-1", 400, "[holger-sync-11] lab-1 syncs `via: Front`").unwrap_err().contains("via: Front"));
assert!(invited("lab-1", 200, "{}").unwrap_err().contains("no invite"));
let mut b = read(200, ALARM).unwrap();
b.downstreams[1].via = "wireguard".into();
assert!(view(&b).rows.iter().all(|r| r.invitable), "every downstream is invited, front and wireguard (HOL-PRODUCTS-7)");
}
/// ★ Holger lane HOL-PRODUCTS-7: a plain server invites a `front` downstream directly:
/// holger signs a v1 invite whose door is its sync door, and the console says the door
/// (a wireguard one's address stays its line).
#[test]
fn a_front_downstream_gets_a_direct_invite_with_its_door() {
let body = r#"{"invite":{"schema":"holger.sync-signed/v1","kind":"invite","signer":"hq"},"address":null,"schema":"holger.downstream-invite/v1","door":"http://127.0.0.1:8080","upstream":"hq","key":"ab12"}"#;
let got = invited("lab", 200, body).unwrap();
assert_eq!((got.door.as_str(), got.address.as_str(), got.schema.as_str()), ("http://127.0.0.1:8080", "", "holger.downstream-invite/v1"));
assert_eq!(got.upstream_line(), r#"(name: "hq", key: "ab12", invite: Some("lab.invite.json"))"#);
assert!(got.said().contains("door http://127.0.0.1:8080"), "{}", got.said());
let wg = invited("lab-2", 200, r#"{"invite":{"kind":"invite"},"address":"10.45.0.2","door":"http://10.45.0.1:50070"}"#).unwrap();
assert!(wg.said().contains("at 10.45.0.2"), "{}", wg.said());
}
/// ★ The Revoke button's answer is said: holger's record of the revocation, or its
/// refusal with the status, verbatim.
#[test]
fn a_revocation_says_holgers_answer_or_its_refusal() {
let ok = r#"{"revoked":{"name":"lab-2","key":"ab","via":"Front","revoked_at":1,"revoked_by":"owner"}}"#;
assert_eq!(revoke_said("lab-2", 200, ok), Ok("lab-2 revoked by owner".to_string()));
assert_eq!(revoke_said("lab-2", 200, r#"{"revoked":{}}"#), Ok("lab-2 revoked".to_string()));
let no = "[holger-sync-9] 'lab-2' is not a live downstream of this Holger (never enrolled, or revoked already)";
assert_eq!(revoke_said("lab-2", 404, no), Err(format!("holger refused the revocation of lab-2 (404): {no}")));
assert_eq!(revoke_said("lab-2", 403, " "), Err("holger refused the revocation of lab-2 (403): holger gave no reason".to_string()));
assert!(revoke_said("lab-2", 200, "not json").unwrap_err().contains("unreadable"));
}
fn line(r: &Row) -> (&str, &str, Tone, &str, &str, &str, &str, bool) {
(r.name.as_str(), r.state.as_str(), r.tone, r.last_applied.as_str(), r.lag.as_str(), r.invalidations.as_str(), r.headline.as_str(), r.alarm)
}
/// ★ Per downstream: the last sync applied, the lag, whether the invalidation reached
/// it, and the ALARM for the one that did not acknowledge in time; every word the server's.
#[test]
fn the_board_draws_lag_reach_and_the_alarm_for_an_unacked_invalidation() {
let v = view(&read(200, ALARM).unwrap());
assert_eq!(
v.alarm.as_deref(),
Some("ALARM: dc3: 1 invalidation(s) not acknowledged within 86400 s"),
"the banner names the downstream and says the server's headline"
);
assert_eq!(v.header, "upstream cloud · generation 3 · acknowledgement deadline 24 h");
assert_eq!(
v.rows.iter().map(line).collect::<Vec<_>>(),
vec![
("onprem", "current", Tone::Good, "generation 3", "none", "1 of 1 reached", "at generation 3", false),
("dc2", "behind", Tone::Warn, "generation 1", "2 generation(s), 1 h", "1 of 1 reached", "behind 2 generation(s), 3600 s", false),
("dc3", "alarm", Tone::Bad, "generation 1", "2 generation(s), 25 h", "0 of 1 reached", "1 invalidation(s) not acknowledged within 86400 s", true),
("lab", "never", Tone::Neutral, "none yet", "3 generation(s), 25 h", "0 of 1 reached", "no acknowledgement received yet", false),
]
);
assert_eq!(v.rows[2].pending, ["pypi/demo_pkg-1.0.0-py3-none-any.whl (malware), waiting 25 h — OVERDUE"]);
assert_eq!(v.rows[3].pending, ["pypi/demo_pkg-1.0.0-py3-none-any.whl (malware), waiting 25 h"], "not overdue: the server did not say so");
assert_eq!(v.rows[0].last_ack, "24 h ago");
assert_eq!(v.rows[3].last_ack, "never");
assert_eq!(
v.invalidations,
[InvalidationLine {
coordinate: "pypi/demo_pkg-1.0.0-py3-none-any.whl".into(),
why: "malware: found after its approval (generation 2)".into(),
reached: "onprem, dc2".into(),
waiting: "dc3, lab".into(),
}]
);
}
/// ★ The e2e gate's board: the invalidation reached `onprem`, nothing waits, no alarm.
#[test]
fn the_e2e_gates_board_shows_the_invalidation_reached_the_downstream() {
let v = view(&read(200, REACHED).unwrap());
assert_eq!(v.alarm, None);
assert_eq!(v.rows.iter().map(line).collect::<Vec<_>>(), vec![("onprem", "current", Tone::Good, "generation 2", "none", "1 of 1 reached", "at generation 2", false)]);
assert!(v.rows[0].pending.is_empty());
assert_eq!(v.invalidations[0].reached, "onprem");
assert_eq!(v.invalidations[0].waiting, "none");
}
#[test]
fn a_holger_without_sync_or_a_refusal_is_said_in_words() {
assert_eq!(read(404, "no sync").unwrap_err(), "this Holger runs no sync (no `sync:` block): there is no board to draw");
assert_eq!(read(403, "denied").unwrap_err(), "holger refused the sync board (403): denied");
assert!(read(200, "{not json").unwrap_err().starts_with("holger answered something unreadable"));
assert!(read(200, r#"{"schema":"holger.sync-board/v9"}"#).unwrap_err().contains("holger.sync-board/v9"));
assert_eq!(state_tone("resyncing"), Tone::Neutral, "a word the console does not know is shown untoned");
assert_eq!((span(45), span(720), span(90_000), span(259_200), span(-5)), ("45 s".into(), "12 min".into(), "25 h".into(), "3 days".into(), "0 s".into()));
}
}