server/lib/tests/mirror_change_token.rs
//! **A znippy-backed repository settles.**
//!
//! `mirror_declaration.rs` proves the cycle with a fake upstream and, for the
//! holger→holger case, with a fake backend that can publish a `sha256`. That
//! left the case that is actually 15 of holger's 17 backends untested by
//! anything but a comment: a `znippy-*` repository, which has neither a
//! modification time nor a content digest in its file table, and which
//! therefore landed in `unproven` on every tick and re-fetched its whole
//! repository every 300 s under the default `Refetch` policy.
//!
//! Everything here is REAL except the transport: a real `.znippy` archive
//! sealed by `znippy-compress`, served by the real `ZnippyRepoZnippy` backend,
//! listed through the real `RepositoryBackendTrait::list`, converted by the
//! real `holger_listing`, and diffed by the real `ChangeMap`. The four proofs:
//!
//! 1. a second cycle over an unchanged znippy repository reports
//! `unchanged N, to_fetch 0` under the DEFAULT policy,
//! 2. changing ONE artifact upstream makes exactly one `changed`,
//! 3. a token from a DIFFERENT source is never compared — it is `unproven`,
//! never `changed` and never `unchanged`,
//! 4. the listing that produces the tokens reads the archive INDEX and not one
//! byte of any artifact, counted rather than asserted.
use std::collections::BTreeMap;
use std::path::{Path, PathBuf};
use std::sync::atomic::{AtomicUsize, Ordering};
use std::sync::{Arc, Mutex};
use anyhow::Result;
use holger_server_lib::mirror_cycle::{
coord_to_id, id_to_coord, run_pair, MirrorPair, MirrorSettings, MirrorUpstream, PairReport,
RonUnprovenPolicy,
};
use repository_znippy_package::ZnippyRepoZnippy;
use traits::{
ArtifactEntry, ArtifactFormat, ArtifactId, Bytes, RemoteAsset, RepositoryBackendTrait,
};
// ───────────────────────────── a real archive ─────────────────────────────
/// Seal a real `.znippy` holding `(path, bytes)` entries. Returns the archive
/// path. `znippy-compress` appends the extension itself.
fn seal(base: &PathBuf, entries: &[(&str, Vec<u8>)]) -> PathBuf {
let compressor = znippy_compress::compress_stream(base, true).expect("compressor");
let sender = compressor.sender().clone();
for (path, data) in entries {
sender
.send(znippy_compress::ArchiveEntry {
relative_path: (*path).into(),
data: data.clone(),
pkg_type: None,
repo: None,
})
.expect("send entry");
}
drop(sender);
compressor.finish().expect("seal archive");
base.with_extension("znippy")
}
/// The three artifacts every proof here starts from. The `znippy-package`
/// backend parses `{name}-{version}.znippy`, so these are real coordinates.
fn corpus() -> Vec<(&'static str, Vec<u8>)> {
vec![
("alpha-1.0.znippy", b"alpha-bytes-one".to_vec()),
("beta-2.0.znippy", b"beta-bytes-two".to_vec()),
("gamma-3.0.znippy", b"gamma-bytes-three".to_vec()),
]
}
// ───────────────────────── the upstream, for real ─────────────────────────
/// A holger upstream whose repository is a REAL znippy archive behind the REAL
/// `ZnippyRepoZnippy` backend. Nothing about the provenance is faked: what the
/// listing can prove is whatever that backend can prove.
struct ZnippyUpstream {
repo: Mutex<Arc<ZnippyRepoZnippy>>,
endpoint: String,
fetches: Arc<AtomicUsize>,
}
impl ZnippyUpstream {
fn open(endpoint: &str, archive: &Path) -> Self {
Self {
repo: Mutex::new(Arc::new(
ZnippyRepoZnippy::with_archive("pkgs".into(), archive.to_path_buf())
.expect("open archive"),
)),
endpoint: endpoint.to_string(),
fetches: Arc::new(AtomicUsize::new(0)),
}
}
/// Point the same upstream at a re-sealed archive — an upstream that
/// published something new.
fn reopen(&self, archive: &Path) {
*self.repo.lock().unwrap() = Arc::new(
ZnippyRepoZnippy::with_archive("pkgs".into(), archive.to_path_buf())
.expect("re-open archive"),
);
}
fn fetch_count(&self) -> usize {
self.fetches.load(Ordering::SeqCst)
}
}
#[async_trait::async_trait]
impl MirrorUpstream for ZnippyUpstream {
async fn list(&self, repository: &str, _page_size: u32) -> Result<Vec<RemoteAsset>> {
let repo = Arc::clone(&*self.repo.lock().unwrap());
let entries: Vec<ArtifactEntry> = repo.list(None, 0)?;
// THE REAL CONVERSION — this is where the scope is stamped.
Ok(holger_server_lib::mirror_cycle::holger_listing(
&self.endpoint,
repository,
entries,
))
}
async fn fetch(&self, _repository: &str, id: &ArtifactId) -> Result<Option<Bytes>> {
self.fetches.fetch_add(1, Ordering::SeqCst);
let repo = Arc::clone(&*self.repo.lock().unwrap());
repo.fetch(id)
}
}
/// A writable in-memory destination, so the cycle's `put` is real without a
/// second archive.
#[derive(Default)]
struct MemBackend {
held: Mutex<BTreeMap<String, Vec<u8>>>,
}
impl RepositoryBackendTrait for MemBackend {
fn caps(&self) -> traits::Caps {
traits::Caps {
listing: true,
..traits::Caps::NONE
}
}
fn name(&self) -> &str {
"mem"
}
fn format(&self) -> ArtifactFormat {
ArtifactFormat::Znippy
}
fn is_writable(&self) -> bool {
true
}
fn fetch(&self, id: &ArtifactId) -> Result<Option<Bytes>> {
Ok(self
.held
.lock()
.unwrap()
.get(&id_to_coord(id))
.map(|v| Bytes::from(v.clone())))
}
fn put(&self, id: &ArtifactId, data: &[u8]) -> Result<()> {
self.held
.lock()
.unwrap()
.insert(id_to_coord(id), data.to_vec());
Ok(())
}
fn handle_http2_request(
&self,
_m: &str,
_p: &str,
_b: &[u8],
) -> Result<(u16, Vec<(String, String)>, Vec<u8>)> {
anyhow::bail!("the mirror never goes through the HTTP door")
}
fn list(&self, _f: Option<&str>, _l: usize) -> Result<Vec<ArtifactEntry>> {
Ok(self
.held
.lock()
.unwrap()
.iter()
.filter_map(|(c, v)| {
coord_to_id(c).map(|id| ArtifactEntry {
id,
size_bytes: v.len() as i64,
content_type: String::new(),
..ArtifactEntry::UNKNOWN_PROVENANCE
})
})
.collect())
}
}
async fn cycle(
up: &dyn MirrorUpstream,
source_name: &str,
local: &Arc<dyn RepositoryBackendTrait>,
state: &Path,
) -> PairReport {
let pair = MirrorPair {
ron_from: "pkgs".into(),
ron_to: "pkgs".into(),
// The DEFAULT. The point of every proof here is that no policy was
// written to make the mirror settle.
ron_unproven: RonUnprovenPolicy::Refetch,
};
run_pair(
up,
source_name,
&pair,
local,
state,
1000,
4,
holger_server_lib::mirror_cycle::now_nanos(),
)
.await
.expect("cycle")
}
// ───────────────────────────── the proofs ─────────────────────────────
/// **★ A znippy-backed repository reaches `unchanged`, with no policy written.**
///
/// This is ?46 stated as a cycle. A `znippy-*` backend answers `None` to
/// `published_unix_ms` and `None` to `sha256` — it has neither in its file
/// table — so before the change token every coordinate of every one of the 15
/// znippy-backed repositories landed in `unproven`, and the DEFAULT
/// `UnprovenPolicy::Refetch` re-fetched all of them on every tick, for ever.
#[tokio::test]
async fn a_znippy_backed_mirror_settles_on_the_second_cycle_under_the_default_policy() {
let tmp = tempfile::tempdir().expect("tempdir");
let archive = seal(&tmp.path().join("up"), &corpus());
let up = ZnippyUpstream::open("http://up:50051", &archive);
let mem = Arc::new(MemBackend::default());
let local = mem.clone() as Arc<dyn RepositoryBackendTrait>;
let state = tmp.path().join("state");
let first = cycle(&up, "up", &local, &state).await;
assert_eq!(
first.listed,
3,
"the backend enumerated the archive: {}",
first.line()
);
assert_eq!(first.added, 3, "{}", first.line());
assert_eq!(first.fetched, 3, "{}", first.line());
let second = cycle(&up, "up", &local, &state).await;
assert_eq!(
second.unchanged,
3,
"a znippy listing carries a change token, so the second cycle PROVES sameness: {}",
second.line()
);
assert_eq!(
second.unproven,
0,
"nothing left in doubt: {}",
second.line()
);
assert_eq!(second.changed, 0, "{}", second.line());
assert_eq!(second.added, 0, "{}", second.line());
assert_eq!(second.to_fetch, 0, "THE WHOLE POINT: {}", second.line());
assert!(second.moved_nothing(), "{}", second.line());
assert_eq!(
up.fetch_count(),
3,
"the upstream was never asked for a byte again"
);
}
/// One artifact changes upstream, and exactly one coordinate is re-fetched.
///
/// The other two must stay `unchanged`: a token that moved for every row when
/// one row moved would be a per-ARCHIVE token, and that is a different — and
/// useless — thing.
#[tokio::test]
async fn changing_one_artifact_upstream_changes_exactly_one_coordinate() {
let tmp = tempfile::tempdir().expect("tempdir");
let archive = seal(&tmp.path().join("up"), &corpus());
let up = ZnippyUpstream::open("http://up:50051", &archive);
let mem = Arc::new(MemBackend::default());
let local = mem.clone() as Arc<dyn RepositoryBackendTrait>;
let state = tmp.path().join("state");
assert_eq!(cycle(&up, "up", &local, &state).await.fetched, 3);
assert_eq!(cycle(&up, "up", &local, &state).await.to_fetch, 0);
// Re-seal with BETA's bytes replaced. Same path, same coordinate, different
// bytes — the case a size comparison can miss and a token cannot.
let mut changed = corpus();
changed[1].1 = b"beta-bytes-TWO".to_vec();
assert_eq!(
changed[1].1.len(),
corpus()[1].1.len(),
"a SAME-SIZE edit, so only the token can see it"
);
let next = seal(&tmp.path().join("up2"), &changed);
up.reopen(&next);
let third = cycle(&up, "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!(
mem.held
.lock()
.unwrap()
.get("/beta@2.0")
.map(|v| v.as_slice()),
Some(&b"beta-bytes-TWO"[..]),
"and the new bytes landed"
);
}
/// **★ A token is never compared across two different sources.**
///
/// Two upstreams, the same coordinates, the same source NAME — so the same
/// change map — and different source identities. Their tokens are not
/// comparable, and the honest verdict is `unproven`: NOT `changed` (the
/// artifacts may be identical and usually are), and NOT `unchanged` (nothing
/// was proved).
///
/// This is not a hypothetical: two holgers mirroring the same crates publish
/// different tokens for identical bytes as a matter of course, because the fold
/// binds the archive-relative path and each server lays its archive out its own
/// way. Without the scope check, repointing a source at its own replica would
/// report the entire repository as `changed` on every tick, for ever — the
/// exact failure this lane exists to remove, with the sign flipped.
#[tokio::test]
async fn a_token_from_another_source_is_unproven_never_changed() {
let tmp = tempfile::tempdir().expect("tempdir");
let state = tmp.path().join("state");
let mem = Arc::new(MemBackend::default());
let local = mem.clone() as Arc<dyn RepositoryBackendTrait>;
// Source A: its own archive, its own endpoint.
let a_archive = seal(&tmp.path().join("a"), &corpus());
let a = ZnippyUpstream::open("http://a:50051", &a_archive);
let first = cycle(&a, "up", &local, &state).await;
assert_eq!(first.added, 3);
// Source B: byte-identical artifacts, a different server. Sealed into a
// second archive so the two are genuinely two sources.
let b_archive = seal(&tmp.path().join("b"), &corpus());
let b = ZnippyUpstream::open("http://b:50051", &b_archive);
let second = cycle(&b, "up", &local, &state).await;
assert_eq!(
second.changed,
0,
"a cross-source token comparison must NEVER be read as a change: {}",
second.line()
);
assert_eq!(
second.unchanged,
0,
"and must never be read as sameness either — nothing was proved: {}",
second.line()
);
assert_eq!(
second.unproven,
3,
"it is the `unproven` bucket, which is what that bucket is for: {}",
second.line()
);
assert_eq!(
second.to_fetch, 3,
"and the default policy re-fetches the doubt"
);
}
// ───────────────────── the listing reads no artifact byte ─────────────────
/// A `ZnippyReader` that COUNTS. Everything it answers comes from the real
/// archive; what it adds is a tally of how often each door was opened.
struct CountingReader {
inner: traits::znippy_common::ZnippyArchive,
extracts: AtomicUsize,
listings: AtomicUsize,
sizes: AtomicUsize,
}
impl traits::znippy_common::ZnippyReader for CountingReader {
fn list_files(&self) -> Result<Vec<String>> {
self.listings.fetch_add(1, Ordering::SeqCst);
self.inner.list_files()
}
fn extract_file(&self, p: &str) -> Result<Vec<u8>> {
self.extracts.fetch_add(1, Ordering::SeqCst);
self.inner.extract_file(p)
}
fn contains(&self, p: &str) -> bool {
self.inner.contains(p)
}
fn file_size(&self, p: &str) -> Option<u64> {
self.sizes.fetch_add(1, Ordering::SeqCst);
self.inner.file_size(p)
}
}
/// **★ The listing stamps a token on every row and never extracts a byte.**
///
/// COUNTED, the way `the_listing_reads_manifests_and_not_the_store_counted`
/// counts for OCI. The token is a fold of the per-chunk BLAKE3s that the
/// archive's lookup sub-index already carries, so the whole listing costs one
/// index read and then pure memory. If it ever started reading artifacts, this
/// goes red on the first row rather than on somebody's 50 000-file archive.
#[test]
fn a_znippy_listing_stamps_a_token_and_never_extracts_a_byte() {
let tmp = tempfile::tempdir().expect("tempdir");
let archive = seal(&tmp.path().join("counted"), &corpus());
let reader = CountingReader {
inner: traits::znippy_common::ZnippyArchive::open(&archive).expect("open"),
extracts: AtomicUsize::new(0),
listings: AtomicUsize::new(0),
sizes: AtomicUsize::new(0),
};
let tokens = traits::ArchiveTokens::at(&archive);
let index = tokens
.open()
.expect("a v0.7 archive carries a lookup sub-index");
let entries = traits::list_archive_entries(
&reader,
Some(&index),
None,
0,
"application/octet-stream",
|p| traits::coordinate_from_name_dash_version(p, ".znippy"),
)
.expect("listing");
assert_eq!(entries.len(), 3, "every artifact was listed");
for e in &entries {
let t = e
.change_token
.as_deref()
.unwrap_or_else(|| panic!("{}: a znippy listing must stamp a change token", e.id.name));
assert_eq!(
t.len(),
64,
"{}: the fold is 64 hex chars, got {t:?}",
e.id.name
);
assert!(t.chars().all(|c| c.is_ascii_hexdigit()));
assert!(
e.published_unix_ms.is_none(),
"a znippy file table still has no mtime"
);
assert!(
e.sha256.is_none(),
"and still no content digest — the token is not one"
);
}
// Distinct artifacts, distinct tokens. A token that collided would report
// two different artifacts as the same one.
let mut seen: Vec<&str> = entries
.iter()
.filter_map(|e| e.change_token.as_deref())
.collect();
seen.sort_unstable();
seen.dedup();
assert_eq!(seen.len(), 3, "three artifacts, three tokens");
// THE COUNT.
assert_eq!(
reader.extracts.load(Ordering::SeqCst),
0,
"the listing must not extract ONE artifact — that is the defect a content \
digest would have been"
);
assert_eq!(
reader.listings.load(Ordering::SeqCst),
1,
"one walk of the file table"
);
assert_eq!(
reader.sizes.load(Ordering::SeqCst),
3,
"one index size lookup per row"
);
assert!(
index.rows() >= 3,
"the index carries at least one chunk row per artifact"
);
}
/// The same bytes at the same path fold to the same token, and different bytes
/// do not — measured on two real archives rather than argued.
#[test]
fn the_token_moves_with_the_bytes_and_not_with_anything_else() {
let tmp = tempfile::tempdir().expect("tempdir");
let one = seal(&tmp.path().join("one"), &corpus());
let two = seal(&tmp.path().join("two"), &corpus());
let mut edited = corpus();
edited[0].1 = b"alpha-bytes-TWO".to_vec();
let three = seal(&tmp.path().join("three"), &edited);
let tok = |archive: &Path, path: &str| -> String {
traits::ArchiveTokens::at(archive)
.open()
.expect("lookup sub-index")
.token(path)
.expect("token")
};
assert_eq!(
tok(&one, "alpha-1.0.znippy"),
tok(&two, "alpha-1.0.znippy"),
"same path, same bytes, same token — a re-seal of unchanged content must not \
report a change"
);
assert_ne!(
tok(&one, "alpha-1.0.znippy"),
tok(&three, "alpha-1.0.znippy"),
"a SAME-SIZE edit must move the token; that is the edit `size` cannot see"
);
assert_eq!(
tok(&one, "beta-2.0.znippy"),
tok(&three, "beta-2.0.znippy"),
"and it must move for THAT artifact only"
);
assert_ne!(
tok(&one, "alpha-1.0.znippy"),
tok(&one, "beta-2.0.znippy"),
"two artifacts in one archive are two tokens"
);
}
/// A backend with no archive PATH — one built from a bare reader — answers no
/// token at all, rather than a fabricated one. `MirrorSettings` is named here
/// only so the import that documents where the policy lives is not unused.
#[test]
fn a_backend_with_no_archive_path_answers_no_token() {
let _ = std::mem::size_of::<MirrorSettings>();
let none = traits::ArchiveTokens::none();
assert!(none.path().is_none());
assert!(
none.open().is_none(),
"no path, no index, no token — never an invented value"
);
let tmp = tempfile::tempdir().expect("tempdir");
let missing = traits::ArchiveTokens::at(tmp.path().join("not-here.znippy"));
assert!(
missing.open().is_none(),
"an archive that cannot be read is 'cannot answer', not an error and not a guess"
);
}