server/lib/src/upstreams.rs
//! **Where this box's bytes come from** (lane A13, 2026-10-23): the upstreams the
//! estate board draws above the main box, and the evidence of each one's state.
//!
//! Two kinds, each declared in the config and registered when it is wired:
//!
//! * **`proxy`** — a repository another repository names in `ron_upstreams`: a
//! registry this box pulls through (crates.io as `rust-remote`, PyPI or npm
//! behind a `nexus-remote`/`artifactory-remote`/…, or a local repository).
//! [`crate::proxy::ProxyBackend`] records every member's outcome here as it
//! walks the chain: a hit, a clean miss, or an error, with the time. The
//! numbers are SINCE THIS BOOT, like `/metrics`, and the board says so.
//! * **`mirror`** — a `mirror.ron_sources` entry (another holger, paired or
//! not). Its evidence is the mirror's own health
//! ([`crate::metrics::MirrorSourceHealth`]: cycles, artifacts fetched, the last
//! success, the pin check), read at gather time; this registry only keeps
//! what the declaration says (endpoint, cadence, the pairing's name).
//!
//! Nothing here dials an upstream: the board is drawn from what the walks and the
//! cycles already recorded. One process-wide registry, like the metrics.
use crate::lock::LockOrRecover as _;
use std::collections::BTreeMap;
use std::sync::Mutex;
pub const KIND_PROXY: &str = "proxy";
pub const KIND_MIRROR: &str = "mirror";
/// A `sources:` entry the pull-through ingests into the steps (`npm`, `python`, `maven`, …).
pub const KIND_PULL: &str = "pull";
/// A `sources:` upload door (or the intake chain's `incoming` when no source is named).
pub const KIND_UPLOAD: &str = "upload";
/// A downstream box's `sync.upstreams` entry: the upstream whose signed manifests it applies.
pub const KIND_SYNC: &str = "sync";
/// What one walk of the proxy chain learnt of one member.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum Outcome<'a> {
/// It served the artifact.
Hit,
/// It answered, and does not hold it.
Miss,
/// It did not answer (connection refused, DNS, TLS, a non-404 status).
Error(&'a str),
}
/// **One upstream as the board reads it**, at one instant.
#[derive(Clone, Debug, Default, PartialEq, Eq)]
pub struct UpstreamFact {
pub name: String,
/// [`KIND_PROXY`] | [`KIND_MIRROR`].
pub kind: String,
/// The repository type (`rust-remote`, `nexus-remote`, …) or the mirror
/// source's kind (`Holger`, …).
pub speaks: String,
/// Its address as declared; empty when the declaration names none.
pub address: String,
/// The local repositories it feeds.
pub feeds: Vec<String>,
/// A mirror whose upstream is a PAIRED holger: the pairing's name for it.
pub paired_as: String,
/// A mirror's declared cycle, seconds.
pub every_secs: u64,
/// When it was registered (Unix ms): the floor of "never since boot".
pub registered_ms: i64,
/// Answers (proxy: hits + clean misses; mirror: completed cycles).
pub answered: u64,
/// Failures (proxy: errors; mirror: failed cycles).
pub failed: u64,
/// Artifacts taken from it (proxy: hits; mirror: artifacts fetched).
pub fetched: u64,
/// The last answer (Unix ms); 0 = none since boot.
pub last_ok_ms: i64,
/// The last failure (Unix ms); 0 = none since boot.
pub last_err_ms: i64,
/// The last failure's words, short.
pub last_error: String,
/// A mirror: failures since its last completed cycle.
pub consecutive_failures: u64,
/// A mirror of a paired holger: the pin check's word (`paired`,
/// `key-mismatch`, `unreachable`, `not yet checked`); empty otherwise.
pub link: String,
}
static REG: Mutex<BTreeMap<String, UpstreamFact>> = Mutex::new(BTreeMap::new());
fn key(kind: &str, name: &str) -> String {
format!("{kind}\u{0}{name}")
}
/// **A proxy member, as the wiring makes it**: `name` (its backend's name, the
/// one the walk records under), its repository type and declared address, and
/// the local repository it feeds. Registering twice adds the second feed.
pub fn register_proxy(name: &str, speaks: &str, address: &str, feeds: &str, now_ms: i64) {
let mut reg = REG.lock_or_recover();
let f = reg
.entry(key(KIND_PROXY, name))
.or_insert_with(|| UpstreamFact {
name: name.to_string(),
kind: KIND_PROXY.into(),
speaks: speaks.to_string(),
address: address.to_string(),
registered_ms: now_ms,
..Default::default()
});
if !feeds.is_empty() && !f.feeds.iter().any(|x| x == feeds) {
f.feeds.push(feeds.to_string());
f.feeds.sort();
}
}
/// **What a walk learnt of proxy member `name`.** A name nothing registered is
/// ignored: the registry only holds what the config declares.
pub fn record(name: &str, outcome: Outcome<'_>, now_ms: i64) {
let mut reg = REG.lock_or_recover();
let Some(f) = reg.get_mut(&key(KIND_PROXY, name)) else {
return;
};
match outcome {
Outcome::Hit => {
f.answered += 1;
f.fetched += 1;
f.last_ok_ms = now_ms;
}
Outcome::Miss => {
f.answered += 1;
f.last_ok_ms = now_ms;
}
Outcome::Error(e) => {
f.failed += 1;
f.last_err_ms = now_ms;
f.last_error = e.chars().take(120).collect();
}
}
}
/// **What the pull-through learnt of `sources:` entry `name`** (HBOARD-1): `fetched`
/// counts one artifact taken from it; `failed` a failure; any other outcome an answer.
pub fn record_pull(name: &str, outcome: &str, now_ms: i64) {
let mut reg = REG.lock_or_recover();
let f = reg
.entry(key(KIND_PULL, name))
.or_insert_with(|| UpstreamFact {
name: name.to_string(),
kind: KIND_PULL.into(),
registered_ms: now_ms,
..Default::default()
});
match outcome {
"fetched" => {
f.answered += 1;
f.fetched += 1;
f.last_ok_ms = now_ms;
}
"failed" => {
f.failed += 1;
f.last_err_ms = now_ms;
}
_ => {
f.answered += 1;
f.last_ok_ms = now_ms;
}
}
}
/// **A mirror source, as its timer starts.**
pub fn register_mirror(
name: &str,
speaks: &str,
endpoint: &str,
feeds: &[String],
every_secs: u64,
paired_as: &str,
now_ms: i64,
) {
let mut reg = REG.lock_or_recover();
let f = reg
.entry(key(KIND_MIRROR, name))
.or_insert_with(|| UpstreamFact {
registered_ms: now_ms,
..Default::default()
});
f.name = name.to_string();
f.kind = KIND_MIRROR.into();
f.speaks = speaks.to_string();
f.address = endpoint.to_string();
f.feeds = feeds.to_vec();
f.feeds.sort();
f.every_secs = every_secs;
f.paired_as = paired_as.to_string();
}
/// **Every upstream now**: proxies as recorded, mirrors with their health read
/// from the metrics at this instant. Ordered by kind, then name.
pub fn snapshot() -> Vec<UpstreamFact> {
let mut out: Vec<UpstreamFact> = REG.lock_or_recover().values().cloned().collect();
let health = crate::metrics::global().mirror_sources();
for f in out.iter_mut().filter(|f| f.kind == KIND_MIRROR) {
if let Some(h) = health.iter().find(|h| h.name() == f.name) {
let (ok, failed, fetched, _, _) = h.counts();
f.answered = ok;
f.failed = failed;
f.fetched = fetched;
f.last_ok_ms = h.last_success_unix().map(|s| s as i64 * 1000).unwrap_or(0);
f.consecutive_failures = h.consecutive_failures();
if !f.paired_as.is_empty() {
f.link = h.link_state().word().to_string();
}
}
}
out
}
#[cfg(test)]
mod tests {
use super::*;
/// ★ The walk's outcomes land on the member they belong to, and a member
/// the config never declared is not invented.
#[test]
fn a_walks_outcomes_are_kept_per_declared_member() {
register_proxy(
"t-a13-crates-io",
"rust-remote",
"https://index.crates.io",
"crates",
1,
);
register_proxy(
"t-a13-crates-io",
"rust-remote",
"https://index.crates.io",
"cargo-cache",
1,
);
record("t-a13-crates-io", Outcome::Hit, 10);
record("t-a13-crates-io", Outcome::Hit, 11);
record("t-a13-crates-io", Outcome::Miss, 12);
record("t-a13-crates-io", Outcome::Error("connection refused"), 13);
record("t-a13-never-declared", Outcome::Hit, 14);
let all = snapshot();
let f = all.iter().find(|f| f.name == "t-a13-crates-io").unwrap();
assert_eq!(
(f.answered, f.failed, f.fetched, f.last_ok_ms, f.last_err_ms),
(3, 1, 2, 12, 13)
);
assert_eq!(
f.feeds,
vec!["cargo-cache".to_string(), "crates".to_string()]
);
assert_eq!(f.last_error, "connection refused");
assert!(all.iter().all(|f| f.name != "t-a13-never-declared"));
}
/// ★ A mirror's numbers are its health's, read at the instant of the snapshot.
#[test]
fn a_mirror_reads_its_health() {
register_mirror(
"t-a13-cloud",
"Holger",
"https://cloud:50051",
&["B".into()],
60,
"cloud-holger",
1,
);
let h = crate::metrics::global().mirror_source("t-a13-cloud");
h.record_success(7, 700, 0);
h.record_failure();
h.set_link_state(crate::peers::LinkState::Paired);
let f = snapshot()
.into_iter()
.find(|f| f.name == "t-a13-cloud")
.unwrap();
assert_eq!(
(
f.kind.as_str(),
f.answered,
f.failed,
f.fetched,
f.consecutive_failures,
f.link.as_str()
),
(KIND_MIRROR, 1, 1, 7, 1, "paired")
);
assert!(f.last_ok_ms > 0);
}
}