server/lib/src/mirror_cycle.rs
//! **The mirror's DECLARATION and its cycle** — the `mirror:` block of the RON
//! config, and the caller [`crate::mirror`] was written for.
//!
//! # Why this module exists
//!
//! [`crate::mirror`] is the engine: a persisted per-`(upstream, repository)`
//! change map, a parallel merge-join diff, and a three-way verdict. It was
//! merged as a **library with no configuration and no caller** — nothing in a
//! holger binary ever built one. This module is the missing half:
//!
//! * a **RON declaration** (`mirror:`) that says *this holger mirrors that
//! upstream, these repositories, this often, and this is what it does about a
//! coordinate it cannot prove*, and
//! * a **cycle** that turns one declared pair into one pass of
//! list → diff → fetch → put → adopt → store.
//!
//! It is deliberately NOT a second mirror engine. Every verdict is
//! [`ChangeMap::diff`]'s; every fetch decision is [`MirrorDiff::to_fetch`]'s
//! under the declared [`UnprovenPolicy`]. What is added here is configuration,
//! a transport, and a loop.
//!
//! # The block
//!
//! ```ron
//! mirror: Some((
//! ron_state_dir: "holger-mirror",
//! ron_sources: [
//! (
//! ron_name: "holger-rs",
//! ron_kind: Holger,
//! ron_endpoint: "http://holger.rs:50051",
//! ron_credential: Anonymous,
//! ron_every_secs: 300,
//! ron_repositories: [
//! ( ron_from: "crates", ron_to: "crates" ),
//! ( ron_from: "bundles", ron_to: "bundles", ron_unproven: TrustSize ),
//! ],
//! ),
//! ],
//! )),
//! ```
//!
//! An **on-prem mirror against a real cloud holger** writes the same block with
//! a credential and a TLS endpoint — this is the shape an operator deploys:
//!
//! ```ron
//! mirror: Some((
//! ron_state_dir: "holger-mirror",
//! ron_sources: [
//! (
//! ron_name: "cloud",
//! ron_kind: Holger,
//! ron_endpoint: "https://holger.example.com:50051",
//! ron_credential: Mtls(
//! ron_client_cert: "/etc/holger/pki/mirror.crt",
//! ron_client_key: "/etc/holger/pki/mirror.key",
//! ron_ca_cert: Some("/etc/holger/pki/estate-ca.crt"),
//! ),
//! ron_every_secs: 300,
//! ron_repositories: [
//! ( ron_from: "crates", ron_to: "crates" ),
//! ( ron_from: "bundles", ron_to: "bundles" ),
//! ],
//! ),
//! ],
//! )),
//! ```
//!
//! The CN of `mirror.crt` is the principal the upstream authorizes. On the
//! UPSTREAM side that is one line in the role map — and, because `repo_roles`
//! outranks `roles`, it can be one line per repository.
//!
//! A mirror that only wants the upstream's OPEN repositories still has to speak
//! TLS to ask for them, and says so with
//! `ron_credential: Tls(ron_ca_cert: Some("…"))` — verify the server, present
//! nothing. `Anonymous` is cleartext, and holger refuses the combination of
//! `Anonymous` with an `https://` endpoint (and of TLS material with an
//! `http://` one) at boot rather than letting the two halves of one statement
//! disagree.
//!
//! **Absent ⇒ no mirror**, which is exactly where every holger written before
//! this key existed is: no upstream is listed, nothing is fetched, nothing is
//! written, and not one byte of prior behaviour changes. The key is
//! `#[serde(default)]` so those files keep parsing; what they do NOT do is
//! quietly acquire a mirror.
//!
//! # Resilience — what a cycle does when it fails, and how anyone finds out
//!
//! A mirror lives on somebody else's network, so a failing cycle is not an
//! exception: it is Tuesday, and the loop is written for it.
//!
//! * **A failed cycle never ends the timer.** There is no `return` in the loop;
//! the only way a source's timer stops is the process stopping.
//! * **A failed cycle is retried sooner than the period, never later.** The
//! delay starts at `ron_retry_secs` (5 s), doubles per consecutive failure,
//! and is clamped at `ron_every_secs`. An operator who wrote
//! `ron_every_secs: 300` is promised the upstream is asked at least that
//! often; backoff may only beat that promise, never break it. Writing
//! `ron_retry_secs: 300` restores one-attempt-per-period exactly.
//! * **A connect is bounded** by `ron_connect_timeout_secs` (20 s, clamped to
//! the period). `tonic`'s connect has no timeout, and a connect that never
//! returns is the one way this loop could genuinely stop without saying so.
//! * **Every failure says whether retrying can help** — see [`Transience`]. The
//! upstream half is transient (`warn!`, retried); a change map that cannot be
//! read or written is permanent (`error!`, repeated). A transient failure
//! that has outlived one declared period escalates to `error!`, because "the
//! upstream restarted" and "this mirror has not mirrored since Tuesday" are
//! not the same fact.
//! * **A mirror that is not mirroring is visible from outside the process.** A
//! completed cycle logs at `info`; this module is therefore named in
//! [`crate::DEFAULT_LOG_FILTER`] so those lines survive the `warn` floor a
//! deployment that never configured logging runs at — otherwise silence in a
//! journal means both "fine" and "stopped". `/metrics` publishes
//! `holger_mirror_last_success_timestamp_seconds{source=…}` (0 ⇒ never), the
//! consecutive-failure gauge and the cycle/artifact/byte counters, per
//! declared source and from boot. `time() - last_success > 2 * ron_every_secs`
//! is the alert; nothing systemd can see says it.
//!
//! # `ron_unproven` — the field the engine's third verdict forced
//!
//! The engine classifies a coordinate present on both sides three ways, and the
//! reasoning is [`crate::mirror`]'s, restated here only because the config key
//! is meaningless without it:
//!
//! 1. digests on **both** sides — equal ⇒ `unchanged`, differ ⇒ `changed`. The
//! only proof of sameness there is.
//! 2. sizes on both sides that **differ** ⇒ `changed`. A size difference proves
//! a change with no digest anywhere.
//! 3. **everything else is `unproven`** — never `unchanged`. Equal size with no
//! digest does not exclude a same-size edit (one byte flipped in a jar, a
//! rebuilt wheel), and calling that `unchanged` is how a mirror serves stale
//! bytes forever.
//!
//! So the declaration has to be able to say what happens to bucket 3, and
//! [`RonUnprovenPolicy`] is that and nothing else:
//!
//! * **`Refetch`** — the DEFAULT, and the default because it is the safe one.
//! Re-fetch the unproven coordinate. Bandwidth is cheaper than a stale mirror.
//! A pair that says nothing gets this.
//! * **`TrustSize`** — opt-in, per pair: "this upstream publishes immutable
//! coordinates, a matching size is good enough". It buys bandwidth and pays
//! for it by missing a same-size edit, and an operator has to write the word.
//!
//! Under **either** value the `unproven` bucket is still counted and still
//! reported ([`PairReport::unproven`]): the policy changes what is fetched, never
//! what is admitted. An operator can always see the size of the doubt.
//!
//! A value that is neither word is a **boot error** — RON refuses an unknown
//! enum variant — rather than a silent fall back to something weaker.
//!
//! # What a holger upstream can and cannot prove — MEASURED, not assumed
//!
//! **This section used to say that a holger → holger mirror can never reach
//! `unchanged`. That was true and it is not true any more.**
//!
//! `RepositoryService.ListArtifacts` answers `ArtifactEntry`, and that message
//! now carries an optional `sha256` beside `id`, `size_bytes` and
//! `content_type`. [`holger_listing`] carries it into [`RemoteAsset::sha256`],
//! so a coordinate the upstream could prove reaches verdict 1 above — equal
//! digests ⇒ `unchanged` — with no policy written at all, exactly like a Nexus
//! or Artifactory source.
//!
//! What has NOT changed is that the answer depends on which backend serves the
//! upstream repository, because the two sides of that are genuinely different
//! costs (see [`traits::ArtifactEntry`]):
//!
//! | upstream backend | digest in its listing | verdict for a size-unchanged coordinate |
//! |---|---|---|
//! | `file-oci` | yes — the manifest is content-addressed | `unchanged` |
//! | `file-rust` | when the checksum is already memoized | `unchanged`, else `unproven` |
//! | `raw`, every `znippy-*` | no — answering would mean hashing the store | `unproven` |
//!
//! So `ron_unproven` is still per-PAIR and still needed, and it is still
//! `Refetch` by default. What it no longer has to paper over is holger's own
//! listing being poorer than the registries it mirrors. A pair whose upstream
//! can prove sameness now proves it; a pair whose upstream cannot still leaves
//! the operator the real choice — pay the bandwidth (`Refetch`) or accept the
//! same-size blind spot (`TrustSize`).
//!
//! An upstream built **before** these fields existed sends neither of them, and
//! proto3 explicit presence means that arrives as `None` = unknown rather than
//! as an empty digest. An old upstream therefore behaves exactly as it did
//! before this change: `unproven`, and the policy decides.
//!
//! # Credentials — and why an on-prem mirror needed one
//!
//! [`MirrorCredential::Anonymous`] is the default, and it is the only thing that
//! ever ran. It works against `holger.rs` for exactly one reason: that server
//! serves reads OPEN. A customer mirroring from a real cloud holger does not get
//! that, and an anonymous mirror against a gated upstream is not a degraded
//! mirror — it is no mirror at all.
//!
//! [`MirrorCredential::Mtls`] is the answer, and it is the whole answer: the
//! upstream already accepts a client certificate (`AuthMethodConfig::Mtls`, the
//! leaf CN becomes the principal), already scopes that principal per repository
//! (`repo_roles` outranks `roles` in both directions), and the client this
//! module already goes through already knows how to present one
//! ([`crate::object::RemoteHolger::connect_with_tls`]). Nothing was built here.
//! The three were wired together.
//!
//! ## The ruling: FILE, not vault — and what the other way would have cost
//!
//! The identity could have come from `nornir-vault`, the way
//! `agent/lib/src/creds.rs` takes its connector passwords. It does not, and the
//! reasoning is a difference in kind between the two secrets, plus a measured
//! dependency cost:
//!
//! * **A client certificate is not a password.** The certificate half is public
//! by construction — it is what gets handed to the server on every handshake,
//! and it is what an operator pastes into a ticket when asking for the CN to
//! be granted a role. The private key beside it is protected the way every
//! other private key on this box is protected: by the filesystem. That is the
//! SAME protection the vault's own file relies on (`nornir-vault` writes 0600
//! and then trusts the OS). Putting a key inside a 0600 vault file that sits
//! next to a 0600 key file does not raise the floor; it adds a lock to a door
//! in the same wall.
//! * **`holger-server` links no vault, and the airgap tiers are why.** The
//! `embedded` and `mini` tiers deliberately link nothing network-ish;
//! `nornir-vault` sits behind the agent's `online` feature. Pulling it into
//! `holger-server` to read two PEM files would put a new crate (and its
//! closure) into every tier that ships — including the ones whose entire
//! selling point is that they do not have one.
//! * **The seam the vault variant exists for is still guarded.** A password IS
//! the kind of secret a vault is for, which is why `agent`'s `creds.rs`
//! refuses a literal one by name. Here there is no literal secret to refuse:
//! the RON names PATHS, and a path is not a credential.
//!
//! What the vault would have bought: one place to rotate, and an operator who
//! already runs a vault for connector passwords keeping one habit instead of
//! two. That is real, and it is why [`MirrorCredential::Vault`] keeps its
//! spelling and keeps **refusing at boot by name** rather than quietly becoming
//! a second-class option that silently downgrades to anonymous. When
//! `holger-server` grows a tier that links the vault, that arm stops being a
//! refusal and starts being a loader — and no RON written today has to change.
//!
//! ## Never anonymous by accident
//!
//! A source that DECLARES a credential and cannot load it **refuses the boot**,
//! named ([`MirrorSource::load_credential`], called from [`MirrorSettings::validate`],
//! which `wire_holger` runs before anything is served). Missing file, unreadable
//! file, a file that is not the kind of PEM it was declared as — all three stop
//! the server. They do not become a warning and a cycle that reads open: a
//! mirror whose security posture nobody can see is worse than one that will not
//! start.
//!
//! ## No key material ever leaves this module
//!
//! The RON carries paths. The loaded bytes live in [`MirrorClientIdentity`],
//! which has a HAND-WRITTEN `Debug` that prints the two paths and the word
//! `<redacted>` — because a derived `Debug` printing a private key into a log
//! line is not hypothetical: `ConnectorAuth` derived one and printed passwords
//! and bearer tokens in full. Errors name paths, never contents.
//! `no_key_material_in_any_rendering_of_a_loaded_identity` is the canary.
use std::path::{Path, PathBuf};
use std::sync::Arc;
use std::time::Duration;
use anyhow::{bail, Context, Result};
use rustls::pki_types::pem::PemObject as _;
use serde::{Deserialize, Serialize};
use traits::{ArtifactId, RemoteAsset, RepositoryBackendTrait};
use crate::mirror::{ChangeMap, MirrorDiff, UnprovenPolicy};
/// Default directory for the change maps, relative to the data root.
fn default_state_dir() -> String {
"holger-mirror".to_string()
}
/// Default seconds between cycles: five minutes. Long enough that a mirror is
/// not a load generator, short enough that "it appears on the next cycle" means
/// minutes and not a shift.
fn default_every_secs() -> u64 {
300
}
/// Default listing page size. `ListArtifacts` pages; this bounds one message.
fn default_page_size() -> u32 {
1000
}
/// How many fetches are in flight at once. Network concurrency, not CPU
/// parallelism — the diff is where the cores go, and that is `gatling`'s.
fn default_in_flight() -> usize {
8
}
/// Default seconds before the FIRST retry after a failed cycle, doubling to a
/// ceiling of [`MirrorSource::ron_every_secs`].
///
/// Five seconds, and the number was measured, not chosen. On `oden` the on-prem
/// mirror boots one second after the network interface and the first cycle's
/// connect fails; with a 300 s period and no retry the next attempt was 300 s
/// later, so a reboot cost five minutes of not mirroring **every time** while
/// systemd reported the unit active. Five seconds turns that into five seconds.
///
/// It can never make the mirror *slower* than declared and it can never make it
/// faster than declared on the SUCCESS path: the retry floor only applies after
/// a failure, and it is clamped to the period, so `ron_every_secs: 5` with this
/// default retries at 5 s, not at 5 s twice.
fn default_retry_secs() -> u64 {
5
}
/// Default seconds a connect may take before it is abandoned as failed.
///
/// Twenty. `tonic`'s `Endpoint::connect` has no timeout of its own: against a
/// black-holed address (a firewall that drops SYN rather than refusing) it waits
/// on the kernel's TCP retry ladder, and against a TLS-terminating middlebox it
/// can wait indefinitely. A connect with no bound is the one thing in this loop
/// that could genuinely wedge the timer forever — every other stage returns.
fn default_connect_timeout_secs() -> u64 {
20
}
// ───────────────────────────── the declaration ─────────────────────────────
/// The `mirror:` block. Unknown keys are refused for the reason every other
/// block in this config refuses them: a key holger does not understand is a key
/// holger is not honouring while the file says it is.
#[derive(Serialize, Deserialize, Clone, Debug, PartialEq, Eq)]
#[serde(deny_unknown_fields)]
pub struct MirrorSettings {
/// Where the per-`(upstream, repository)` change maps are persisted. A
/// relative path anchors under the data root, like every other directory a
/// holger config names.
///
/// This directory IS the thing that makes a restart cheap: it is what
/// [`ChangeMap::load`] reads, and a mirror whose state dir was wiped
/// re-fetches everything (correctly — it has no evidence it holds anything).
#[serde(default = "default_state_dir")]
pub ron_state_dir: String,
/// The upstreams this holger mirrors. Empty ⇒ the block is declared and no
/// cycle runs, which is a legal thing to write and is not the same as
/// omitting the block: it says an operator meant to have none.
pub ron_sources: Vec<MirrorSource>,
}
/// One upstream, and the repositories mirrored from it.
#[derive(Serialize, Deserialize, Clone, Debug, PartialEq, Eq)]
#[serde(deny_unknown_fields)]
pub struct MirrorSource {
/// The upstream's name. It is the change map's namespace — [`ChangeMap`]
/// files are keyed `(upstream, repository)` — so RENAMING it orphans the
/// maps and the next cycle re-fetches everything. Pick it once.
pub ron_name: String,
/// What the upstream speaks. See [`MirrorSourceKind`].
pub ron_kind: MirrorSourceKind,
/// The upstream's endpoint, e.g. `http://holger.rs:50051`.
pub ron_endpoint: String,
/// How this holger authenticates to it. Default [`MirrorCredential::Anonymous`].
#[serde(default)]
pub ron_credential: MirrorCredential,
/// Seconds between cycles when the server runs the mirror on its own timer.
/// `holger-server mirror --once` ignores it.
#[serde(default = "default_every_secs")]
pub ron_every_secs: u64,
/// Listing page size for one `ListArtifacts` call.
#[serde(default = "default_page_size")]
pub ron_page_size: u32,
/// How many artifact fetches run concurrently within one pair.
#[serde(default = "default_in_flight")]
pub ron_in_flight: usize,
/// Seconds before the first retry after a FAILED cycle, doubling each
/// consecutive failure and **clamped at [`Self::ron_every_secs`]**.
///
/// The clamp is the promise: an operator who wrote `ron_every_secs: 300` is
/// told the upstream is asked at least every 300 s, and no amount of backoff
/// may turn that into a four-hour gap. Backoff here only ever makes a
/// recovery FASTER than the declared period, never a gap longer than it.
///
/// Setting it equal to `ron_every_secs` restores the behaviour this key was
/// added to change: one attempt per period, failure or not.
#[serde(default = "default_retry_secs")]
pub ron_retry_secs: u64,
/// Seconds a connect to this upstream may take before it counts as failed.
/// Bounds the one stage of a cycle that has no bound of its own; see
/// [`default_connect_timeout_secs`].
#[serde(default = "default_connect_timeout_secs")]
pub ron_connect_timeout_secs: u64,
/// The (upstream repo → local repo) pairs.
pub ron_repositories: Vec<MirrorPair>,
}
/// What the upstream speaks.
///
/// An unknown word here is a RON parse error, not a fallback: a `ron_kind` this
/// binary cannot drive must stop the boot, because the alternative is a holger
/// that reports a configured mirror and runs none.
#[derive(Serialize, Deserialize, Clone, Copy, Debug, PartialEq, Eq)]
pub enum MirrorSourceKind {
/// Another holger, over gRPC (`RepositoryService`). The wired kind.
Holger,
/// Sonatype Nexus. Its connector lives in the AGENT (`agent/connector/nexus`,
/// behind the agent's `online` feature) and is reachable from
/// `holger-agent`'s one-shot `migrate`, not from `holger-server`. Declaring
/// it here is refused at boot by name rather than ignored.
Nexus,
/// JFrog Artifactory. Same story as [`MirrorSourceKind::Nexus`].
Artifactory,
}
/// Where the upstream credential comes from. **No variant carries a secret**:
/// every one of them names a PATH, and a path is not a credential. The vault
/// variant additionally carries an ENTRY NAME, exactly the shape
/// `agent/lib/src/creds.rs` defines, so a config written for one is readable by
/// the other.
///
/// An unknown variant, or an unknown key inside one, is a RON parse error. That
/// is the point of `deny_unknown_fields` here: `ron_client_key` misspelled must
/// stop the boot, not leave a mirror that reads open because the field it needed
/// was silently absent.
#[derive(Serialize, Deserialize, Clone, Debug, PartialEq, Eq, Default)]
#[serde(deny_unknown_fields)]
pub enum MirrorCredential {
/// No credential and **no TLS** — a cleartext gRPC channel. The default, and
/// what a mirror of a public `http://` holger needs.
///
/// Declaring it against an `https://` endpoint is a **boot refusal**, not a
/// connect failure five minutes later: the two spellings disagree about
/// whether this mirror is talking over TLS, and only one of them can be
/// right.
#[default]
Anonymous,
/// **TLS, with no client identity.** The upstream's certificate is verified;
/// nothing is presented in return.
///
/// This is the certless half of the per-repository posture: a cloud holger
/// that serves `bundles` open and `crates` behind a certificate is still an
/// HTTPS server, and a mirror that only wants `bundles` still has to speak
/// TLS to ask for it. Before this variant existed the only non-mTLS spelling
/// was cleartext, so "open repository over a real cloud endpoint" had no
/// spelling at all.
Tls {
/// PEM path to the CA that signed the UPSTREAM's server certificate.
/// Omitted ⇒ the platform / webpki roots.
#[serde(default)]
ron_ca_cert: Option<String>,
/// **The name the upstream's certificate is checked against** (lane OX-WGDS3), and the
/// calls' `:authority`, when `ron_endpoint` is an ADDRESS: a downstream reaching its
/// estate's front at `https://10.0.2.2:<port>` names `repo.<estate>`, the name the
/// front's certificate and its repository route carry. Omitted ⇒ the endpoint's host.
#[serde(default)]
ron_server_name: Option<String>,
/// **The calls' `:authority`, when it is not [`ron_server_name`]** (lane OX-WGDS3): a front
/// whose certificate is `<estate>`'s routes `repo.<estate>` by `:authority`, so a downstream
/// dialling it at an address names the certificate's name and the route's apart.
/// Omitted ⇒ `ron_server_name` (else the endpoint's host).
#[serde(default)]
ron_authority: Option<String>,
},
/// **A client certificate — mTLS.** The wired credentialled path, and the
/// one an on-prem mirror against a real cloud holger uses.
///
/// The upstream turns the leaf certificate's CN into the principal, and that
/// principal resolves through the ordinary role map — including
/// `repo_roles`, so ONE upstream can gate `crates` behind a certificate
/// while `bundles` stays open to anyone.
///
/// All three fields are filesystem paths. See the module docs for why the
/// key is read from a file rather than a vault.
Mtls {
/// PEM path to the CLIENT certificate this mirror presents. Its CN is
/// the subject the upstream's `roles` / `repo_roles` maps key on.
ron_client_cert: String,
/// PEM path to the private key for `ron_client_cert`. Protect it the way
/// the box protects every other private key: 0600, owned by the service
/// user. holger reads it and never renders it.
ron_client_key: String,
/// PEM path to the CA that signed the UPSTREAM's server certificate.
///
/// Omitted ⇒ the platform / webpki roots, which is right for a cloud
/// holger behind a public CA and wrong for a private PKI. An on-prem
/// mirror against an estate CA names it.
#[serde(default)]
ron_ca_cert: Option<String>,
},
/// A named entry in a `nornir-vault`. Parses here; refused at boot in
/// `holger-server`, which links no vault.
Vault {
/// Path to the vault file.
ron_vault_path: String,
/// The entry's name inside it.
ron_entry: String,
},
/// **This box's installed replication credential** (`nordisk-twin`, stinky
/// 14 §P3) — the directory `InstallTwinCertificate` wrote: the key born here,
/// the leaf the source's issuer signed, the PINNED issuer and the source's
/// door record. Read per connect attempt, so a certification (or a renewal)
/// installed while the timer runs is picked up on the next tick.
///
/// Installed ⇒ TLS 1.3 to the first recorded door that answers, the pinned
/// CA the ONLY root, `ron_endpoint` a label. Not installed ⇒ a LOOPBACK
/// `ron_endpoint` is dialled in the clear (the in-process tests' road) and a
/// remote one is refused, `holger-twin-13`, every tick — never followed in
/// the clear.
Twin {
/// The credential directory (`<data_root>/holger-twin-tls`).
ron_dir: String,
},
}
// ───────────────────────────── the LOADED credential ─────────────────────────
/// A client identity with its PEM bytes in memory: what
/// [`MirrorCredential::Mtls`] becomes once the declared files have been read and
/// proved to be the kind of PEM they were declared as.
///
/// **It does not derive `Debug`, and that is load-bearing.** `key_pem` is a
/// private key. A derived `Debug` would put it into any `{:?}` — a log line, an
/// `anyhow` chain, an audit record, a panic message. That exact class of leak
/// was found in `ConnectorAuth`, which derived `Debug` and printed passwords and
/// bearer tokens in full. The hand-written impl below prints the PATHS, which an
/// operator needs in order to fix a misconfiguration, and the word `<redacted>`
/// for everything else.
#[derive(Clone, PartialEq, Eq)]
pub struct MirrorClientIdentity {
/// The certificate PEM. Public by construction — it crosses the wire on
/// every handshake — but rendered as `<redacted>` anyway, so that no future
/// edit has to remember which half of a keypair was safe.
cert_pem: Vec<u8>,
/// The private key PEM. Never rendered, never logged, never in an error.
key_pem: Vec<u8>,
/// The upstream server's CA PEM, when one was named.
ca_pem: Option<Vec<u8>>,
/// Where `cert_pem` came from. Diagnostics only.
cert_path: String,
/// Where `key_pem` came from. Diagnostics only.
key_path: String,
/// Where `ca_pem` came from, when one was named. Diagnostics only.
ca_path: Option<String>,
}
impl std::fmt::Debug for MirrorClientIdentity {
/// Paths and a redaction marker. Never bytes. See the type's docs.
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.debug_struct("MirrorClientIdentity")
.field("cert_path", &self.cert_path)
.field("key_path", &self.key_path)
.field("ca_path", &self.ca_path)
.field("cert_pem", &"<redacted>")
.field("key_pem", &"<redacted>")
.field("ca_pem", &self.ca_pem.as_ref().map(|_| "<redacted>"))
.finish()
}
}
impl MirrorClientIdentity {
/// The `(cert_pem, key_pem)` pair [`crate::object::RemoteHolger::connect_with_tls`]
/// takes. The ONLY accessor that hands out key bytes, and it hands them
/// straight to the TLS stack.
pub fn client_identity(&self) -> (Vec<u8>, Vec<u8>) {
(self.cert_pem.clone(), self.key_pem.clone())
}
/// The upstream server's CA PEM, when one was declared.
pub fn ca_pem(&self) -> Option<Vec<u8>> {
self.ca_pem.clone()
}
/// The certificate's path, for a diagnostic. Safe to render.
pub fn cert_path(&self) -> &str {
&self.cert_path
}
}
/// A [`MirrorCredential`] that has been RESOLVED — the files read, the PEMs
/// proved, the refusals already taken. Produced by
/// [`MirrorSource::load_credential`], at boot by [`MirrorSettings::validate`]
/// and again per connect attempt by [`HolgerConnector::connect`].
///
/// Resolving twice is deliberate. The boot pass is the REFUSAL: a declared
/// credential that will not load stops the server, in front of the operator,
/// before anything is served. The per-attempt pass is what a long-running timer
/// needs: a rotated certificate is picked up without a restart, and a rotation
/// that broke the files becomes a `Transience::Permanent` error the timer names
/// on every tick. Neither pass has an arm that yields
/// [`LoadedCredential::Anonymous`] from a declaration that asked for a
/// certificate.
///
/// `Debug` is derived and that is safe: the only variant carrying bytes carries
/// [`MirrorClientIdentity`], whose `Debug` redacts them.
#[derive(Clone, Debug, PartialEq, Eq, Default)]
pub enum LoadedCredential {
/// Cleartext, nothing presented.
#[default]
Anonymous,
/// TLS with the server verified (against `ca_pem`, or the platform roots
/// when `None`) and NO client identity.
Tls {
ca_pem: Option<Vec<u8>>,
server_name: Option<String>,
authority: Option<String>,
},
/// A loaded client certificate + key.
Mtls(MirrorClientIdentity),
/// This box's replication credential, `None` before it is certified —
/// see [`MirrorCredential::Twin`].
Twin {
cred: Option<crate::twin_tls::Credential>,
dir: String,
},
}
impl LoadedCredential {
/// `true` when this mirror will present a client certificate.
pub fn is_mtls(&self) -> bool {
matches!(self, LoadedCredential::Mtls(_))
}
/// `true` when the channel will be TLS at all.
pub fn is_tls(&self) -> bool {
!matches!(self, LoadedCredential::Anonymous)
}
}
/// Read one declared PEM file, or refuse NAMING THE PATH AND NOTHING ELSE.
///
/// The error deliberately carries the OS error and the path and never one byte
/// of what was (or was not) read: a "this file is not a certificate" message
/// that quotes the file is a message that prints a private key when the operator
/// has swapped the two paths — which is the single most likely way to get here.
fn read_pem(path: &str, role: &str, source: &str) -> Result<Vec<u8>> {
std::fs::read(path).map_err(|e| {
anyhow::anyhow!(
"mirror source '{source}': ron_credential Mtls names {role} \"{path}\", which could \
not be read ({e}). This REFUSES the boot rather than mirroring anonymously: a \
source that declares a credential and runs without one is a mirror whose security \
posture nobody can see. Fix the path or its permissions, or declare Anonymous if \
the upstream really does serve reads open."
)
})
}
impl MirrorSource {
/// Read and prove one declared CA PEM — the UPSTREAM's server-certificate
/// root, shared by the `Tls` and `Mtls` arms.
fn load_ca(&self, path: &str) -> Result<Vec<u8>> {
let bytes = read_pem(path, "ron_ca_cert", &self.ron_name)?;
let ok = rustls::pki_types::CertificateDer::pem_slice_iter(&bytes)
.filter_map(|c| c.ok())
.next()
.is_some();
if !ok {
bail!(
"mirror source '{}': ron_ca_cert \"{path}\" holds no PEM CERTIFICATE. It must \
be the CA that signed the UPSTREAM's server certificate — omit the key \
entirely to use the platform roots.",
self.ron_name
);
}
Ok(bytes)
}
/// **The endpoint's scheme and the credential must agree**, and a
/// disagreement is a boot refusal.
///
/// The two are one statement written twice, and each half is useless without
/// the other:
///
/// * `Anonymous` against `https://` opens a CLEARTEXT channel to a TLS port.
/// tonic fails the connect, five minutes into a timer, with a transport
/// error that says nothing about the config. Refusing at boot turns that
/// into one sentence an operator can act on.
/// * `Tls` / `Mtls` against `http://` is worse, because it does NOT fail: a
/// cleartext channel is opened and the certificate is simply never
/// presented. A config that reads as authenticated, serving as open — the
/// exact shape every refusal in this module exists to prevent.
fn refuse_scheme_mismatch(&self) -> Result<()> {
let ep = self.ron_endpoint.trim();
let https = ep.starts_with("https://");
let http = ep.starts_with("http://");
match &self.ron_credential {
MirrorCredential::Anonymous if https => bail!(
"mirror source '{}': ron_endpoint \"{ep}\" is https, but ron_credential is \
Anonymous, which opens a CLEARTEXT channel and cannot complete a TLS \
handshake. Declare Tls(ron_ca_cert: …) for a server this mirror only needs \
to verify, or Mtls(…) for one that demands a client certificate.",
self.ron_name
),
MirrorCredential::Tls { .. } | MirrorCredential::Mtls { .. } if http => bail!(
"mirror source '{}': ron_credential declares TLS material but ron_endpoint \
\"{ep}\" is http. A cleartext channel presents no certificate at all, so this \
config READS as authenticated and would RUN as open. Use an https:// endpoint, \
or declare Anonymous and mean it.",
self.ron_name
),
_ => Ok(()),
}
}
/// Resolve [`Self::ron_credential`] into something the transport can use, or
/// REFUSE BY NAME.
///
/// Called from [`MirrorSettings::validate`], which `wire_holger` runs at
/// boot — so every refusal below happens before a single byte is served, not
/// on the first cycle. The four ways a declared credential fails are all
/// here and all fatal:
///
/// * the endpoint's SCHEME contradicts the credential —
/// [`Self::refuse_scheme_mismatch`],
/// * the file is **missing or unreadable** — [`read_pem`],
/// * the file is **the wrong kind of PEM** (a key where a certificate was
/// declared, or the reverse — the classic swapped-paths mistake, which
/// otherwise surfaces as an opaque TLS handshake failure at run time),
/// * the credential is a [`MirrorCredential::Vault`], which this binary
/// links no vault for.
///
/// Never falls back to [`LoadedCredential::Anonymous`]. The only way to get
/// that is to declare it.
pub fn load_credential(&self) -> Result<LoadedCredential> {
self.refuse_scheme_mismatch()?;
match &self.ron_credential {
MirrorCredential::Anonymous => Ok(LoadedCredential::Anonymous),
MirrorCredential::Tls {
ron_ca_cert,
ron_server_name,
ron_authority,
} => Ok(LoadedCredential::Tls {
ca_pem: match ron_ca_cert {
None => None,
Some(p) => Some(self.load_ca(p)?),
},
server_name: ron_server_name.clone().filter(|n| !n.trim().is_empty()),
authority: ron_authority.clone().filter(|n| !n.trim().is_empty()),
}),
MirrorCredential::Mtls {
ron_client_cert,
ron_client_key,
ron_ca_cert,
} => {
let cert_pem = read_pem(ron_client_cert, "ron_client_cert", &self.ron_name)?;
let key_pem = read_pem(ron_client_key, "ron_client_key", &self.ron_name)?;
// KIND checks. rustls' `PemObject` is already in this crate's tree
// (`exposed::tls` loads the server's own material with it), so
// this costs no dependency. The parse result is thrown away —
// only the ANSWER is kept, so nothing that was parsed can reach
// an error string.
let has_cert = rustls::pki_types::CertificateDer::pem_slice_iter(&cert_pem)
.filter_map(|c| c.ok())
.next()
.is_some();
if !has_cert {
bail!(
"mirror source '{}': ron_client_cert \"{ron_client_cert}\" holds no PEM \
CERTIFICATE. If that path is the private key, the two are swapped — \
ron_client_cert is the public half the upstream reads the CN from.",
self.ron_name
);
}
let has_key = rustls::pki_types::PrivateKeyDer::from_pem_slice(&key_pem).is_ok();
if !has_key {
bail!(
"mirror source '{}': ron_client_key \"{ron_client_key}\" holds no PEM \
PRIVATE KEY. If that path is the certificate, the two are swapped.",
self.ron_name
);
}
// HOL-NCSC (NCSC P3): the client key is the process's alone (after its kind,
// so a swapped pair is said as one).
crate::keyfile::check_private(
std::path::Path::new(ron_client_key),
&format!("mirror source '{}': ron_client_key", self.ron_name),
)?;
let ca_pem = match ron_ca_cert {
None => None,
Some(p) => Some(self.load_ca(p)?),
};
Ok(LoadedCredential::Mtls(MirrorClientIdentity {
cert_pem,
key_pem,
ca_pem,
cert_path: ron_client_cert.clone(),
key_path: ron_client_key.clone(),
ca_path: ron_ca_cert.clone(),
}))
}
MirrorCredential::Twin { ron_dir } => Ok(LoadedCredential::Twin {
cred: crate::twin_tls::load_dir(std::path::Path::new(ron_dir))
.map_err(|e| anyhow::anyhow!("mirror source '{}': {e}", self.ron_name))?,
dir: ron_dir.clone(),
}),
MirrorCredential::Vault { ron_entry, .. } => bail!(
"mirror source '{}': ron_credential Vault(ron_entry: \"{ron_entry}\") names a \
vault, and holger-server links none (`nornir-vault` sits behind the agent's \
`online` feature). This is REFUSED rather than run anonymously: a config that \
reads as authenticated must not serve as open. Use Mtls, which needs no vault \
and is the wired credentialled path, or Anonymous for an upstream that serves \
reads open.",
self.ron_name
),
}
}
}
/// One mirrored repository.
#[derive(Serialize, Deserialize, Clone, Debug, PartialEq, Eq)]
#[serde(deny_unknown_fields)]
pub struct MirrorPair {
/// The repository's name ON THE UPSTREAM.
pub ron_from: String,
/// The LOCAL repository the bytes land in. It must be declared in this same
/// config and must be writable, or the boot is refused naming both.
pub ron_to: String,
/// What to do with a coordinate whose sameness cannot be proved. See the
/// module docs; the default re-fetches.
#[serde(default)]
pub ron_unproven: RonUnprovenPolicy,
}
/// The config spelling of [`UnprovenPolicy`]. One word, two values, and it is a
/// separate type from the engine's only so that the RON vocabulary is owned by
/// the config and cannot drift into the engine.
#[derive(Serialize, Deserialize, Clone, Copy, Debug, PartialEq, Eq, Default)]
pub enum RonUnprovenPolicy {
/// Re-fetch it. **The default.**
#[default]
Refetch,
/// Treat "same size, no digest" as unchanged and skip it. Opt-in; accepts a
/// same-size edit will be missed.
TrustSize,
}
impl From<RonUnprovenPolicy> for UnprovenPolicy {
fn from(p: RonUnprovenPolicy) -> Self {
match p {
RonUnprovenPolicy::Refetch => UnprovenPolicy::Refetch,
RonUnprovenPolicy::TrustSize => UnprovenPolicy::TrustSize,
}
}
}
impl MirrorSettings {
/// Every check that can be made without touching the network, made BEFORE
/// anything runs.
///
/// `local_repos` is `(name, writable)` for every repository this config
/// declares — [`crate::Holger`] hands it in so this stays testable without a
/// wired server.
///
/// Refusals, all by name:
/// * a `ron_kind` this binary cannot drive,
/// * a `ron_credential` this binary cannot read — a missing, unreadable or
/// wrong-kind PEM, or a vault entry (this binary links no vault). The
/// files are opened HERE so a broken credential stops the boot instead of
/// turning into a cycle that reads anonymously,
/// * a `ron_to` that names no declared repository,
/// * a `ron_to` that names a repository that is not writable,
/// * a duplicate `(source, from)` pair — two declarations sharing one change
/// map would each adopt over the other's state,
/// * `ron_every_secs: 0` when a timer would use it.
pub fn validate(&self, local_repos: &[(String, bool)]) -> Result<()> {
let mut seen: Vec<(String, String)> = Vec::new();
for src in &self.ron_sources {
match src.ron_kind {
MirrorSourceKind::Holger => {}
other => bail!(
"mirror source '{}': ron_kind {other:?} is not wired into holger-server. \
Only `Holger` is. The Nexus and Artifactory connectors live behind the \
agent's `online` feature and are driven by `holger-agent migrate`, a \
one-shot bulk pull — not by this continuous mirror. Declare `Holger`, or \
remove the source.",
src.ron_name
),
}
// The credential is RESOLVED here, at boot, not at the first cycle.
// A missing / unreadable / wrong-kind PEM, and a vault this binary
// cannot open, both stop the server — see
// [`MirrorSource::load_credential`]. The loaded value is dropped:
// this is the refusal, `plan` does the keeping.
src.load_credential()?;
if src.ron_every_secs == 0 {
bail!(
"mirror source '{}': ron_every_secs is 0. A zero period is a busy loop \
against the upstream, not a mirror; omit the key for the {}s default.",
src.ron_name,
default_every_secs()
);
}
if src.ron_retry_secs == 0 {
bail!(
"mirror source '{}': ron_retry_secs is 0. A zero first-retry is a busy loop \
against an upstream that is already down — the one thing a backoff exists \
to prevent. Omit the key for the {}s default, or name a real number; it is \
clamped to ron_every_secs either way.",
src.ron_name,
default_retry_secs()
);
}
if src.ron_connect_timeout_secs == 0 {
bail!(
"mirror source '{}': ron_connect_timeout_secs is 0 — every connect would be \
abandoned before it started and the source could never mirror. Omit the key \
for the {}s default.",
src.ron_name,
default_connect_timeout_secs()
);
}
if src.ron_repositories.is_empty() {
bail!(
"mirror source '{}' declares no ron_repositories — it would connect and \
mirror nothing. Name at least one pair, or remove the source.",
src.ron_name
);
}
for pair in &src.ron_repositories {
let key = (src.ron_name.clone(), pair.ron_from.clone());
if seen.contains(&key) {
bail!(
"mirror: ({}, {}) is declared twice. Two pairs sharing one upstream \
repository share ONE change map, and each cycle would adopt over the \
other's state — so the second would re-fetch what the first just took.",
src.ron_name,
pair.ron_from
);
}
seen.push(key);
match local_repos.iter().find(|(n, _)| *n == pair.ron_to) {
None => bail!(
"mirror source '{}': ron_to: \"{}\" names no repository in this config. \
The mirror writes into a LOCAL repository; declare it in `repositories`.",
src.ron_name,
pair.ron_to
),
Some((_, false)) => bail!(
"mirror source '{}': local repository '{}' is not writable, so the mirror \
could list and diff and then fail every single put. Give it a \
`ron_data_dir` (or a writable archive) and an `ron_out`.",
src.ron_name,
pair.ron_to
),
Some((_, true)) => {}
}
}
}
Ok(())
}
/// The state dir, anchored under `data_root` when it is relative.
pub fn state_dir(&self, data_root: Option<&Path>) -> PathBuf {
let p = Path::new(&self.ron_state_dir);
match data_root {
Some(root) if p.is_relative() => root.join(p),
_ => p.to_path_buf(),
}
}
/// The state dir AND, when it cannot be trusted to be the server's, the
/// reason — for the caller to log before it does anything else.
///
/// # The defect this exists for
///
/// `ron_state_dir: "holger-mirror"` is relative, and a relative path with no
/// `data_root` resolves against the **current working directory**. The
/// server always has one (`HOLGER_DATA`, which the installer sets and the
/// unit carries), so the service reads and writes
/// `$HOLGER_DATA/holger-mirror`. A hand-run `holger-server mirror --once`
/// from a shell usually does NOT, so it silently used
/// `$PWD/holger-mirror` — a different, empty directory. It found an empty
/// change map, reported `map 0 -> 1`, and re-fetched an artifact the mirror
/// already held.
///
/// Nothing was lost and nothing was corrupt; the operator was simply looking
/// somewhere else and the tool did not say so. On a metered link that is a
/// bill. So now it says so.
pub fn state_dir_resolved(&self, data_root: Option<&Path>) -> (PathBuf, Option<String>) {
let dir = self.state_dir(data_root);
let unanchored = Path::new(&self.ron_state_dir).is_relative() && data_root.is_none();
let warning = unanchored.then(|| {
format!(
"mirror: ron_state_dir \"{}\" is relative and no data root is set (HOLGER_DATA is \
unset), so the change maps resolve against the CURRENT DIRECTORY — {}. The \
server resolves them under its data root, so this run may not see what the \
service persisted and may re-fetch what it already holds. Set HOLGER_DATA to \
the server's data root, or make ron_state_dir absolute.",
self.ron_state_dir,
dir.display()
)
});
(dir, warning)
}
}
impl MirrorSource {
/// [`Self::ron_every_secs`] as a `Duration`.
pub fn interval(&self) -> Duration {
Duration::from_secs(self.ron_every_secs.max(1))
}
/// The FIRST retry delay after a failed cycle — [`Self::ron_retry_secs`],
/// never zero (that is a busy loop against someone else's network) and never
/// longer than [`Self::interval`] (that is a gap the operator did not
/// declare).
pub fn retry_floor(&self) -> Duration {
Duration::from_secs(self.ron_retry_secs.max(1)).min(self.interval())
}
/// How long a connect may take before it is abandoned. Clamped to the period
/// for the same reason [`Self::retry_floor`] is: a connect that outlives the
/// cadence has already cost the cycle it was meant to serve.
pub fn connect_timeout(&self) -> Duration {
Duration::from_secs(self.ron_connect_timeout_secs.max(1)).min(self.interval())
}
/// The next delay after `previous` failed: double it, and stop at the
/// declared period. The whole backoff policy, in one place, so the timer and
/// its test cannot hold different opinions about it.
pub fn next_backoff(&self, previous: Duration) -> Duration {
(previous * 2).min(self.interval()).max(self.retry_floor())
}
}
// ───────────────────────────── transient vs permanent ─────────────────────────────
/// Whether a failed cycle can be expected to fix itself.
///
/// This is attached to the error as `anyhow` context at the point the failure
/// happens — the STAGE knows, and nothing downstream has to guess from a string.
/// [`transience_of`] reads it back out of the chain.
///
/// The distinction is not cosmetic: it decides what the log says and how loud it
/// is. A [`Transience::Transient`] failure is the condition a mirror exists to
/// survive (the upstream is restarting, the link is down, the box just booted
/// and the network is one second behind), so it is a `warn!` and the timer keeps
/// its cadence. A [`Transience::Permanent`] failure will still be there on the
/// next tick and every tick after it — a change-map file that cannot be read or
/// written, a coordinate that is not a coordinate — so it is an `error!` and it
/// is repeated rather than said once and forgotten.
///
/// Most permanent *declaration* errors never reach a cycle at all:
/// [`MirrorSettings::validate`] refuses an unwired `ron_kind`, a vault
/// credential this binary cannot read, a `ron_to` that names no repository and a
/// `ron_to` that is not writable, at BOOT. What is left here is the small set a
/// running process can still walk into.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum Transience {
/// Retrying is the right answer. Network, upstream, link.
Transient,
/// Retrying will do the same thing. Local state, or a declaration a running
/// process could not be refused at boot.
Permanent,
}
impl std::fmt::Display for Transience {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.write_str(match self {
Transience::Transient => "transient",
Transience::Permanent => "permanent",
})
}
}
/// Read a cycle failure's [`Transience`] back out of its `anyhow` chain.
///
/// An error carrying no marker is **transient**, deliberately: an unclassified
/// failure is one nobody has reasoned about yet, and the cost of retrying
/// something that was really permanent is one log line per period, while the
/// cost of NOT retrying something that was really transient is the defect this
/// module was rewritten to close.
pub fn transience_of(err: &anyhow::Error) -> Transience {
err.downcast_ref::<Transience>()
.copied()
.unwrap_or(Transience::Transient)
}
// ───────────────────────────── coordinates ─────────────────────────────
/// The change map's key for one artifact: `"<namespace>/<name>@<version>"`, with
/// an empty namespace when there is none.
///
/// It is a STRING because the change map is a sorted column of strings an
/// operator can open in any Arrow tool and grep. The encoding is parsed back by
/// [`coord_to_id`]: split at the FIRST `/` and the LAST `@`, which is why an
/// npm scope (`@scope/pkg`) and a Maven group (`org.example`) both survive the
/// round trip.
pub fn id_to_coord(id: &ArtifactId) -> String {
format!(
"{}/{}@{}",
id.namespace.as_deref().unwrap_or(""),
id.name,
id.version
)
}
/// The inverse of [`id_to_coord`]. `None` when the string is not a coordinate —
/// which for a map on disk means the file was written by something else, and is
/// reported rather than guessed at.
pub fn coord_to_id(coord: &str) -> Option<ArtifactId> {
let (ns, rest) = coord.split_once('/')?;
let (name, version) = rest.rsplit_once('@')?;
Some(ArtifactId {
namespace: (!ns.is_empty()).then(|| ns.to_string()),
name: name.to_string(),
version: version.to_string(),
})
}
/// Turn one page of a holger listing into [`RemoteAsset`]s.
///
/// `sha256` is carried through from the entry when the upstream backend could
/// prove one, and stays `None` when it could not. It is **not** an omission
/// either way, and the two states must not be collapsed:
///
/// * `Some(hex)` — the upstream published a digest; a coordinate whose digest
/// matches the change map reaches `unchanged` with no policy at all.
/// * `None` — the upstream backend does not know (a znippy-backed repository
/// cannot answer without reading and hashing every artifact; see
/// [`traits::ArtifactEntry`]). The coordinate lands in `unproven` and
/// `ron_unproven` decides what happens to it.
///
/// The digest is normalized on the way in — 64 hex characters, lowercased, or
/// nothing. A malformed value is DROPPED rather than carried, because the diff
/// compares against whatever is here: a wrong digest would mean either a
/// permanent re-fetch or, worse, an `unchanged` verdict about bytes that moved.
/// This is the same rule the connector path applies to the digests it parses
/// out of a Nexus or Artifactory listing.
///
/// `change_token` is carried too, and it is the field that makes a mirror of a
/// **znippy-backed** repository settle at all: 15 of holger's 17 backends are
/// `znippy-*`, none of them can answer `sha256`, and before this every one of
/// them landed in `unproven` on every tick and re-fetched its whole repository
/// every cycle under the default policy.
///
/// The token is SCOPE-STAMPED here, with `"{endpoint}/{repository}"` — the
/// identity of the source this listing was asked of. That is what makes
/// "a token is never compared across sources" a property of the code: the
/// change map stores the scoped value, and
/// [`crate::mirror::change_tokens_comparable`] refuses a comparison whose two scopes
/// differ. It has to happen HERE rather than in the backend, because the
/// backend knows which repository it is and not which server the asker
/// reached it as.
pub fn holger_listing(
endpoint: &str,
repository: &str,
entries: Vec<traits::ArtifactEntry>,
) -> Vec<RemoteAsset> {
// One scope string for the whole page, not one per row.
let scope = format!("{endpoint}/{repository}");
entries
.into_iter()
.map(|e| {
let coord = id_to_coord(&e.id);
RemoteAsset {
download_url: format!("{endpoint}/{repository}/{coord}"),
path: coord,
content_type: (!e.content_type.is_empty()).then(|| e.content_type.clone()),
size: (e.size_bytes >= 0).then_some(e.size_bytes as u64),
// Whatever the upstream backend could prove, and nothing when
// it could not. Normalized here, so a malformed digest never
// becomes something the diff trusts.
sha256: e.sha256.as_deref().and_then(traits::normalize_sha256),
// Same discipline, and the scope on top. A token the upstream
// did not publish stays absent — never stamped, because a
// scope with nothing behind it is not a token.
change_token: e
.change_token
.as_deref()
.and_then(|t| crate::mirror::scope_change_token(&scope, t)),
}
})
.collect()
}
// ───────────────────────────── the report ─────────────────────────────
/// What one pair's cycle did. Counted, not timed — the same choice
/// [`crate::mirror::DiffStats`] makes, for the same reason.
#[derive(Debug, Clone, Default, Serialize, Deserialize, PartialEq, Eq)]
pub struct PairReport {
/// The upstream's declared name.
pub source: String,
/// The repository on the upstream.
pub from: String,
/// The local repository the bytes landed in.
pub to: String,
/// Coordinates the upstream listed this cycle.
pub listed: usize,
/// Entries the change map held BEFORE this cycle. `0` on a first run.
pub map_before: usize,
/// Coordinates the map had never seen.
pub added: usize,
/// Coordinates PROVED different.
pub changed: usize,
/// Coordinates PROVED identical.
pub unchanged: usize,
/// Coordinates whose sameness could not be proved. Reported under EVERY
/// policy — the policy decides the fetch, never the admission.
pub unproven: usize,
/// Map entries the upstream no longer lists. The mirror does not delete
/// them locally; it drops them from the map, so a re-appearance is `added`.
pub removed: usize,
/// What [`MirrorDiff::to_fetch`] asked for under the declared policy.
pub to_fetch: usize,
/// What was actually fetched and written.
pub fetched: usize,
/// Bytes written.
pub bytes: u64,
/// Fetches that failed. Their previous map entry is carried over verbatim,
/// so the next cycle asks for them again.
pub failed: usize,
/// Entries the change map holds AFTER this cycle.
pub map_after: usize,
/// The coordinates behind `failed` whose fetch or put failed — what a
/// timer with a job queue hands on as `Job::MirrorAsset`, one each.
pub failed_coords: Vec<String>,
/// Coordinates not fetched this cycle because a `Job::MirrorAsset` for
/// them is already open (counted in `failed` too: the map keeps them).
pub in_queue: usize,
/// The policy this pair ran under.
pub unproven_policy: RonUnprovenPolicy,
/// What this cycle WROTE, oldest first, at most [`WRITTEN_KEPT`] (the
/// newest): the coordinate, the lowercase-hex sha256 of the bytes written,
/// their length — a twin's evidence of what it holds, which its primary
/// keeps (`holger.grow.v1.HeldArtifact`, the functest's read of the twin).
#[serde(default, skip_serializing_if = "Vec::is_empty")]
pub written: Vec<(String, String, u64)>,
}
/// How many written artifacts one [`PairReport`] names.
pub const WRITTEN_KEPT: usize = 256;
impl PairReport {
/// One line, all the numbers, for a log or a terminal.
pub fn line(&self) -> String {
format!(
"mirror {}:{} -> {} [{:?}]: listed {} (map {} -> {}) | added {} changed {} \
unchanged {} unproven {} removed {} | to_fetch {} fetched {} failed {} bytes {}",
self.source,
self.from,
self.to,
self.unproven_policy,
self.listed,
self.map_before,
self.map_after,
self.added,
self.changed,
self.unchanged,
self.unproven,
self.removed,
self.to_fetch,
self.fetched,
self.failed,
self.bytes,
)
}
/// `true` when this cycle moved no bytes at all.
pub fn moved_nothing(&self) -> bool {
self.fetched == 0 && self.bytes == 0
}
}
// ───────────────────────────── the cycle ─────────────────────────────
/// What one cycle needs that this module will not go and find for itself: the
/// upstream transport and the local backend. Handed in so the cycle is testable
/// against an in-memory pair with no network and no server.
#[async_trait::async_trait]
pub trait MirrorUpstream: Send + Sync {
/// Every coordinate the upstream repository holds, paged to exhaustion.
async fn list(&self, repository: &str, page_size: u32) -> Result<Vec<RemoteAsset>>;
/// One artifact's bytes. `None` ⇒ the upstream no longer has it.
///
/// `Bytes`, not `Vec<u8>`, and on this path it is not cosmetic: the holger
/// transport already HAS a `Bytes` (`HolgerObject::fetch` returns one), and
/// a `Vec` here made it pay a full `.to_vec()` copy of every artifact a
/// downstream pulls. `Bytes::from(Vec<u8>)` takes the allocation, so a fake
/// upstream that owns a `Vec` still wraps for free -- the copy only ever
/// went one way. See `.nornir/vec-to-bytes-migration.md`.
async fn fetch(&self, repository: &str, id: &ArtifactId) -> Result<Option<traits::Bytes>>;
/// **Tell a PAIRED upstream how far this mirror has synced**
/// (`holger.grow.v1.DownstreamReport`, see [`crate::peers`]) and learn its
/// position `(boot, head)` now. `Ok(None)`: this transport has no such door
/// (a fake, a Nexus) — nothing is said and nothing is learnt. `proof`: the
/// signed proof a SYNC downstream's report carries (`x-holger-sync-report`,
/// lane OX-WGDS2), its name for an upstream it is not paired with.
async fn report_sync(
&self,
_req: crate::grpc::holger_grow_proto::DownstreamReportRequest,
_proof: Option<String>,
) -> Result<Option<(u64, u64)>> {
Ok(None)
}
}
/// **What a paired mirror has applied, per upstream repository** — the twin's
/// [`crate::grow::FollowPosition`], one per pair: the upstream's `(boot, head)`
/// at the START of the last cycle of that repository that landed whole, and
/// when that cycle ended (Unix ms). Kept by the source's timer.
#[derive(Debug, Default, Clone)]
pub struct SyncPoints {
points: std::collections::BTreeMap<String, (u64, u64, i64)>,
}
impl SyncPoints {
/// A cycle of `repository` that started at `at` landed whole at `now_ms`.
pub fn completed(&mut self, repository: &str, at: (u64, u64), now_ms: i64) {
self.points
.insert(repository.to_string(), (at.0, at.1, now_ms));
}
/// The report for every pair of `plan` (a pair never completed says 0).
pub fn request(
&self,
plan: &SourcePlan,
) -> crate::grpc::holger_grow_proto::DownstreamReportRequest {
use crate::grpc::holger_grow_proto as pb;
pb::DownstreamReportRequest {
points: plan
.pairs
.iter()
.map(|(pair, _)| {
let (boot, seq, done) =
self.points.get(&pair.ron_from).copied().unwrap_or_default();
pb::DownstreamPoint {
repository: pair.ron_from.clone(),
applied_boot: boot,
applied_seq: seq,
last_complete_unix_ms: done,
}
})
.collect(),
every_secs: u32::try_from(plan.source.ron_every_secs).unwrap_or(u32::MAX),
summary: plan.summary.clone(),
summary_proof: String::new(),
console_proof: String::new(),
}
}
}
/// Run ONE cycle for one pair: list, diff, fetch what the policy asks for, write
/// it into the local backend, adopt the listing into the map, persist the map.
///
/// The order is the load-bearing part. The map is adopted **after** the fetches
/// and the failures are named, so a coordinate the mirror could not take keeps
/// its previous entry and is asked for again next cycle. A map written before
/// the bytes landed would claim artifacts this holger does not hold.
#[allow(clippy::too_many_arguments)]
pub async fn run_pair(
upstream: &dyn MirrorUpstream,
source_name: &str,
pair: &MirrorPair,
local: &Arc<dyn RepositoryBackendTrait>,
state_dir: &Path,
page_size: u32,
in_flight: usize,
now_nanos: i64,
) -> Result<PairReport> {
run_pair_skipping(
upstream,
source_name,
pair,
local,
state_dir,
page_size,
in_flight,
now_nanos,
&|_| false,
)
.await
}
/// [`run_pair`], leaving out every coordinate `skip` names — the ones a
/// `Job::MirrorAsset` already owns, so the cycle and the job do not fetch the
/// same bytes twice. A skipped coordinate counts as `failed` (the map keeps its
/// previous entry and asks again next cycle) and in `in_queue`.
#[allow(clippy::too_many_arguments)]
pub async fn run_pair_skipping(
upstream: &dyn MirrorUpstream,
source_name: &str,
pair: &MirrorPair,
local: &Arc<dyn RepositoryBackendTrait>,
state_dir: &Path,
page_size: u32,
in_flight: usize,
now_nanos: i64,
skip: &(dyn Fn(&str) -> bool + Sync),
) -> Result<PairReport> {
// The upstream end of the cycle: every way this fails is a way the network
// or the other holger failed, and every one of them is over by the next
// tick or is not, but retrying is always the right answer.
let fresh = upstream
.list(&pair.ron_from, page_size)
.await
.map_err(|e| e.context(Transience::Transient))
.with_context(|| format!("mirror {source_name}:{}: listing", pair.ron_from))?;
// The LOCAL end: a change map that cannot be read is a file on this disk.
// The next tick will find the same file in the same state, so this is said
// at `error!` and said again every cycle until somebody fixes it.
let mut map = ChangeMap::load(state_dir, source_name, &pair.ron_from)
.map_err(|e| e.context(Transience::Permanent))
.with_context(|| format!("mirror {source_name}:{}: load change map", pair.ron_from))?;
let map_before = map.len();
let diff: MirrorDiff = map.diff(&fresh, 0);
let policy: UnprovenPolicy = pair.ron_unproven.into();
let want = diff.to_fetch(policy);
let mut report = PairReport {
source: source_name.to_string(),
from: pair.ron_from.clone(),
to: pair.ron_to.clone(),
listed: fresh.len(),
map_before,
added: diff.added.len(),
changed: diff.changed.len(),
unchanged: diff.unchanged.len(),
unproven: diff.unproven.len(),
removed: diff.removed.len(),
to_fetch: want.len(),
unproven_policy: pair.ron_unproven,
..Default::default()
};
let mut failed: Vec<u32> = Vec::new();
let cap = in_flight.max(1);
// Network concurrency only. The CPU work in a cycle is the diff, and that
// is already fanned out over gatling inside the engine; rayon appears
// nowhere in this path.
let mut i = 0usize;
while i < want.len() {
let hi = (i + cap).min(want.len());
let mut batch = Vec::with_capacity(hi - i);
for &fi in &want[i..hi] {
let asset = &fresh[fi as usize];
if skip(&asset.path) {
failed.push(fi);
report.in_queue += 1;
continue;
}
let Some(id) = coord_to_id(&asset.path) else {
log::warn!(
"mirror {source_name}:{}: coordinate '{}' is not a holger coordinate — skipped",
pair.ron_from,
asset.path
);
failed.push(fi);
continue;
};
batch.push((fi, id));
}
let mut pending = Vec::with_capacity(batch.len());
for (fi, id) in batch {
pending.push(async move {
let got = upstream.fetch(&pair.ron_from, &id).await;
(fi, id, got)
});
}
for out in futures_util::future::join_all(pending).await {
let (fi, id, got) = out;
match got {
Ok(Some(bytes)) => match local.put(&id, &bytes) {
Ok(()) => {
report.fetched += 1;
report.bytes += bytes.len() as u64;
if report.written.len() == WRITTEN_KEPT {
report.written.remove(0);
}
report.written.push((
id_to_coord(&id),
nornir_hash::sha256_hex(&bytes),
bytes.len() as u64,
));
}
Err(e) => {
log::warn!(
"mirror {source_name}:{} -> {}: put {} failed: {e}",
pair.ron_from,
pair.ron_to,
id_to_coord(&id)
);
failed.push(fi);
report.failed_coords.push(id_to_coord(&id));
}
},
Ok(None) => {
log::warn!(
"mirror {source_name}:{}: upstream listed {} and then did not have it",
pair.ron_from,
id_to_coord(&id)
);
failed.push(fi);
report.failed_coords.push(id_to_coord(&id));
}
Err(e) => {
log::warn!(
"mirror {source_name}:{}: fetch {} failed: {e}",
pair.ron_from,
id_to_coord(&id)
);
failed.push(fi);
report.failed_coords.push(id_to_coord(&id));
}
}
}
i = hi;
}
report.failed = failed.len();
map.adopt(&fresh, &diff, &failed, now_nanos);
map.store(state_dir)
.map_err(|e| e.context(Transience::Permanent))
.with_context(|| format!("mirror {source_name}:{}: persist change map", pair.ron_from))?;
report.map_after = map.len();
Ok(report)
}
/// **Backfill one pair** (stinky SB #35): forget every change-map entry whose
/// bytes `local` does not hold any more — a twin whose store lost an artifact
/// (a disk replaced under it, a restore from an older copy, a delete by hand)
/// would otherwise read the map's "unchanged" for ever and never fetch it
/// again. The next [`run_pair`] fetches what was forgotten. A backend that
/// cannot list is left alone (nothing proves what it holds). Answers how many
/// entries were forgotten.
pub fn backfill_pair(
source_name: &str,
pair: &MirrorPair,
local: &Arc<dyn RepositoryBackendTrait>,
state_dir: &Path,
) -> Result<usize> {
let mut map = ChangeMap::load(state_dir, source_name, &pair.ron_from)
.with_context(|| format!("backfill {source_name}:{}: load change map", pair.ron_from))?;
if map.is_empty() || !local.caps().listing {
return Ok(0);
}
let held: std::collections::HashSet<String> = local
.list(None, usize::MAX)
.with_context(|| {
format!(
"backfill {source_name}:{}: list `{}`",
pair.ron_from, pair.ron_to
)
})?
.into_iter()
.map(|e| id_to_coord(&e.id))
.collect();
let forgotten =
map.retain(|c| coord_to_id(c).is_none_or(|id| held.contains(&id_to_coord(&id))));
if forgotten > 0 {
map.store(state_dir).with_context(|| {
format!(
"backfill {source_name}:{}: persist change map",
pair.ron_from
)
})?;
}
Ok(forgotten)
}
/// Nanoseconds since the Unix epoch, for [`run_pair`]'s `now_nanos`.
pub fn now_nanos() -> i64 {
std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.map(|d| d.as_nanos() as i64)
.unwrap_or(0)
}
// ───────────────────────────── the holger transport ─────────────────────────────
/// A [`MirrorUpstream`] that speaks to another holger over gRPC, through the
/// `RemoteHolger` client this crate already has. No new client, no new crate.
#[cfg(feature = "fetch")]
pub struct HolgerUpstream {
endpoint: String,
remote: crate::object::RemoteHolger,
}
#[cfg(feature = "fetch")]
impl HolgerUpstream {
/// Connect anonymously — no TLS material, no client certificate. What a
/// mirror of a public holger needs, and what every mirror did before
/// [`MirrorCredential::Mtls`] existed.
pub async fn connect(endpoint: &str) -> Result<Self> {
Self::connect_with(endpoint, &LoadedCredential::Anonymous).await
}
/// Connect **with the resolved credential**.
///
/// This is the seam the whole lane exists for. `RemoteHolger` already knew
/// how to present a client certificate
/// ([`crate::object::RemoteHolger::connect_with_tls`], which builds
/// `tls.identity(Identity::from_pem(cert, key))`); this module went through
/// `RemoteHolger` and never called it. Nothing in the client changed — the
/// declaration was connected to it.
///
/// * [`LoadedCredential::Anonymous`] ⇒ the plain channel, byte for byte the
/// prior behaviour.
/// * [`LoadedCredential::Mtls`] ⇒ TLS with our client identity, and the
/// declared CA pinned for the SERVER's certificate when one was named.
///
/// There is no third arm and no fallback: an `Mtls` credential that cannot
/// build a channel returns `Err`, which the caller reports. It never becomes
/// an anonymous connection.
pub async fn connect_with(endpoint: &str, cred: &LoadedCredential) -> Result<Self> {
let remote = match cred {
LoadedCredential::Anonymous => {
crate::object::RemoteHolger::connect(endpoint.to_string())
.await
.map_err(|e| e.context(Transience::Transient))
.with_context(|| format!("mirror: connect upstream {endpoint}"))?
}
LoadedCredential::Tls {
ca_pem,
server_name,
authority,
} => {
// TLS with the server verified and NOTHING presented. The
// certless half of a per-repository posture: the open
// repositories on an HTTPS upstream are still HTTPS.
crate::object::RemoteHolger::connect_with_tls_named(
endpoint.to_string(),
ca_pem.clone(),
None,
None,
server_name.clone(),
authority.clone(),
)
.await
.map_err(|e| e.context(Transience::Transient))
.with_context(|| format!("mirror: connect upstream {endpoint} over TLS"))?
}
LoadedCredential::Mtls(id) => {
let (cert, key) = id.client_identity();
crate::object::RemoteHolger::connect_with_tls(
endpoint.to_string(),
id.ca_pem(),
Some((cert, key)),
None,
)
.await
// TRANSIENT, like every other connect: a refused TLS handshake
// is most often the upstream restarting, and the one cause that
// is NOT transient — a credential that will never load — has
// already been refused at boot and again by
// `HolgerConnector::connect`, which marks it Permanent.
.map_err(|e| e.context(Transience::Transient))
// The context names the CERTIFICATE PATH, never the material —
// an operator debugging a refused handshake needs to know which
// identity was presented, and that is what a path says.
.with_context(|| {
format!(
"mirror: connect upstream {endpoint} with client certificate {}",
id.cert_path()
)
})?
}
// The twin's credential: mutual TLS 1.3 to the first door that
// answers — never a fallback to the clear once one is installed.
LoadedCredential::Twin { cred: Some(c), .. } => {
let (channel, _at) = crate::twin_tls::channel(c)
.await
.map_err(|e| anyhow::anyhow!(e).context(Transience::Transient))
.with_context(|| {
format!("mirror: connect the replication door for {endpoint}")
})?;
crate::object::RemoteHolger::from_channel(channel)
}
// Not certified yet: the clear only to a LOOPBACK source (tests).
// ... with the primary's admin-read bearer when one is named (lane P1: an owner-only
// repository is never read anonymously, so an uncertified loopback twin follows it
// only as the estate pull does, `twin_follow::BEARER_ENV`).
LoadedCredential::Twin { cred: None, .. } if crate::twin_tls::is_loopback(endpoint) => {
let bearer = std::env::var(crate::twin_follow::BEARER_ENV)
.ok()
.filter(|b| !b.trim().is_empty());
match bearer {
Some(b) => {
crate::object::RemoteHolger::connect_with_token(endpoint.to_string(), b)
.await
}
None => crate::object::RemoteHolger::connect(endpoint.to_string()).await,
}
.map_err(|e| e.context(Transience::Transient))
.with_context(|| format!("mirror: connect upstream {endpoint}"))?
}
LoadedCredential::Twin { cred: None, .. } => {
return Err(anyhow::anyhow!(crate::twin_tls::not_certified(endpoint))
.context(Transience::Transient));
}
};
Ok(Self {
endpoint: endpoint.to_string(),
remote,
})
}
/// Connect, but give up after `within` rather than waiting on the kernel.
///
/// `tonic`'s connect has no timeout, so against an address that drops
/// packets instead of refusing them this call is the only thing standing
/// between a mirror and a task that never returns. A timer that is parked
/// inside a connect looks exactly like a timer that is idle — `is-active:
/// active`, nothing in the journal — which is the shape this whole lane
/// exists to make impossible.
pub async fn connect_within(endpoint: &str, within: Duration) -> Result<Self> {
Self::connect_within_with(endpoint, within, &LoadedCredential::Anonymous).await
}
/// [`connect_within`](Self::connect_within) **presenting the resolved
/// credential** — the call the production connector makes.
///
/// The bound matters more here, not less: a TLS handshake against a host
/// that drops packets parks in exactly the same way a plain connect does,
/// and an mTLS mirror is the one that is talking to something across a
/// customer network rather than a loopback.
pub async fn connect_within_with(
endpoint: &str,
within: Duration,
cred: &LoadedCredential,
) -> Result<Self> {
match tokio::time::timeout(within, Self::connect_with(endpoint, cred)).await {
Ok(r) => r,
Err(_) => Err(anyhow::anyhow!(
"upstream {endpoint} did not answer within {}s — abandoned so the timer keeps \
its cadence (a connect with no bound is a mirror that stops without saying so)",
within.as_secs()
)
.context(Transience::Transient)),
}
}
}
// ───────────────────────────── the connector seam ─────────────────────────────
/// How the timer gets an upstream for a source.
///
/// It exists for one reason: without it, [`spawn_timers`] could only ever be
/// proved against a real gRPC server, and the property that most needed proving
/// — *a cycle whose first connect fails still mirrors on a later tick, with no
/// intervention* — could not be written as a test at all. With it, the proof is
/// a `cargo test`.
#[async_trait::async_trait]
pub trait UpstreamConnector: Send + Sync + 'static {
/// Open a transport for `src`, or fail. Failing is normal.
async fn connect(&self, src: &MirrorSource) -> Result<Box<dyn MirrorUpstream>>;
}
/// The production connector: gRPC to another holger, bounded by
/// [`MirrorSource::connect_timeout`].
/// **Connect to `src`'s upstream Holger** with the production connector — or, in a build
/// without `fetch` (the one-way downstream), refuse by name: there is no client to dial with.
pub async fn connect_upstream(src: &MirrorSource) -> Result<Box<dyn MirrorUpstream>> {
#[cfg(feature = "fetch")]
{
HolgerConnector.connect(src).await
}
#[cfg(not(feature = "fetch"))]
{
Err(anyhow::anyhow!(
"{}",
holger_errcode::BOUNDARY_FETCH_NOT_BUILT.tag(format!(
"source '{}': this build has no fetch code, so it dials no Holger",
src.ron_name
))
))
.context(Transience::Permanent)
}
}
#[cfg(feature = "fetch")]
pub struct HolgerConnector;
#[cfg(feature = "fetch")]
#[async_trait::async_trait]
impl UpstreamConnector for HolgerConnector {
async fn connect(&self, src: &MirrorSource) -> Result<Box<dyn MirrorUpstream>> {
match src.ron_kind {
MirrorSourceKind::Holger => {
// The credential is resolved HERE, per attempt, and a failure to
// resolve it is PERMANENT: a missing or wrong-kind PEM will not
// fix itself on the next tick, so the timer must say so every
// time rather than retrying in silence. Boot `validate` has
// already refused this same failure once — this is the second
// line, for the paths that never went through a wiring pass
// (`mirror --once`) and for a file that moved while the server
// was up. What it NEVER does is fall back to anonymous.
let cred = src
.load_credential()
.map_err(|e| e.context(Transience::Permanent))?;
let up = HolgerUpstream::connect_within_with(
&src.ron_endpoint,
src.connect_timeout(),
&cred,
)
.await?;
Ok(Box::new(up) as Box<dyn MirrorUpstream>)
}
// Refused at boot by `validate`; if one ever reaches here it is
// PERMANENT and the timer says so every tick rather than returning.
other => Err(anyhow::anyhow!(
"mirror source '{}': ron_kind {other:?} is not wired into holger-server, so this \
source can never mirror anything",
src.ron_name
)
.context(Transience::Permanent)),
}
}
}
#[cfg(feature = "fetch")]
#[async_trait::async_trait]
impl MirrorUpstream for HolgerUpstream {
async fn list(&self, repository: &str, page_size: u32) -> Result<Vec<RemoteAsset>> {
use traits::HolgerObject;
let mut out = Vec::new();
let mut token: Option<String> = None;
loop {
let (entries, next) = self
.remote
.list_artifacts(repository, None, page_size, token.clone())
.await?;
let empty = entries.is_empty();
out.extend(holger_listing(&self.endpoint, repository, entries));
if next.is_empty() || empty {
break;
}
token = Some(next);
}
Ok(out)
}
async fn fetch(&self, repository: &str, id: &ArtifactId) -> Result<Option<traits::Bytes>> {
use traits::HolgerObject;
// No `.to_vec()`. The client already handed us a `Bytes`; copying the
// whole body to satisfy a `Vec` signature was one full allocation per
// artifact, on the path every byte a downstream pulls goes through.
self.remote.fetch(repository, id).await
}
async fn report_sync(
&self,
req: crate::grpc::holger_grow_proto::DownstreamReportRequest,
proof: Option<String>,
) -> Result<Option<(u64, u64)>> {
use crate::grpc::holger_grow_proto::grow_service_client::GrowServiceClient;
let mut req = tonic::Request::new(req);
if let Some(v) =
proof.and_then(|p| tonic::metadata::MetadataValue::try_from(p.as_str()).ok())
{
req.metadata_mut().insert(REPORT_HEADER, v);
}
let reply = GrowServiceClient::new(self.remote.channel())
.downstream_report(req)
.await
.map_err(|s| anyhow::anyhow!("DownstreamReport to {}: {s}", self.endpoint))?
.into_inner();
Ok(Some((reply.boot, reply.head_seq)))
}
}
// ───────────────────────────── driving it from a config ─────────────────────────────
/// Run **one cycle over every declared pair**, in declaration order, and return
/// one [`PairReport`] each.
///
/// This is what `holger-server mirror --once` calls and what the timer loop
/// calls per tick. A pair that fails — the upstream is down, the listing
/// refused — is reported as an `Err` in its slot and does **not** stop the
/// others: an upstream restarting under a mirror is a Tuesday, not a defect, and
/// one unreachable source must not cost the rest of the estate its cycle.
///
/// `only` restricts to one source name; `only_repo` to one `ron_from`.
#[cfg(feature = "fetch")]
pub async fn run_once(
holger: &crate::Holger,
data_root: Option<&Path>,
only: Option<&str>,
only_repo: Option<&str>,
) -> Result<Vec<Result<PairReport>>> {
let Some(settings) = &holger.mirror else {
bail!(
"this config declares no `mirror:` block, so there is nothing to mirror. \
A holger with no mirror declaration takes nothing from anywhere — that is \
the default and it is not an error, but `mirror --once` was asked to run a \
cycle and has no pair to run."
);
};
let (state_dir, warning) = settings.state_dir_resolved(data_root);
if let Some(w) = warning {
log::warn!("{w}");
}
log::info!("mirror: change maps under {}", state_dir.display());
std::fs::create_dir_all(&state_dir)
.with_context(|| format!("mirror: create state dir {}", state_dir.display()))?;
let connector = HolgerConnector;
let mut out = Vec::new();
for src in &settings.ron_sources {
if only.is_some_and(|w| w != src.ron_name) {
continue;
}
// The connector resolves the credential — see `HolgerConnector::connect`.
// `run_once` is also reachable from `holger-server mirror --once`, a
// fresh process that has not been through a wiring pass, so the load (and
// its refusal) has to live below this call, not above it.
let upstream = match connector.connect(src).await {
Ok(u) => u,
Err(e) => {
out.push(Err(e));
continue;
}
};
for pair in &src.ron_repositories {
if only_repo.is_some_and(|w| w != pair.ron_from) {
continue;
}
let Some(local) = holger.backend_named(&pair.ron_to) else {
out.push(Err(anyhow::anyhow!(
"mirror {}:{}: local repository '{}' has no backend in this process",
src.ron_name,
pair.ron_from,
pair.ron_to
)));
continue;
};
out.push(
run_pair(
upstream.as_ref(),
&src.ron_name,
pair,
&local,
&state_dir,
src.ron_page_size,
src.ron_in_flight,
now_nanos(),
)
.await,
);
}
}
Ok(out)
}
/// One source's timer plan: everything a background task needs, with no
/// reference to [`crate::Holger`] left in it.
///
/// It exists because `Holger` is **not `Send`** — its wiring pass leaves raw
/// `*const Repository` pointers behind, so an `Arc<Holger>` cannot cross a task
/// boundary. Resolving each pair's backend to an `Arc<dyn RepositoryBackendTrait>`
/// (which is `Send + Sync`) up front is not a workaround for that; it is the
/// right shape anyway, because it makes every "this pair has no backend" failure
/// happen ONCE, at boot, in front of the operator, instead of every cycle inside
/// a log line nobody reads.
/// **Signs a mirror's sync report as the sync downstream this box is** (lane OX-WGDS2,
/// `crate::sync::reports`): the report's text in, the `x-holger-sync-report` value out
/// (`None`: nothing to sign with yet).
pub type ReportProof = Arc<
dyn Fn(&crate::grpc::holger_grow_proto::DownstreamReportRequest) -> Option<String>
+ Send
+ Sync,
>;
pub struct SourcePlan {
/// The source, verbatim from the config.
pub source: MirrorSource,
/// One `(pair, local backend)` per declared repository, in order.
pub pairs: Vec<(MirrorPair, Arc<dyn RepositoryBackendTrait>)>,
/// Where the change maps live.
pub state_dir: PathBuf,
/// The pairing with this source's upstream holger, when there is one
/// ([`crate::peers::UpstreamRecord`]). Each cycle first checks the upstream
/// presents the pinned key, and refuses (`holger-peer-3`) when it does not.
pub pinned: Option<crate::peers::UpstreamRecord>,
/// **This box is a SYNC downstream of this source** (lane OX-WGDS2): its reports carry
/// the proof this signs, so the upstream counts them by the downstream's enrolled
/// name, through the front's public door or its WireGuard sync door alike.
pub report_as: Option<ReportProof>,
/// **What this box is**, from its own RON (HOL-FLOW-SMART): every report to a holger
/// upstream carries it, so the upstream's flow can unfold this downstream. No secret.
pub summary: Option<crate::grpc::holger_grow_proto::DownstreamSummary>,
/// **The summary's own proof** (HOL-SYNC-SKEW): signs it into `summary_proof`, apart from
/// the report's proof, so the text `report_as` signs is the same in every holger version.
pub summary_as: Option<ReportProof>,
/// **This box's console address, and its proof** (HOL-FLOW-FRONT): read at each report
/// from what this box's front last reported, written into the summary and signed into
/// `console_proof`, so the upstream's flow can link to this box's own flow.
pub console: Option<(ConsoleOf, ReportProof)>,
}
/// **This box's console address now** (`crate::front_report::console_url`); empty when none.
pub type ConsoleOf = Arc<dyn Fn() -> String + Send + Sync>;
/// The gRPC metadata a sync downstream's report proof travels in
/// (`crate::sync::reports::REPORT_HEADER`; here so a build without `sync` names it too).
pub const REPORT_HEADER: &str = "x-holger-sync-report";
/// Resolve the declared mirror into per-source plans, or refuse by name.
///
/// Called at boot, before anything is spawned. `Ok(vec![])` when no `mirror:`
/// block is declared — the default, and not an error.
#[cfg(feature = "fetch")]
pub fn plan(holger: &crate::Holger, data_root: Option<&Path>) -> Result<Vec<SourcePlan>> {
let Some(settings) = &holger.mirror else {
return Ok(Vec::new());
};
let (state_dir, warning) = settings.state_dir_resolved(data_root);
if let Some(w) = warning {
log::warn!("{w}");
}
log::info!("mirror: change maps under {}", state_dir.display());
std::fs::create_dir_all(&state_dir)
.with_context(|| format!("mirror: create state dir {}", state_dir.display()))?;
// The pairings this box has recorded with its upstream holgers. Unreadable
// is a refusal: a pin that silently vanished is a mirror that stops
// checking who it pulls from.
let upstream_pins = crate::peers::load(&holger.peers_root())
.map_err(|e| anyhow::anyhow!(e))?
.1;
let mut plans = Vec::new();
for src in &settings.ron_sources {
let mut pairs = Vec::new();
for pair in &src.ron_repositories {
let backend = holger.backend_named(&pair.ron_to).with_context(|| {
format!(
"mirror source '{}': local repository '{}' has no backend in this process",
src.ron_name, pair.ron_to
)
})?;
pairs.push((pair.clone(), backend));
}
// Refuses by name — `?`, not a warning. A planned source whose declared
// credential will not load must not become a spawned timer; the loaded
// value is dropped, because `HolgerConnector::connect` resolves it again
// per attempt (so a rotated certificate is picked up without a restart,
// and a broken one is a PERMANENT error the timer names every tick).
let pinned = upstream_pins
.iter()
.find(|u| u.source == src.ron_name)
.cloned();
if pinned.is_some() && matches!(src.ron_credential, MirrorCredential::Anonymous) {
anyhow::bail!(
"mirror source '{}' is paired with an upstream holger, but its credential is \
Anonymous (cleartext): there is no key to check against the pin. Give it a \
`Tls` or `Mtls` credential.",
src.ron_name
);
}
src.load_credential()?;
// A box whose `sync:` block trusts an upstream reports to it as that downstream
// (lane OX-WGDS2): a holger source, the upstream of its name or the only one.
#[cfg(feature = "sync")]
let report_as = holger
.sync
.as_ref()
.filter(|c| {
src.ron_kind == MirrorSourceKind::Holger
&& (c.upstreams.iter().any(|u| u.name == src.ron_name)
|| c.upstreams.len() == 1)
})
.map(|_| crate::sync::reports::proof_for(&holger.data_root(), &src.ron_name));
// The summary's own proof, from the same key (HOL-SYNC-SKEW).
#[cfg(feature = "sync")]
let summary_as = report_as
.as_ref()
.map(|_| crate::sync::reports::summary_proof_for(&holger.data_root(), &src.ron_name));
// The console address and its proof (HOL-FLOW-FRONT), from the same key.
#[cfg(feature = "sync")]
let console = report_as.as_ref().map(|_| {
let root = holger.data_root();
let of: ConsoleOf = Arc::new(move || crate::front_report::console_url(&root));
(
of,
crate::sync::reports::console_proof_for(&holger.data_root(), &src.ron_name),
)
});
#[cfg(not(feature = "sync"))]
let (report_as, summary_as, console) = (None, None, None);
plans.push(SourcePlan {
source: src.clone(),
pairs,
state_dir: state_dir.clone(),
pinned,
report_as,
summary_as,
console,
// What this box is, for its upstream holger's flow (HOL-FLOW-SMART).
summary: (src.ron_kind == MirrorSourceKind::Holger)
.then(|| crate::flow::ron_summary(holger).into()),
});
}
Ok(plans)
}
/// One `tokio` task per planned source, each on its own `ron_every_secs` timer.
///
/// Returns the handles. Dropping them does not stop the tasks; the server holds
/// them for the life of the process, which is the whole point — a mirror that
/// stopped when its handle went out of scope would be a mirror that ran once.
///
/// A cycle that fails is LOGGED, COUNTED and RETRIED. It does not give up, and
/// — since the reboot that produced this rewrite — it does not wait the whole
/// declared period to try again either: a failed tick schedules the next one
/// [`MirrorSource::retry_floor`] away, doubling per consecutive failure and
/// clamped at [`MirrorSource::interval`]. Backoff here can only ever make the
/// recovery sooner than the operator declared, never the gap longer.
///
/// There is no `return` anywhere in the loop. The only way a source's timer
/// stops is the process stopping.
#[cfg(feature = "fetch")]
pub fn spawn_timers(plans: Vec<SourcePlan>) -> Vec<tokio::task::JoinHandle<()>> {
spawn_timers_with(plans, Arc::new(HolgerConnector))
}
/// [`spawn_timers`] with the transport handed in — the seam the resilience
/// proofs drive. Production calls [`spawn_timers`]; the tests hand in a
/// connector that refuses the first N connects.
#[cfg(feature = "fetch")]
pub fn spawn_timers_with(
plans: Vec<SourcePlan>,
connector: Arc<dyn UpstreamConnector>,
) -> Vec<tokio::task::JoinHandle<()>> {
spawn_timers_with_handoff(plans, connector, None)
}
/// [`spawn_timers`] with a place to hand failed assets to — the job queue
/// ([`Handoff`], `Job::MirrorAsset`). `None` is [`spawn_timers`] exactly.
#[cfg(feature = "fetch")]
pub fn spawn_timers_handing_off(
plans: Vec<SourcePlan>,
handoff: Option<Arc<Handoff>>,
) -> Vec<tokio::task::JoinHandle<()>> {
spawn_timers_with_handoff(plans, Arc::new(HolgerConnector), handoff)
}
/// Both seams at once (the tests' door).
#[cfg(feature = "fetch")]
pub fn spawn_timers_with_handoff(
plans: Vec<SourcePlan>,
connector: Arc<dyn UpstreamConnector>,
handoff: Option<Arc<Handoff>>,
) -> Vec<tokio::task::JoinHandle<()>> {
let mut handles = Vec::new();
for plan in plans {
let connector = connector.clone();
let handoff = handoff.clone();
// A32-14, stays a task: one timer per mirror plan; its handle is kept and aborted at the stop.
handles.push(tokio::spawn(async move {
run_timer(plan, connector, handoff).await;
}));
}
handles
}
/// One source's timer, as a plain future so a test can drive it under
/// `tokio::time` without a process.
///
/// **This function never returns.** Every failure path inside it logs, counts,
/// schedules the next attempt and loops; that is the whole fix. The previous
/// shape had one `return` (an unwired `ron_kind`) and an easy misreading that a
/// connect failure ended the timer, and a mirror whose timer has ended is
/// indistinguishable from a healthy one from the outside — `is-active: active`,
/// `NRestarts: 0`, and nothing in the journal, forever.
#[cfg(feature = "fetch")]
async fn run_timer(
plan: SourcePlan,
connector: Arc<dyn UpstreamConnector>,
handoff: Option<Arc<Handoff>>,
) {
let src = plan.source.clone();
let health = crate::metrics::global().mirror_source(&src.ron_name);
// The estate board's upstream row (lane A13): what the declaration says; the
// health above is its evidence.
crate::upstreams::register_mirror(
&src.ron_name,
&format!("{:?}", src.ron_kind),
&src.ron_endpoint,
&plan
.pairs
.iter()
.map(|(p, _)| p.ron_to.clone())
.collect::<Vec<_>>(),
src.ron_every_secs,
plan.pinned.as_ref().map(|p| p.name.as_str()).unwrap_or(""),
crate::clock::now_unix_ms(),
);
let period = src.interval();
let mut backoff = src.retry_floor();
// The first attempt is immediate, as it always was: a mirror that waited a
// period before its first cycle would mirror nothing across a short boot.
let mut next = tokio::time::Instant::now();
// What this mirror has applied of a PAIRED upstream, for its reports.
let mut synced = SyncPoints::default();
loop {
tokio::time::sleep_until(next).await;
let outcome = run_timer_once(
&src,
&plan,
connector.as_ref(),
&health,
handoff.as_deref(),
&mut synced,
)
.await;
match outcome {
// Every pair completed. Back to the declared cadence.
true => {
backoff = src.retry_floor();
next = tokio::time::Instant::now() + period;
}
// Something failed. Ask again sooner, then less often, never later
// than the period the operator wrote.
false => {
next = tokio::time::Instant::now() + backoff;
backoff = src.next_backoff(backoff);
}
}
}
}
/// One attempt for one source: connect, then every pair. `true` when the whole
/// attempt completed with no failed pair and no failed artifact.
#[cfg(feature = "fetch")]
async fn run_timer_once(
src: &MirrorSource,
plan: &SourcePlan,
connector: &dyn UpstreamConnector,
health: &crate::metrics::MirrorSourceHealth,
handoff: Option<&Handoff>,
synced: &mut SyncPoints,
) -> bool {
// ★ A PAIRED upstream must present the key pinned at pairing, before one
// byte is pulled from it. See `crate::peers`.
if let Some(pin) = &plan.pinned {
let identity = match src.load_credential() {
Ok(LoadedCredential::Mtls(id)) => Some(id.client_identity()),
_ => None,
};
let presented =
crate::peers::probe_upstream_key(&src.ron_endpoint, identity, src.connect_timeout())
.await;
let state = crate::peers::verdict(pin, presented.as_deref());
health.set_link_state(state);
if state == crate::peers::LinkState::KeyMismatch {
let e = anyhow::anyhow!(holger_errcode::PEER_KEY_MISMATCH.tag(format!(
"upstream `{}` at {} presented {} but was paired with {} — refusing to pull from it",
pin.name,
src.ron_endpoint,
crate::peers::short(presented.as_deref().unwrap_or("")),
crate::peers::short(&pin.fingerprint)
)))
.context(Transience::Transient);
health.record_failure();
report_failure(src, &format!("mirror {}", src.ron_name), &e, health);
return false;
}
}
let upstream = match connector.connect(src).await {
Ok(u) => u,
Err(e) => {
health.record_failure();
report_failure(src, &format!("mirror {}", src.ron_name), &e, health);
return false;
}
};
// ★ A PAIRED upstream hears where this mirror stands BEFORE the cycle and
// answers its position now — the twin's pattern: a pair that lands whole
// has applied the upstream's head as it was at this moment.
let at = if plan.pinned.is_some() || plan.report_as.is_some() {
report_sync(src, plan, upstream.as_ref(), synced).await
} else {
None
};
let mut moved_point = false;
let mut all_ok = true;
for (pair, local) in &plan.pairs {
// A coordinate a `Job::MirrorAsset` already owns is left to the job.
let skip = |coord: &str| {
handoff.is_some_and(|h| {
h.asset_is_open(&src.ron_name, &pair.ron_from, &pair.ron_to, coord)
})
};
match run_pair_skipping(
upstream.as_ref(),
&src.ron_name,
pair,
local,
&plan.state_dir,
src.ron_page_size,
src.ron_in_flight,
now_nanos(),
&skip,
)
.await
{
Ok(r) => {
// A cycle that took nothing is still a cycle that RAN, and the
// difference between the two is exactly what `/metrics` is for.
health.record_success(r.fetched as u64, r.bytes, r.failed as u64);
if let (Some(at), 0) = (at, r.failed) {
synced.completed(&pair.ron_from, at, now_nanos() / 1_000_000);
moved_point = true;
}
// ★ Each asset that did not land gets its OWN job — its own
// status, reason and retry — so the cycle moves on instead of
// backing the whole source off for one bad artifact. The
// per-cycle backoff is kept for what is the cycle's: a listing,
// a connect, a change map.
let mut handed = 0usize;
if let Some(h) = handoff {
for coord in &r.failed_coords {
match h.hand_off_asset(&src.ron_name, &pair.ron_from, &pair.ron_to, coord) {
Ok(job) => {
handed += 1;
log::info!(
"mirror {}:{}: {coord} did not land — handed to job {job}",
src.ron_name,
pair.ron_from
);
}
Err(e) => log::warn!(
"mirror {}:{}: could not queue {coord} ({e:#}); the next cycle asks again",
src.ron_name,
pair.ron_from
),
}
}
}
let owned_elsewhere = r.in_queue + handed;
if r.failed > owned_elsewhere {
all_ok = false;
log::warn!(
"mirror {}:{}: {} artifact(s) did not land and are asked for again next \
cycle — {}",
src.ron_name,
pair.ron_from,
r.failed,
r.line()
);
} else {
log::info!("{}", r.line());
}
if owned_elsewhere > 0 {
log::info!(
"mirror {}:{}: {} asset(s) owned by the job queue ({} handed on now, {} already queued)",
src.ron_name,
pair.ron_from,
owned_elsewhere,
handed,
r.in_queue
);
}
}
Err(e) => {
all_ok = false;
health.record_failure();
report_failure(
src,
&format!("mirror {}:{} cycle failed", src.ron_name, pair.ron_from),
&e,
health,
);
}
}
}
// ...and the point it reached, at once, so the upstream need not wait a
// whole cycle for the word.
if moved_point {
report_sync(src, plan, upstream.as_ref(), synced).await;
}
if all_ok {
// A mirror that was down and is not any more says so ONCE, at a level an
// operator sees by default. Success is otherwise `info`, and the default
// log floor is `warn` — which is precisely why 80 minutes of healthy
// cycles looked, from the journal, like 80 minutes of nothing.
if let Some(n) = health.take_recovery() {
log::warn!(
"mirror {}: RECOVERED — a cycle completed after {n} consecutive failure(s)",
src.ron_name
);
}
}
all_ok
}
/// One sync report to a paired upstream; the upstream's `(boot, head)`, or `None`
/// (said at `warn`) when it could not be told. A report that fails never fails
/// the cycle: the bytes are the mirror's job, the word is the operator's.
#[cfg(feature = "fetch")]
async fn report_sync(
src: &MirrorSource,
plan: &SourcePlan,
upstream: &dyn MirrorUpstream,
synced: &SyncPoints,
) -> Option<(u64, u64)> {
let mut req = synced.request(plan);
// This box's console address (HOL-FLOW-FRONT): in the summary, with its own proof.
if let (Some(s), Some((of, _))) = (req.summary.as_mut(), plan.console.as_ref()) {
s.console_url = of();
}
// The summary's proof first: it is no part of the text the report's proof signs.
if let Some(p) = plan.summary_as.as_ref().and_then(|sign| sign(&req)) {
req.summary_proof = p;
}
if let Some(p) = plan.console.as_ref().and_then(|(_, sign)| sign(&req)) {
req.console_proof = p;
}
let proof = plan.report_as.as_ref().and_then(|sign| sign(&req));
match upstream.report_sync(req, proof).await {
Ok(at) => at,
Err(e) => {
log::warn!(
"mirror {}: sync point not reported to the paired upstream: {e:#}",
src.ron_name
);
None
}
}
}
/// Say what failed, say whether retrying can help, and say it louder the longer
/// it has been going on.
///
/// * a PERMANENT failure is an `error!` every time it happens — it is not going
/// to stop by itself and one line at boot would be lost,
/// * a TRANSIENT failure is a `warn!`, and becomes an `error!` once the source
/// has been failing for longer than one declared period. "The upstream
/// restarted" and "this mirror has not mirrored since Tuesday" are not the
/// same fact and must not be the same line.
#[cfg(feature = "fetch")]
fn report_failure(
src: &MirrorSource,
what: &str,
e: &anyhow::Error,
health: &crate::metrics::MirrorSourceHealth,
) {
let kind = transience_of(e);
let stalled = health.stalled_for_secs().unwrap_or(0);
let period = src.ron_every_secs;
match kind {
Transience::Permanent => {
log::error!("{what}: PERMANENT — retrying will do the same thing: {e:#}")
}
Transience::Transient if stalled > period => log::error!(
"{what}: transient, but this source has not completed a cycle for {stalled}s \
({} consecutive failures) against a declared period of {period}s — it is NOT \
mirroring: {e:#}",
health.consecutive_failures()
),
Transience::Transient => log::warn!("{what}: transient, will retry: {e:#}"),
}
}
// ───────────────────────────── one asset, its own job ─────────────────────────────
/// **Where a cycle hands an asset that did not land** — the job queue in a
/// served process (`Job::MirrorAsset`), nothing in `mirror --once`.
#[cfg(feature = "jobs")]
pub type Handoff = crate::jobs::JobQueue;
/// Without the job queue there is nowhere to hand an asset: no value of this
/// type exists, so every `handoff` is `None`.
#[cfg(not(feature = "jobs"))]
pub enum Handoff {}
/// Only the timer cycle (`fetch`) hands assets off.
#[cfg(all(not(feature = "jobs"), feature = "fetch"))]
impl Handoff {
fn hand_off_asset(&self, _: &str, _: &str, _: &str, _: &str) -> Result<String> {
match *self {}
}
fn asset_is_open(&self, _: &str, _: &str, _: &str, _: &str) -> bool {
match *self {}
}
}
#[cfg(feature = "jobs")]
fn asset_job(source: &str, from_repo: &str, to_repo: &str, coord: &str) -> crate::jobs::Job {
crate::jobs::Job::MirrorAsset {
source: source.to_string(),
from_repo: from_repo.to_string(),
to_repo: to_repo.to_string(),
coord: coord.to_string(),
requested_by: format!("mirror:{source}"),
}
}
#[cfg(feature = "jobs")]
impl crate::jobs::JobQueue {
/// Queue the asset; the job id. Idempotent per coordinate.
pub fn hand_off_asset(
&self,
source: &str,
from_repo: &str,
to_repo: &str,
coord: &str,
) -> Result<String> {
Ok(self
.enqueue(asset_job(source, from_repo, to_repo, coord))?
.0)
}
/// Whether a job already owns this coordinate (the cycle then leaves it).
pub fn asset_is_open(&self, source: &str, from_repo: &str, to_repo: &str, coord: &str) -> bool {
self.open_job_for(&asset_job(source, from_repo, to_repo, coord).content_id())
.is_some()
}
}
/// **The mirror's transport for one asset**, as the job queue's runner reaches
/// it (`Job::MirrorAsset`): the declared source, connected through the
/// same [`HolgerConnector`] a cycle uses (same credential, same endpoint), on
/// the server's runtime.
#[cfg_attr(not(feature = "jobs"), allow(dead_code))]
pub struct MirrorJobs {
sources: Vec<MirrorSource>,
rt: tokio::runtime::Handle,
connector: Arc<dyn UpstreamConnector>,
}
impl MirrorJobs {
#[cfg(feature = "fetch")]
pub fn new(sources: Vec<MirrorSource>, rt: tokio::runtime::Handle) -> Self {
Self::with_connector(sources, rt, Arc::new(HolgerConnector))
}
/// The transport handed in (a test's seam).
pub fn with_connector(
sources: Vec<MirrorSource>,
rt: tokio::runtime::Handle,
connector: Arc<dyn UpstreamConnector>,
) -> Self {
Self {
sources,
rt,
connector,
}
}
}
#[cfg(feature = "jobs")]
impl MirrorJobs {
/// The bytes of `coord` in `from_repo` on the declared source `source`;
/// `Ok(None)` when the upstream no longer has it. Blocks: it runs on a
/// job-pool thread.
pub fn fetch_asset(
&self,
source: &str,
from_repo: &str,
coord: &str,
) -> Result<Option<traits::Bytes>> {
let src = self
.sources
.iter()
.find(|s| s.ron_name == source)
.ok_or_else(|| anyhow::anyhow!("mirror source '{source}' is not declared any more"))?;
let id = coord_to_id(coord)
.ok_or_else(|| anyhow::anyhow!("'{coord}' is not a holger coordinate"))?;
// Runs on a job-pool thread (not a runtime worker), so blocking on the
// server's runtime here is the ordinary bridge, not a nested runtime.
self.rt.block_on(async {
let up = tokio::time::timeout(src.connect_timeout(), self.connector.connect(src))
.await
.map_err(|_| anyhow::anyhow!("connect to mirror source '{source}' timed out"))??;
up.fetch(from_repo, &id).await
})
}
}
#[cfg(all(test, feature = "fetch"))]
mod tests {
use super::*;
fn repos() -> Vec<(String, bool)> {
vec![("crates".to_string(), true), ("frozen".to_string(), false)]
}
fn one_source(body: &str) -> String {
one_source_at("http://holger.rs:50051", body)
}
/// `one_source` with the endpoint spelled out — the credential arms care
/// about the SCHEME (see `refuse_scheme_mismatch`), so TLS declarations need
/// an `https://` upstream to be legal at all.
fn one_source_at(endpoint: &str, body: &str) -> String {
format!(
"( ron_sources: [ ( ron_name: \"up\", ron_kind: Holger, \
ron_endpoint: \"{endpoint}\", {body} ) ] )"
)
}
/// ★ **A gap is backfilled** (stinky SB #35): a twin's first cycle in a
/// process, a primary that restarted and a head that went backwards are
/// gaps; a backfill forgets every change-map entry whose bytes the store no
/// longer holds, so the next cycle fetches it, and keeps the rest.
#[test]
fn a_gap_is_detected_and_backfilled_against_what_the_store_holds() {
use crate::grow::{FollowGap, FollowPosition};
let p = FollowPosition::default();
assert_eq!(
p.gap((7, 10)),
Some(FollowGap::NeverFollowedHere),
"nothing completed here yet"
);
p.completed((7, 10));
assert_eq!(p.gap((7, 12)), None, "the same boot, ahead: no gap");
assert_eq!(
p.gap((8, 3)),
Some(FollowGap::PrimaryRestarted {
applied_boot: 7,
boot: 8
})
);
assert_eq!(
p.gap((7, 4)),
Some(FollowGap::Rewound {
applied_seq: 10,
head: 4
})
);
let dir = tempfile::tempdir().unwrap();
let local: Arc<dyn RepositoryBackendTrait> = Arc::new(crate::raw::RawBackend::new(
"crates".into(),
dir.path().join("crates"),
));
let held = coord_to_id("/a.tar@").unwrap();
local.put(&held, b"a").unwrap();
let state = dir.path().join("state");
std::fs::create_dir_all(&state).unwrap();
let fresh: Vec<RemoteAsset> = ["/a.tar@", "/b.tar@"]
.iter()
.map(|c| RemoteAsset {
path: c.to_string(),
download_url: String::new(),
content_type: None,
size: Some(1),
sha256: None,
change_token: None,
})
.collect();
let mut map = ChangeMap::empty("twin-of-primary", "crates");
let d = map.diff(&fresh, 0);
map.adopt(&fresh, &d, &[], 1);
map.store(&state).unwrap();
let pair = MirrorPair {
ron_from: "crates".into(),
ron_to: "crates".into(),
ron_unproven: Default::default(),
};
assert_eq!(
backfill_pair("twin-of-primary", &pair, &local, &state).unwrap(),
1,
"b.tar is claimed and not held"
);
let after = ChangeMap::load(&state, "twin-of-primary", "crates").unwrap();
assert!(
after.get("/a.tar@").is_some() && after.get("/b.tar@").is_none(),
"only what is held stays claimed"
);
assert_eq!(
backfill_pair("twin-of-primary", &pair, &local, &state).unwrap(),
0,
"a second backfill forgets nothing"
);
}
#[test]
fn a_pair_that_says_nothing_refetches_the_unproven() {
let s: MirrorSettings = ron::from_str(&one_source(
"ron_repositories: [ ( ron_from: \"crates\", ron_to: \"crates\" ) ]",
))
.expect("parse");
assert_eq!(
s.ron_sources[0].ron_repositories[0].ron_unproven,
RonUnprovenPolicy::Refetch
);
assert_eq!(
UnprovenPolicy::from(s.ron_sources[0].ron_repositories[0].ron_unproven),
UnprovenPolicy::Refetch,
"the config's default must be the ENGINE's default — the safe one"
);
assert_eq!(s.ron_state_dir, "holger-mirror");
assert_eq!(s.ron_sources[0].ron_every_secs, 300);
assert_eq!(s.ron_sources[0].ron_credential, MirrorCredential::Anonymous);
s.validate(&repos()).expect("valid");
}
#[test]
fn trust_size_is_a_word_an_operator_has_to_write() {
let s: MirrorSettings = ron::from_str(&one_source(
"ron_repositories: [ ( ron_from: \"crates\", ron_to: \"crates\", \
ron_unproven: TrustSize ) ]",
))
.expect("parse");
assert_eq!(
UnprovenPolicy::from(s.ron_sources[0].ron_repositories[0].ron_unproven),
UnprovenPolicy::TrustSize
);
}
#[test]
fn an_unknown_unproven_value_refuses_rather_than_degrades() {
let e = ron::from_str::<MirrorSettings>(&one_source(
"ron_repositories: [ ( ron_from: \"crates\", ron_to: \"crates\", \
ron_unproven: TrustEverything ) ]",
))
.expect_err("an unknown verdict policy must not parse");
let msg = format!("{e}");
assert!(!msg.is_empty(), "the refusal must say something: {msg}");
}
#[test]
fn a_typo_in_the_block_is_a_boot_error() {
let e = ron::from_str::<MirrorSettings>("( ron_state_dirr: \"x\", ron_sources: [] )")
.expect_err("deny_unknown_fields must refuse ron_state_dirr");
assert!(format!("{e}").contains("ron_state_dirr") || format!("{e}").contains("unknown"));
}
#[test]
fn a_typo_in_a_pair_is_a_boot_error() {
ron::from_str::<MirrorSettings>(&one_source(
"ron_repositories: [ ( ron_from: \"crates\", ron_too: \"crates\" ) ]",
))
.expect_err("deny_unknown_fields must refuse ron_too");
}
#[test]
fn a_vault_credential_is_refused_by_name_never_run_anonymously() {
let s: MirrorSettings = ron::from_str(&one_source(
"ron_credential: Vault( ron_vault_path: \"/var/lib/holger/agent.vault\", \
ron_entry: \"holger-rs\" ), \
ron_repositories: [ ( ron_from: \"crates\", ron_to: \"crates\" ) ]",
))
.expect("the shape parses — the operator must be told WHY, not `unknown variant`");
let e = s.validate(&repos()).expect_err("must refuse");
let msg = format!("{e}");
assert!(msg.contains("holger-rs"), "names the entry: {msg}");
assert!(msg.contains("REFUSED"), "says it refuses: {msg}");
}
// ── the mTLS credential ───────────────────────────────────────────────────
/// A throwaway CA + a client cert signed by it, in memory. `rcgen` is already
/// a dev-dependency of this crate (`tests/integration_oci_tls.rs` uses it).
fn client_pair(cn: &str) -> (String, String) {
let ck = rcgen::generate_simple_self_signed(vec![cn.to_string()]).expect("gen");
(ck.cert.pem(), ck.signing_key.serialize_pem())
}
fn write(dir: &std::path::Path, name: &str, body: &str) -> String {
let p = dir.join(name);
std::fs::write(&p, body).expect("write");
std::fs::set_permissions(
&p,
<std::fs::Permissions as std::os::unix::fs::PermissionsExt>::from_mode(0o600),
)
.unwrap();
p.to_string_lossy().into_owned()
}
fn mtls_source(cert: &str, key: &str, ca: Option<&str>) -> MirrorSettings {
let ca_ron = match ca {
Some(c) => format!("ron_ca_cert: Some(\"{c}\"), "),
None => String::new(),
};
ron::from_str(&one_source_at(
"https://holger.example.com:50051",
&format!(
"ron_credential: Mtls( ron_client_cert: \"{cert}\", ron_client_key: \"{key}\", \
{ca_ron}), ron_repositories: [ ( ron_from: \"crates\", ron_to: \"crates\" ) ]"
),
))
.expect("the Mtls shape parses")
}
/// The happy path: paths that exist and hold what they say they hold pass
/// `validate` and resolve to a real client identity.
#[test]
fn an_mtls_credential_whose_files_are_real_loads_and_validates() {
let dir = tempfile::tempdir().expect("tmp");
let (cert, key) = client_pair("mirror-node");
let cp = write(dir.path(), "mirror.crt", &cert);
let kp = write(dir.path(), "mirror.key", &key);
let s = mtls_source(&cp, &kp, Some(&cp));
s.validate(&repos())
.expect("a loadable credential must not refuse the boot");
let loaded = s.ron_sources[0].load_credential().expect("loads");
assert!(
loaded.is_mtls(),
"an Mtls declaration must not resolve to Anonymous"
);
let LoadedCredential::Mtls(id) = &loaded else {
unreachable!()
};
assert_eq!(id.cert_path(), cp);
assert!(
id.ca_pem().is_some(),
"the declared CA must be carried, not dropped"
);
let (c, k) = id.client_identity();
assert!(!c.is_empty() && !k.is_empty());
}
/// **The hard requirement.** A declared-but-unloadable credential refuses at
/// BOOT (`validate` is what `wire_holger` runs), naming the source and the
/// path — and never becomes an anonymous mirror.
#[test]
fn a_declared_credential_that_cannot_be_loaded_refuses_the_boot_by_name() {
let s = mtls_source("/nonexistent/mirror.crt", "/nonexistent/mirror.key", None);
let e = s
.validate(&repos())
.expect_err("a missing client cert must stop the boot");
let msg = format!("{e:#}");
assert!(msg.contains("up"), "names the source: {msg}");
assert!(
msg.contains("/nonexistent/mirror.crt"),
"names the path: {msg}"
);
assert!(msg.contains("REFUSES"), "says it refuses: {msg}");
assert!(
s.ron_sources[0].load_credential().is_err(),
"and it is an Err, not a silent LoadedCredential::Anonymous"
);
}
/// The swapped-paths mistake — the most likely way to get a credential
/// wrong — is caught at boot as a KIND error, not at the first handshake as
/// an opaque TLS failure.
#[test]
fn a_pem_of_the_wrong_kind_is_refused_at_boot_not_at_the_handshake() {
let dir = tempfile::tempdir().expect("tmp");
let (cert, key) = client_pair("mirror-node");
let cp = write(dir.path(), "mirror.crt", &cert);
let kp = write(dir.path(), "mirror.key", &key);
// cert and key swapped.
let s = mtls_source(&kp, &cp, None);
let e = s.validate(&repos()).expect_err("swapped paths must refuse");
let msg = format!("{e:#}");
assert!(msg.contains("no PEM CERTIFICATE"), "names the kind: {msg}");
assert!(msg.contains("swapped"), "names the likely cause: {msg}");
// ron_ca_cert pointed at a key.
let s = mtls_source(&cp, &kp, Some(&kp));
let e = s
.validate(&repos())
.expect_err("a CA that is not a certificate must refuse");
assert!(format!("{e:#}").contains("ron_ca_cert"), "{e:#}");
}
/// **THE CANARY.** `ConnectorAuth` derived `Debug` and printed passwords and
/// bearer tokens in full. Nothing that carries key material in this module
/// may render it — not `Debug`, not an error chain, not a report line.
///
/// A distinctive marker is planted inside the PEM files and then hunted for
/// in every rendering a loaded credential can reach.
#[test]
fn no_key_material_in_any_rendering_of_a_loaded_identity() {
const CANARY: &str = "CANARY-mirror-mtls-private-key-must-never-be-printed";
let dir = tempfile::tempdir().expect("tmp");
let (cert, key) = client_pair("mirror-node");
// The marker rides in a PEM comment line, so the files still parse.
let cp = write(dir.path(), "mirror.crt", &format!("# {CANARY}\n{cert}"));
let kp = write(dir.path(), "mirror.key", &format!("# {CANARY}\n{key}"));
let s = mtls_source(&cp, &kp, Some(&cp));
let loaded = s.ron_sources[0].load_credential().expect("loads");
// 1. Debug of the loaded credential and of the identity inside it.
let rendered = format!("{loaded:?}");
assert!(
!rendered.contains(CANARY),
"Debug leaked key material: {rendered}"
);
assert!(!rendered.contains("BEGIN"), "Debug leaked PEM: {rendered}");
assert!(
rendered.contains("<redacted>"),
"and it must SAY it redacted: {rendered}"
);
assert!(
rendered.contains(&cp),
"paths are what an operator needs: {rendered}"
);
// 2. Debug of the DECLARATION: it carries paths, so it is safe — but it
// must be safe on purpose, and a future field holding bytes breaks this.
let decl = format!("{:?}", s.ron_sources[0].ron_credential);
assert!(
!decl.contains(CANARY),
"the declaration leaked key material: {decl}"
);
// 3. The error chain of a connect failure against a dead endpoint: the
// only place key bytes get near a `with_context`.
let rt = tokio::runtime::Builder::new_current_thread()
.enable_all()
.build()
.expect("rt");
let err = rt.block_on(async {
match HolgerUpstream::connect_with("http://127.0.0.1:1/", &loaded).await {
Err(e) => e,
Ok(_) => panic!("port 1 must refuse the connection"),
}
});
let chain = format!("{err:#}");
assert!(
!chain.contains(CANARY),
"the error chain leaked key material: {chain}"
);
assert!(
!chain.contains("BEGIN"),
"the error chain leaked PEM: {chain}"
);
// 4. The report line every cycle logs.
let line = PairReport {
source: "up".into(),
from: "crates".into(),
to: "crates".into(),
..Default::default()
}
.line();
assert!(
!line.contains(CANARY),
"the cycle report leaked key material: {line}"
);
}
/// **The endpoint's scheme and the credential are one statement written
/// twice, and a disagreement refuses the boot.**
///
/// The `http://` + TLS-material direction is the dangerous one: it does not
/// fail at all. A cleartext channel opens and the certificate is simply
/// never presented — a config that READS as authenticated, RUNNING as open.
#[test]
fn the_endpoint_scheme_and_the_credential_must_agree() {
let pairs: [(&str, &str, &str); 3] = [
(
"https://holger.example.com:50051",
"ron_credential: Anonymous",
"CLEARTEXT",
),
(
"http://holger.rs:50051",
"ron_credential: Tls( ron_ca_cert: None )",
"READS as authenticated",
),
(
"http://holger.rs:50051",
"ron_credential: Mtls( ron_client_cert: \"/a.crt\", ron_client_key: \"/a.key\" )",
"READS as authenticated",
),
];
for (endpoint, cred, expect) in pairs {
let s: MirrorSettings = ron::from_str(&one_source_at(
endpoint,
&format!(
"{cred}, ron_repositories: [ ( ron_from: \"crates\", ron_to: \"crates\" ) ]"
),
))
.expect("the shape parses — the operator must be told WHY");
let e = s
.validate(&repos())
.expect_err("a scheme/credential mismatch must stop the boot");
let msg = format!("{e:#}");
assert!(msg.contains(expect), "{endpoint} + {cred}: {msg}");
assert!(msg.contains(endpoint), "names the endpoint: {msg}");
}
// …and the two matching combinations are fine.
for (endpoint, cred) in [
("http://holger.rs:50051", "ron_credential: Anonymous"),
(
"https://holger.example.com:50051",
"ron_credential: Tls( ron_ca_cert: None )",
),
] {
let s: MirrorSettings = ron::from_str(&one_source_at(
endpoint,
&format!(
"{cred}, ron_repositories: [ ( ron_from: \"crates\", ron_to: \"crates\" ) ]"
),
))
.expect("parse");
s.validate(&repos()).expect("a matching pair must boot");
}
}
/// `Tls` is TLS with nothing presented — the certless half of a
/// per-repository posture, and NOT the same thing as `Anonymous`.
#[test]
fn tls_without_a_client_identity_is_its_own_spelling() {
let s: MirrorSettings = ron::from_str(&one_source_at(
"https://holger.example.com:50051",
"ron_credential: Tls( ron_ca_cert: None ), \
ron_repositories: [ ( ron_from: \"crates\", ron_to: \"crates\" ) ]",
))
.expect("parse");
let loaded = s.ron_sources[0].load_credential().expect("loads");
assert!(loaded.is_tls(), "Tls must build a TLS channel");
assert!(!loaded.is_mtls(), "…and must present no client identity");
assert_ne!(loaded, LoadedCredential::Anonymous);
}
/// The RON vocabulary refuses a typo inside the credential rather than
/// dropping the field and mirroring anonymously.
#[test]
fn a_misspelled_credential_field_refuses_instead_of_degrading() {
ron::from_str::<MirrorSettings>(&one_source(
"ron_credential: Mtls( ron_client_cert: \"/a\", ron_client_keyy: \"/b\" ), \
ron_repositories: [ ( ron_from: \"crates\", ron_to: \"crates\" ) ]",
))
.expect_err("deny_unknown_fields must refuse ron_client_keyy");
}
#[test]
fn a_kind_this_binary_cannot_drive_is_refused_by_name() {
let s: MirrorSettings = ron::from_str(
"( ron_sources: [ ( ron_name: \"nx\", ron_kind: Nexus, \
ron_endpoint: \"http://nexus:8081\", \
ron_repositories: [ ( ron_from: \"a\", ron_to: \"crates\" ) ] ) ] )",
)
.expect("parse");
let e = s.validate(&repos()).expect_err("must refuse");
assert!(format!("{e}").contains("Nexus"), "{e}");
}
#[test]
fn ron_to_must_name_a_declared_writable_repository() {
let s: MirrorSettings = ron::from_str(&one_source(
"ron_repositories: [ ( ron_from: \"crates\", ron_to: \"nope\" ) ]",
))
.expect("parse");
assert!(format!("{}", s.validate(&repos()).unwrap_err()).contains("nope"));
let s: MirrorSettings = ron::from_str(&one_source(
"ron_repositories: [ ( ron_from: \"crates\", ron_to: \"frozen\" ) ]",
))
.expect("parse");
let e = format!("{}", s.validate(&repos()).unwrap_err());
assert!(e.contains("frozen") && e.contains("not writable"), "{e}");
}
#[test]
fn two_pairs_may_not_share_one_change_map() {
let s: MirrorSettings = ron::from_str(&one_source(
"ron_repositories: [ ( ron_from: \"crates\", ron_to: \"crates\" ), \
( ron_from: \"crates\", ron_to: \"crates\" ) ]",
))
.expect("parse");
assert!(format!("{}", s.validate(&repos()).unwrap_err()).contains("twice"));
}
#[test]
fn a_zero_period_is_refused() {
let s: MirrorSettings = ron::from_str(&one_source(
"ron_every_secs: 0, ron_repositories: [ ( ron_from: \"crates\", ron_to: \"crates\" ) ]",
))
.expect("parse");
assert!(format!("{}", s.validate(&repos()).unwrap_err()).contains("busy loop"));
}
#[test]
fn coordinates_round_trip_including_an_npm_scope() {
for id in [
ArtifactId {
namespace: None,
name: "serde".into(),
version: "1.0.0".into(),
},
ArtifactId {
namespace: Some("org.example".into()),
name: "widget".into(),
version: "2.1".into(),
},
ArtifactId {
namespace: None,
name: "@scope/pkg".into(),
version: "0.1.0".into(),
},
] {
let coord = id_to_coord(&id);
assert_eq!(
coord_to_id(&coord).as_ref(),
Some(&id),
"round trip {coord}"
);
}
assert_eq!(coord_to_id("not-a-coordinate"), None);
}
/// **A holger listing now carries the digest its backend could prove, and
/// still carries nothing when the backend could not.**
///
/// This test used to be called `a_holger_listing_publishes_no_digest_and_
/// says_so` and it asserted `sha256 == None` unconditionally, because
/// `ArtifactEntry` had no digest field at all. It has one now, so the true
/// statement is the two-sided one: what the entry knows is carried through
/// verbatim, and what it does not know is still `None` — never an empty
/// string, never a fabricated hash.
///
/// Both halves are RED-when-broken. Losing the digest puts every
/// size-unchanged coordinate back in `unproven` forever; inventing one
/// makes `unchanged` a lie.
#[test]
fn a_holger_listing_carries_the_digest_it_was_given_and_nothing_when_it_was_not() {
let known = "b".repeat(64);
let assets = holger_listing(
"http://holger.rs:50051",
"crates",
vec![
// A backend that CAN prove sameness — file-oci, or file-rust
// with the checksum already memoized.
traits::ArtifactEntry::unknown_provenance(
ArtifactId {
namespace: None,
name: "a".into(),
version: "1".into(),
},
12,
"application/octet-stream",
)
.with_sha256(Some(&known)),
// A backend that cannot — every znippy-backed one.
traits::ArtifactEntry::unknown_provenance(
ArtifactId {
namespace: None,
name: "b".into(),
version: "1".into(),
},
12,
"application/octet-stream",
),
],
);
assert_eq!(assets[0].path, "/a@1");
assert_eq!(assets[0].size, Some(12));
assert_eq!(
assets[0].sha256.as_deref(),
Some(known.as_str()),
"a digest the listing published must reach the diff — it is the only thing that \
can turn a size-unchanged coordinate into `unchanged` instead of `unproven`"
);
assert_eq!(
assets[1].sha256, None,
"a backend that does not know must stay unknown — an invented or empty digest \
would make `unchanged` a lie about bytes nobody compared"
);
}
/// A malformed digest is DROPPED, not carried. The listing is the boundary
/// where a value from elsewhere becomes something the diff will trust, so
/// anything that is not 64 hex characters stops here.
#[test]
fn a_malformed_digest_in_a_listing_becomes_unknown_rather_than_being_carried() {
for bad in [
"",
"sha256:aa",
&"z".repeat(64),
&"a".repeat(63),
&"a".repeat(65),
] {
let mut e = traits::ArtifactEntry::unknown_provenance(
ArtifactId {
namespace: None,
name: "a".into(),
version: "1".into(),
},
12,
"application/octet-stream",
);
// Set the field directly: `with_sha256` would already have refused
// it, and the point here is that the listing refuses it too.
e.sha256 = Some(bad.to_string());
let assets = holger_listing("http://holger.rs:50051", "crates", vec![e]);
assert_eq!(
assets[0].sha256, None,
"{bad:?} must not survive into a RemoteAsset"
);
}
}
/// An UPPERCASE digest is the same digest. It is lowercased rather than
/// dropped, so a peer that spells hex the other way still reaches
/// `unchanged` instead of re-fetching forever.
#[test]
fn a_digest_is_lowercased_so_spelling_does_not_cost_a_refetch() {
let e = traits::ArtifactEntry::unknown_provenance(
ArtifactId {
namespace: None,
name: "a".into(),
version: "1".into(),
},
12,
"application/octet-stream",
)
.with_sha256(Some("A".repeat(64)));
let assets = holger_listing("http://holger.rs:50051", "crates", vec![e]);
assert_eq!(assets[0].sha256.as_deref(), Some("a".repeat(64).as_str()));
}
#[test]
fn the_backoff_starts_low_doubles_and_never_passes_the_declared_period() {
let s: MirrorSettings = ron::from_str(&one_source(
"ron_every_secs: 300, ron_repositories: [ ( ron_from: \"crates\", ron_to: \"crates\" ) ]",
))
.expect("parse");
let src = &s.ron_sources[0];
assert_eq!(src.ron_retry_secs, 5, "the default first retry");
assert_eq!(src.retry_floor(), Duration::from_secs(5));
// 5 → 10 → 20 → 40 → 80 → 160 → 300 → 300 → 300 …
let mut d = src.retry_floor();
let mut seen = vec![d];
for _ in 0..10 {
d = src.next_backoff(d);
seen.push(d);
}
assert_eq!(
seen,
[5, 10, 20, 40, 80, 160, 300, 300, 300, 300, 300]
.map(Duration::from_secs)
.to_vec(),
"the backoff doubles and then STOPS at the declared period"
);
for d in &seen {
assert!(
*d <= src.interval(),
"an operator who wrote ron_every_secs: 300 never consented to a {d:?} gap"
);
}
}
#[test]
fn a_retry_floor_longer_than_the_period_is_clamped_to_the_period() {
// A period SHORTER than the retry floor must not stretch the cadence.
let s: MirrorSettings = ron::from_str(&one_source(
"ron_every_secs: 2, ron_repositories: [ ( ron_from: \"crates\", ron_to: \"crates\" ) ]",
))
.expect("parse");
let src = &s.ron_sources[0];
assert_eq!(src.retry_floor(), Duration::from_secs(2));
assert_eq!(src.next_backoff(src.retry_floor()), Duration::from_secs(2));
assert_eq!(src.connect_timeout(), Duration::from_secs(2));
}
#[test]
fn writing_the_period_into_the_retry_restores_the_old_one_attempt_per_period() {
let s: MirrorSettings = ron::from_str(&one_source(
"ron_every_secs: 300, ron_retry_secs: 300, \
ron_repositories: [ ( ron_from: \"crates\", ron_to: \"crates\" ) ]",
))
.expect("parse");
let src = &s.ron_sources[0];
assert_eq!(src.retry_floor(), Duration::from_secs(300));
assert_eq!(
src.next_backoff(src.retry_floor()),
Duration::from_secs(300)
);
s.validate(&repos()).expect("valid");
}
#[test]
fn a_zero_retry_or_zero_connect_timeout_is_refused() {
let zero_retry: MirrorSettings = ron::from_str(&one_source(
"ron_retry_secs: 0, ron_repositories: [ ( ron_from: \"crates\", ron_to: \"crates\" ) ]",
))
.expect("parse");
assert!(
format!("{}", zero_retry.validate(&repos()).unwrap_err()).contains("busy loop"),
"a zero retry must be refused by name"
);
let zero_ct: MirrorSettings = ron::from_str(&one_source(
"ron_connect_timeout_secs: 0, \
ron_repositories: [ ( ron_from: \"crates\", ron_to: \"crates\" ) ]",
))
.expect("parse");
assert!(
format!("{}", zero_ct.validate(&repos()).unwrap_err()).contains("never mirror"),
"a zero connect timeout must be refused by name"
);
}
#[test]
fn a_config_written_before_the_resilience_keys_existed_still_parses() {
// The live `holger-c.ron` on oden, trimmed to the source. Neither new key
// appears in it, and it must keep working with the documented defaults.
let s: MirrorSettings = ron::from_str(
"( ron_state_dir: \"holger-mirror\", ron_sources: [ ( ron_name: \"holger-rs\", \
ron_kind: Holger, ron_endpoint: \"http://holger.rs:50051\", \
ron_credential: Anonymous, ron_every_secs: 300, ron_repositories: [ \
( ron_from: \"crates\", ron_to: \"crates\" ) ] ) ] )",
)
.expect("a file written before these keys must still parse");
assert_eq!(s.ron_sources[0].ron_every_secs, 300);
assert_eq!(s.ron_sources[0].ron_retry_secs, default_retry_secs());
assert_eq!(
s.ron_sources[0].ron_connect_timeout_secs,
default_connect_timeout_secs()
);
}
#[test]
fn a_failure_says_whether_retrying_can_help() {
// The stage attaches the verdict; nothing downstream parses a string.
let transient = anyhow::anyhow!("connection refused")
.context(Transience::Transient)
.context("mirror up: connect upstream http://x:1");
assert_eq!(transience_of(&transient), Transience::Transient);
// The outer context is still the readable one; the verdict rides inside.
assert_eq!(
format!("{transient}"),
"mirror up: connect upstream http://x:1"
);
assert!(format!("{transient:#}").contains("transient"));
let permanent = anyhow::anyhow!("invalid Arrow IPC footer")
.context(Transience::Permanent)
.context("mirror up:crates: load change map");
assert_eq!(transience_of(&permanent), Transience::Permanent);
// An unclassified failure is retried, deliberately: not retrying a
// transient failure is the defect; retrying a permanent one is a log line.
let unmarked = anyhow::anyhow!("something nobody has classified yet");
assert_eq!(transience_of(&unmarked), Transience::Transient);
}
#[test]
fn an_unanchored_state_dir_says_it_is_using_the_current_directory() {
let s: MirrorSettings = ron::from_str("( ron_sources: [] )").expect("parse");
let (dir, warning) = s.state_dir_resolved(Some(Path::new("/home/alice/holger-c")));
assert_eq!(dir, PathBuf::from("/home/alice/holger-c/holger-mirror"));
assert!(warning.is_none(), "the server's own resolution is silent");
let (dir, warning) = s.state_dir_resolved(None);
assert_eq!(dir, PathBuf::from("holger-mirror"));
let w = warning.expect(
"a hand-run with no HOLGER_DATA reads a DIFFERENT map and must say so — this is \
what made a mirror re-fetch an artifact it already held",
);
assert!(w.contains("CURRENT DIRECTORY"), "{w}");
assert!(w.contains("HOLGER_DATA"), "{w}");
// An absolute state dir cannot drift, so it is never warned about.
let abs: MirrorSettings =
ron::from_str("( ron_state_dir: \"/srv/maps\", ron_sources: [] )").expect("parse");
assert!(abs.state_dir_resolved(None).1.is_none());
}
#[test]
fn the_state_dir_anchors_under_the_data_root() {
let s: MirrorSettings = ron::from_str("( ron_sources: [] )").expect("parse");
assert_eq!(
s.state_dir(Some(Path::new("/var/lib/holger"))),
PathBuf::from("/var/lib/holger/holger-mirror")
);
let abs: MirrorSettings =
ron::from_str("( ron_state_dir: \"/srv/maps\", ron_sources: [] )").expect("parse");
assert_eq!(
abs.state_dir(Some(Path::new("/var/lib/holger"))),
PathBuf::from("/srv/maps")
);
}
}