server/lib/tests/mirror_scaling.rs
//! The mirror change-map diff, MEASURED — counted, not timed.
//!
//! The diff is the mirror's hot path: it runs over the whole upstream listing,
//! every cycle, and a real Maven or PyPI proxy lists hundreds of thousands of
//! coordinates. This test is the evidence that the parallel merge-join in
//! `holger_server_lib::mirror` is actually cheaper than the obvious serial
//! `HashMap` diff, and that it stays cheaper as the listing grows.
//!
//! **It counts rather than times.** A build box shared by six lanes has a load
//! average that moves by an order of magnitude inside one test run, so a wall
//! clock there measures the other five lanes. Allocated bytes, allocation
//! count, key comparisons and key hashes do not move with load: they are the
//! same number on an idle box and a wedged one, and they are the numbers this
//! prints and asserts on. Wall clock is printed too, marked for what it is —
//! context, never a gate.
//!
//! Run it as a report:
//!
//! ```text
//! cargo test -p holger-server-lib --test mirror_scaling -- --nocapture
//! ```
//!
//! It is ONE `#[test]` on purpose. The counting allocator below is global, so a
//! second test running beside it would have its allocations folded into these
//! numbers.
use std::alloc::{GlobalAlloc, Layout, System};
use std::sync::atomic::{AtomicU64, Ordering};
use std::time::Instant;
use holger_server_lib::mirror::{diff_reference, ChangeMap, MirrorDiff, UnprovenPolicy};
use traits::RemoteAsset;
// ───────────────────────── the counting allocator ─────────────────────────
static ALLOC_BYTES: AtomicU64 = AtomicU64::new(0);
static ALLOC_COUNT: AtomicU64 = AtomicU64::new(0);
/// `System`, plus two counters. Deallocation is deliberately NOT counted: the
/// question is how much memory traffic a diff generates, not what its high-water
/// mark is, and every allocation is traffic whether or not it is freed promptly.
struct Counting;
unsafe impl GlobalAlloc for Counting {
unsafe fn alloc(&self, layout: Layout) -> *mut u8 {
ALLOC_BYTES.fetch_add(layout.size() as u64, Ordering::Relaxed);
ALLOC_COUNT.fetch_add(1, Ordering::Relaxed);
unsafe { System.alloc(layout) }
}
unsafe fn dealloc(&self, ptr: *mut u8, layout: Layout) {
unsafe { System.dealloc(ptr, layout) }
}
unsafe fn realloc(&self, ptr: *mut u8, layout: Layout, new_size: usize) -> *mut u8 {
// A grow is traffic too — count the delta, which is what a `Vec` that
// was not preallocated pays over and over.
if new_size > layout.size() {
ALLOC_BYTES.fetch_add((new_size - layout.size()) as u64, Ordering::Relaxed);
ALLOC_COUNT.fetch_add(1, Ordering::Relaxed);
}
unsafe { System.realloc(ptr, layout, new_size) }
}
}
#[global_allocator]
static ALLOCATOR: Counting = Counting;
/// Run `f`, and report what it allocated and how long it took.
fn measured<T>(f: impl FnOnce() -> T) -> (T, u64, u64, f64) {
let b0 = ALLOC_BYTES.load(Ordering::Relaxed);
let c0 = ALLOC_COUNT.load(Ordering::Relaxed);
let t0 = Instant::now();
let out = f();
let ms = t0.elapsed().as_secs_f64() * 1000.0;
(
out,
ALLOC_BYTES.load(Ordering::Relaxed) - b0,
ALLOC_COUNT.load(Ordering::Relaxed) - c0,
ms,
)
}
// ───────────────────────── the listing ─────────────────────────
fn xorshift(state: &mut u64) -> u64 {
let mut x = *state;
x ^= x << 13;
x ^= x >> 7;
x ^= x << 17;
*state = x;
x
}
fn hexdigest(seed: u64) -> String {
let mut s = String::with_capacity(64);
for i in 0..32u64 {
s.push_str(&format!(
"{:02x}",
(seed.wrapping_mul(31).wrapping_add(i)) as u8
));
}
s
}
/// A listing shaped like a Maven proxy's: deep group paths, versioned
/// artifacts, and a quarter of the coordinates publishing no digest — which is
/// the honest shape, not a convenient one.
fn listing(seed: u64, n: usize) -> Vec<RemoteAsset> {
let mut s = seed | 1;
(0..n)
.map(|i| {
let g = xorshift(&mut s) % 4_000;
let v = xorshift(&mut s) % 200;
let path = format!(
"org/example/group{g:04}/artifact-{i:07}/1.{v}.0/artifact-{i:07}-1.{v}.0.jar"
);
let d = xorshift(&mut s);
RemoteAsset {
download_url: format!("https://upstream.invalid/{path}"),
path,
content_type: None,
size: Some(xorshift(&mut s) % 40_000_000),
sha256: (!d.is_multiple_of(4)).then(|| hexdigest(d)),
change_token: None,
}
})
.collect()
}
/// What an upstream looks like one cycle later: a few percent re-published, a
/// few percent dropped, a few percent added.
fn next_cycle(seed: u64, base: &[RemoteAsset]) -> Vec<RemoteAsset> {
let mut s = seed | 1;
let mut out = Vec::with_capacity(base.len() + base.len() / 50);
for a in base {
match xorshift(&mut s) % 100 {
0..=1 => {} // dropped upstream
2..=4 => {
// re-published under the same coordinate
let d = xorshift(&mut s);
out.push(RemoteAsset {
sha256: a.sha256.as_ref().map(|_| hexdigest(d)),
change_token: None,
size: Some(xorshift(&mut s) % 40_000_000),
..a.clone()
});
}
_ => out.push(a.clone()),
}
if xorshift(&mut s).is_multiple_of(50) {
let d = xorshift(&mut s);
let path = format!(
"org/example/new{:05}/artifact-{:07}/1.0.0/a.jar",
d % 9_000,
d % 99_999
);
out.push(RemoteAsset {
download_url: format!("https://upstream.invalid/{path}"),
path,
content_type: None,
size: Some(1234),
sha256: Some(hexdigest(d)),
change_token: None,
});
}
}
out
}
fn same_verdict(a: &MirrorDiff, b: &MirrorDiff) -> bool {
a.added == b.added
&& a.changed == b.changed
&& a.unchanged == b.unchanged
&& a.unproven == b.unproven
&& a.removed == b.removed
}
// ───────────────────────── the measurement ─────────────────────────
#[test]
fn diff_scaling_counted_against_the_serial_hashmap_baseline() {
println!();
println!("mirror change-map diff — COUNTED measurement");
println!("(allocated bytes / allocation count / key comparisons / key hashes are");
println!(" load-independent; the ms column is context on a shared box, not a gate)");
println!();
println!(
"{:>9} {:>9} {:<12} {:>12} {:>10} {:>12} {:>10} {:>9}",
"assets", "map", "impl", "alloc bytes", "allocs", "key cmps", "key hash", "ms"
);
let scales = [25_000usize, 50_000, 100_000, 200_000, 400_000];
let mut last_baseline_bytes = 0u64;
let mut last_join_bytes = 0u64;
for n in scales {
let base = listing(0xB16_u64.wrapping_add(n as u64), n);
let fresh = next_cycle(0xC0FFEE, &base);
// Seed the map from the first listing — the state a mirror is in at the
// start of its second cycle.
let mut map = ChangeMap::empty("nexus-upstream", "maven-central");
let seed_diff = map.diff(&base, 0);
map.adopt(&base, &seed_diff, &[], 1_000_000_000);
assert_eq!(map.len(), base.len());
let (baseline, b_bytes, b_allocs, b_ms) = measured(|| diff_reference(&map, &fresh));
let (join, j_bytes, j_allocs, j_ms) = measured(|| map.diff(&fresh, 0));
println!(
"{:>9} {:>9} {:<12} {:>12} {:>10} {:>12} {:>10} {:>9.1}",
fresh.len(),
map.len(),
"hashmap",
b_bytes,
b_allocs,
baseline.stats.key_comparisons,
baseline.stats.key_hashes,
b_ms
);
println!(
"{:>9} {:>9} {:<12} {:>12} {:>10} {:>12} {:>10} {:>9.1} workers={}",
fresh.len(),
map.len(),
"merge-join",
j_bytes,
j_allocs,
join.stats.key_comparisons,
join.stats.key_hashes,
j_ms,
join.stats.workers
);
// ── the load-independent gates ──
assert!(
same_verdict(&baseline, &join),
"the parallel join disagreed with the serial oracle at n={n}"
);
assert_eq!(
join.stats.key_hashes, 0,
"the merge-join must hash nothing at n={n}"
);
assert_eq!(
baseline.stats.key_hashes,
(map.len() + join.order().len()) as u64,
"the baseline hashes every map slot and every asset — that is the cost it pays"
);
let units = (join.stats.fresh_assets + join.stats.map_entries) as u64;
assert!(
join.stats.key_comparisons <= 2 * units,
"{} comparisons over {units} units at n={n} is not a merge-join",
join.stats.key_comparisons
);
assert!(
j_bytes < b_bytes,
"the merge-join allocated {j_bytes} B against the baseline's {b_bytes} B at n={n}"
);
// A steady-state cycle fetches a small fraction of the listing — that is
// the entire point of the map.
let fetch = join.to_fetch(UnprovenPolicy::Refetch).len();
let trust = join.to_fetch(UnprovenPolicy::TrustSize).len();
assert!(
trust < fetch,
"TrustSize must fetch strictly less than Refetch while unproven coordinates exist"
);
assert!(
fetch < fresh.len() / 2,
"a mirror that re-fetches {fetch} of {} is not fetching only what changed",
fresh.len()
);
if n == *scales.last().unwrap() {
last_baseline_bytes = b_bytes;
last_join_bytes = j_bytes;
println!();
println!(
" at {} assets / {} map entries: fetch {} of {} ({:.2}%) under Refetch, \
{} under TrustSize",
fresh.len(),
map.len(),
fetch,
fresh.len(),
100.0 * fetch as f64 / fresh.len() as f64,
trust
);
println!(
" added {} changed {} unchanged {} unproven {} removed {}",
join.added.len(),
join.changed.len(),
join.unchanged.len(),
join.unproven.len(),
join.removed.len()
);
}
}
println!();
println!(
" largest scale: merge-join allocated {} B vs the hashmap baseline's {} B \
({:.2}x less), and hashed 0 keys against the baseline's every key.",
last_join_bytes,
last_baseline_bytes,
last_baseline_bytes as f64 / last_join_bytes.max(1) as f64
);
println!();
}