server/lib/src/mirror.rs
//! Mirror change map — "fetch only what changed" for a continuous on-prem mirror.
//!
//! # What this is
//!
//! Holger on-prem runs as a **mirror** of an upstream: another holger, or a
//! Nexus / Artifactory. A mirror cycle asks the upstream for its listing
//! (`ConnectorTrait::list_assets`, hundreds of thousands of [`RemoteAsset`]s for
//! a real Maven or PyPI proxy) and must then fetch **only what changed since
//! last cycle**. Everything else is the difference between a nightly sync and a
//! nightly re-download of the whole world.
//!
//! This module is that difference: a persisted per-`(upstream, repository)`
//! **change map** — coordinate → digest (or size, when the source publishes no
//! digest) → when last seen — and a **diff** that classifies a fresh listing
//! against it into added / changed / unchanged / unproven / removed.
//!
//! It is deliberately a leaf: no gRPC, no CLI, no connector calls. It takes a
//! listing someone else fetched and a directory someone else chose. The caller
//! (agent, service, scheduler) wires it.
//!
//! # Why a LOCAL map, even against another holger
//!
//! Against another holger the upstream already knows what it sent us:
//! [`crate::audit::query::sent_to_by`] answers "what did this server send to
//! this downstream, and when". It is tempting to skip the local map and ask.
//! We do not, for three reasons, and none of them is effort:
//!
//! 1. **It records sends, not contents.** The audit log is a log of transfers.
//! It can reconstruct what the downstream *received*; it cannot see an
//! artifact the upstream gained, or one whose bytes the upstream replaced,
//! because neither event is a send. A mirror needs "what does the upstream
//! HAVE now", which is exactly the fresh listing diffed against local state.
//! 2. **It is the upstream's log, read over a privileged API.** A mirror that
//! depends on being allowed to read its upstream's audit trail is a mirror
//! that stops working the moment the upstream tightens its roles — and
//! reads are open by default here, so the downstream would frequently be
//! one `anonymous` bucket among many (see `SentToKey::Ident`'s own note).
//! 3. **It does not exist for Nexus or Artifactory.** The owner's requirement
//! is one mirror that works "klockrent" against another holger *and* against
//! Nexus/Artifactory. A mechanism that only exists on one of the three is
//! not the mechanism; the local map is, so it is the only one built.
//!
//! `sent_to_by` stays what it is: an operator's answer to "who pulled what".
//!
//! # Store: sorted JSON lines, not a key-value database
//!
//! A change map is a wide, cold, append-mostly column of coordinates, and this
//! diff wants it **sorted** (the hot path is a merge-join, not hashing half a
//! million keys) and **contiguous** in memory ([`ChangeMap`] keeps five `Vec`s
//! for the whole map, never one allocation per coordinate). On disk it is one
//! JSON header line plus one JSON array per entry, ascending — greppable with
//! any tool, and written beside the target then renamed into place.
//!
//! The only dependency this module adds is `gatling` (package `rotaryengine`),
//! the leaf crate holding `znippy_zoomies::gatling` — the root `znippy-zoomies`
//! re-exports this very crate (`pub use ::gatling::{…}`), so depending on the
//! leaf is the same engine without dragging the OSM index and the codecs into
//! `holger-server-lib`. Rayon is not used and must not be.
//!
//! # What happens when the source publishes no sha256 — and what a znippy
//! # repository publishes instead
//!
//! [`RemoteAsset::sha256`] is populated by both connectors now (Nexus reads
//! `checksum.sha256`; Artifactory asks AQL for `actual_sha256`), but `None` is
//! still common and honest: it means *this source published no digest*. It is
//! also the permanent answer of 15 of holger's 17 repository backends, because
//! a znippy file table carries neither a modification time nor a per-entry
//! SHA-256 and producing one would mean extracting and hashing the whole
//! repository on a listing.
//!
//! What those backends CAN publish, for free, is a **change token**
//! ([`RemoteAsset::change_token`]): an opaque value folded from the per-chunk
//! BLAKE3 hashes the archive index already holds. It is not a content digest —
//! it is BLAKE3 and it binds the archive-relative path — so it is never
//! compared against a `sha256`, and because two sources lay their archives out
//! their own way it is never compared across sources either. The value carries
//! its own SCOPE for exactly that reason, and a comparison whose two scopes
//! differ is REFUSED rather than answered.
//!
//! This module never pretends. A coordinate is classified by:
//!
//! 1. **Both sides have a digest** → equal ⇒ `unchanged`, differ ⇒ `changed`.
//! The strongest proof there is, and the only one comparable across
//! sources.
//! 2. **Both sides have a change token with the SAME scope** → equal ⇒
//! `unchanged`, differ ⇒ `changed`. This is the one that makes a
//! znippy-backed mirror settle. Below the digest, because it proves less.
//! 3. **Both sides have a size and the sizes differ** → `changed`. A size
//! difference proves a change even with no digest and no token anywhere.
//! 4. **Otherwise** → `unproven`. Not `unchanged`. Sizes that match without a
//! digest do **not** establish sameness: a same-size edit — one byte flipped
//! in a jar, a rebuilt wheel — is invisible to `size`, and silently calling
//! that `unchanged` is how a mirror serves stale bytes forever. A token on
//! ONE side only lands here too, and so does a pair of tokens from two
//! different sources: neither proves a change and neither proves sameness.
//!
//! `unproven` is a bucket of its own, never folded into `unchanged`, so a caller
//! can always see how much of a sync it cannot prove. What the mirror *does*
//! about it is [`UnprovenPolicy`], and the default is
//! [`UnprovenPolicy::Refetch`]: re-fetch it. Fetching bytes we already had costs
//! bandwidth; skipping bytes that changed costs correctness, and a mirror exists
//! to be correct. [`UnprovenPolicy::TrustSize`] is available for an operator who
//! knows their upstream publishes immutable coordinates and will accept the
//! risk — it is opt-in, and the bucket stays visible either way.
//!
//! # The hot path
//!
//! [`ChangeMap::diff`] is O(n log n) for the sort plus O(n + m) for the join,
//! with both phases fanned out over `gatling`:
//!
//! * `gatling_sort_unstable_by` builds a coordinate-ascending permutation of the
//! fresh listing — a `Vec<u32>` of indices, so no [`RemoteAsset`] is moved or
//! cloned.
//! * `gatling_reduce` splits that permutation into one contiguous partition per
//! worker, binary-searches the (already sorted) map for each partition's
//! coordinate boundary, and runs an independent **merge-join** per partition.
//! Partitions are disjoint by construction, merged back in ascending partition
//! order, so every output bucket comes out coordinate-sorted and the result is
//! byte-identical to the serial reference.
//!
//! The reference implementation, [`diff_reference`], is a plain serial
//! `HashMap` lookup per asset. It is kept public and is the oracle every test
//! compares against — and the baseline the scaling measurement counts against.
//! It allocates one hash table of `m` entries and hashes every one of `n`
//! coordinates; the merge-join allocates the index permutation and the output
//! vectors and hashes nothing.
use std::collections::HashMap;
use std::fs::{self, File};
use std::io::{BufWriter, Write as _};
use std::path::{Path, PathBuf};
use anyhow::{bail, Context, Result};
use gatling::gatling_forkjoin::{default_workers, gatling_reduce};
use gatling::gatling_sort::gatling_sort_unstable_by;
use traits::RemoteAsset;
/// The on-disk format this reader and writer speak. Anything else is REFUSED:
/// silently reading an unknown format with the wrong rule is how a mirror
/// serves stale bytes.
const MAP_FORMAT: &str = "holger.mirror.map/v3";
/// `flags` bit: this entry carries a sha256.
const FLAG_DIGEST: u8 = 0b0000_0001;
/// `flags` bit: this entry carries a size.
const FLAG_SIZE: u8 = 0b0000_0010;
/// `flags` bit: this entry carries a change token.
const FLAG_TOKEN: u8 = 0b0000_0100;
// ───────────────────────────── policy ─────────────────────────────
/// What the mirror does with a coordinate whose sameness cannot be **proved** —
/// see the module docs, "What happens when the source publishes no sha256".
///
/// This steers [`MirrorDiff::to_fetch`] only. The diff itself is always total
/// and always honest: `unproven` is reported as its own bucket under every
/// policy, so an operator can see the size of the doubt regardless.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
pub enum UnprovenPolicy {
/// **Default.** Re-fetch it. Bandwidth is cheaper than a stale mirror.
#[default]
Refetch,
/// Treat "same size, no digest" as unchanged and skip the fetch. Opt-in,
/// for an upstream whose coordinates are known-immutable. Accepts, in
/// exchange for the bandwidth, that a same-size edit will be missed.
TrustSize,
}
// ───────────────────────────── stats ─────────────────────────────
/// What a diff cost, counted rather than timed.
///
/// A wall clock on a build box shared by six lanes measures the other five.
/// These counters do not: they are the same on an idle box and a loaded one,
/// and they are what the scaling test reports.
#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
pub struct DiffStats {
/// Assets in the fresh listing.
pub fresh_assets: usize,
/// Entries in the stored map.
pub map_entries: usize,
/// Coordinates the fresh listing repeated. The first in sort order wins;
/// the rest are not classified, because they are the same coordinate.
pub duplicate_coords: usize,
/// Partitions the join ran over (= gatling workers used).
pub workers: usize,
/// Coordinate string comparisons performed by the join, including the
/// binary searches that place the partition boundaries. This is the number
/// the merge-join exists to hold down. The SORT that precedes the join is
/// not counted here: both implementations pay the same one, so it is not a
/// difference between them.
pub key_comparisons: u64,
/// 32-byte digest comparisons performed.
pub digest_comparisons: u64,
/// Change-token comparisons performed. Counted apart from the digest
/// comparisons because they are a different KIND of answer: a digest proves
/// sameness anywhere, a token proves it only at one source.
pub token_comparisons: u64,
/// Coordinate hashes performed. Zero for the merge-join; one per map entry
/// plus one per fresh asset for [`diff_reference`].
pub key_hashes: u64,
}
// ───────────────────────────── diff ─────────────────────────────
/// The classification of one fresh listing against one stored [`ChangeMap`].
///
/// Every bucket holds **indices**, never copies: `added` / `changed` /
/// `unchanged` / `unproven` index into the `fresh: &[RemoteAsset]` slice that
/// was diffed, and `removed` indexes into the [`ChangeMap`]. Nothing in a diff
/// owns a coordinate string, so a diff of half a million assets is a few
/// megabytes of `u32`.
#[derive(Debug, Clone, Default)]
pub struct MirrorDiff {
/// Fresh coordinates the map has never seen. Always fetch.
pub added: Vec<u32>,
/// Fresh coordinates PROVED different (digests differ, or sizes differ).
/// Always fetch.
pub changed: Vec<u32>,
/// Fresh coordinates PROVED identical (both digests present and equal).
/// Never fetch.
pub unchanged: Vec<u32>,
/// Fresh coordinates whose sameness could not be proved, because at least
/// one side published no digest. [`UnprovenPolicy`] decides the fetch.
pub unproven: Vec<u32>,
/// Map entries with no matching fresh coordinate: the upstream dropped
/// them. Indices into the [`ChangeMap`] that produced the diff.
pub removed: Vec<u32>,
/// Counted cost of producing this diff.
pub stats: DiffStats,
/// Fresh indices in coordinate-ascending order, duplicates dropped. Kept so
/// [`ChangeMap::adopt`] can rebuild the map without sorting a second time.
order: Vec<u32>,
}
impl MirrorDiff {
/// The fetch list under `policy`, coordinate-ascending within each bucket:
/// added, then changed, then — unless [`UnprovenPolicy::TrustSize`] —
/// unproven.
pub fn to_fetch(&self, policy: UnprovenPolicy) -> Vec<u32> {
let extra = match policy {
UnprovenPolicy::Refetch => self.unproven.len(),
UnprovenPolicy::TrustSize => 0,
};
let mut out = Vec::with_capacity(self.added.len() + self.changed.len() + extra);
out.extend_from_slice(&self.added);
out.extend_from_slice(&self.changed);
if policy == UnprovenPolicy::Refetch {
out.extend_from_slice(&self.unproven);
}
out
}
/// `true` when the upstream is byte-for-byte where we left it *and we can
/// prove it* — nothing added, nothing changed, nothing removed and nothing
/// unproven. A cycle that sees this can do literally no work.
pub fn is_clean(&self) -> bool {
self.added.is_empty()
&& self.changed.is_empty()
&& self.removed.is_empty()
&& self.unproven.is_empty()
}
/// The coordinate-ascending permutation of the fresh listing, duplicates
/// dropped. Exposed because a caller that wants to walk the listing in map
/// order should not re-sort it.
pub fn order(&self) -> &[u32] {
&self.order
}
}
// ───────────────────────────── the map ─────────────────────────────
/// One entry of a [`ChangeMap`], handed out by value at the edges. The map does
/// **not** store these — see [`ChangeMap`]'s columnar layout.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct MapEntry<'a> {
/// The coordinate — [`RemoteAsset::path`], the upstream's own key.
pub coord: &'a str,
/// The digest last seen for it, when the source published one.
pub sha256: Option<[u8; 32]>,
/// The size last seen for it, when the source published one.
pub size: Option<u64>,
/// The opaque, SOURCE-SCOPED change token last seen for it, when the source
/// published one. Compared only against another token with the same scope
/// (see [`change_tokens_comparable`]) — never against a digest and
/// never across sources.
pub change_token: Option<&'a str>,
/// When this coordinate was last seen in a listing, nanoseconds since the
/// Unix epoch.
pub last_seen_nanos: i64,
}
/// The persisted change map for one `(upstream, repository)` pair.
///
/// **Columnar and sorted**, on disk and in memory. Coordinates live
/// concatenated in one `Vec<u8>` with a `Vec<u32>` offset column — the exact
/// layout an Arrow `Utf8` array has — so a map of 500 000 coordinates is five
/// allocations, not 500 000. The rows are kept in ascending coordinate order at
/// all times, which is what lets [`diff`](ChangeMap::diff) merge-join instead of
/// hash.
#[derive(Debug, Clone)]
pub struct ChangeMap {
upstream: String,
repository: String,
/// Concatenated coordinates, ascending.
coord_bytes: Vec<u8>,
/// `n + 1` offsets into `coord_bytes`.
coord_offsets: Vec<u32>,
/// `n` digests; the slot is meaningless unless `FLAG_DIGEST` is set.
digests: Vec<[u8; 32]>,
/// `n` sizes; meaningless unless `FLAG_SIZE` is set.
sizes: Vec<u64>,
/// Concatenated change tokens, in entry order; the slot is meaningless
/// unless `FLAG_TOKEN` is set. Same concat+offsets layout as the
/// coordinates, for the same reason: a map of half a million tokens is two
/// allocations, not half a million `String`s.
token_bytes: Vec<u8>,
/// `n + 1` offsets into `token_bytes`.
token_offsets: Vec<u32>,
/// `n` presence flags, one byte per entry.
flags: Vec<u8>,
/// `n` last-seen timestamps, nanoseconds since the Unix epoch.
last_seen: Vec<i64>,
}
impl ChangeMap {
/// A map that has never seen anything. The first sync against it reports
/// every fresh coordinate as `added`, which is correct.
pub fn empty(upstream: impl Into<String>, repository: impl Into<String>) -> Self {
Self {
upstream: upstream.into(),
repository: repository.into(),
coord_bytes: Vec::new(),
coord_offsets: vec![0],
digests: Vec::new(),
sizes: Vec::new(),
token_bytes: Vec::new(),
token_offsets: vec![0],
flags: Vec::new(),
last_seen: Vec::new(),
}
}
/// Which upstream this map belongs to.
pub fn upstream(&self) -> &str {
&self.upstream
}
/// Which repository on that upstream.
pub fn repository(&self) -> &str {
&self.repository
}
/// Number of coordinates in the map.
pub fn len(&self) -> usize {
self.flags.len()
}
/// `true` when the map has no coordinates.
pub fn is_empty(&self) -> bool {
self.flags.is_empty()
}
/// The `i`th coordinate, ascending. Panics on out of range.
#[inline]
pub fn coord(&self, i: usize) -> &str {
let lo = self.coord_offsets[i] as usize;
let hi = self.coord_offsets[i + 1] as usize;
// SAFETY-free: the bytes came from `&str`s, and the offsets are only
// ever written at entry boundaries, so every slice is valid UTF-8.
// We still go through the checked conversion — this is not a hot loop
// relative to the comparisons it feeds.
std::str::from_utf8(&self.coord_bytes[lo..hi]).unwrap_or("")
}
/// The `i`th change token, as stored. Meaningless unless `FLAG_TOKEN` is
/// set for that entry — [`entry`](Self::entry) is what checks the flag.
/// Panics on out of range.
#[inline]
pub fn token(&self, i: usize) -> &str {
let lo = self.token_offsets[i] as usize;
let hi = self.token_offsets[i + 1] as usize;
std::str::from_utf8(&self.token_bytes[lo..hi]).unwrap_or("")
}
/// The `i`th entry.
pub fn entry(&self, i: usize) -> MapEntry<'_> {
MapEntry {
coord: self.coord(i),
sha256: (self.flags[i] & FLAG_DIGEST != 0).then(|| self.digests[i]),
size: (self.flags[i] & FLAG_SIZE != 0).then(|| self.sizes[i]),
change_token: (self.flags[i] & FLAG_TOKEN != 0).then(|| self.token(i)),
last_seen_nanos: self.last_seen[i],
}
}
/// Every entry, ascending by coordinate.
pub fn entries(&self) -> impl Iterator<Item = MapEntry<'_>> + '_ {
(0..self.len()).map(move |i| self.entry(i))
}
/// Look one coordinate up. `O(log n)`.
pub fn get(&self, coord: &str) -> Option<MapEntry<'_>> {
let (i, hit, _) = self.lower_bound(coord);
hit.then(|| self.entry(i))
}
/// Index of the first coordinate `>= key`, whether it is an exact hit, and
/// how many string comparisons it took.
#[inline]
fn lower_bound(&self, key: &str) -> (usize, bool, u64) {
let (mut lo, mut hi) = (0usize, self.len());
let mut cmps = 0u64;
while lo < hi {
let mid = lo + (hi - lo) / 2;
cmps += 1;
if self.coord(mid) < key {
lo = mid + 1;
} else {
hi = mid;
}
}
let hit = lo < self.len() && {
cmps += 1;
self.coord(lo) == key
};
(lo, hit, cmps)
}
// ───────────────────────── the diff ─────────────────────────
/// Classify a fresh listing against this map.
///
/// Parallel merge-join over `gatling`; see the module docs for the shape and
/// [`DiffStats`] for what it cost. Byte-identical to [`diff_reference`] —
/// `diff_matches_reference_on_random_listings` is the test that says so.
///
/// `workers == 0` means one partition per core.
pub fn diff(&self, fresh: &[RemoteAsset], workers: usize) -> MirrorDiff {
let n = fresh.len();
// ── phase 1: a coordinate-ascending permutation of the listing ──
// Indices, not assets: nothing is moved and nothing is cloned.
let mut order: Vec<u32> = (0..n as u32).collect();
gatling_sort_unstable_by(&mut order, |a, b| {
fresh[*a as usize].path.cmp(&fresh[*b as usize].path)
});
// Drop repeated coordinates: the first in sort order wins. A listing
// that names the same coordinate twice names one coordinate.
let mut duplicate_coords = 0usize;
if n > 1 {
let mut w = 1usize;
for r in 1..order.len() {
if fresh[order[r] as usize].path == fresh[order[w - 1] as usize].path {
duplicate_coords += 1;
} else {
order[w] = order[r];
w += 1;
}
}
order.truncate(w);
}
let mut stats = DiffStats {
fresh_assets: n,
map_entries: self.len(),
duplicate_coords,
workers: 1,
..Default::default()
};
// ── degenerate ends, handled without waking a pool ──
if order.is_empty() {
return MirrorDiff {
removed: (0..self.len() as u32).collect(),
stats,
order,
..Default::default()
};
}
if self.is_empty() {
return MirrorDiff {
added: order.clone(),
stats,
order,
..Default::default()
};
}
// ── phase 2: partition the listing, merge-join each partition ──
let w = resolve_workers(workers, order.len());
stats.workers = w;
let base = order.len() / w;
let rem = order.len() % w;
let lo_of = |p: usize| p * base + p.min(rem);
let acc = gatling_reduce(
w,
w,
Part::default,
|acc: &mut Part, p: usize| {
let lo_f = lo_of(p);
let hi_f = lo_of(p + 1);
if lo_f == hi_f {
return;
}
// The map range this partition owns is bounded by the same
// coordinate keys that bound its listing range, so the
// partitions tile both inputs exactly once with no overlap.
let (lo_m, cl) = if p == 0 {
(0usize, 0u64)
} else {
let (i, _, c) = self.lower_bound(&fresh[order[lo_f] as usize].path);
(i, c)
};
let (hi_m, ch) = if p + 1 == w {
(self.len(), 0u64)
} else {
let (i, _, c) = self.lower_bound(&fresh[order[hi_f] as usize].path);
(i, c)
};
acc.cmps += cl + ch;
acc.reserve(hi_f - lo_f);
self.join_range(fresh, &order, lo_f, hi_f, lo_m, hi_m, acc);
},
|acc: &mut Part, part: Part| acc.absorb(part),
);
stats.key_comparisons = acc.cmps;
stats.digest_comparisons = acc.dcmps;
stats.token_comparisons = acc.tcmps;
MirrorDiff {
added: acc.added,
changed: acc.changed,
unchanged: acc.unchanged,
unproven: acc.unproven,
removed: acc.removed,
stats,
order,
}
}
/// One partition's merge-join: two sorted runs walked in lockstep, one
/// coordinate comparison per step, no hashing and no allocation beyond the
/// output vectors the partition already reserved.
#[allow(clippy::too_many_arguments)]
fn join_range(
&self,
fresh: &[RemoteAsset],
order: &[u32],
lo_f: usize,
hi_f: usize,
lo_m: usize,
hi_m: usize,
acc: &mut Part,
) {
let (mut f, mut m) = (lo_f, lo_m);
while f < hi_f && m < hi_m {
let fi = order[f] as usize;
acc.cmps += 1;
match fresh[fi].path.as_str().cmp(self.coord(m)) {
std::cmp::Ordering::Less => {
// The listing has a coordinate the map does not.
acc.added.push(order[f]);
f += 1;
}
std::cmp::Ordering::Greater => {
// The map has a coordinate the listing does not.
acc.removed.push(m as u32);
m += 1;
}
std::cmp::Ordering::Equal => {
self.classify(fresh, fi, m, order[f], acc);
f += 1;
m += 1;
}
}
}
while f < hi_f {
acc.added.push(order[f]);
f += 1;
}
while m < hi_m {
acc.removed.push(m as u32);
m += 1;
}
}
/// The three-way verdict for one coordinate present on both sides. The rule
/// and the reason it is this rule are in the module docs.
#[inline]
fn classify(&self, fresh: &[RemoteAsset], fi: usize, mi: usize, out_idx: u32, acc: &mut Part) {
let asset = &fresh[fi];
let flags = self.flags[mi];
// 1. A digest on both sides is the strongest proof of sameness there
// is, and the only one that is comparable ACROSS sources.
if flags & FLAG_DIGEST != 0 {
if let Some(fresh_digest) = asset.sha256.as_deref().and_then(decode_sha256) {
acc.dcmps += 1;
if fresh_digest == self.digests[mi] {
acc.unchanged.push(out_idx);
} else {
acc.changed.push(out_idx);
}
return;
}
}
// 2. A CHANGE TOKEN on both sides, with the SAME source scope, is the
// other proof there is — and it is the only one 15 of holger's 17
// backends can supply. It is deliberately below the digest: a digest
// is comparable across sources and a token is not, so where both
// exist the stronger one answers.
//
// Scope is checked, not assumed. Two sources holding byte-identical
// artifacts publish DIFFERENT tokens as a matter of course, so a
// cross-source comparison would report `changed` for ever about
// something that never moved. When the scopes differ the honest
// verdict is neither `changed` nor `unchanged`; it falls through to
// the size rule and then to `unproven`.
if flags & FLAG_TOKEN != 0 {
if let Some(fresh_token) = asset.change_token.as_deref() {
let stored = self.token(mi);
if change_tokens_comparable(fresh_token, stored) {
acc.tcmps += 1;
if fresh_token == stored {
acc.unchanged.push(out_idx);
} else {
acc.changed.push(out_idx);
}
return;
}
}
}
// 3. A size difference proves a CHANGE even with no digest anywhere.
if flags & FLAG_SIZE != 0 {
if let Some(fresh_size) = asset.size {
if fresh_size != self.sizes[mi] {
acc.changed.push(out_idx);
return;
}
}
}
// 4. Anything else is not proved. It is NOT `unchanged`: equal sizes
// with no digest do not exclude a same-size edit, and a token on one
// side only proves nothing in either direction — a source that
// STOPPED publishing tokens must not read as "everything changed",
// and one that just started must not read as "nothing changed".
acc.unproven.push(out_idx);
}
// ───────────────────────── adopting a sync ─────────────────────────
/// Replace this map with what the fresh listing says, so the next cycle
/// diffs against the truth.
///
/// `failed` names the fresh indices the mirror could **not** fetch this
/// cycle. Their previous entry is carried over unchanged when they had one
/// (so the next diff still reports them as needing work) and they are
/// omitted when they did not — the map never claims a coordinate whose
/// bytes the mirror does not hold. Pass `&[]` when the whole cycle
/// succeeded.
///
/// `diff` must be the diff this map produced for this `fresh`; its cached
/// sort order is reused, so adopting costs no second sort.
pub fn adopt(
&mut self,
fresh: &[RemoteAsset],
diff: &MirrorDiff,
failed: &[u32],
now_nanos: i64,
) {
let mut failed_sorted: Vec<u32> = failed.to_vec();
failed_sorted.sort_unstable();
let n = diff.order.len();
let mut next = Self {
upstream: std::mem::take(&mut self.upstream),
repository: std::mem::take(&mut self.repository),
coord_bytes: Vec::with_capacity(self.coord_bytes.len().max(n * 48)),
coord_offsets: Vec::with_capacity(n + 1),
digests: Vec::with_capacity(n),
sizes: Vec::with_capacity(n),
token_bytes: Vec::with_capacity(self.token_bytes.len().max(n * 8)),
token_offsets: Vec::with_capacity(n + 1),
flags: Vec::with_capacity(n),
last_seen: Vec::with_capacity(n),
};
next.coord_offsets.push(0);
next.token_offsets.push(0);
for &fi in &diff.order {
let asset = &fresh[fi as usize];
if failed_sorted.binary_search(&fi).is_ok() {
// Not fetched: carry the previous entry verbatim, or drop it.
if let Some(prev) = self.get(&asset.path) {
next.push_raw(
prev.coord,
prev.sha256,
prev.size,
prev.change_token,
prev.last_seen_nanos,
);
}
continue;
}
next.push_raw(
&asset.path,
asset.sha256.as_deref().and_then(decode_sha256),
asset.size,
asset.change_token.as_deref(),
now_nanos,
);
}
*self = next;
}
/// **Forget every coordinate `keep` says no to** (stinky SB #35, the twin's
/// backfill): an entry whose bytes the store no longer holds is dropped, so
/// the next diff reports it `added` and the cycle fetches it again — the map
/// never claims a coordinate whose bytes the mirror does not hold. Order is
/// kept. Answers how many were dropped.
pub fn retain(&mut self, keep: impl Fn(&str) -> bool) -> usize {
let mut next = Self::empty(
std::mem::take(&mut self.upstream),
std::mem::take(&mut self.repository),
);
let mut dropped = 0;
for i in 0..self.len() {
let e = self.entry(i);
if keep(e.coord) {
next.push_raw(e.coord, e.sha256, e.size, e.change_token, e.last_seen_nanos);
} else {
dropped += 1;
}
}
*self = next;
dropped
}
/// Append one entry. The caller guarantees ascending coordinate order —
/// every call site walks a sorted permutation.
#[inline]
fn push_raw(
&mut self,
coord: &str,
sha256: Option<[u8; 32]>,
size: Option<u64>,
change_token: Option<&str>,
ts: i64,
) {
self.coord_bytes.extend_from_slice(coord.as_bytes());
self.coord_offsets.push(self.coord_bytes.len() as u32);
let mut flags = 0u8;
self.digests.push(match sha256 {
Some(d) => {
flags |= FLAG_DIGEST;
d
}
None => [0u8; 32],
});
self.sizes.push(match size {
Some(s) => {
flags |= FLAG_SIZE;
s
}
None => 0,
});
// An ABSENT token appends no bytes and leaves the offset where it was,
// so an empty slice and "no token" are the same on disk and in memory —
// and the FLAG is what tells them apart, exactly as it does for the
// digest. An empty token that survived would compare equal to another
// empty token, which is the one thing a change token must never do.
if let Some(t) = change_token {
flags |= FLAG_TOKEN;
self.token_bytes.extend_from_slice(t.as_bytes());
}
self.token_offsets.push(self.token_bytes.len() as u32);
self.flags.push(flags);
self.last_seen.push(ts);
}
// ───────────────────────── persistence ─────────────────────────
/// The file this map lives in under `dir`.
pub fn path_in(dir: &Path, upstream: &str, repository: &str) -> PathBuf {
dir.join(format!("{}.map", slug(upstream, repository)))
}
/// Load the map for `(upstream, repository)` from `dir`.
///
/// **A missing file is not an error**: it is a mirror that has never synced
/// this repository, and the honest answer is [`ChangeMap::empty`], which
/// makes the first cycle a full fetch. A file that exists and is corrupt,
/// or carries a format version this reader does not understand, IS an
/// error — silently treating it as empty would turn one bad file into a
/// silent full re-download.
pub fn load(dir: &Path, upstream: &str, repository: &str) -> Result<Self> {
let path = Self::path_in(dir, upstream, repository);
if !path.exists() {
return Ok(Self::empty(upstream, repository));
}
Self::read_file(&path)
}
/// Read a map from an explicit path.
///
/// The file is JSON lines: one header object
/// (`{"format":"holger.mirror.map/v3","upstream":..,"repository":..,"entries":n}`)
/// then one `[coord, sha256|null, size|null, last_seen_nanos, token|null]`
/// array per entry, ascending by coordinate.
pub fn read_file(path: &Path) -> Result<Self> {
use std::io::BufRead as _;
let file = File::open(path)
.with_context(|| format!("mirror change map: open {}", path.display()))?;
let mut lines = std::io::BufReader::new(file).lines();
let header: serde_json::Value = match lines.next() {
Some(l) => serde_json::from_str(&l?)
.with_context(|| format!("mirror change map: bad header in {}", path.display()))?,
None => bail!(
"mirror change map {}: empty file — it is not a change map",
path.display()
),
};
match header.get("format").and_then(|v| v.as_str()) {
Some(MAP_FORMAT) => {}
Some(other) => bail!(
"mirror change map {}: format {other}, this reader understands only {MAP_FORMAT}",
path.display()
),
None => bail!(
"mirror change map {}: no format header — it is not a change map",
path.display()
),
}
let hs = |k: &str| {
header
.get(k)
.and_then(|v| v.as_str())
.unwrap_or_default()
.to_string()
};
let mut map = Self::empty(hs("upstream"), hs("repository"));
if let Some(n) = header.get("entries").and_then(|v| v.as_u64()) {
map.reserve(n as usize);
}
for (row, line) in lines.enumerate() {
let line =
line.with_context(|| format!("mirror change map: read {}", path.display()))?;
if line.is_empty() {
continue;
}
let (coord, hex, size, seen, token): (
String,
Option<String>,
Option<u64>,
i64,
Option<String>,
) = serde_json::from_str(&line)
.with_context(|| format!("mirror change map {}: bad row {row}", path.display()))?;
let digest = match hex {
None => None,
Some(hex) => match decode_sha256(&hex) {
Some(d) => Some(d),
None => bail!(
"mirror change map {}: coordinate {coord:?} carries {hex:?}, which is not \
64 lowercase hex chars",
path.display()
),
},
};
// A present but EMPTY token is dropped to absent: an empty token
// compares equal to another empty token, and that is the one
// comparison a change token must never win.
let token = token.filter(|t| !t.is_empty());
map.push_raw(&coord, digest, size, token.as_deref(), seen);
}
// The whole diff rests on the file being sorted and unique. Verify it
// rather than trust it.
for i in 1..map.len() {
if map.coord(i - 1) >= map.coord(i) {
bail!(
"mirror change map {}: coordinates are not strictly ascending at row {i} \
({:?} then {:?})",
path.display(),
map.coord(i - 1),
map.coord(i)
);
}
}
Ok(map)
}
/// Reserve room for `n` entries up front.
fn reserve(&mut self, n: usize) {
self.coord_bytes.reserve(n * 48);
self.coord_offsets.reserve(n);
self.digests.reserve(n);
self.sizes.reserve(n);
self.token_bytes.reserve(n * 8);
self.token_offsets.reserve(n);
self.flags.reserve(n);
self.last_seen.reserve(n);
}
/// Write the map for `(upstream, repository)` into `dir`.
///
/// Writes a temporary beside the target and renames it into place, so a
/// crash mid-write leaves the previous map intact rather than a truncated
/// one. A truncated change map is worse than none: it would report every
/// coordinate past the truncation as `removed`.
pub fn store(&self, dir: &Path) -> Result<()> {
fs::create_dir_all(dir)
.with_context(|| format!("mirror change map: create {}", dir.display()))?;
let final_path = Self::path_in(dir, &self.upstream, &self.repository);
let tmp_path = final_path.with_extension("map.tmp");
{
let file = File::create(&tmp_path)
.with_context(|| format!("mirror change map: create {}", tmp_path.display()))?;
let mut w = BufWriter::new(file);
let header = serde_json::json!({
"format": MAP_FORMAT,
"upstream": self.upstream,
"repository": self.repository,
"entries": self.len(),
});
writeln!(w, "{header}").context("mirror change map: write header")?;
let mut hexbuf = [0u8; 64];
for i in 0..self.len() {
let hex = (self.flags[i] & FLAG_DIGEST != 0).then(|| {
encode_sha256(&self.digests[i], &mut hexbuf);
// Written from a fixed hex alphabet, so it is ASCII.
std::str::from_utf8(&hexbuf).unwrap_or("").to_string()
});
let size = (self.flags[i] & FLAG_SIZE != 0).then_some(self.sizes[i]);
let token = (self.flags[i] & FLAG_TOKEN != 0).then(|| self.token(i));
let row = serde_json::json!([self.coord(i), hex, size, self.last_seen[i], token]);
writeln!(w, "{row}").context("mirror change map: write row")?;
}
w.flush().context("mirror change map: flush")?;
}
fs::rename(&tmp_path, &final_path).with_context(|| {
format!(
"mirror change map: rename {} -> {}",
tmp_path.display(),
final_path.display()
)
})?;
Ok(())
}
}
// ───────────────────────── the reference diff ─────────────────────────
/// The serial `HashMap` diff: the oracle every test checks
/// [`ChangeMap::diff`] against, and the baseline the scaling measurement counts
/// against.
///
/// Same verdicts, same rule, same buckets — and the buckets come out in the same
/// coordinate order, so the two are comparable field by field. It is kept
/// public rather than hidden in `#[cfg(test)]` because a parallel hot path whose
/// simple twin is not runnable in production is a parallel hot path nobody can
/// bisect against when it disagrees with reality.
pub fn diff_reference(map: &ChangeMap, fresh: &[RemoteAsset]) -> MirrorDiff {
let mut index: HashMap<&str, u32> = HashMap::with_capacity(map.len());
for i in 0..map.len() {
index.insert(map.coord(i), i as u32);
}
let n = fresh.len();
let mut order: Vec<u32> = (0..n as u32).collect();
order.sort_unstable_by(|a, b| fresh[*a as usize].path.cmp(&fresh[*b as usize].path));
let mut duplicate_coords = 0usize;
if n > 1 {
let mut w = 1usize;
for r in 1..order.len() {
if fresh[order[r] as usize].path == fresh[order[w - 1] as usize].path {
duplicate_coords += 1;
} else {
order[w] = order[r];
w += 1;
}
}
order.truncate(w);
}
let mut out = MirrorDiff {
stats: DiffStats {
fresh_assets: n,
map_entries: map.len(),
duplicate_coords,
workers: 1,
// One hash to build each map slot, one to probe for each asset.
key_hashes: map.len() as u64 + order.len() as u64,
..Default::default()
},
..Default::default()
};
let mut seen = vec![false; map.len()];
let mut acc = Part::default();
for &fi in &order {
match index.get(fresh[fi as usize].path.as_str()) {
Some(&mi) => {
seen[mi as usize] = true;
map.classify(fresh, fi as usize, mi as usize, fi, &mut acc);
}
None => acc.added.push(fi),
}
}
out.added = acc.added;
out.changed = acc.changed;
out.unchanged = acc.unchanged;
out.unproven = acc.unproven;
// A lower bound, and labelled as one: a successful `HashMap` probe still
// compares the key for real once the bucket is found, so the reference pays
// at least one string comparison per matched coordinate ON TOP of hashing
// every key. Collision chains make the true number larger, never smaller.
out.stats.key_comparisons =
(out.changed.len() + out.unchanged.len() + out.unproven.len()) as u64;
out.removed = (0..map.len() as u32)
.filter(|i| !seen[*i as usize])
.collect();
out.stats.digest_comparisons = acc.dcmps;
out.stats.token_comparisons = acc.tcmps;
out.order = order;
out
}
// ───────────────────────── internals ─────────────────────────
/// One partition's output. Merged in ascending partition order, which is what
/// keeps every bucket coordinate-sorted.
#[derive(Debug, Default)]
struct Part {
added: Vec<u32>,
changed: Vec<u32>,
unchanged: Vec<u32>,
unproven: Vec<u32>,
removed: Vec<u32>,
cmps: u64,
dcmps: u64,
tcmps: u64,
}
impl Part {
/// Preallocate for the common case: a steady-state mirror cycle is almost
/// entirely `unchanged`, so that bucket gets the partition's whole span and
/// the rest start small. One allocation per bucket, not one per push.
fn reserve(&mut self, span: usize) {
self.unchanged.reserve(span);
let small = (span / 16).max(8);
self.added.reserve(small);
self.changed.reserve(small);
self.unproven.reserve(small);
self.removed.reserve(small);
}
fn absorb(&mut self, mut other: Part) {
self.added.append(&mut other.added);
self.changed.append(&mut other.changed);
self.unchanged.append(&mut other.unchanged);
self.unproven.append(&mut other.unproven);
self.removed.append(&mut other.removed);
self.cmps += other.cmps;
self.dcmps += other.dcmps;
self.tcmps += other.tcmps;
}
}
#[inline]
fn resolve_workers(requested: usize, units: usize) -> usize {
let w = if requested == 0 {
default_workers()
} else {
requested
};
w.max(1).min(units.max(1))
}
/// Decode 64 lowercase hex chars into 32 bytes. Anything else is `None` — a
/// malformed digest is treated as "no digest", never as a digest that happens
/// not to match, because the second would report a spurious `changed` forever.
///
/// Uppercase is accepted on read: the connectors normalise to lowercase, but a
/// map hand-written by an operator should not be silently unreadable.
#[inline]
fn decode_sha256(hex: &str) -> Option<[u8; 32]> {
let b = hex.as_bytes();
if b.len() != 64 {
return None;
}
let mut out = [0u8; 32];
for (i, slot) in out.iter_mut().enumerate() {
let hi = hexval(b[2 * i])?;
let lo = hexval(b[2 * i + 1])?;
*slot = (hi << 4) | lo;
}
Some(out)
}
#[inline]
fn hexval(c: u8) -> Option<u8> {
match c {
b'0'..=b'9' => Some(c - b'0'),
b'a'..=b'f' => Some(c - b'a' + 10),
b'A'..=b'F' => Some(c - b'A' + 10),
_ => None,
}
}
/// Write 32 bytes as 64 lowercase hex chars into a caller-owned buffer — no
/// `String` per row on the store path.
#[inline]
fn encode_sha256(d: &[u8; 32], out: &mut [u8; 64]) {
const HEX: &[u8; 16] = b"0123456789abcdef";
for (i, byte) in d.iter().enumerate() {
out[2 * i] = HEX[(byte >> 4) as usize];
out[2 * i + 1] = HEX[(byte & 0x0f) as usize];
}
}
/// A filesystem-safe, collision-free name for one `(upstream, repository)`.
///
/// The readable part is for the operator who has to find the file; the blake3
/// suffix is what makes it correct, because two different pairs can sanitise to
/// the same readable part (`a/b` + `c` and `a` + `b/c`) and a change map served
/// under the wrong pair's name would report the entire repository as changed.
fn slug(upstream: &str, repository: &str) -> String {
let mut readable = String::with_capacity(upstream.len() + repository.len() + 1);
for part in [upstream, repository] {
if !readable.is_empty() {
readable.push('-');
}
for ch in part.chars() {
readable.push(match ch {
'a'..='z' | 'A'..='Z' | '0'..='9' | '.' | '_' | '-' => ch,
_ => '_',
});
}
}
readable.truncate(96);
// Length-prefixed, so `a/b`+`c` and `a`+`b/c` cannot hash alike either.
let mut h = blake3::Hasher::new();
h.update(b"holger.mirror.map.v1\x00");
h.update(&(upstream.len() as u64).to_le_bytes());
h.update(upstream.as_bytes());
h.update(&(repository.len() as u64).to_le_bytes());
h.update(repository.as_bytes());
let digest = h.finalize();
format!("{readable}-{}", &digest.to_hex().as_str()[..16])
}
// ───────────────────────────── tests ─────────────────────────────
// ─── the change token's source scope (used only by the mirror) ───
/// The separator between a change token's SCOPE and its opaque part.
///
/// A printable ASCII character on purpose: a change map is a column of strings
/// an operator can open in any Arrow tool and read, and a control byte there
/// would be unreadable. A scope may not contain it (see
/// [`scope_change_token`]); the opaque part may, because the split takes the
/// FIRST occurrence.
pub const CHANGE_TOKEN_SCOPE_SEP: char = '|';
/// Stamp a source scope onto a backend's raw token, producing the
/// `"<scope>|<opaque>"` form [`traits::RemoteAsset::change_token`] carries.
///
/// `scope` names the source the token came FROM — for a holger upstream, its
/// endpoint and repository. It may not be empty and may not contain
/// [`CHANGE_TOKEN_SCOPE_SEP`], because the split that reads it back takes the
/// first separator; a scope that contained one would be read back short, and a
/// short scope is a scope that can collide with another source's.
///
/// `None` when either half is unusable, or when the result would be longer than
/// [`traits::MAX_CHANGE_TOKEN_LEN`]. A token that cannot be scoped is not carried:
/// absent is the honest answer, and the diff already knows what to do with it.
pub fn scope_change_token(scope: &str, raw: &str) -> Option<String> {
let scope = scope.trim();
if scope.is_empty()
|| scope.contains(CHANGE_TOKEN_SCOPE_SEP)
|| scope.chars().any(char::is_control)
{
return None;
}
let raw = traits::normalize_change_token(raw)?;
let mut out = String::with_capacity(scope.len() + 1 + raw.len());
out.push_str(scope);
out.push(CHANGE_TOKEN_SCOPE_SEP);
out.push_str(&raw);
traits::normalize_change_token(&out)
}
/// Split a scoped change token into `(scope, opaque)`. `None` when the value
/// carries no scope at all, which is a token that must not be compared with
/// anything.
///
/// Splits at the FIRST separator, so an opaque part containing one survives
/// intact.
pub fn split_change_token(token: &str) -> Option<(&str, &str)> {
let (scope, rest) = token.split_once(CHANGE_TOKEN_SCOPE_SEP)?;
if scope.is_empty() || rest.is_empty() {
return None;
}
Some((scope, rest))
}
/// Whether two scoped change tokens are comparable at all — that is, whether
/// they came from the SAME source.
///
/// This is the guard the whole concept rests on. Two sources holding
/// byte-identical artifacts publish different tokens as a matter of course
/// (znippy's fold binds the relative path, and two mirrors lay out their
/// archives differently), so comparing across sources would report `changed`
/// forever about artifacts that never move. When this says `false` the honest
/// verdict is `unproven`, never `changed` and never `unchanged`.
pub fn change_tokens_comparable(a: &str, b: &str) -> bool {
match (split_change_token(a), split_change_token(b)) {
(Some((sa, _)), Some((sb, _))) => sa == sb,
_ => false,
}
}
#[cfg(test)]
mod change_token_scope_tests {
use super::*;
/// The scope is part of the value, and a scope that could not be read back
/// unambiguously is refused outright rather than carried short.
#[test]
fn a_scoped_token_carries_its_source_and_splits_back_exactly() {
let t = scope_change_token("http://up:50051/pkgs", "abcdef").expect("scoped");
assert_eq!(t, "http://up:50051/pkgs|abcdef");
assert_eq!(
split_change_token(&t),
Some(("http://up:50051/pkgs", "abcdef"))
);
// An opaque half containing the separator survives: the split takes the
// FIRST one.
let odd = scope_change_token("src", "a|b").expect("scoped");
assert_eq!(split_change_token(&odd), Some(("src", "a|b")));
// A scope that contains the separator could be read back short, and a
// short scope can collide with another source's.
assert_eq!(scope_change_token("a|b", "x"), None);
assert_eq!(
scope_change_token("", "x"),
None,
"an unnamed source is not a scope"
);
assert_eq!(
scope_change_token("src", ""),
None,
"a scope with nothing behind it"
);
}
/// **The guard the whole concept rests on.** Two tokens are comparable only
/// when they came from the same source; an unscoped value is comparable
/// with nothing at all.
#[test]
fn tokens_are_comparable_only_within_one_source() {
let a = scope_change_token("http://a/pkgs", "same").unwrap();
let b = scope_change_token("http://b/pkgs", "same").unwrap();
let a2 = scope_change_token("http://a/pkgs", "other").unwrap();
assert!(
change_tokens_comparable(&a, &a2),
"one source, two values: comparable"
);
assert_ne!(a, a2);
assert!(
!change_tokens_comparable(&a, &b),
"identical opaque halves from two sources must NOT be comparable — two servers \
holding the same artifact publish different tokens as a matter of course"
);
assert!(
!change_tokens_comparable("bare", &a),
"an unscoped value compares with nothing"
);
assert!(!change_tokens_comparable(&a, "bare"));
}
}
#[cfg(test)]
mod tests {
use super::*;
fn asset(path: &str, sha: Option<&str>, size: Option<u64>) -> RemoteAsset {
RemoteAsset {
path: path.to_string(),
download_url: format!("https://upstream.invalid/{path}"),
content_type: None,
size,
sha256: sha.map(|s| s.to_string()),
change_token: None,
}
}
fn digest(seed: u8) -> String {
let mut s = String::with_capacity(64);
for i in 0..32u8 {
s.push_str(&format!("{:02x}", seed.wrapping_add(i)));
}
s
}
/// Build a map by adopting a listing wholesale — the normal way a map comes
/// into being after the first successful cycle.
fn map_of(upstream: &str, repo: &str, assets: &[RemoteAsset]) -> ChangeMap {
let mut m = ChangeMap::empty(upstream, repo);
let d = m.diff(assets, 0);
m.adopt(assets, &d, &[], 1_000);
m
}
#[test]
fn empty_map_reports_every_coordinate_as_added() {
let fresh = vec![
asset("b/2.jar", Some(&digest(2)), Some(20)),
asset("a/1.jar", Some(&digest(1)), Some(10)),
];
let m = ChangeMap::empty("nexus", "maven-central");
let d = m.diff(&fresh, 0);
assert_eq!(d.added.len(), 2);
assert!(d.changed.is_empty() && d.unchanged.is_empty() && d.removed.is_empty());
// Added comes out coordinate-ascending, so "a/1" precedes "b/2".
assert_eq!(fresh[d.added[0] as usize].path, "a/1.jar");
assert_eq!(d.to_fetch(UnprovenPolicy::Refetch).len(), 2);
assert!(!d.is_clean());
}
#[test]
fn a_second_identical_listing_is_clean_and_fetches_nothing() {
let fresh = vec![
asset("a/1.jar", Some(&digest(1)), Some(10)),
asset("b/2.jar", Some(&digest(2)), Some(20)),
];
let m = map_of("nexus", "maven-central", &fresh);
let d = m.diff(&fresh, 0);
assert!(d.is_clean(), "unchanged upstream must produce a clean diff");
assert_eq!(d.unchanged.len(), 2);
assert!(d.to_fetch(UnprovenPolicy::Refetch).is_empty());
}
#[test]
fn a_new_digest_on_a_known_coordinate_is_changed() {
let before = vec![asset("a/1.jar", Some(&digest(1)), Some(10))];
let m = map_of("holger-upstream", "rust", &before);
let after = vec![asset("a/1.jar", Some(&digest(9)), Some(10))];
let d = m.diff(&after, 0);
assert_eq!(d.changed.len(), 1, "same size, different digest ⇒ changed");
assert!(d.unchanged.is_empty() && d.unproven.is_empty());
}
#[test]
fn a_dropped_coordinate_is_removed_and_a_new_one_is_added() {
let before = vec![
asset("a/1.jar", Some(&digest(1)), Some(10)),
asset("b/2.jar", Some(&digest(2)), Some(20)),
];
let m = map_of("nexus", "maven", &before);
let after = vec![
asset("b/2.jar", Some(&digest(2)), Some(20)),
asset("c/3.jar", Some(&digest(3)), Some(30)),
];
let d = m.diff(&after, 0);
assert_eq!(d.removed.len(), 1);
assert_eq!(m.coord(d.removed[0] as usize), "a/1.jar");
assert_eq!(d.added.len(), 1);
assert_eq!(after[d.added[0] as usize].path, "c/3.jar");
assert_eq!(d.unchanged.len(), 1);
}
#[test]
fn no_digest_and_equal_size_is_unproven_never_unchanged() {
let before = vec![asset("a/1.jar", None, Some(10))];
let m = map_of("artifactory", "libs-release", &before);
let after = vec![asset("a/1.jar", None, Some(10))];
let d = m.diff(&after, 0);
assert!(
d.unchanged.is_empty(),
"equal size without a digest must never be reported as PROVED unchanged"
);
assert_eq!(d.unproven.len(), 1);
// Default policy re-fetches it; TrustSize is the opt-in that does not.
assert_eq!(d.to_fetch(UnprovenPolicy::Refetch).len(), 1);
assert!(d.to_fetch(UnprovenPolicy::TrustSize).is_empty());
assert!(!d.is_clean(), "an unproven coordinate is not a clean cycle");
}
#[test]
fn no_digest_but_a_different_size_is_proof_of_change() {
let before = vec![asset("a/1.jar", None, Some(10))];
let m = map_of("artifactory", "libs-release", &before);
let after = vec![asset("a/1.jar", None, Some(11))];
let d = m.diff(&after, 0);
assert_eq!(d.changed.len(), 1);
assert!(d.unproven.is_empty());
}
#[test]
fn a_digest_appearing_for_the_first_time_is_unproven_not_changed() {
// The map was written before the connectors published sha256; the fresh
// listing has one. There is nothing to compare it to, so it is not
// PROVED different — it is unproven, and the default re-fetches.
let before = vec![asset("a/1.jar", None, Some(10))];
let m = map_of("nexus", "maven", &before);
let after = vec![asset("a/1.jar", Some(&digest(4)), Some(10))];
let d = m.diff(&after, 0);
assert!(d.changed.is_empty());
assert_eq!(d.unproven.len(), 1);
}
#[test]
fn a_source_that_stops_publishing_digests_falls_back_to_unproven() {
let before = vec![asset("a/1.jar", Some(&digest(1)), Some(10))];
let m = map_of("nexus", "maven", &before);
let after = vec![asset("a/1.jar", None, Some(10))];
let d = m.diff(&after, 0);
assert_eq!(d.unproven.len(), 1);
assert!(d.unchanged.is_empty());
}
#[test]
fn a_malformed_digest_is_treated_as_absent_not_as_a_mismatch() {
let before = vec![asset("a/1.jar", Some(&digest(1)), Some(10))];
let m = map_of("nexus", "maven", &before);
let after = vec![asset("a/1.jar", Some("not-a-digest"), Some(10))];
let d = m.diff(&after, 0);
assert!(
d.changed.is_empty(),
"garbage must not masquerade as a mismatch"
);
assert_eq!(d.unproven.len(), 1);
}
#[test]
fn an_absent_size_never_proves_anything() {
let before = vec![asset("a/1.jar", None, None)];
let m = map_of("nexus", "maven", &before);
let after = vec![asset("a/1.jar", None, Some(10))];
let d = m.diff(&after, 0);
assert_eq!(d.unproven.len(), 1);
assert!(d.changed.is_empty());
}
#[test]
fn a_repeated_coordinate_in_a_listing_is_one_coordinate() {
let fresh = vec![
asset("a/1.jar", Some(&digest(1)), Some(10)),
asset("a/1.jar", Some(&digest(1)), Some(10)),
];
let m = ChangeMap::empty("nexus", "maven");
let d = m.diff(&fresh, 0);
assert_eq!(d.stats.duplicate_coords, 1);
assert_eq!(d.added.len(), 1);
}
#[test]
fn adopt_carries_a_failed_fetch_over_so_the_next_cycle_retries_it() {
let before = vec![asset("a/1.jar", Some(&digest(1)), Some(10))];
let mut m = map_of("nexus", "maven", &before);
// Upstream changed it, but the fetch failed.
let after = vec![asset("a/1.jar", Some(&digest(7)), Some(10))];
let d = m.diff(&after, 0);
assert_eq!(d.changed.len(), 1);
let failed = d.changed.clone();
m.adopt(&after, &d, &failed, 2_000);
// The map must still hold the OLD digest, so the next diff still says
// "changed" — a failed fetch must not be recorded as a success.
assert_eq!(m.get("a/1.jar").unwrap().sha256, decode_sha256(&digest(1)));
let d2 = m.diff(&after, 0);
assert_eq!(d2.changed.len(), 1, "the retry must still be pending");
}
#[test]
fn adopt_omits_a_new_coordinate_whose_fetch_failed() {
let mut m = ChangeMap::empty("nexus", "maven");
let fresh = vec![asset("a/1.jar", Some(&digest(1)), Some(10))];
let d = m.diff(&fresh, 0);
m.adopt(&fresh, &d, &[0], 2_000);
assert!(
m.is_empty(),
"the map must never claim a coordinate the mirror does not hold"
);
}
#[test]
fn adopt_refreshes_last_seen_on_an_unchanged_coordinate() {
let fresh = vec![asset("a/1.jar", Some(&digest(1)), Some(10))];
let mut m = map_of("nexus", "maven", &fresh);
assert_eq!(m.get("a/1.jar").unwrap().last_seen_nanos, 1_000);
let d = m.diff(&fresh, 0);
m.adopt(&fresh, &d, &[], 5_000);
assert_eq!(m.get("a/1.jar").unwrap().last_seen_nanos, 5_000);
}
#[test]
fn store_then_load_round_trips_every_column_including_the_absences() {
let dir = tempfile::tempdir().unwrap();
let fresh = vec![
asset("a/1.jar", Some(&digest(1)), Some(10)),
asset("b/2.jar", None, Some(20)),
asset("c/3.jar", Some(&digest(3)), None),
asset("d/4.jar", None, None),
];
let m = map_of("nexus-prod", "maven-central", &fresh);
m.store(dir.path()).unwrap();
let back = ChangeMap::load(dir.path(), "nexus-prod", "maven-central").unwrap();
assert_eq!(back.len(), m.len());
assert_eq!(back.upstream(), "nexus-prod");
assert_eq!(back.repository(), "maven-central");
for i in 0..m.len() {
assert_eq!(back.entry(i), m.entry(i), "row {i} did not round-trip");
}
// And the reloaded map produces the identical verdict.
assert!(back.diff(&fresh, 0).is_clean() == m.diff(&fresh, 0).is_clean());
}
#[test]
fn a_map_that_was_never_synced_loads_as_empty_rather_than_failing() {
let dir = tempfile::tempdir().unwrap();
let m = ChangeMap::load(dir.path(), "nexus", "never-synced").unwrap();
assert!(m.is_empty());
assert_eq!(m.upstream(), "nexus");
}
#[test]
fn a_corrupt_map_file_is_an_error_not_a_silent_full_resync() {
let dir = tempfile::tempdir().unwrap();
let p = ChangeMap::path_in(dir.path(), "nexus", "maven");
std::fs::write(&p, b"this is not an arrow ipc stream").unwrap();
let err = ChangeMap::load(dir.path(), "nexus", "maven").unwrap_err();
let msg = format!("{err:#}");
assert!(msg.contains("mirror change map"), "unhelpful error: {msg}");
}
#[test]
fn two_upstreams_with_the_same_repository_name_get_separate_files() {
let dir = tempfile::tempdir().unwrap();
let a = vec![asset("a/1.jar", Some(&digest(1)), Some(10))];
let b = vec![asset("z/9.jar", Some(&digest(9)), Some(90))];
map_of("nexus-a", "maven", &a).store(dir.path()).unwrap();
map_of("nexus-b", "maven", &b).store(dir.path()).unwrap();
let ma = ChangeMap::load(dir.path(), "nexus-a", "maven").unwrap();
let mb = ChangeMap::load(dir.path(), "nexus-b", "maven").unwrap();
assert_eq!(ma.len(), 1);
assert_eq!(mb.len(), 1);
assert!(ma.get("a/1.jar").is_some());
assert!(mb.get("z/9.jar").is_some());
}
#[test]
fn a_slash_in_an_upstream_name_cannot_collide_with_one_in_the_repository() {
assert_ne!(slug("a/b", "c"), slug("a", "b/c"));
}
// ── the parallel path against the serial oracle ──
/// Deterministic pseudo-random listings, so a failure is reproducible from
/// the seed printed in the assertion rather than from luck.
fn xorshift(state: &mut u64) -> u64 {
let mut x = *state;
x ^= x << 13;
x ^= x >> 7;
x ^= x << 17;
*state = x;
x
}
fn scenario(seed: u64, n: usize) -> (ChangeMap, Vec<RemoteAsset>) {
let mut s = seed | 1;
let base: Vec<RemoteAsset> = (0..n)
.map(|i| {
let d = (xorshift(&mut s) % 251) as u8;
asset(
&format!(
"org/example/pkg{:06}/artifact-{}.jar",
xorshift(&mut s) % 1_000_000,
i
),
(!xorshift(&mut s).is_multiple_of(4))
.then(|| digest(d))
.as_deref(),
(!xorshift(&mut s).is_multiple_of(8)).then(|| xorshift(&mut s) % 100_000),
)
})
.collect();
let m = map_of("nexus", "maven", &base);
// Mutate the listing the way an upstream does between cycles.
let mut fresh = Vec::with_capacity(n);
for (i, a) in base.iter().enumerate() {
match xorshift(&mut s) % 10 {
0 => {} // dropped upstream ⇒ removed
1 => fresh.push(asset(
&a.path,
Some(&digest((xorshift(&mut s) % 251) as u8)),
a.size,
)), // re-published
2 => fresh.push(asset(&a.path, a.sha256.as_deref(), Some(i as u64 + 7))), // resized
3 => fresh.push(asset(&a.path, None, a.size)), // digest withdrawn
_ => fresh.push(a.clone()),
}
if xorshift(&mut s).is_multiple_of(12) {
fresh.push(asset(
&format!(
"org/example/new{:06}/artifact-{i}.jar",
xorshift(&mut s) % 999_983
),
Some(&digest((xorshift(&mut s) % 251) as u8)),
Some(1234),
));
}
}
(m, fresh)
}
#[test]
fn diff_matches_reference_on_random_listings() {
for seed in [1u64, 2, 3, 12345, 999_331] {
let (m, fresh) = scenario(seed, 2_000);
let want = diff_reference(&m, &fresh);
for workers in [1usize, 2, 3, 7, 16] {
let got = m.diff(&fresh, workers);
assert_eq!(
got.added, want.added,
"added, seed {seed}, workers {workers}"
);
assert_eq!(
got.changed, want.changed,
"changed, seed {seed}, workers {workers}"
);
assert_eq!(
got.unchanged, want.unchanged,
"unchanged, seed {seed}, workers {workers}"
);
assert_eq!(
got.unproven, want.unproven,
"unproven, seed {seed}, workers {workers}"
);
assert_eq!(
got.removed, want.removed,
"removed, seed {seed}, workers {workers}"
);
assert_eq!(got.stats.duplicate_coords, want.stats.duplicate_coords);
}
}
}
#[test]
fn worker_count_never_changes_the_verdict_or_its_order() {
let (m, fresh) = scenario(4242, 5_000);
let one = m.diff(&fresh, 1);
for workers in [2usize, 5, 13, 32, 0] {
let many = m.diff(&fresh, workers);
assert_eq!(one.added, many.added, "workers {workers}");
assert_eq!(one.changed, many.changed, "workers {workers}");
assert_eq!(one.unchanged, many.unchanged, "workers {workers}");
assert_eq!(one.unproven, many.unproven, "workers {workers}");
assert_eq!(one.removed, many.removed, "workers {workers}");
}
}
#[test]
fn every_fresh_coordinate_lands_in_exactly_one_bucket() {
let (m, fresh) = scenario(77, 3_000);
let d = m.diff(&fresh, 0);
let mut all: Vec<u32> = Vec::new();
all.extend_from_slice(&d.added);
all.extend_from_slice(&d.changed);
all.extend_from_slice(&d.unchanged);
all.extend_from_slice(&d.unproven);
all.sort_unstable();
let mut want = d.order().to_vec();
want.sort_unstable();
assert_eq!(all, want, "the buckets must partition the listing exactly");
// And removed partitions the map against what the listing matched.
assert_eq!(
d.removed.len() + d.changed.len() + d.unchanged.len() + d.unproven.len(),
m.len()
);
}
#[test]
fn the_merge_join_hashes_nothing_and_compares_near_linearly() {
let (m, fresh) = scenario(31337, 20_000);
let d = m.diff(&fresh, 0);
let r = diff_reference(&m, &fresh);
assert_eq!(d.stats.key_hashes, 0, "the merge-join must not hash keys");
assert!(
r.stats.key_hashes > 0,
"the reference is the hashing baseline"
);
// The join walks two sorted runs: one comparison per step, plus O(log m)
// per partition boundary. Two comparisons per unit is a generous ceiling.
let units = (d.stats.fresh_assets + d.stats.map_entries) as u64;
assert!(
d.stats.key_comparisons <= 2 * units,
"{} comparisons for {units} units is not a merge-join",
d.stats.key_comparisons
);
}
#[test]
fn a_full_cycle_converges_to_clean() {
let (mut m, fresh) = scenario(5150, 4_000);
let d = m.diff(&fresh, 0);
assert!(!d.is_clean());
m.adopt(&fresh, &d, &[], 9_000);
let d2 = m.diff(&fresh, 0);
// Everything that CAN be proved is now proved; what remains unproven is
// exactly the coordinates the source publishes no digest for.
assert!(d2.added.is_empty() && d2.changed.is_empty() && d2.removed.is_empty());
let no_digest = fresh.iter().filter(|a| a.sha256.is_none()).count();
assert!(
d2.unproven.len() <= no_digest,
"only digest-less coordinates may stay unproven"
);
assert!(d2.to_fetch(UnprovenPolicy::TrustSize).is_empty());
}
#[test]
fn a_large_map_round_trips_across_batch_boundaries() {
let dir = tempfile::tempdir().unwrap();
let n = 66_536;
let fresh: Vec<RemoteAsset> = (0..n)
.map(|i| {
asset(
&format!("p/{i:08}.jar"),
Some(&digest((i % 251) as u8)),
Some(i as u64),
)
})
.collect();
let m = map_of("nexus", "big", &fresh);
m.store(dir.path()).unwrap();
let back = ChangeMap::load(dir.path(), "nexus", "big").unwrap();
assert_eq!(back.len(), n);
assert!(back.diff(&fresh, 0).is_clean());
}
// ───────────────────── the change token ─────────────────────
/// An asset carrying an opaque, source-scoped change token and nothing
/// else — the shape a `znippy-*` listing produces.
fn tokened(path: &str, scope: &str, raw: &str, size: Option<u64>) -> RemoteAsset {
RemoteAsset {
path: path.to_string(),
download_url: format!("https://upstream.invalid/{path}"),
content_type: None,
size,
sha256: None,
change_token: scope_change_token(scope, raw),
}
}
/// The whole point, at unit scale: equal tokens from the SAME source are
/// `unchanged`, with no digest anywhere and no policy written.
#[test]
fn an_equal_token_from_the_same_source_is_unchanged() {
let fresh = vec![tokened("a/1.znippy", "http://up/pkgs", "aaaa", Some(10))];
let map = map_of("up", "pkgs", &fresh);
let d = map.diff(&fresh, 0);
assert_eq!(
d.unchanged,
vec![0],
"equal token at one source PROVES sameness"
);
assert!(d.unproven.is_empty());
assert!(
d.to_fetch(UnprovenPolicy::Refetch).is_empty(),
"the DEFAULT fetches nothing"
);
assert_eq!(
d.stats.token_comparisons, 1,
"counted, and it is a TOKEN comparison"
);
assert_eq!(d.stats.digest_comparisons, 0, "no digest was involved");
}
/// A different token at the same source is a change, even at the same size.
#[test]
fn a_different_token_at_the_same_source_is_changed_even_at_the_same_size() {
let before = vec![tokened("a/1.znippy", "http://up/pkgs", "aaaa", Some(10))];
let map = map_of("up", "pkgs", &before);
let after = vec![tokened("a/1.znippy", "http://up/pkgs", "bbbb", Some(10))];
let d = map.diff(&after, 0);
assert_eq!(
d.changed,
vec![0],
"a same-size edit is invisible to `size` and not to a token"
);
assert!(d.unchanged.is_empty());
}
/// **★ A token is never compared across sources.**
///
/// The opaque halves are IDENTICAL here, which is the trap: without the
/// scope check this would read as `unchanged` — a mirror told that a
/// different server's artifact is the one it already holds. With the scope
/// check it is `unproven`, and the default policy fetches and finds out.
#[test]
fn a_token_from_another_source_is_never_compared() {
let mine = vec![tokened("a/1.znippy", "http://a/pkgs", "same", Some(10))];
let map = map_of("up", "pkgs", &mine);
let theirs = vec![tokened("a/1.znippy", "http://b/pkgs", "same", Some(10))];
let d = map.diff(&theirs, 0);
assert!(
d.unchanged.is_empty(),
"identical opaque halves from two sources prove NOTHING"
);
assert!(
d.changed.is_empty(),
"and must not be reported as a change either"
);
assert_eq!(d.unproven, vec![0]);
assert_eq!(
d.stats.token_comparisons, 0,
"the comparison was REFUSED, not lost"
);
}
/// A token on one side only proves nothing in either direction — a source
/// that stops publishing tokens must not read as "everything changed", and
/// one that starts must not read as "nothing changed".
#[test]
fn a_token_on_one_side_only_is_unproven() {
let with = vec![tokened("a/1.znippy", "http://up/pkgs", "aaaa", Some(10))];
let map = map_of("up", "pkgs", &with);
let without = vec![asset("a/1.znippy", None, Some(10))];
assert_eq!(
map.diff(&without, 0).unproven,
vec![0],
"the source stopped publishing"
);
let map2 = map_of("up", "pkgs", &without);
assert_eq!(
map2.diff(&with, 0).unproven,
vec![0],
"the source started publishing"
);
}
/// A digest outranks a token, because a digest is comparable across sources
/// and a token is not. Here the digests MATCH and the tokens differ: the
/// digest answers, and the answer is `unchanged`.
#[test]
fn a_digest_outranks_a_token_when_both_are_present() {
let mut before = vec![tokened("a/1.znippy", "http://up/pkgs", "aaaa", Some(10))];
before[0].sha256 = Some(digest(7));
let map = map_of("up", "pkgs", &before);
let mut after = vec![tokened("a/1.znippy", "http://up/pkgs", "zzzz", Some(10))];
after[0].sha256 = Some(digest(7));
let d = map.diff(&after, 0);
assert_eq!(
d.unchanged,
vec![0],
"the digest is the stronger proof and it answers"
);
assert_eq!(d.stats.digest_comparisons, 1);
assert_eq!(
d.stats.token_comparisons, 0,
"the token was not even reached"
);
}
/// A size difference still proves a change when the tokens are not
/// comparable. The token arm does not swallow the rule below it.
#[test]
fn a_size_difference_still_proves_a_change_under_an_incomparable_token() {
let mine = vec![tokened("a/1.znippy", "http://a/pkgs", "same", Some(10))];
let map = map_of("up", "pkgs", &mine);
let theirs = vec![tokened("a/1.znippy", "http://b/pkgs", "same", Some(11))];
assert_eq!(map.diff(&theirs, 0).changed, vec![0]);
}
/// The token survives store/load, and so does its ABSENCE — the two are
/// different states on disk, exactly as they are for the digest.
#[test]
fn the_token_column_round_trips_including_the_absences() {
let dir = tempfile::tempdir().unwrap();
let fresh = vec![
tokened("a/1.znippy", "http://up/pkgs", "aaaa", Some(10)),
asset("b/2.znippy", None, Some(20)),
tokened("c/3.znippy", "http://up/pkgs", "cccc", None),
];
let m = map_of("up", "pkgs", &fresh);
m.store(dir.path()).unwrap();
let back = ChangeMap::load(dir.path(), "up", "pkgs").unwrap();
for i in 0..m.len() {
assert_eq!(back.entry(i), m.entry(i), "row {i} did not round-trip");
}
assert!(
back.entry(1).change_token.is_none(),
"absent stays absent, never empty"
);
let d = back.diff(&fresh, 0);
assert_eq!(
d.unchanged.len(),
2,
"the two tokened rows are PROVED same after a reload"
);
assert_eq!(
d.unproven.len(),
1,
"and the row with neither digest nor token stays in doubt"
);
}
/// An unknown format is refused by name, and the refusal names the format
/// this reader does understand.
#[test]
fn a_map_from_the_future_is_still_refused_by_name() {
let dir = tempfile::tempdir().unwrap();
let path = ChangeMap::path_in(dir.path(), "up", "pkgs");
std::fs::write(&path, "{\"format\":\"holger.mirror.map/v99\"}\n").unwrap();
let err = ChangeMap::read_file(&path).expect_err("v99 must be refused");
let msg = err.to_string();
assert!(msg.contains("v99") && msg.contains(MAP_FORMAT), "{msg}");
}
// stinky 1.4 (2026-09-21): the maven backend is optional (`fmt-maven`), so
// the real-maven settle proof below is one nested module behind that feature
// (in server-lib's `default`, so `cargo test` still runs it).
#[cfg(feature = "fmt-maven")]
mod maven_settles {
use super::*;
// ───────── a typed-view upstream settles, and refetches only what moved ─────────
//
// `tests/mirror_change_token.rs` proves the cycle over the `znippy-package`
// backend, which always had its rows' paths in hand. The six backends that
// list through a TYPED VIEW (maven, npm, gem, conda, python) or a manifest
// (build-thing) could not name their rows until `znippy-common 0.9.16`,
// and stamped nothing: a mirror of any of them re-fetched the whole
// repository every tick. That proof is not table-driven over formats, so
// this is the same cycle over the REAL maven backend and the real maven
// plugin, for maven only; the other five are proven per backend in their
// own crates (`<format>_entries_carry_a_change_token_that_moves_only_when_the_bytes_do`)
// and share `holger_listing` and the diff with this one.
/// A `.pom` at its maven path. The plugin parses a bare `.pom` directly, so
/// no jar is needed for the view to carry the GAV columns.
fn pom(group: &str, artifact: &str, version: &str, extra: &str) -> (String, Vec<u8>) {
(
format!(
"{}/{artifact}/{version}/{artifact}-{version}.pom",
group.replace('.', "/")
),
format!(
"<?xml version=\"1.0\"?><project>\
<groupId>{group}</groupId><artifactId>{artifact}</artifactId>\
<version>{version}</version><packaging>pom</packaging>{extra}</project>"
)
.into_bytes(),
)
}
/// Seal a REAL maven archive with the REAL maven plugin, so the typed view
/// has its coordinates and the backend lists through it.
fn seal_maven(base: &Path, files: &[(String, Vec<u8>)]) -> PathBuf {
let src = base.join("src");
for (rel, bytes) in files {
let p = src.join(rel);
fs::create_dir_all(p.parent().unwrap()).unwrap();
fs::write(&p, bytes).unwrap();
}
let out = base.join("maven.znippy");
znippy_compress::compress_dir(
&src,
&out,
false,
Some(&znippy_common::plugin::PluginRegistry::with_plugin(
Box::new(znippy_plugin_maven::NativeMavenPlugin),
)),
None,
None,
)
.expect("compress_dir with the maven plugin");
out
}
fn maven_corpus(rewrite_guava: bool) -> Vec<(String, Vec<u8>)> {
let extra = if rewrite_guava {
"<description>rewritten</description>"
} else {
""
};
vec![
pom("org.apache.commons", "commons-lang3", "3.12.0", ""),
pom("com.google.guava", "guava", "32.1.3-jre", extra),
pom("org.slf4j", "slf4j-api", "2.0.9", ""),
]
}
/// The upstream: the real backend behind the real conversion. Only the
/// transport is missing.
struct MavenUpstream {
repo: std::sync::Mutex<std::sync::Arc<repository_znippy_maven::MavenRepoZnippy>>,
fetches: std::sync::atomic::AtomicUsize,
}
impl MavenUpstream {
fn open(archive: &Path) -> Self {
Self {
repo: std::sync::Mutex::new(std::sync::Arc::new(
repository_znippy_maven::MavenRepoZnippy::with_archive(
"maven".into(),
archive.to_path_buf(),
)
.expect("open archive"),
)),
fetches: std::sync::atomic::AtomicUsize::new(0),
}
}
fn reopen(&self, archive: &Path) {
*self.repo.lock().unwrap() = std::sync::Arc::new(
repository_znippy_maven::MavenRepoZnippy::with_archive(
"maven".into(),
archive.to_path_buf(),
)
.expect("re-open archive"),
);
}
}
#[async_trait::async_trait]
impl crate::mirror_cycle::MirrorUpstream for MavenUpstream {
async fn list(&self, repository: &str, _page_size: u32) -> Result<Vec<RemoteAsset>> {
let repo = std::sync::Arc::clone(&*self.repo.lock().unwrap());
let entries = traits::RepositoryBackendTrait::list(&*repo, None, 0)?;
Ok(crate::mirror_cycle::holger_listing(
"http://up:50051",
repository,
entries,
))
}
async fn fetch(
&self,
_repository: &str,
id: &traits::ArtifactId,
) -> Result<Option<traits::Bytes>> {
self.fetches
.fetch_add(1, std::sync::atomic::Ordering::SeqCst);
let repo = std::sync::Arc::clone(&*self.repo.lock().unwrap());
traits::RepositoryBackendTrait::fetch(&*repo, id)
}
}
/// A writable in-memory destination, so the cycle's `put` is real.
#[derive(Default)]
struct MemBackend {
held: std::sync::Mutex<std::collections::BTreeMap<String, Vec<u8>>>,
}
impl traits::RepositoryBackendTrait for MemBackend {
fn caps(&self) -> traits::Caps {
traits::Caps {
listing: true,
..traits::Caps::NONE
}
}
fn name(&self) -> &str {
"mem"
}
fn format(&self) -> traits::ArtifactFormat {
traits::ArtifactFormat::Maven3
}
fn is_writable(&self) -> bool {
true
}
fn fetch(&self, id: &traits::ArtifactId) -> Result<Option<traits::Bytes>> {
Ok(self
.held
.lock()
.unwrap()
.get(&crate::mirror_cycle::id_to_coord(id))
.map(|v| traits::Bytes::from(v.clone())))
}
fn put(&self, id: &traits::ArtifactId, data: &[u8]) -> Result<()> {
self.held
.lock()
.unwrap()
.insert(crate::mirror_cycle::id_to_coord(id), data.to_vec());
Ok(())
}
fn handle_http2_request(
&self,
_m: &str,
_p: &str,
_b: &[u8],
) -> Result<(u16, Vec<(String, String)>, Vec<u8>)> {
bail!("the mirror never goes through the HTTP door")
}
fn list(&self, _f: Option<&str>, _l: usize) -> Result<Vec<traits::ArtifactEntry>> {
Ok(self
.held
.lock()
.unwrap()
.iter()
.filter_map(|(c, v)| {
crate::mirror_cycle::coord_to_id(c).map(|id| traits::ArtifactEntry {
id,
size_bytes: v.len() as i64,
content_type: String::new(),
..traits::ArtifactEntry::UNKNOWN_PROVENANCE
})
})
.collect())
}
}
async fn maven_cycle(
up: &MavenUpstream,
local: &std::sync::Arc<dyn traits::RepositoryBackendTrait>,
state: &Path,
) -> crate::mirror_cycle::PairReport {
let pair = crate::mirror_cycle::MirrorPair {
ron_from: "maven".into(),
ron_to: "maven".into(),
// The DEFAULT: no policy written to make the mirror settle.
ron_unproven: crate::mirror_cycle::RonUnprovenPolicy::Refetch,
};
crate::mirror_cycle::run_pair(
up,
"up",
&pair,
local,
state,
1000,
4,
crate::mirror_cycle::now_nanos(),
)
.await
.expect("cycle")
}
/// **A typed-view (maven) upstream settles on the second cycle, and a
/// rewritten pom re-fetches exactly one coordinate.**
#[tokio::test]
async fn a_stamped_maven_upstream_settles_and_refetches_only_the_changed_entry() {
let tmp = tempfile::tempdir().expect("tempdir");
let archive = seal_maven(&tmp.path().join("up"), &maven_corpus(false));
let up = MavenUpstream::open(&archive);
let mem = std::sync::Arc::new(MemBackend::default());
let local = mem.clone() as std::sync::Arc<dyn traits::RepositoryBackendTrait>;
let state = tmp.path().join("state");
let first = maven_cycle(&up, &local, &state).await;
assert_eq!(
first.listed,
3,
"the typed view enumerated the archive: {}",
first.line()
);
assert_eq!(first.added, 3, "{}", first.line());
assert_eq!(first.fetched, 3, "{}", first.line());
let second = maven_cycle(&up, &local, &state).await;
assert_eq!(
second.unchanged,
3,
"a maven listing carries a change token now: {}",
second.line()
);
assert_eq!(
second.unproven,
0,
"nothing left in doubt: {}",
second.line()
);
assert_eq!(second.to_fetch, 0, "THE WHOLE POINT: {}", second.line());
assert!(second.moved_nothing(), "{}", second.line());
assert_eq!(
up.fetches.load(std::sync::atomic::Ordering::SeqCst),
3,
"the upstream was never asked for a byte again"
);
// Re-seal with guava's pom rewritten: same GAV, same path, different bytes.
let next = seal_maven(&tmp.path().join("up2"), &maven_corpus(true));
up.reopen(&next);
let third = maven_cycle(&up, &local, &state).await;
assert_eq!(
third.changed,
1,
"exactly one artifact moved: {}",
third.line()
);
assert_eq!(
third.unchanged,
2,
"the other two are still PROVED same: {}",
third.line()
);
assert_eq!(third.unproven, 0, "{}", third.line());
assert_eq!(third.to_fetch, 1, "{}", third.line());
assert_eq!(
up.fetches.load(std::sync::atomic::Ordering::SeqCst),
4,
"one more fetch, not three"
);
let held = mem.held.lock().unwrap();
let guava = held
.iter()
.find(|(c, _)| c.contains("guava"))
.map(|(_, v)| String::from_utf8_lossy(v).into_owned())
.expect("guava landed");
assert!(
guava.contains("rewritten"),
"and the new bytes landed: {guava}"
);
}
}
}