server/lib/src/peers.rs
//! **PAIRED HOLGERS** — a downstream holger that mirrors from an upstream holger,
//! known to both of them by name and by key.
//!
//! The case this exists for: cloud holger promotes `A → B`; an on-prem holger's
//! `C` mirrors `B` on a schedule (`mirror.ron_sources`). Before this module the
//! cloud side saw the on-prem box only as one more hashed address in the
//! "observed pullers" band, and the on-prem side saw the cloud only as a generic
//! "upstream" box. Neither knew the other was a holger, let alone which one.
//!
//! ## A pairing is a RECORD, made once — the twin's pattern
//!
//! Exactly the shape [`crate::twin::elect`] uses for the role: a small RON file
//! under the data root, written `create_new`, so a second pairing under the same
//! name is a refusal by name (`holger-peer-1`) rather than an overwrite. The
//! server reads the records at boot; a pairing takes effect at the next start,
//! the same as an election.
//!
//! * The UPSTREAM records *"downstream holger `<name>`, key fingerprint F, may
//! pull repositories [B]"* — [`DownstreamRecord`], file
//! `holger-peers/downstream-<name>.ron`.
//! * The DOWNSTREAM records *"upstream holger `<name>`, key fingerprint G, pulls
//! through mirror source `<source>`"* — [`UpstreamRecord`], file
//! `holger-peers/upstream-<source>.ron`.
//!
//! ## No new crypto — what is reused
//!
//! * **The key is the mTLS client certificate the mirror ALREADY presents**
//! ([`crate::mirror_cycle::MirrorCredential::Mtls`]). The upstream's TLS layer
//! (rustls, via tonic, against the configured `client_ca`) has already verified
//! the chain before a request reaches a handler; pairing adds a PIN on top: the
//! SHA-256 of the leaf certificate's DER, the same digest `openssl x509
//! -fingerprint -sha256` prints. The CN still resolves through the ordinary
//! role map, unchanged.
//! * **The upstream's key is its TLS server certificate**, pinned the same way.
//! The downstream checks it with one extra rustls handshake before each cycle
//! ([`probe_upstream_key`]) — rustls and tokio-rustls, the crates the HTTP
//! gateway already terminates TLS with.
//!
//! So a pairing needs a TLS door on the upstream. A cleartext upstream cannot be
//! paired, and says so: there is no key to pin.
//!
//! ## What a pairing changes
//!
//! * A pull whose client certificate matches a paired downstream is attributed
//! to that NAME (the audit row's principal is `holger-peer:<name>`), counted on
//! the peer's own row of the flow picture, and **excluded** from the anonymous
//! visitor count and from the hashed puller band. A pull of a repository the
//! record does not list is refused by name (`holger-peer-2`).
//! * A pull WITHOUT a paired certificate is exactly what it was: anonymous, and
//! drawn only as a per-day hash.
//! * On the downstream, a cycle against a paired upstream whose certificate does
//! NOT match the pin is refused (`holger-peer-3`) — a mirror must not pull from
//! a box that is not the one it paired with — and the picture names the state
//! `key-mismatch`.
//!
//! ## How far a downstream has synced — the twin's report, per repository
//!
//! After each mirror cycle the downstream tells its upstream its LAST SYNC
//! POINT (`holger.grow.v1.DownstreamReport`), the shape of the twin's
//! `TwinReport`: for every paired repository, the upstream boot and the
//! upstream head ([`crate::grow::Hold`]'s sequence) at the START of its last
//! complete cycle of it, and when that cycle completed. The position is the
//! upstream's own sequence and not the downstream's change map: a change map is
//! a set of coordinates on the downstream's disk and says nothing the upstream
//! can compare with; the upstream's head is the one number both sides share.
//! The downstream learns it from the report's answer, exactly as a twin does.
//!
//! The report is authenticated by the paired client certificate ALONE: a caller
//! whose leaf matches no pin is refused by name (`holger-peer-4`), a repository
//! the pairing does not list by `holger-peer-2`. The upstream keeps the last
//! report per downstream DURABLY (`holger-peers/sync-<name>.ron`, overwritten
//! in place), so a restart keeps "last synced at" — while a position from the
//! previous boot counts nothing against the new head ([`crate::follow_point`],
//! the rule twins use). [`PeerBoard::rows`] is the peer picture (the admin-read
//! `holger.grow.v1.Downstreams`): lag, `ok | behind | stalled | silent`, and
//! today's pulls.
use crate::lock::LockOrRecover as _;
use std::collections::BTreeMap;
use std::path::{Path, PathBuf};
use std::sync::Mutex;
use rustls::pki_types::pem::PemObject as _;
use serde::{Deserialize, Serialize};
/// The directory, under the data root, the pairing records live in.
pub const PEERS_DIR: &str = "holger-peers";
/// The audit principal prefix a paired pull is recorded under.
pub const PEER_PRINCIPAL_PREFIX: &str = "holger-peer:";
/// **The upstream's record of one downstream holger.**
#[derive(Serialize, Deserialize, Clone, Debug, PartialEq, Eq)]
pub struct DownstreamRecord {
/// The downstream's name, as it is drawn.
pub name: String,
/// `sha256:<64 hex>` of the downstream's client certificate (DER).
pub fingerprint: String,
/// The repositories on THIS box it may pull.
pub repos: Vec<String>,
/// Unix seconds.
pub paired_at: u64,
/// Who made the pairing (the OS user of the local verb).
pub paired_by: String,
}
/// **The downstream's record of its upstream holger.**
#[derive(Serialize, Deserialize, Clone, Debug, PartialEq, Eq)]
pub struct UpstreamRecord {
/// The upstream's name, as it is drawn.
pub name: String,
/// `sha256:<64 hex>` of the upstream's TLS server certificate (DER).
pub fingerprint: String,
/// The `mirror.ron_sources[].ron_name` that pulls from it.
pub source: String,
/// Unix seconds.
pub paired_at: u64,
/// Who made the pairing.
pub paired_by: String,
}
/// **The upstream's durable copy of one downstream's last sync report**, file
/// `holger-peers/sync-<name>.ron`.
#[derive(Serialize, Deserialize, Clone, Debug, Default, PartialEq, Eq)]
pub struct DownstreamSync {
/// The downstream's name (the pairing's).
pub name: String,
/// The downstream's declared cycle, seconds (0 = the mirror default).
pub every_secs: u32,
/// When this upstream last heard the downstream (Unix ms).
pub last_seen_ms: i64,
/// Per repository on THIS box.
pub repos: BTreeMap<String, RepoSync>,
/// **Artifacts delivered to this downstream, ever** (lane A13): kept here so
/// a restart keeps the count. Written at each report and by [`PeerBoard::flush`].
/// A record from before A13 has none and reads as 0.
#[serde(default)]
pub fetched_total: u64,
/// Their bytes.
#[serde(default)]
pub bytes_total: u64,
/// What it said of its own RON in its last report (HOL-FLOW-SMART).
#[serde(default)]
pub summary: Option<crate::flow::RonSummary>,
}
/// One paired repository's last reported point.
#[derive(Serialize, Deserialize, Clone, Copy, Debug, Default, PartialEq, Eq)]
pub struct RepoSync {
pub point: crate::follow_point::FollowPoint,
/// When the downstream's last complete cycle of it ended (Unix ms; 0 = never).
pub last_complete_ms: i64,
}
/// Why a sync report was refused.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum ReportRefused {
/// No paired downstream presented this certificate (`holger-peer-4`).
NotPaired(String),
/// The pairing does not list a reported repository (`holger-peer-2`).
RepoNotPaired(String),
/// The record could not be kept.
Unwritable(String),
}
impl ReportRefused {
/// The refusal on the wire.
pub fn status(self) -> tonic::Status {
match self {
ReportRefused::NotPaired(m) => crate::grpc::coded(
&holger_errcode::PEER_NOT_PAIRED,
tonic::Code::PermissionDenied,
m,
),
ReportRefused::RepoNotPaired(m) => crate::grpc::coded(
&holger_errcode::PEER_REPO_NOT_PAIRED,
tonic::Code::PermissionDenied,
m,
),
ReportRefused::Unwritable(m) => crate::grpc::internal_err(m),
}
}
}
/// `sha256:<64 hex>` of a certificate's DER bytes.
pub fn fingerprint_der(der: &[u8]) -> String {
use sha2::Digest as _;
format!("sha256:{}", hex::encode(sha2::Sha256::digest(der)))
}
/// The short form the picture prints: the first 12 hex digits.
pub fn short(fingerprint: &str) -> String {
let hex = fingerprint.strip_prefix("sha256:").unwrap_or(fingerprint);
hex.chars().take(12).collect()
}
/// The fingerprint of the FIRST certificate in a PEM file's bytes.
pub fn fingerprint_pem(pem: &[u8]) -> Result<String, String> {
let first = rustls::pki_types::CertificateDer::pem_slice_iter(pem)
.next()
.ok_or_else(|| "no CERTIFICATE block in the PEM".to_string())?
.map_err(|e| format!("the PEM does not parse: {e}"))?;
Ok(fingerprint_der(first.as_ref()))
}
fn peers_dir(data_root: &Path) -> PathBuf {
data_root.join(PEERS_DIR)
}
fn sync_path(dir: &Path, name: &str) -> PathBuf {
dir.join(format!("sync-{name}.ron"))
}
/// The four state words, back to their `'static` form.
fn word(s: &str) -> &'static str {
match s {
"ok" => "ok",
"behind" => "behind",
"stalled" => "stalled",
_ => "silent",
}
}
/// Overwrite a record in place: written beside it, synced, renamed over it —
/// a reader sees the old record or the new one, never half of one.
fn write_replace<T: Serialize>(path: &Path, record: &T) -> Result<(), String> {
let text = ron::ser::to_string_pretty(record, ron::ser::PrettyConfig::default())
.map_err(|e| format!("serialising {}: {e}", path.display()))?;
if let Some(dir) = path.parent() {
std::fs::create_dir_all(dir).map_err(|e| format!("creating {}: {e}", dir.display()))?;
}
let tmp = path.with_extension("ron.tmp");
{
use std::io::Write as _;
let mut f =
std::fs::File::create(&tmp).map_err(|e| format!("writing {}: {e}", tmp.display()))?;
f.write_all(text.as_bytes())
.map_err(|e| format!("writing {}: {e}", tmp.display()))?;
f.sync_all()
.map_err(|e| format!("syncing {}: {e}", tmp.display()))?;
}
std::fs::rename(&tmp, path).map_err(|e| format!("renaming {} into place: {e}", path.display()))
}
fn check_name(what: &str, name: &str) -> Result<(), String> {
let ok = !name.is_empty()
&& name.len() <= 64
&& name
.chars()
.all(|c| c.is_ascii_alphanumeric() || c == '-' || c == '_' || c == '.')
&& !name.starts_with('.');
if ok {
Ok(())
} else {
Err(format!(
"{what} {name:?} is not a usable name: 1-64 characters of [A-Za-z0-9._-], not starting with '.'"
))
}
}
fn write_once<T: Serialize>(path: &Path, record: &T, what: &str) -> Result<(), String> {
let text = ron::ser::to_string_pretty(record, ron::ser::PrettyConfig::default())
.map_err(|e| format!("serialising the {what}: {e}"))?;
if let Some(dir) = path.parent() {
std::fs::create_dir_all(dir).map_err(|e| format!("creating {}: {e}", dir.display()))?;
}
// `create_new`: the role record's mechanism. A pairing is made once; a
// second one under the same name is a refusal, never an overwrite.
let mut f = std::fs::OpenOptions::new()
.write(true)
.create_new(true)
.open(path)
.map_err(|e| match e.kind() {
std::io::ErrorKind::AlreadyExists => holger_errcode::PEER_ALREADY_PAIRED.tag(format!(
"{} already exists — this {what} was made once and is not re-made; remove the \
file by hand to un-pair",
path.display()
)),
_ => format!("writing {}: {e}", path.display()),
})?;
use std::io::Write as _;
f.write_all(text.as_bytes())
.map_err(|e| format!("writing {}: {e}", path.display()))?;
f.sync_all()
.map_err(|e| format!("syncing {}: {e}", path.display()))
}
/// **On the UPSTREAM: record a downstream holger.** `cert_pem` is the client
/// certificate the downstream's mirror presents.
pub fn pair_downstream(
data_root: &Path,
name: &str,
cert_pem: &[u8],
repos: &[String],
paired_by: &str,
now_unix: u64,
) -> Result<DownstreamRecord, String> {
check_name("downstream", name)?;
if repos.is_empty() {
return Err(
"a pairing names the repositories the downstream may pull; none were given".into(),
);
}
let record = DownstreamRecord {
name: name.to_string(),
fingerprint: fingerprint_pem(cert_pem)?,
repos: repos.to_vec(),
paired_at: now_unix,
paired_by: paired_by.to_string(),
};
write_once(
&peers_dir(data_root).join(format!("downstream-{name}.ron")),
&record,
"pairing",
)?;
Ok(record)
}
/// **On the DOWNSTREAM: record the upstream holger a mirror source pulls from.**
/// `cert_pem` is the upstream's TLS server certificate.
pub fn pair_upstream(
data_root: &Path,
source: &str,
name: &str,
cert_pem: &[u8],
paired_by: &str,
now_unix: u64,
) -> Result<UpstreamRecord, String> {
check_name("mirror source", source)?;
check_name("upstream", name)?;
let record = UpstreamRecord {
name: name.to_string(),
fingerprint: fingerprint_pem(cert_pem)?,
source: source.to_string(),
paired_at: now_unix,
paired_by: paired_by.to_string(),
};
write_once(
&peers_dir(data_root).join(format!("upstream-{source}.ron")),
&record,
"pairing",
)?;
Ok(record)
}
/// The directory, under [`PEERS_DIR`], a re-pair moves the replaced record into.
pub const REPLACED_DIR: &str = "replaced";
/// **Re-pair a KNOWN peer with its new key** (lane E2): the peer lost its key (or its
/// operator rolled its certificate) and presents a new one. The record keeps its name and
/// what it may pull; only the pin, `paired_at` and `paired_by` change, and the replaced
/// record is kept beside it (`holger-peers/replaced/<file>-<now>.ron`) for the audit. The
/// lost key is refused from the next start: a re-pair never keeps the old pin.
///
/// A name that was never paired is refused by name (`holger-peer-5`): a first pairing is
/// [`pair_downstream`] / [`pair_upstream`], made once. The pin that is already there is
/// refused too (`holger-peer-1`): there is nothing to re-pair.
fn repair<T: Serialize + serde::de::DeserializeOwned + Clone>(
data_root: &Path,
file: &str,
what: &str,
now_unix: u64,
with: impl FnOnce(&T) -> Result<T, String>,
) -> Result<(T, T), String> {
let path = peers_dir(data_root).join(format!("{file}.ron"));
let text = match std::fs::read_to_string(&path) {
Ok(t) => t,
Err(e) if e.kind() == std::io::ErrorKind::NotFound => {
return Err(holger_errcode::PEER_NOT_KNOWN.tag(format!(
"{what} is not paired on this box ({} does not exist): a re-pair replaces a known peer's key; pair it first",
path.display()
)))
}
Err(e) => return Err(format!("reading {}: {e}", path.display())),
};
let old: T = ron::from_str(&text)
.map_err(|e| format!("the pairing record {} is not readable: {e}", path.display()))?;
let new = with(&old)?;
write_replace(
&peers_dir(data_root)
.join(REPLACED_DIR)
.join(format!("{file}-{now_unix}.ron")),
&old,
)?;
write_replace(&path, &new)?;
Ok((new, old))
}
fn same_pin(what: &str, fingerprint: &str) -> String {
holger_errcode::PEER_ALREADY_PAIRED.tag(format!(
"{what} is already paired with this key ({}): there is nothing to re-pair",
short(fingerprint)
))
}
/// **On the UPSTREAM: re-pair downstream `name` with its new client certificate.**
/// Answers `(new, replaced)`.
pub fn repair_downstream(
data_root: &Path,
name: &str,
cert_pem: &[u8],
paired_by: &str,
now_unix: u64,
) -> Result<(DownstreamRecord, DownstreamRecord), String> {
check_name("downstream", name)?;
let fingerprint = fingerprint_pem(cert_pem)?;
let what = format!("downstream holger '{name}'");
repair(
data_root,
&format!("downstream-{name}"),
&what,
now_unix,
|old: &DownstreamRecord| {
if old.fingerprint == fingerprint {
return Err(same_pin(&what, &fingerprint));
}
Ok(DownstreamRecord {
fingerprint: fingerprint.clone(),
paired_at: now_unix,
paired_by: paired_by.to_string(),
..old.clone()
})
},
)
}
/// **On the DOWNSTREAM: re-pair the upstream of mirror source `source` with its new
/// server certificate.** Answers `(new, replaced)`.
pub fn repair_upstream(
data_root: &Path,
source: &str,
cert_pem: &[u8],
paired_by: &str,
now_unix: u64,
) -> Result<(UpstreamRecord, UpstreamRecord), String> {
check_name("mirror source", source)?;
let fingerprint = fingerprint_pem(cert_pem)?;
let what = format!("the upstream of mirror source '{source}'");
repair(
data_root,
&format!("upstream-{source}"),
&what,
now_unix,
|old: &UpstreamRecord| {
if old.fingerprint == fingerprint {
return Err(same_pin(&what, &fingerprint));
}
Ok(UpstreamRecord {
fingerprint: fingerprint.clone(),
paired_at: now_unix,
paired_by: paired_by.to_string(),
..old.clone()
})
},
)
}
/// Every record under the data root. An unreadable record is an `Err`, never a
/// silent skip: a pairing that quietly vanished would turn a known peer back
/// into an anonymous visitor with nothing said.
pub fn load(data_root: &Path) -> Result<(Vec<DownstreamRecord>, Vec<UpstreamRecord>), String> {
let dir = peers_dir(data_root);
let mut down = Vec::new();
let mut up = Vec::new();
let entries = match std::fs::read_dir(&dir) {
Ok(e) => e,
Err(e) if e.kind() == std::io::ErrorKind::NotFound => return Ok((down, up)),
Err(e) => return Err(format!("reading {}: {e}", dir.display())),
};
let mut paths: Vec<PathBuf> = entries.filter_map(|e| e.ok().map(|e| e.path())).collect();
paths.sort();
for p in paths {
let Some(file) = p.file_name().and_then(|f| f.to_str()) else {
continue;
};
if !file.ends_with(".ron") {
continue;
}
let text =
std::fs::read_to_string(&p).map_err(|e| format!("reading {}: {e}", p.display()))?;
if file.starts_with("downstream-") {
down.push(
ron::from_str::<DownstreamRecord>(&text).map_err(|e| {
format!("the pairing record {} is not readable: {e}", p.display())
})?,
);
} else if file.starts_with("upstream-") {
up.push(
ron::from_str::<UpstreamRecord>(&text).map_err(|e| {
format!("the pairing record {} is not readable: {e}", p.display())
})?,
);
}
}
Ok((down, up))
}
// ─────────────────────────── the upstream's live tally ───────────────────────
/// One paired downstream's numbers.
#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
pub struct PeerCount {
/// Unix seconds of the last pull, if any since this process started.
pub last_pull_unix: Option<u64>,
/// Requests in the current UTC day.
pub requests_today: u64,
/// Bytes sent in the current UTC day.
pub bytes_today: u64,
/// Artifacts delivered in the current UTC day (lane A13).
pub fetched_today: u64,
/// Artifacts delivered, ever: the durable count read back at boot plus every
/// delivery since (lane A13).
pub fetched_total: u64,
/// Their bytes.
pub bytes_total: u64,
}
#[derive(Default)]
struct Tally {
day: u64,
counts: Vec<PeerCount>,
/// Per downstream: delivered since its totals were last written.
dirty: Vec<bool>,
}
/// **What this box knows about its peers**, for serving and for drawing.
///
/// One per server, shared by every door, exactly like the visitor tally.
#[derive(Default)]
pub struct PeerBoard {
downstreams: Vec<DownstreamRecord>,
upstreams: Vec<UpstreamRecord>,
tally: Mutex<Tally>,
/// Where the sync records live (`<data_root>/holger-peers`); `None` keeps
/// them in memory only (a box with no data root, a fixture).
sync_dir: Option<PathBuf>,
/// The last sync report per downstream, index-aligned with `downstreams`.
sync: Mutex<Vec<Option<DownstreamSync>>>,
}
impl PeerBoard {
/// A board over these records.
pub fn new(downstreams: Vec<DownstreamRecord>, upstreams: Vec<UpstreamRecord>) -> PeerBoard {
let n = downstreams.len();
PeerBoard {
downstreams,
upstreams,
tally: Mutex::new(Tally {
day: 0,
counts: vec![PeerCount::default(); n],
dirty: vec![false; n],
}),
sync_dir: None,
sync: Mutex::new(vec![None; n]),
}
}
/// **Keep the sync records under this data root**, reading what a previous
/// boot kept. An unreadable record is an `Err`, never a silent skip — the
/// rule [`load`] keeps for the pairings.
pub fn with_sync_root(mut self, data_root: &Path) -> Result<PeerBoard, String> {
let dir = peers_dir(data_root);
let mut kept = vec![None; self.downstreams.len()];
for (i, d) in self.downstreams.iter().enumerate() {
let p = sync_path(&dir, &d.name);
match std::fs::read_to_string(&p) {
Ok(text) => {
kept[i] = Some(ron::from_str::<DownstreamSync>(&text).map_err(|e| {
format!("the sync record {} is not readable: {e}", p.display())
})?)
}
Err(e) if e.kind() == std::io::ErrorKind::NotFound => {}
Err(e) => return Err(format!("reading {}: {e}", p.display())),
}
}
{
// The totals a previous boot kept are where this boot counts from.
let mut t = self.tally.lock_or_recover();
for (c, k) in t.counts.iter_mut().zip(&kept) {
if let Some(k) = k {
c.fetched_total = k.fetched_total;
c.bytes_total = k.bytes_total;
}
}
}
self.sync = Mutex::new(kept);
self.sync_dir = Some(dir);
Ok(self)
}
/// **A paired downstream's sync report**, on the upstream: the caller is
/// named by its leaf certificate and nothing else; every reported
/// repository must be one its pairing lists. Recorded (durably when the
/// board has a data root), and answered with this box's position.
pub fn report(
&self,
leaf_der: Option<&[u8]>,
req: &crate::grpc::holger_grow_proto::DownstreamReportRequest,
boot: u64,
head: u64,
now_ms: i64,
) -> Result<crate::grpc::holger_grow_proto::TwinReportReply, ReportRefused> {
let idx = leaf_der.and_then(|der| self.downstream_by_cert(der)).ok_or_else(|| {
ReportRefused::NotPaired(match leaf_der {
None => "a sync report must come over mutual TLS with the downstream's paired client \
certificate; this caller presented none"
.to_string(),
Some(der) => format!(
"the client certificate {} matches no paired downstream holger — pair it on this \
box first (`holger-server pair downstream`)",
short(&fingerprint_der(der))
),
})
})?;
let rec = &self.downstreams[idx];
if let Some(bad) = req
.points
.iter()
.find(|p| !rec.repos.iter().any(|r| r == &p.repository))
{
return Err(ReportRefused::RepoNotPaired(format!(
"downstream holger `{}` is paired for [{}], not for `{}`",
rec.name,
rec.repos.join(", "),
bad.repository
)));
}
let mut all = self.sync.lock_or_recover();
let prev = all.get(idx).cloned().flatten();
let (fetched_total, bytes_total) = self.totals_taken(idx);
let mut next = DownstreamSync {
name: rec.name.clone(),
every_secs: req.every_secs,
last_seen_ms: now_ms,
repos: prev.as_ref().map(|p| p.repos.clone()).unwrap_or_default(),
fetched_total,
bytes_total,
summary: req.summary.as_ref().map(crate::flow::RonSummary::from),
};
for p in &req.points {
let before = next.repos.get(&p.repository).map(|r| r.point);
let point = crate::follow_point::FollowPoint::heard(
before.as_ref(),
p.applied_boot,
p.applied_seq,
boot,
head,
now_ms,
);
next.repos.insert(
p.repository.clone(),
RepoSync {
point,
last_complete_ms: p.last_complete_unix_ms,
},
);
}
if let Some(dir) = &self.sync_dir {
write_replace(&sync_path(dir, &rec.name), &next).map_err(ReportRefused::Unwritable)?;
}
if let Some(slot) = all.get_mut(idx) {
*slot = Some(next);
}
Ok(crate::grpc::holger_grow_proto::TwinReportReply {
boot,
head_seq: head,
})
}
/// Downstream `idx`'s totals now, marking them written.
fn totals_taken(&self, idx: usize) -> (u64, u64) {
let mut t = self.tally.lock_or_recover();
if let Some(d) = t.dirty.get_mut(idx) {
*d = false;
}
t.counts
.get(idx)
.map(|c| (c.fetched_total, c.bytes_total))
.unwrap_or_default()
}
/// **Write every downstream's totals that moved since they were last
/// written** into its sync record (lane A13; the minute timer and the
/// shutdown call it, off the request path). A downstream that never
/// reported gets a record that holds its totals and says it was never heard
/// (`last_seen_ms: 0`). Records written; 0 on a board with no data root.
pub fn flush(&self) -> Result<usize, String> {
let Some(dir) = &self.sync_dir else {
return Ok(0);
};
let mut n = 0;
for (idx, d) in self.downstreams.iter().enumerate() {
let moved = self
.tally
.lock_or_recover()
.dirty
.get(idx)
.copied()
.unwrap_or(false);
if !moved {
continue;
}
let mut all = self.sync.lock_or_recover();
let (fetched_total, bytes_total) = self.totals_taken(idx);
let mut next = all
.get(idx)
.cloned()
.flatten()
.unwrap_or_else(|| DownstreamSync {
name: d.name.clone(),
..Default::default()
});
(next.fetched_total, next.bytes_total) = (fetched_total, bytes_total);
write_replace(&sync_path(dir, &d.name), &next)?;
if let Some(slot) = all.get_mut(idx) {
*slot = Some(next);
}
n += 1;
}
Ok(n)
}
/// The last sync report of downstream `idx`, if one was ever kept.
pub fn sync_of(&self, idx: usize) -> Option<DownstreamSync> {
self.sync.lock_or_recover().get(idx).cloned().flatten()
}
/// **The peer picture**: every paired downstream with its last sync point
/// judged against this box's `(boot, head)` now — the twin's words and rule
/// ([`crate::follow_point`]), silent after 3 of the downstream's declared
/// cycles, stalled after 5 — and today's pulls.
pub fn rows(
&self,
boot: u64,
head: u64,
now_ms: i64,
) -> Vec<crate::grpc::holger_grow_proto::DownstreamRow> {
use crate::grpc::holger_grow_proto as pb;
let counts = self.snapshot((now_ms.max(0) / 1000) as u64);
let sync = self.sync.lock_or_recover().clone();
self.downstreams
.iter()
.enumerate()
.map(|(i, d)| {
let count = counts.get(i).map(|c| c.1).unwrap_or_default();
let mut row = pb::DownstreamRow {
name: d.name.clone(),
fingerprint: d.fingerprint.clone(),
state: "silent".into(),
requests_today: count.requests_today,
bytes_today: count.bytes_today,
last_pull_unix: count.last_pull_unix.unwrap_or(0) as i64,
fetched_total: count.fetched_total,
bytes_total: count.bytes_total,
fetched_today: count.fetched_today,
..Default::default()
};
// A record that only keeps totals (never reported) says nothing of a sync point.
let Some(s) = sync
.get(i)
.cloned()
.flatten()
.filter(|s| s.last_seen_ms > 0)
else {
row.repositories = d
.repos
.iter()
.map(|r| pb::DownstreamRepoRow {
repository: r.clone(),
state: "silent".into(),
head_seq: head,
lag_entries: head,
..Default::default()
})
.collect();
row.lag_entries = head;
return row;
};
let every = u64::from(if s.every_secs == 0 { 300 } else { s.every_secs }) * 1000;
let mut state: &'static str = "ok";
for r in &d.repos {
let rr = match s.repos.get(r) {
Some(rs) => {
let j = rs.point.judge(boot, head, now_ms, 3 * every, 5 * every);
pb::DownstreamRepoRow {
repository: r.clone(),
state: j.state.into(),
lag_entries: j.lag,
applied_seq: j.applied,
head_seq: head,
last_complete_unix_ms: rs.last_complete_ms,
}
}
// Paired, and never reported: it says nothing of this one.
None => pb::DownstreamRepoRow {
repository: r.clone(),
state: "silent".into(),
head_seq: head,
lag_entries: head,
..Default::default()
},
};
state = crate::follow_point::worse(state, word(&rr.state));
row.lag_entries = row.lag_entries.max(rr.lag_entries);
row.repositories.push(rr);
}
row.state = state.into();
row.stalled = state == "stalled";
row.last_seen_unix_ms = s.last_seen_ms;
row.seen_ago_ms = (now_ms - s.last_seen_ms).max(0) as u64;
row.every_secs = s.every_secs;
row.summary = s.summary.clone().map(Into::into);
row
})
.collect()
}
/// The paired downstreams.
pub fn downstreams(&self) -> &[DownstreamRecord] {
&self.downstreams
}
/// The paired upstreams.
pub fn upstreams(&self) -> &[UpstreamRecord] {
&self.upstreams
}
/// The pairing for a mirror source, if it has one.
pub fn upstream_for_source(&self, source: &str) -> Option<&UpstreamRecord> {
self.upstreams.iter().find(|u| u.source == source)
}
/// Which paired downstream presented this leaf certificate, if any.
pub fn downstream_by_cert(&self, leaf_der: &[u8]) -> Option<usize> {
if self.downstreams.is_empty() {
return None;
}
let fp = fingerprint_der(leaf_der);
self.downstreams.iter().position(|d| d.fingerprint == fp)
}
/// Count one pull by downstream `idx`. `bytes` may be 0 when unknown.
pub fn record_pull(&self, idx: usize, bytes: u64, now_unix: u64) {
let mut t = self.tally.lock_or_recover();
let day = now_unix / crate::visitors::WINDOW_SECS;
if day != t.day {
t.day = day;
for c in t.counts.iter_mut() {
c.requests_today = 0;
c.bytes_today = 0;
c.fetched_today = 0;
}
}
if let Some(c) = t.counts.get_mut(idx) {
c.requests_today += 1;
c.bytes_today += bytes;
c.last_pull_unix = Some(now_unix);
}
}
/// **One artifact delivered** to downstream `idx` (a pull already counted
/// by [`PeerBoard::record_pull`]): its bytes, and one package fetched —
/// today and in the durable total (lane A13).
pub fn delivered(&self, idx: usize, bytes: u64) {
let mut t = self.tally.lock_or_recover();
if let Some(c) = t.counts.get_mut(idx) {
c.bytes_today += bytes;
c.fetched_today += 1;
c.fetched_total += 1;
c.bytes_total += bytes;
}
if let Some(d) = t.dirty.get_mut(idx) {
*d = true;
}
}
/// Each downstream with its numbers, as of `now_unix`.
pub fn snapshot(&self, now_unix: u64) -> Vec<(DownstreamRecord, PeerCount)> {
let t = self.tally.lock_or_recover();
let same_day = t.day == now_unix / crate::visitors::WINDOW_SECS;
self.downstreams
.iter()
.zip(t.counts.iter())
.map(|(d, c)| {
let mut c = *c;
if !same_day {
c.requests_today = 0;
c.bytes_today = 0;
c.fetched_today = 0;
}
(d.clone(), c)
})
.collect()
}
}
/// The boot's board, for the clean stop's flush (lane A13).
static LIVE: std::sync::OnceLock<std::sync::Arc<PeerBoard>> = std::sync::OnceLock::new();
/// Name the boot's board (once; a second call is ignored).
pub fn set_live(board: std::sync::Arc<PeerBoard>) {
let _ = LIVE.set(board);
}
/// Flush the boot's board, if one was named.
pub fn flush_live() -> Result<usize, String> {
LIVE.get().map(|b| b.flush()).unwrap_or(Ok(0))
}
// ─────────────────────────── the downstream's pin check ──────────────────────
/// What the downstream knows about its paired upstream's key.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum LinkState {
/// No cycle has checked it yet.
Unchecked,
/// The upstream presented the pinned key.
Paired,
/// The upstream could not be reached to check.
Unreachable,
/// The upstream presented a DIFFERENT key. Cycles are refused.
KeyMismatch,
}
impl LinkState {
/// The word the picture prints.
pub fn word(self) -> &'static str {
match self {
LinkState::Unchecked => "not yet checked",
LinkState::Paired => "paired",
LinkState::Unreachable => "unreachable",
LinkState::KeyMismatch => "key-mismatch",
}
}
pub(crate) fn to_u8(self) -> u8 {
match self {
LinkState::Unchecked => 0,
LinkState::Paired => 1,
LinkState::Unreachable => 2,
LinkState::KeyMismatch => 3,
}
}
pub(crate) fn from_u8(v: u8) -> LinkState {
match v {
1 => LinkState::Paired,
2 => LinkState::Unreachable,
3 => LinkState::KeyMismatch,
_ => LinkState::Unchecked,
}
}
}
/// **The verdict on a presented upstream key.** Pure, so the one rule is testable
/// without a socket.
pub fn verdict(pinned: &UpstreamRecord, presented: Option<&str>) -> LinkState {
match presented {
None => LinkState::Unreachable,
Some(fp) if fp == pinned.fingerprint => LinkState::Paired,
Some(_) => LinkState::KeyMismatch,
}
}
/// A verifier that ACCEPTS whatever the server presents, so the probe can read
/// the certificate and compare it with the pin. The pin comparison IS the check;
/// the real data channel is a separate connection that still verifies the chain
/// against the configured CA exactly as before. Signatures are still verified —
/// a server that cannot prove it holds the key for the certificate it shows
/// fails the handshake.
#[derive(Debug)]
struct CaptureOnly(std::sync::Arc<rustls::crypto::CryptoProvider>);
impl rustls::client::danger::ServerCertVerifier for CaptureOnly {
fn verify_server_cert(
&self,
_end_entity: &rustls::pki_types::CertificateDer<'_>,
_intermediates: &[rustls::pki_types::CertificateDer<'_>],
_server_name: &rustls::pki_types::ServerName<'_>,
_ocsp: &[u8],
_now: rustls::pki_types::UnixTime,
) -> Result<rustls::client::danger::ServerCertVerified, rustls::Error> {
Ok(rustls::client::danger::ServerCertVerified::assertion())
}
fn verify_tls12_signature(
&self,
message: &[u8],
cert: &rustls::pki_types::CertificateDer<'_>,
dss: &rustls::DigitallySignedStruct,
) -> Result<rustls::client::danger::HandshakeSignatureValid, rustls::Error> {
rustls::crypto::verify_tls12_signature(
message,
cert,
dss,
&self.0.signature_verification_algorithms,
)
}
fn verify_tls13_signature(
&self,
message: &[u8],
cert: &rustls::pki_types::CertificateDer<'_>,
dss: &rustls::DigitallySignedStruct,
) -> Result<rustls::client::danger::HandshakeSignatureValid, rustls::Error> {
rustls::crypto::verify_tls13_signature(
message,
cert,
dss,
&self.0.signature_verification_algorithms,
)
}
fn supported_verify_schemes(&self) -> Vec<rustls::SignatureScheme> {
self.0.signature_verification_algorithms.supported_schemes()
}
}
/// `host:port` out of `https://host:port/...`.
fn host_port(endpoint: &str) -> Option<(String, String)> {
let rest = endpoint
.split_once("://")
.map(|(_, r)| r)
.unwrap_or(endpoint);
let authority = rest.split('/').next()?;
let (host, port) = match authority.rsplit_once(':') {
Some((h, p)) if p.chars().all(|c| c.is_ascii_digit()) && !p.is_empty() => {
(h, p.to_string())
}
_ => (authority, "443".to_string()),
};
let host = host.trim_start_matches('[').trim_end_matches(']');
Some((host.to_string(), format!("{host}:{port}")))
}
/// **Read the upstream's certificate fingerprint** with one TLS handshake,
/// presenting `client_identity` (cert PEM, key PEM) when the upstream demands a
/// client certificate. `None` when it could not be reached or spoke no TLS.
pub async fn probe_upstream_key(
endpoint: &str,
client_identity: Option<(Vec<u8>, Vec<u8>)>,
within: std::time::Duration,
) -> Option<String> {
let (host, addr) = host_port(endpoint)?;
let provider = std::sync::Arc::new(rustls::crypto::ring::default_provider());
// TLS 1.3 only, box to box (stinky SB #39).
let builder = rustls::ClientConfig::builder_with_provider(provider.clone())
.with_protocol_versions(&[&rustls::version::TLS13])
.ok()?
.dangerous()
.with_custom_certificate_verifier(std::sync::Arc::new(CaptureOnly(provider)));
let config = match client_identity {
Some((cert, key)) => {
let certs: Vec<_> = rustls::pki_types::CertificateDer::pem_slice_iter(&cert[..])
.filter_map(|c| c.ok())
.collect();
let key = rustls::pki_types::PrivateKeyDer::from_pem_slice(&key[..]).ok()?;
builder.with_client_auth_cert(certs, key).ok()?
}
None => builder.with_no_client_auth(),
};
let name = rustls::pki_types::ServerName::try_from(host).ok()?;
let connector = tokio_rustls::TlsConnector::from(std::sync::Arc::new(config));
let fut = async move {
crate::egress::check("peer probe", &addr).ok()?;
let tcp = tokio::net::TcpStream::connect(addr).await.ok()?;
let tls = connector.connect(name, tcp).await.ok()?;
let (_, conn) = tls.get_ref();
let leaf = conn.peer_certificates()?.first()?.clone();
Some(fingerprint_der(leaf.as_ref()))
};
tokio::time::timeout(within, fut).await.ok().flatten()
}
/// Per-source link states, for the picture — keyed by mirror source name.
pub type LinkStates = BTreeMap<String, LinkState>;
#[cfg(test)]
mod tests {
use super::*;
fn cert_pem(cn: &str) -> Vec<u8> {
let ck = rcgen::generate_simple_self_signed(vec![cn.to_string()]).expect("a cert");
ck.cert.pem().into_bytes()
}
/// ★ A pairing is made ONCE: the second one under the same name is refused
/// by name, and the record survives a reload byte for byte.
#[test]
fn a_pairing_is_recorded_once_and_reloads() {
let dir = tempfile::tempdir().unwrap();
let pem = cert_pem("onprem-1");
let d = pair_downstream(dir.path(), "onprem-1", &pem, &["B".into()], "ops", 100).unwrap();
assert!(d.fingerprint.starts_with("sha256:") && d.fingerprint.len() == 7 + 64);
let again =
pair_downstream(dir.path(), "onprem-1", &pem, &["B".into()], "ops", 200).unwrap_err();
assert!(again.contains("holger-peer-1"), "{again}");
let u = pair_upstream(
dir.path(),
"cloud",
"holger-rs",
&cert_pem("cloud"),
"ops",
300,
)
.unwrap();
let (down, up) = load(dir.path()).unwrap();
assert_eq!(down, vec![d]);
assert_eq!(up, vec![u]);
assert!(pair_downstream(dir.path(), "../x", &pem, &["B".into()], "ops", 1).is_err());
}
/// ★ The key check is a named state: the pin matches → paired; another key
/// → key-mismatch; no answer → unreachable.
#[test]
fn the_upstream_key_verdict_names_the_state() {
let rec = UpstreamRecord {
name: "cloud".into(),
fingerprint: fingerprint_pem(&cert_pem("cloud")).unwrap(),
source: "cloud".into(),
paired_at: 0,
paired_by: "t".into(),
};
assert_eq!(verdict(&rec, Some(&rec.fingerprint)), LinkState::Paired);
let other = fingerprint_pem(&cert_pem("impostor")).unwrap();
assert_eq!(verdict(&rec, Some(&other)), LinkState::KeyMismatch);
assert_eq!(LinkState::KeyMismatch.word(), "key-mismatch");
assert_eq!(verdict(&rec, None), LinkState::Unreachable);
}
/// The board finds a downstream by its certificate and by nothing else.
#[test]
fn a_downstream_is_found_by_its_certificate() {
let ck = rcgen::generate_simple_self_signed(vec!["onprem-1".to_string()]).unwrap();
let der = ck.cert.der().to_vec();
let rec = DownstreamRecord {
name: "onprem-1".into(),
fingerprint: fingerprint_der(&der),
repos: vec!["B".into()],
paired_at: 0,
paired_by: "t".into(),
};
let board = PeerBoard::new(vec![rec], vec![]);
assert_eq!(board.downstream_by_cert(&der), Some(0));
let other = rcgen::generate_simple_self_signed(vec!["x".to_string()]).unwrap();
assert_eq!(board.downstream_by_cert(other.cert.der()), None);
board.record_pull(0, 10, 86_400 * 3 + 5);
board.record_pull(0, 5, 86_400 * 3 + 6);
let s = board.snapshot(86_400 * 3 + 7);
assert_eq!(
s[0].1,
PeerCount {
last_pull_unix: Some(86_400 * 3 + 6),
requests_today: 2,
bytes_today: 15,
..Default::default()
}
);
assert_eq!(
board.snapshot(86_400 * 4 + 1)[0].1.requests_today,
0,
"a new day starts at zero"
);
}
fn paired_board(dir: &Path) -> (PeerBoard, Vec<u8>) {
let ck = rcgen::generate_simple_self_signed(vec!["onprem-1".to_string()]).unwrap();
let der = ck.cert.der().to_vec();
let rec = DownstreamRecord {
name: "onprem-1".into(),
fingerprint: fingerprint_der(&der),
repos: vec!["B".into(), "C".into()],
paired_at: 0,
paired_by: "t".into(),
};
(
PeerBoard::new(vec![rec], vec![])
.with_sync_root(dir)
.unwrap(),
der,
)
}
fn point(
repo: &str,
boot: u64,
seq: u64,
done: i64,
) -> crate::grpc::holger_grow_proto::DownstreamPoint {
crate::grpc::holger_grow_proto::DownstreamPoint {
repository: repo.into(),
applied_boot: boot,
applied_seq: seq,
last_complete_unix_ms: done,
}
}
/// ★ **A paired downstream's RON summary is kept with its report** (HOL-FLOW-SMART),
/// handed on in its board row, and read back after a restart.
#[test]
fn a_paired_downstreams_ron_summary_is_kept_and_in_its_row() {
use crate::grpc::holger_grow_proto::{DownstreamReportRequest, DownstreamSummary};
let dir = tempfile::tempdir().unwrap();
let (board, der) = paired_board(dir.path());
let summary = DownstreamSummary {
boundary: "Fetch".into(),
repositories: vec!["mirror-b".into()],
first_import: "admin".into(),
..Default::default()
};
let req = DownstreamReportRequest {
points: vec![point("B", 7, 10, 5_000)],
every_secs: 60,
summary: Some(summary.clone()),
summary_proof: String::new(),
console_proof: String::new(),
};
board.report(Some(&der), &req, 7, 15, 6_000).unwrap();
assert_eq!(board.rows(7, 15, 6_000)[0].summary, Some(summary.clone()));
let (again, _) = paired_board(dir.path());
assert_eq!(
again.rows(7, 15, 7_000)[0].summary,
Some(summary),
"read back after a restart"
);
}
/// ★ **Only the paired key may say how far it synced**, and only of what it
/// is paired for — each refusal by name; the word it may say is kept under
/// the data root and read back after a restart, judged against the NEW
/// boot's head (a point from another boot counts nothing).
#[test]
fn a_sync_report_is_held_to_the_pairing_and_survives_a_restart() {
use crate::grpc::holger_grow_proto::DownstreamReportRequest;
let dir = tempfile::tempdir().unwrap();
let (board, der) = paired_board(dir.path());
let req = DownstreamReportRequest {
points: vec![point("B", 7, 10, 5_000)],
every_secs: 60,
summary: None,
summary_proof: String::new(),
console_proof: String::new(),
};
let none = board.report(None, &req, 7, 15, 6_000).unwrap_err();
assert!(matches!(none, ReportRefused::NotPaired(_)));
assert!(none.status().message().contains("holger-peer-4"));
let other = rcgen::generate_simple_self_signed(vec!["x".to_string()]).unwrap();
let stranger = board
.report(Some(other.cert.der()), &req, 7, 15, 6_000)
.unwrap_err();
assert!(
stranger
.clone()
.status()
.message()
.contains("holger-peer-4"),
"{stranger:?}"
);
let off_list = DownstreamReportRequest {
points: vec![point("staging", 7, 10, 5_000)],
every_secs: 60,
summary: None,
summary_proof: String::new(),
console_proof: String::new(),
};
let refused = board
.report(Some(&der), &off_list, 7, 15, 6_000)
.unwrap_err();
assert!(refused.status().message().contains("holger-peer-2"));
assert!(board.sync_of(0).is_none(), "a refused report keeps nothing");
let reply = board.report(Some(&der), &req, 7, 15, 6_000).unwrap();
assert_eq!((reply.boot, reply.head_seq), (7, 15));
let rows = board.rows(7, 15, 6_000);
assert_eq!(rows.len(), 1);
let b = &rows[0].repositories[0];
assert_eq!(
(
b.repository.as_str(),
b.state.as_str(),
b.applied_seq,
b.lag_entries,
b.last_complete_unix_ms
),
("B", "behind", 10, 5, 5_000)
);
assert_eq!(
rows[0].repositories[1].state, "silent",
"C is paired and was never reported"
);
assert_eq!(rows[0].state, "silent");
assert_eq!(rows[0].lag_entries, 15);
assert!(dir
.path()
.join(PEERS_DIR)
.join("sync-onprem-1.ron")
.is_file());
// The restart: a new board over the same root reads the last word back.
let again = PeerBoard::new(board.downstreams().to_vec(), vec![])
.with_sync_root(dir.path())
.unwrap();
assert_eq!(again.sync_of(0), board.sync_of(0));
let after = again.rows(8, 3, 7_000);
let b = &after[0].repositories[0];
assert_eq!(
(b.applied_seq, b.lag_entries, b.last_complete_unix_ms),
(0, 3, 5_000),
"another boot's point counts nothing; when it synced is kept"
);
// The pairing loader is not confused by the sync record beside it.
assert_eq!(
load(dir.path()).unwrap().0.len(),
0,
"this fixture paired in memory only"
);
}
/// ★ **A downstream that keeps reporting the same point behind the head is
/// `stalled`** after five of its declared cycles; one that moves is
/// `behind`; one that catches up is `ok`; one that goes quiet is `silent`.
#[test]
fn a_downstream_that_repeats_its_point_is_stalled() {
use crate::grpc::holger_grow_proto::DownstreamReportRequest;
let dir = tempfile::tempdir().unwrap();
let (board, der) = paired_board(dir.path());
let both = |seq| DownstreamReportRequest {
points: vec![point("B", 7, seq, 1), point("C", 7, seq, 1)],
every_secs: 1,
summary: None,
summary_proof: String::new(),
console_proof: String::new(),
};
board.report(Some(&der), &both(10), 7, 20, 0).unwrap();
board.report(Some(&der), &both(10), 7, 20, 5_500).unwrap();
let r = &board.rows(7, 20, 5_500)[0];
assert!(r.stalled && r.state == "stalled", "{r:?}");
board.report(Some(&der), &both(12), 7, 20, 6_000).unwrap();
assert_eq!(board.rows(7, 20, 6_000)[0].state, "behind");
board.report(Some(&der), &both(20), 7, 20, 6_500).unwrap();
let r = &board.rows(7, 20, 6_500)[0];
assert_eq!(
(r.state.as_str(), r.lag_entries, r.stalled),
("ok", 0, false)
);
assert_eq!(board.rows(7, 20, 60_000)[0].state, "silent");
}
/// ★ **How many packages a downstream has fetched survives a restart** (lane
/// A13): every delivery counts one, today and in total; the total is kept in
/// its sync record (at its report, and by the flush for one that never
/// reported) and read back by the next boot, which counts on from it. A sync
/// record written before A13 still reads, with a total of 0.
#[test]
fn a_downstreams_fetched_total_survives_a_restart() {
use crate::grpc::holger_grow_proto::DownstreamReportRequest;
let dir = tempfile::tempdir().unwrap();
let (board, der) = paired_board(dir.path());
let now = 86_400 * 10;
for bytes in [100, 200, 300] {
board.record_pull(0, 0, now);
board.delivered(0, bytes);
}
board.record_pull(0, 0, now); // a listing: a pull, not a package
let r = &board.rows(7, 20, (now * 1000) as i64)[0];
assert_eq!(
(
r.fetched_total,
r.fetched_today,
r.bytes_total,
r.requests_today
),
(3, 3, 600, 4)
);
assert_eq!(
r.state, "silent",
"delivered, and never reported: no sync point"
);
// Never reported: the flush keeps the totals anyway.
assert_eq!(board.flush().unwrap(), 1);
assert_eq!(board.flush().unwrap(), 0, "nothing moved since");
let again = PeerBoard::new(board.downstreams().to_vec(), vec![])
.with_sync_root(dir.path())
.unwrap();
let r = &again.rows(7, 20, ((now + 86_400) * 1000) as i64)[0];
assert_eq!(
(
r.fetched_total,
r.bytes_total,
r.fetched_today,
r.last_seen_unix_ms,
r.state.as_str()
),
(3, 600, 0, 0, "silent"),
"the total is kept; today is a new day; still never heard"
);
// It reports, then takes one more, then a restart: 4.
again.delivered(0, 50);
again
.report(
Some(&der),
&DownstreamReportRequest {
points: vec![point("B", 7, 20, 1)],
every_secs: 60,
summary: None,
summary_proof: String::new(),
console_proof: String::new(),
},
7,
20,
((now + 86_400) * 1000) as i64,
)
.unwrap();
let third = PeerBoard::new(board.downstreams().to_vec(), vec![])
.with_sync_root(dir.path())
.unwrap();
let r = &third.rows(7, 20, ((now + 86_400) * 1000) as i64)[0];
assert_eq!(
(r.fetched_total, r.bytes_total),
(4, 650),
"the report wrote the total"
);
// A record from before A13 reads, with no total.
std::fs::write(
dir.path().join(PEERS_DIR).join("sync-onprem-1.ron"),
"(name: \"onprem-1\", every_secs: 60, last_seen_ms: 5, repos: {})",
)
.unwrap();
let old = PeerBoard::new(board.downstreams().to_vec(), vec![])
.with_sync_root(dir.path())
.unwrap();
assert_eq!(old.rows(7, 20, 10)[0].fetched_total, 0);
}
}