server/lib/tests/mirror_resilience.rs
//! **The mirror's TIMER, across failures** — the half `mirror_declaration.rs`
//! does not reach.
//!
//! `mirror_declaration.rs` proves one cycle: list, diff, fetch, adopt, store. It
//! calls `run_pair` directly, so it can say nothing at all about the loop around
//! it — and the loop is where the field defect was.
//!
//! # The defect these tests are the regression for
//!
//! On `oden`, 2026-09-20 15:03:50, one second after a reboot:
//!
//! ```text
//! WARN holger_server_lib::mirror_cycle] mirror holger-rs: connect failed:
//! mirror: connect upstream http://holger.rs:50051
//! ```
//!
//! The network was one second behind the service, which is normal. What was NOT
//! normal is what an operator could then learn from outside the process: the
//! unit said `active`, `NRestarts: 0`, and the journal said nothing for eighty
//! minutes — because a completed cycle logs at `info` and the default log floor
//! is `warn`. Silence meant both "mirroring fine" and "not mirroring at all",
//! and no reading of the process could tell them apart.
//!
//! Three things had to become true, and each is a test here:
//!
//! 1. a cycle whose upstream is unreachable at the first tick mirrors on a later
//! tick **without intervention**, and sooner than the declared period,
//! 2. a restart re-fetches **nothing** it already holds — including a restart
//! that goes through the timer rather than through `run_pair`,
//! 3. a mirror that is not mirroring is visible from **outside** the process.
use std::collections::BTreeMap;
use std::sync::{Arc, Mutex};
use std::time::Duration;
use anyhow::Result;
use holger_server_lib::mirror_cycle::{
coord_to_id, id_to_coord, spawn_timers_with, MirrorPair, MirrorSource, MirrorSourceKind,
MirrorUpstream, RonUnprovenPolicy, SourcePlan, UpstreamConnector,
};
use traits::{
ArtifactEntry, ArtifactFormat, ArtifactId, Bytes, RemoteAsset, RepositoryBackendTrait,
};
// ───────────────────────────── the fakes ─────────────────────────────
/// An upstream holding `coord -> bytes` and counting its fetches, so
/// "re-fetched nothing" is a number and not a feeling.
#[derive(Clone, Default)]
struct FakeUpstream {
items: Arc<Mutex<BTreeMap<String, Vec<u8>>>>,
fetches: Arc<Mutex<usize>>,
/// Every sync report it was told, with the proof it carried (lane OX-WGDS2).
reports: Arc<Mutex<Vec<Told>>>,
}
/// One sync report as the fake upstream was told it: the applied sequences, the proof.
type Told = (Vec<u64>, Option<String>);
impl FakeUpstream {
fn put(&self, coord: &str, bytes: &[u8]) {
self.items
.lock()
.unwrap()
.insert(coord.to_string(), bytes.to_vec());
}
fn fetch_count(&self) -> usize {
*self.fetches.lock().unwrap()
}
}
#[async_trait::async_trait]
impl MirrorUpstream for FakeUpstream {
async fn list(&self, _repository: &str, _page_size: u32) -> Result<Vec<RemoteAsset>> {
let items = self.items.lock().unwrap();
Ok(items
.iter()
.map(|(coord, bytes)| RemoteAsset {
path: coord.clone(),
download_url: format!("fake://{coord}"),
content_type: None,
size: Some(bytes.len() as u64),
// The holger shape: a listing with no digest. Every seen
// coordinate lands in `unproven`, which is why these pairs
// declare `TrustSize` — see `mirror_cycle`'s module docs.
sha256: None,
change_token: None,
})
.collect())
}
async fn fetch(&self, _repository: &str, id: &ArtifactId) -> Result<Option<traits::Bytes>> {
*self.fetches.lock().unwrap() += 1;
Ok(self
.items
.lock()
.unwrap()
.get(&id_to_coord(id))
.cloned()
.map(traits::Bytes::from))
}
async fn report_sync(
&self,
req: holger_server_lib::grpc::holger_grow_proto::DownstreamReportRequest,
proof: Option<String>,
) -> Result<Option<(u64, u64)>> {
self.reports
.lock()
.unwrap()
.push((req.points.iter().map(|p| p.applied_seq).collect(), proof));
// The upstream's position now: boot 7, head 42.
Ok(Some((7, 42)))
}
}
/// A connector that REFUSES its first `refusals` connects — the boot race, as a
/// fixture: the box is up, the service is up, the network is not yet.
struct FlakyConnector {
up: FakeUpstream,
refusals: Mutex<usize>,
attempts: Arc<Mutex<usize>>,
}
#[async_trait::async_trait]
impl UpstreamConnector for FlakyConnector {
async fn connect(&self, src: &MirrorSource) -> Result<Box<dyn MirrorUpstream>> {
*self.attempts.lock().unwrap() += 1;
let mut left = self.refusals.lock().unwrap();
if *left > 0 {
*left -= 1;
// The exact error shape the real transport produces on a box whose
// interface is not up yet.
return Err(anyhow::anyhow!(
"tcp connect error: Network is unreachable (os error 101)"
)
.context(format!("mirror: connect upstream {}", src.ron_endpoint)));
}
Ok(Box::new(self.up.clone()) as Box<dyn MirrorUpstream>)
}
}
/// A writable in-memory repository, so the cycle's `put` is real without a disk.
#[derive(Default)]
struct MemBackend {
held: Mutex<BTreeMap<String, Vec<u8>>>,
}
impl MemBackend {
fn holds(&self, coord: &str) -> Option<Vec<u8>> {
self.held.lock().unwrap().get(coord).cloned()
}
}
impl RepositoryBackendTrait for MemBackend {
fn caps(&self) -> traits::Caps {
traits::Caps {
listing: true,
..traits::Caps::NONE
}
}
fn name(&self) -> &str {
"mem"
}
fn format(&self) -> ArtifactFormat {
ArtifactFormat::Raw
}
fn is_writable(&self) -> bool {
true
}
fn fetch(&self, id: &ArtifactId) -> Result<Option<Bytes>> {
Ok(self
.held
.lock()
.unwrap()
.get(&id_to_coord(id))
.map(|v| Bytes::from(v.clone())))
}
fn put(&self, id: &ArtifactId, data: &[u8]) -> Result<()> {
self.held
.lock()
.unwrap()
.insert(id_to_coord(id), data.to_vec());
Ok(())
}
fn handle_http2_request(
&self,
_m: &str,
_p: &str,
_b: &[u8],
) -> Result<(u16, Vec<(String, String)>, Vec<u8>)> {
anyhow::bail!("the mirror never goes through the HTTP door")
}
fn list(&self, _f: Option<&str>, _l: usize) -> Result<Vec<ArtifactEntry>> {
Ok(self
.held
.lock()
.unwrap()
.iter()
.filter_map(|(c, v)| {
coord_to_id(c).map(|id| ArtifactEntry {
id,
size_bytes: v.len() as i64,
content_type: String::new(),
// The mirror DESTINATION: an in-memory store that knows
// neither a publication time nor a digest, and says so.
..ArtifactEntry::UNKNOWN_PROVENANCE
})
})
.collect())
}
}
/// One source declared the way the live `holger-c.ron` declares it, but on a
/// one-second cadence so a test is a test and not a coffee break.
fn source(name: &str) -> MirrorSource {
MirrorSource {
ron_name: name.to_string(),
ron_kind: MirrorSourceKind::Holger,
ron_endpoint: "http://unreachable.invalid:50051".to_string(),
ron_credential: Default::default(),
ron_every_secs: 1,
ron_page_size: 1000,
ron_in_flight: 4,
ron_retry_secs: 1,
ron_connect_timeout_secs: 1,
ron_repositories: vec![MirrorPair {
ron_from: "bundles".into(),
ron_to: "bundles".into(),
ron_unproven: RonUnprovenPolicy::TrustSize,
}],
}
}
/// Poll `f` every 25 ms for up to `secs`. Returns false on timeout. No sleeps
/// longer than the thing being waited for, so a green run is fast and a red one
/// is bounded.
async fn within<F: FnMut() -> bool>(secs: u64, mut f: F) -> bool {
let deadline = std::time::Instant::now() + Duration::from_secs(secs);
while std::time::Instant::now() < deadline {
if f() {
return true;
}
tokio::time::sleep(Duration::from_millis(25)).await;
}
f()
}
// ───────────── lane OX-WGDS2: a sync downstream's mirror reports, signed ─────────────
/// ★ A box that is a SYNC downstream of its source (not paired with it) reports its sync
/// point before and after each cycle, every report carrying the proof its sync key signs
/// over that report; the point it reports is the upstream's head at the cycle's start.
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn a_sync_downstreams_mirror_reports_its_point_with_its_proof() {
let dir = tempfile::tempdir().expect("tempdir");
let up = FakeUpstream::default();
up.put("/bundle@1", b"bundle bytes");
let mem = Arc::new(MemBackend::default());
let connector = Arc::new(FlakyConnector {
up: up.clone(),
refusals: Mutex::new(0),
attempts: Arc::new(Mutex::new(0)),
});
let src = source("sync-downstream-reports");
let signed: holger_server_lib::mirror_cycle::ReportProof = Arc::new(|req| {
Some(format!(
"proof-over-{}",
req.points
.iter()
.map(|p| p.applied_seq.to_string())
.collect::<Vec<_>>()
.join(",")
))
});
let plan = SourcePlan {
source: src.clone(),
pairs: vec![(
src.ron_repositories[0].clone(),
mem.clone() as Arc<dyn RepositoryBackendTrait>,
)],
state_dir: dir.path().to_path_buf(),
pinned: None,
report_as: Some(signed),
summary: None,
summary_as: None,
console: None,
};
let handles = spawn_timers_with(vec![plan], connector);
assert!(
within(20, || up.reports.lock().unwrap().len() >= 2).await,
"it reported before and after its first cycle"
);
let reports = up.reports.lock().unwrap().clone();
assert_eq!(
reports[0],
(vec![0], Some("proof-over-0".to_string())),
"before the first cycle: nothing applied, signed"
);
assert_eq!(
reports[1],
(vec![42], Some("proof-over-42".to_string())),
"after it: the head at the cycle's start, signed"
);
assert!(mem.holds("/bundle@1").is_some());
for h in handles {
h.abort();
}
}
// ───────────────────── PROOF 1: a failed tick is not the end ─────────────────────
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn a_mirror_whose_first_connect_fails_mirrors_on_a_later_tick_without_intervention() {
let dir = tempfile::tempdir().expect("tempdir");
let up = FakeUpstream::default();
up.put(
"/crossing-proof@1.0",
b"thirty-nine bytes of downstream proof!!",
);
let mem = Arc::new(MemBackend::default());
let attempts = Arc::new(Mutex::new(0usize));
let connector = Arc::new(FlakyConnector {
up: up.clone(),
// Three refusals: the boot tick and two retries. The mirror gets no help
// in between — nobody restarts it, nobody runs a cycle by hand.
refusals: Mutex::new(3),
attempts: attempts.clone(),
});
let src = source("first-tick-fails");
let plan = SourcePlan {
source: src.clone(),
pairs: vec![(
src.ron_repositories[0].clone(),
mem.clone() as Arc<dyn RepositoryBackendTrait>,
)],
state_dir: dir.path().to_path_buf(),
pinned: None,
report_as: None,
summary: None,
summary_as: None,
console: None,
};
let handles = spawn_timers_with(vec![plan], connector.clone());
let landed = within(20, || mem.holds("/crossing-proof@1.0").is_some()).await;
assert!(
landed,
"the upstream was unreachable for the first three ticks and then came back; a mirror \
that needs a human after that is the defect"
);
assert_eq!(
mem.holds("/crossing-proof@1.0").as_deref(),
Some(&b"thirty-nine bytes of downstream proof!!"[..]),
"and the bytes are the upstream's"
);
assert!(
*attempts.lock().unwrap() >= 4,
"it must have RETRIED, not succeeded on one attempt: {} attempt(s)",
attempts.lock().unwrap()
);
// The health the operator reads from outside says the same thing.
let health = holger_server_lib::metrics::global().mirror_source("first-tick-fails");
let (ok, failed, fetched, _, bytes) = health.counts();
assert!(ok >= 1, "a completed cycle was counted");
assert!(failed >= 3, "and so were the failures: {failed}");
assert_eq!(fetched, 1);
assert_eq!(bytes, 39);
assert_eq!(
health.consecutive_failures(),
0,
"the streak is cleared by a completed cycle"
);
assert!(
health.last_success_unix().is_some(),
"and the timestamp moved"
);
for h in handles {
h.abort();
}
}
// ───────────── PROOF 1b: the retry is SOONER than the declared period ─────────────
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn the_retry_after_a_failure_comes_sooner_than_the_declared_period() {
let dir = tempfile::tempdir().expect("tempdir");
let up = FakeUpstream::default();
up.put("/a@1", b"x");
let mem = Arc::new(MemBackend::default());
let attempts = Arc::new(Mutex::new(0usize));
// A TEN SECOND period with a one second first retry: if a failed tick had to
// wait the period, nothing could have landed inside five seconds.
let mut src = source("retry-is-sooner");
src.ron_every_secs = 10;
src.ron_retry_secs = 1;
let connector = Arc::new(FlakyConnector {
up: up.clone(),
refusals: Mutex::new(2),
attempts: attempts.clone(),
});
let plan = SourcePlan {
source: src.clone(),
pairs: vec![(
src.ron_repositories[0].clone(),
mem.clone() as Arc<dyn RepositoryBackendTrait>,
)],
state_dir: dir.path().to_path_buf(),
pinned: None,
report_as: None,
summary: None,
summary_as: None,
console: None,
};
let started = std::time::Instant::now();
let handles = spawn_timers_with(vec![plan], connector);
assert!(
within(9, || mem.holds("/a@1").is_some()).await,
"two failed ticks under a 10s period must not cost 20s"
);
let took = started.elapsed();
assert!(
took < Duration::from_secs(9),
"the backoff (1s, then 2s) must beat the declared period: took {took:?}"
);
for h in handles {
h.abort();
}
}
// ───────────── PROOF 2: a restart re-fetches nothing it already holds ─────────────
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn a_restart_through_the_timer_refetches_nothing_it_already_holds() {
let dir = tempfile::tempdir().expect("tempdir");
let up = FakeUpstream::default();
for i in 0..20 {
up.put(&format!("/pkg{i:02}@1.0"), format!("body-{i}").as_bytes());
}
let mem = Arc::new(MemBackend::default());
let src = source("restart-refetches-nothing");
let plan = |mem: Arc<MemBackend>| SourcePlan {
source: src.clone(),
pairs: vec![(
src.ron_repositories[0].clone(),
mem as Arc<dyn RepositoryBackendTrait>,
)],
state_dir: dir.path().to_path_buf(),
pinned: None,
report_as: None,
summary: None,
summary_as: None,
console: None,
};
// First boot: it takes the world.
let connector = Arc::new(FlakyConnector {
up: up.clone(),
refusals: Mutex::new(0),
attempts: Arc::new(Mutex::new(0)),
});
let first = spawn_timers_with(vec![plan(mem.clone())], connector.clone());
assert!(
within(20, || mem.holds("/pkg19@1.0").is_some()).await,
"first boot mirrors"
);
assert!(
within(20, || up.fetch_count() == 20).await,
"exactly the 20 it did not have: {}",
up.fetch_count()
);
let after_first_boot = up.fetch_count();
for h in first {
h.abort();
}
// Let the aborted task's last cycle (if any) settle before counting again.
tokio::time::sleep(Duration::from_millis(200)).await;
let before_restart = up.fetch_count();
// THE RESTART: a brand new timer, a brand new in-memory map, the SAME state
// dir — which is the only thing a restart is to this engine. The local store
// is the same one, because a restart does not empty the disk.
let second = spawn_timers_with(vec![plan(mem.clone())], connector);
// Give it several full periods to do the wrong thing, if it is going to.
tokio::time::sleep(Duration::from_secs(4)).await;
assert_eq!(
up.fetch_count(),
before_restart,
"a restarted mirror re-pulled {} artifact(s) it already held — on a metered link that \
is the bill a customer notices first (first boot took {after_first_boot})",
up.fetch_count() - before_restart
);
let health = holger_server_lib::metrics::global().mirror_source("restart-refetches-nothing");
let (ok, _, fetched, _, _) = health.counts();
assert!(ok >= 2, "several cycles ran across the restart: {ok}");
assert_eq!(
fetched, 20,
"and exactly twenty artifacts were ever fetched"
);
for h in second {
h.abort();
}
}
// ───────────── PROOF 3: a stopped mirror is visible from outside ─────────────
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn a_mirror_that_cannot_reach_its_upstream_says_so_in_metrics() {
let dir = tempfile::tempdir().expect("tempdir");
let mem = Arc::new(MemBackend::default());
let src = source("never-reachable");
let connector = Arc::new(FlakyConnector {
up: FakeUpstream::default(),
// It never comes back. This is the shape systemd reports as `active`.
refusals: Mutex::new(usize::MAX),
attempts: Arc::new(Mutex::new(0)),
});
let plan = SourcePlan {
source: src.clone(),
pairs: vec![(
src.ron_repositories[0].clone(),
mem.clone() as Arc<dyn RepositoryBackendTrait>,
)],
state_dir: dir.path().to_path_buf(),
pinned: None,
report_as: None,
summary: None,
summary_as: None,
console: None,
};
let handles = spawn_timers_with(vec![plan], connector);
let health = holger_server_lib::metrics::global().mirror_source("never-reachable");
assert!(
within(10, || health.consecutive_failures() >= 2).await,
"the failures must be COUNTED, not merely logged"
);
assert_eq!(
health.last_success_unix(),
None,
"and the success timestamp must stay unset — this is the reading that separates \
'mirroring fine' from 'not mirroring', which no amount of process liveness can"
);
// What a scrape sees. The series EXISTS from boot, so a never-successful
// mirror is a reading of 0 rather than an absent line a dashboard draws as
// a gap.
let text = holger_server_lib::metrics::global().render_prometheus(&[]);
assert!(
text.contains("holger_mirror_last_success_timestamp_seconds{source=\"never-reachable\"} 0"),
"/metrics must publish the never-succeeded reading:\n{text}"
);
assert!(
text.contains("holger_mirror_cycles_total{source=\"never-reachable\",result=\"failed\"}"),
"/metrics must publish the failure count:\n{text}"
);
for h in handles {
h.abort();
}
}
/// ★ **A holger mirror's report carries its own RON summary** (HOL-FLOW-SMART): what the
/// plan says of this box (repositories, boundary, inbox, first import, domain), in every
/// report, so the upstream's flow can unfold it.
#[test]
fn a_holger_mirrors_report_carries_its_ron_summary() {
use holger_server_lib::grpc::holger_grow_proto::DownstreamSummary;
let dir = tempfile::tempdir().unwrap();
let src = source("sync-downstream-summary");
let summary = DownstreamSummary {
boundary: "Fetch".into(),
repositories: vec!["a".into()],
first_import: "admin".into(),
..Default::default()
};
let plan = SourcePlan {
source: src.clone(),
pairs: vec![(
src.ron_repositories[0].clone(),
Arc::new(MemBackend::default()) as Arc<dyn RepositoryBackendTrait>,
)],
state_dir: dir.path().to_path_buf(),
pinned: None,
report_as: None,
summary: Some(summary.clone()),
summary_as: None,
console: None,
};
let req = holger_server_lib::mirror_cycle::SyncPoints::default().request(&plan);
assert_eq!(req.summary, Some(summary));
assert_eq!(req.points.len(), 1);
}