sync/src/ledger.rs
//! **The upstream's ledger**: its generations, its invalidations and, per
//! downstream, what that downstream acknowledged. The board and the alarm are
//! pure functions of the ledger and a clock.
use std::collections::{BTreeMap, BTreeSet};
use serde::{Deserialize, Serialize};
use crate::{
Invalidation, PackageRef, Reason, SyncAck, SyncError, SyncManifest, BOARD_SCHEMA,
MANIFEST_SCHEMA,
};
/// One generation the upstream issued.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct GenerationRecord {
pub generation: u64,
pub issued_at: i64,
/// `pull`, `bundle <id> seq <n>`, `invalidation`.
pub carrier: String,
#[serde(default)]
pub added: Vec<PackageRef>,
/// The ids of the invalidations issued in this generation.
#[serde(default)]
pub invalidations: Vec<String>,
}
/// What one downstream acknowledged.
#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)]
pub struct DownstreamRecord {
/// The newest generation it acknowledged (0: never).
pub last_generation: u64,
/// Unix seconds of its newest ack (0: never).
pub last_ack_at: i64,
pub acks: u64,
/// Packages its acknowledged syncs added, summed.
pub packages_added: u64,
/// The invalidations it confirmed in force.
#[serde(default)]
pub reached: BTreeSet<String>,
/// The invalidations an alarm was raised for and not yet cleared.
#[serde(default)]
pub alarmed: BTreeSet<String>,
/// How its newest ack came: `pull` or `usb`.
#[serde(default)]
pub carrier: String,
}
/// **The ledger.** Kept as one JSON file by the server.
#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)]
pub struct Ledger {
/// This upstream's name.
#[serde(default)]
pub upstream: String,
/// The newest generation issued (0: none yet).
pub generation: u64,
#[serde(default)]
pub generations: Vec<GenerationRecord>,
#[serde(default)]
pub invalidations: Vec<Invalidation>,
#[serde(default)]
pub downstreams: BTreeMap<String, DownstreamRecord>,
}
/// What recording one ack changed.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct AckOutcome {
pub downstream: String,
pub generation: u64,
/// Invalidations this ack confirmed for the first time.
pub reached_now: Vec<String>,
/// Alarms this ack cleared.
pub cleared: Vec<String>,
}
/// An alarm: `downstream` has not acknowledged `invalidation` within the deadline.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct Alarm {
pub downstream: String,
pub invalidation: Invalidation,
/// Seconds since the invalidation was issued.
pub waiting_secs: i64,
pub deadline_secs: u64,
}
impl Alarm {
/// The operator's line.
pub fn line(&self) -> String {
format!(
"ALARM sync: downstream '{}' has not acknowledged invalidation {} of {} ({}) issued {} s ago; the deadline is {} s",
self.downstream,
self.invalidation.id,
self.invalidation.coordinate,
self.invalidation.reason,
self.waiting_secs,
self.deadline_secs
)
}
}
impl Ledger {
/// A ledger for the upstream `name`.
pub fn new(name: impl Into<String>) -> Self {
Self {
upstream: name.into(),
..Default::default()
}
}
fn next_generation(
&mut self,
carrier: &str,
added: Vec<PackageRef>,
invalidations: Vec<String>,
now: i64,
) -> u64 {
self.generation += 1;
self.generations.push(GenerationRecord {
generation: self.generation,
issued_at: now,
carrier: carrier.into(),
added,
invalidations,
});
self.generation
}
/// **Issue an invalidation** in a new generation. The same package and reason
/// again is the invalidation already issued (`false`).
pub fn invalidate(
&mut self,
coordinate: &str,
sha256: Option<String>,
reason: Reason,
detail: &str,
by: &str,
now: i64,
) -> (Invalidation, bool) {
let sha = sha256
.map(|s| s.trim().to_ascii_lowercase())
.filter(|s| !s.is_empty());
if let Some(i) = self
.invalidations
.iter()
.find(|i| i.coordinate == coordinate && i.sha256 == sha && i.reason == reason)
{
return (i.clone(), false);
}
let inv = Invalidation::new(
coordinate,
sha,
reason,
detail,
self.generation + 1,
now,
by,
);
self.next_generation("invalidation", Vec::new(), vec![inv.id.clone()], now);
self.invalidations.push(inv.clone());
(inv, true)
}
/// Is this package in some generation already?
pub fn known(&self, p: &PackageRef) -> bool {
self.generations.iter().any(|g| {
g.added
.iter()
.any(|a| a.sha256 == p.sha256 && a.coordinate == p.coordinate)
})
}
/// A new generation carrying `added` (a bundle export).
pub fn add_generation(&mut self, carrier: &str, added: Vec<PackageRef>, now: i64) -> u64 {
self.next_generation(carrier, added, Vec::new(), now)
}
/// **The pull side's generation**: the packages of `available` no generation
/// carries yet go into a new one. `None`: nothing new.
pub fn absorb_available(&mut self, available: &[PackageRef], now: i64) -> Option<u64> {
let mut new: Vec<PackageRef> = available
.iter()
.filter(|p| !self.known(p))
.cloned()
.collect();
new.sort();
new.dedup();
if new.is_empty() {
return None;
}
Some(self.next_generation("pull", new, Vec::new(), now))
}
/// **The manifest at the newest generation**: every package added after
/// `since`, every invalidation.
pub fn manifest(&self, since: u64, carrier: &str) -> SyncManifest {
let mut added: Vec<PackageRef> = self
.generations
.iter()
.filter(|g| g.generation > since)
.flat_map(|g| g.added.iter().cloned())
.collect();
added.sort();
added.dedup();
self.manifest_with(added, carrier)
}
/// The manifest at the newest generation with exactly `added` (a bundle carries
/// what it carries).
pub fn manifest_with(&self, added: Vec<PackageRef>, carrier: &str) -> SyncManifest {
let issued_at = self.generations.last().map(|g| g.issued_at).unwrap_or(0);
SyncManifest {
schema: MANIFEST_SCHEMA.into(),
upstream: self.upstream.clone(),
generation: self.generation,
issued_at,
carrier: carrier.into(),
added,
invalidations: self.invalidations.clone(),
}
}
/// **Record an ack.** The caller verified its signature and that its signer is
/// enrolled.
pub fn record_ack(&mut self, ack: &SyncAck, now: i64) -> Result<AckOutcome, SyncError> {
if ack.upstream != self.upstream {
return Err(SyncError::Malformed(format!(
"the ack is for upstream '{}', and this is '{}'",
ack.upstream, self.upstream
)));
}
if ack.generation > self.generation {
return Err(SyncError::FutureGeneration {
offered: ack.generation,
issued: self.generation,
});
}
let ids: BTreeSet<&str> = self.invalidations.iter().map(|i| i.id.as_str()).collect();
let rec = self.downstreams.entry(ack.downstream.clone()).or_default();
rec.last_generation = rec.last_generation.max(ack.generation);
rec.last_ack_at = now;
rec.acks += 1;
rec.packages_added += ack.packages_added;
rec.carrier = ack.carrier.clone();
let mut reached_now = Vec::new();
for id in ack
.invalidations_applied
.iter()
.filter(|id| ids.contains(id.as_str()))
{
if rec.reached.insert(id.clone()) {
reached_now.push(id.clone());
}
}
let cleared: Vec<String> = rec
.alarmed
.iter()
.filter(|id| rec.reached.contains(*id))
.cloned()
.collect();
for id in &cleared {
rec.alarmed.remove(id);
}
Ok(AckOutcome {
downstream: ack.downstream.clone(),
generation: ack.generation,
reached_now,
cleared,
})
}
/// The downstreams the board covers: the enrolled ones and every one on record.
fn names(&self, enrolled: &[String]) -> Vec<String> {
let mut n: BTreeSet<String> = enrolled.iter().cloned().collect();
n.extend(self.downstreams.keys().cloned());
n.into_iter().collect()
}
/// The invalidations `downstream` has not confirmed, with how long each waits.
fn pending(&self, downstream: &str, now: i64) -> Vec<(&Invalidation, i64)> {
let reached = self.downstreams.get(downstream).map(|r| &r.reached);
self.invalidations
.iter()
.filter(|i| !reached.is_some_and(|r| r.contains(&i.id)))
.map(|i| (i, (now - i.issued_at).max(0)))
.collect()
}
/// **The alarms raised now**: every (enrolled downstream, invalidation) pair
/// past the deadline that has no alarm yet. Each is raised once; an ack that
/// confirms the invalidation clears it.
pub fn check_alarms(
&mut self,
enrolled: &[String],
now: i64,
deadline_secs: u64,
) -> Vec<Alarm> {
let mut out = Vec::new();
for d in self.names(enrolled) {
let due: Vec<(Invalidation, i64)> = self
.pending(&d, now)
.into_iter()
.filter(|(_, waiting)| *waiting >= deadline_secs as i64)
.map(|(i, w)| (i.clone(), w))
.collect();
let rec = self.downstreams.entry(d.clone()).or_default();
for (inv, waiting) in due {
if rec.alarmed.insert(inv.id.clone()) {
out.push(Alarm {
downstream: d.clone(),
invalidation: inv,
waiting_secs: waiting,
deadline_secs,
});
}
}
}
out
}
/// **The board's data.**
pub fn board(&self, enrolled: &[String], now: i64, deadline_secs: u64) -> SyncBoard {
let mut downstreams = Vec::new();
for d in self.names(enrolled) {
let rec = self.downstreams.get(&d).cloned().unwrap_or_default();
let lag_generations = self.generation.saturating_sub(rec.last_generation);
let lag_secs = self
.generations
.iter()
.find(|g| g.generation > rec.last_generation)
.map(|g| (now - g.issued_at).max(0))
.unwrap_or(0);
// Overdue: past the deadline now, or an alarm raised and not yet cleared by an ack.
let pending: Vec<PendingInvalidation> = self
.pending(&d, now)
.into_iter()
.map(|(i, waiting)| PendingInvalidation {
id: i.id.clone(),
coordinate: i.coordinate.clone(),
reason: i.reason,
issued_at: i.issued_at,
waiting_secs: waiting,
overdue: waiting >= deadline_secs as i64 || rec.alarmed.contains(&i.id),
})
.collect();
let alarm = pending.iter().any(|p| p.overdue);
let (state, headline) = if alarm {
let n = pending.iter().filter(|p| p.overdue).count();
(
"alarm",
format!("{n} invalidation(s) not acknowledged within {deadline_secs} s"),
)
} else if rec.acks == 0 {
("never", "no acknowledgement received yet".to_string())
} else if lag_generations > 0 {
(
"behind",
format!("behind {lag_generations} generation(s), {lag_secs} s"),
)
} else {
("current", format!("at generation {}", rec.last_generation))
};
downstreams.push(DownstreamRow {
name: d,
via: String::new(),
state: state.into(),
headline,
last_generation_applied: rec.last_generation,
lag_generations,
lag_secs,
last_ack_at: rec.last_ack_at,
last_report_at: 0,
entries_behind: 0,
acks: rec.acks,
packages_added: rec.packages_added,
carrier: rec.carrier,
invalidations_reached: rec
.reached
.iter()
.filter(|id| self.invalidations.iter().any(|i| &i.id == *id))
.count(),
invalidations_total: self.invalidations.len(),
pending,
alarm,
});
}
let invalidations = self
.invalidations
.iter()
.map(|i| {
let (reached, pending): (Vec<String>, Vec<String>) = {
let mut r = Vec::new();
let mut p = Vec::new();
for d in &downstreams {
if self
.downstreams
.get(&d.name)
.is_some_and(|x| x.reached.contains(&i.id))
{
r.push(d.name.clone());
} else {
p.push(d.name.clone());
}
}
(r, p)
};
InvalidationRow {
id: i.id.clone(),
coordinate: i.coordinate.clone(),
sha256: i.sha256.clone(),
reason: i.reason,
detail: i.detail.clone(),
generation: i.generation,
issued_at: i.issued_at,
reached,
pending,
}
})
.collect();
SyncBoard {
schema: BOARD_SCHEMA.into(),
upstream: self.upstream.clone(),
generation: self.generation,
generated_at: self.generations.last().map(|g| g.issued_at).unwrap_or(0),
now,
ack_deadline_secs: deadline_secs,
alarm: downstreams.iter().any(|d| d.alarm),
downstreams,
invalidations,
}
}
}
/// **The board's sync data** (JSON): what holger-ui draws for the upstream.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct SyncBoard {
pub schema: String,
pub upstream: String,
pub generation: u64,
/// When the newest generation was issued.
pub generated_at: i64,
pub now: i64,
pub ack_deadline_secs: u64,
/// Any downstream in alarm.
pub alarm: bool,
pub downstreams: Vec<DownstreamRow>,
pub invalidations: Vec<InvalidationRow>,
}
/// One downstream on the board.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct DownstreamRow {
pub name: String,
/// How it reaches the upstream: `front` or `wireguard` (lane HOL-WGDS; the server
/// fills it from its `sync.downstreams`).
#[serde(default)]
pub via: String,
/// `current`, `behind`, `never`, `alarm`.
pub state: String,
pub headline: String,
pub last_generation_applied: u64,
pub lag_generations: u64,
/// Seconds since the oldest generation it has not acknowledged was issued.
pub lag_secs: i64,
pub last_ack_at: i64,
/// When its mirror last reported how far it has synced (Unix seconds; 0 = never;
/// lane OX-WGDS2: the server fills it from the downstream's signed report).
#[serde(default)]
pub last_report_at: i64,
/// How many of the upstream's writes its mirror had not applied at that report.
#[serde(default)]
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<PendingInvalidation>,
pub alarm: bool,
}
/// An invalidation a downstream has not confirmed.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct PendingInvalidation {
pub id: String,
pub coordinate: String,
pub reason: Reason,
pub issued_at: i64,
pub waiting_secs: i64,
pub overdue: bool,
}
/// One invalidation on the board, with the downstreams it reached.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct InvalidationRow {
pub id: String,
pub coordinate: String,
#[serde(default)]
pub sha256: Option<String>,
pub reason: Reason,
pub detail: String,
pub generation: u64,
pub issued_at: i64,
pub reached: Vec<String>,
pub pending: Vec<String>,
}
#[cfg(test)]
mod tests {
use super::*;
use crate::ACK_SCHEMA;
fn pkg(c: &str, s: &str) -> PackageRef {
PackageRef {
coordinate: c.into(),
sha256: s.into(),
}
}
fn ack(l: &Ledger, gen: u64, applied: &[&str]) -> SyncAck {
SyncAck {
schema: ACK_SCHEMA.into(),
downstream: "onprem".into(),
upstream: l.upstream.clone(),
generation: gen,
applied_at: 0,
packages_added: 1,
invalidations_applied: applied.iter().map(|s| s.to_string()).collect(),
quarantined: vec![],
carrier: "usb".into(),
}
}
#[test]
fn generations_count_up_and_a_manifest_carries_what_is_new_and_every_invalidation() {
let mut l = Ledger::new("dmz");
assert_eq!(
l.absorb_available(&[pkg("a", "1"), pkg("b", "2")], 10),
Some(1)
);
assert_eq!(
l.absorb_available(&[pkg("a", "1"), pkg("b", "2")], 11),
None,
"nothing new, no generation"
);
let (inv, new) = l.invalidate("a", Some("1".into()), Reason::Malware, "eicar", "sec", 12);
assert!(new);
assert_eq!(inv.generation, 2);
let (again, new) = l.invalidate("a", Some("1".into()), Reason::Malware, "eicar", "sec", 13);
assert!(!new);
assert_eq!(again, inv);
assert_eq!(l.absorb_available(&[pkg("c", "3")], 14), Some(3));
let m = l.manifest(2, "pull");
assert_eq!(m.generation, 3);
assert_eq!(m.added, vec![pkg("c", "3")]);
assert_eq!(m.invalidations, vec![inv]);
assert_eq!(l.manifest(0, "pull").added.len(), 3);
}
#[test]
fn an_ack_moves_the_downstream_and_a_future_generation_is_refused() {
let mut l = Ledger::new("dmz");
l.absorb_available(&[pkg("a", "1")], 10);
let (inv, _) = l.invalidate(
"a",
Some("1".into()),
Reason::CveAfter,
"CVE-2026-1",
"sec",
20,
);
assert_eq!(l.record_ack(&ack(&l, 9, &[]), 30).unwrap_err().number(), 6);
let out = l
.record_ack(&ack(&l, 2, &[&inv.id, "inv-unknown"]), 30)
.unwrap();
assert_eq!(
out.reached_now,
vec![inv.id.clone()],
"an id this upstream never issued is not counted"
);
let b = l.board(&["onprem".into()], 40, 3600);
let d = &b.downstreams[0];
assert_eq!(
(d.state.as_str(), d.lag_generations, d.invalidations_reached),
("current", 0, 1)
);
assert!(d.pending.is_empty());
assert_eq!(b.invalidations[0].reached, vec!["onprem".to_string()]);
}
#[test]
fn the_alarm_fires_once_past_the_deadline_and_the_ack_clears_it() {
let mut l = Ledger::new("dmz");
l.absorb_available(&[pkg("a", "1")], 0);
l.record_ack(&ack(&l, 1, &[]), 5).unwrap();
let (inv, _) = l.invalidate("a", Some("1".into()), Reason::Pulled, "", "sec", 100);
let enrolled = vec!["onprem".to_string(), "silent".to_string()];
// Before the deadline: nothing.
assert!(l.check_alarms(&enrolled, 150, 60).is_empty());
let b = l.board(&enrolled, 150, 60);
assert!(!b.alarm);
assert_eq!(b.downstreams[0].state, "behind");
assert_eq!(b.downstreams[0].lag_secs, 50);
assert_eq!(b.downstreams[1].state, "never");
// Past it: one alarm per downstream, once.
let a = l.check_alarms(&enrolled, 161, 60);
assert_eq!(a.len(), 2);
assert!(a[0].line().contains(&inv.id), "{}", a[0].line());
assert!(l.check_alarms(&enrolled, 200, 60).is_empty(), "raised once");
let b = l.board(&enrolled, 200, 60);
assert!(b.alarm);
assert!(
l.board(&enrolled, 120, 60).alarm,
"a raised alarm stays on the board until an ack clears it"
);
assert_eq!(b.downstreams[0].state, "alarm");
assert!(b.downstreams[0].pending[0].overdue);
// The ack clears it for that downstream only.
let out = l.record_ack(&ack(&l, 2, &[&inv.id]), 210).unwrap();
assert_eq!(out.cleared, vec![inv.id.clone()]);
let b = l.board(&enrolled, 210, 60);
assert_eq!(b.downstreams[0].state, "current");
assert!(!b.downstreams[0].alarm);
assert!(b.downstreams[1].alarm, "the silent one is still in alarm");
assert!(b.alarm);
}
}