bench-scenarios/src/hops.rs
//! The HOPS — the paths where an artifact crosses from one store to another.
//!
//! ## Why this module exists
//!
//! Stinky-11 executed the `Vec<u8>` → `Bytes` migration across 50 signatures and
//! reported it **FLAT** on every existing arm (`rust_crate_st` 443 405 → 439 893
//! ops/s), and said honestly why: those arms drive `get_file` /
//! `handle_http2_request` **in process**, where the buffer was already moved
//! rather than cloned. The hops the seam exists for — the proxy's
//! cache-and-forward and replication — **had no bench arm in holger at all**.
//! Without one the work cannot be shown to pay, and "it must be faster" is not a
//! measurement. These arms are that instrument.
//!
//! ## What is being measured, precisely
//!
//! Both arms drive the REAL engines, not a re-implementation:
//!
//! * `replication_copy` / `replication_rescan` — `server_lib::replication::replicate_pairs`,
//! the same function the `holger-server replicate` CLI verb and the
//! `ReplicationService.Replicate` RPC call. Per artifact: `source.fetch` →
//! sha256 content id → destination presence probe → `dest.put`.
//! * `proxy_cache_fill` / `proxy_primary_hit` — `server_lib::proxy::ProxyBackend::fetch`,
//! the pull-through cache. A miss walks the upstream chain, runs the cache-fill
//! gate and writes the bytes through into the writable primary before returning
//! them; a hit is served from the primary.
//!
//! Both stores are real `RustRepoZnippy` archives on disk, so the numbers carry
//! the archive's real read and append costs — this is not an in-memory stub.
//!
//! ## Honesty notes that must travel with these numbers
//!
//! * There is **no rival column**. Nexus and Artifactory have no equivalent verb
//! driven here; these are holger-against-holger measurements.
//! * `replication_rescan` is the arm that isolates the PARALLEL half: on a second
//! pass every artifact is already present by content id, so every unit is
//! fetch + hash on both sides and **not one `put` runs**. `replication_copy` is
//! the mixed number, bounded by the destination's single write lock.
//! * Fixed N, never a time-boxed loop, for the same reason `rust_crate_put` is:
//! a cost that grows with the store hides inside a fixed duration by simply
//! completing fewer operations.
use std::path::PathBuf;
use std::sync::Arc;
use std::time::{Duration, Instant};
use anyhow::{Context, Result};
use repository_znippy_rust::RustRepoZnippy;
use traits::{ArtifactId, RepositoryBackendTrait};
use crate::{bench_root, served, Scale, Throughput};
/// Bytes per artifact. Mirrors `stress::PAYLOAD_BYTES` so a hop number and a
/// publish number are over the same unit of work.
const PAYLOAD_BYTES: usize = 16 * 1024;
/// How many artifacts each hop arm moves. **Fixed N**, `HOLGER_BENCH_HOPS`
/// overrides. Both hops write into a growing archive, so the same reasoning that
/// fixed `rust_crate_put`'s N applies here.
fn hop_count(scale: &Scale) -> usize {
if let Ok(v) = std::env::var("HOLGER_BENCH_HOPS") {
if let Ok(n) = v.parse::<usize>() {
if n > 0 {
return n;
}
}
}
scale.entries.clamp(200, 2_000)
}
/// Distinct payload for artifact `i` — deterministic, unique, and NOT
/// compressible to nothing, so the content ids differ and nothing dedups.
fn payload(i: usize) -> Vec<u8> {
let mut v = vec![0u8; PAYLOAD_BYTES];
let mut x = (i as u64)
.wrapping_mul(0x9E37_79B9_7F4A_7C15)
.wrapping_add(1);
for b in v.iter_mut() {
x ^= x << 13;
x ^= x >> 7;
x ^= x << 17;
*b = (x & 0xFF) as u8;
}
v
}
fn ids_and_payloads(n: usize, prefix: &str) -> Vec<(ArtifactId, Vec<u8>)> {
(0..n)
.map(|i| {
(
ArtifactId {
namespace: None,
name: format!("{prefix}-{i:06}"),
version: "1.0.0".into(),
},
payload(i),
)
})
.collect()
}
/// A scratch directory under the bench root, wiped first so a previous run's
/// archive can never be mistaken for this run's fixture.
fn fresh_dir(name: &str) -> Result<PathBuf> {
let dir = bench_root().join(name);
if dir.exists() {
std::fs::remove_dir_all(&dir).ok();
}
std::fs::create_dir_all(&dir).with_context(|| format!("create {}", dir.display()))?;
Ok(dir)
}
fn throughput(ops: usize, bytes: u64, elapsed: Duration) -> Throughput {
let s = elapsed.as_secs_f64().max(f64::MIN_POSITIVE);
Throughput {
ops: ops as u64,
elapsed,
bytes,
ops_sec: ops as f64 / s,
mb_per_sec: bytes as f64 / s / 1_048_576.0,
}
}
/// Both replication numbers from one fixture.
pub struct ReplicationOutcome {
/// First pass: every artifact copied (fetch + hash + probe in parallel, then
/// the serial `put` committer).
pub copy: Throughput,
/// Second pass over the same pair: every artifact already present by content
/// id, so **zero writes** — the pure read/hash fan-out.
pub rescan: Throughput,
}
/// Drive the real replication engine source→dest over N artifacts, twice.
pub fn replication_pair(scale: &Scale) -> Result<ReplicationOutcome> {
use server_lib::replication::{replicate_pairs, Outcome, ReplicationOptions, RepoPair};
let n = hop_count(scale);
let dir = fresh_dir("replication-arena")?;
// Seed the SOURCE. Outside the clock: this is fixture cost, not hop cost.
let corpus = ids_and_payloads(n, "repl-crate");
let source = Arc::new(RustRepoZnippy::with_writable_archive(
"repl-src".into(),
dir.join("source.znippy"),
));
for (id, data) in &corpus {
source
.put(id, data)
.with_context(|| format!("seed source with {}", id.name))?;
}
let dest = Arc::new(RustRepoZnippy::with_writable_archive(
"repl-dst".into(),
dir.join("dest.znippy"),
));
let pair = RepoPair {
source_name: "repl-src".into(),
dest_name: "repl-dst".into(),
source: source.clone() as Arc<dyn RepositoryBackendTrait>,
dest: dest.clone() as Arc<dyn RepositoryBackendTrait>,
};
let opts = ReplicationOptions {
execute: true,
limit: 0,
};
let declared: u64 = corpus.iter().map(|(_, d)| d.len() as u64).sum();
// ── Pass 1: the copy. ──
let t0 = Instant::now();
let first = replicate_pairs(std::slice::from_ref(&pair), &opts);
let copy_elapsed = t0.elapsed();
// ANSWER floor: every artifact must have been COPIED, not skipped, not failed.
let copied = first
.items
.iter()
.filter(|r| r.outcome == Outcome::Copied)
.count();
anyhow::ensure!(
first.items.len() == n && copied == n,
"replication did not copy the corpus: {} rows, {copied} copied, expected {n} of each \
(first failure: {:?})",
first.items.len(),
first
.items
.iter()
.find(|r| r.outcome != Outcome::Copied)
.map(|r| (r.id.name.clone(), r.detail.clone()))
);
// ── Pass 2: the idempotent rescan. Reads and hashes both sides, writes nothing. ──
let t1 = Instant::now();
let second = replicate_pairs(std::slice::from_ref(&pair), &opts);
let rescan_elapsed = t1.elapsed();
let skipped = second
.items
.iter()
.filter(|r| r.outcome == Outcome::Skipped)
.count();
anyhow::ensure!(
skipped == n,
"a second replicate over an unchanged source must be a pure no-op: {skipped} of {n} skipped"
);
// ANSWER floor: the copy is byte-identical, checked through the destination's
// own read path rather than by trusting the report.
let (probe_id, probe_bytes) = &corpus[n - 1];
let got = dest
.fetch(probe_id)
.context("read the last replicated artifact back out of the destination")?
.ok_or_else(|| anyhow::anyhow!("destination does not hold {}", probe_id.name))?;
anyhow::ensure!(
got.as_ref() == probe_bytes.as_slice(),
"replicated bytes differ from the source for {}",
probe_id.name
);
drop(pair);
drop(source);
drop(dest);
std::fs::remove_dir_all(&dir).ok();
Ok(ReplicationOutcome {
copy: throughput(n, declared, copy_elapsed),
rescan: throughput(n, declared, rescan_elapsed),
})
}
/// Both proxy numbers from one fixture.
pub struct ProxyOutcome {
/// Cache MISS: upstream walk + cache-fill gate + write-through `put` into the
/// writable primary + return the bytes.
pub fill: Throughput,
/// Cache HIT: the same coordinates once the primary holds them.
pub hit: Throughput,
}
/// Drive the real pull-through proxy over N artifacts: once cold, once warm.
pub fn proxy_cache(scale: &Scale) -> Result<ProxyOutcome> {
use server_lib::proxy::ProxyBackend;
let n = hop_count(scale);
let dir = fresh_dir("proxy-arena")?;
// The "remote": a second znippy store standing in for an upstream registry, so
// the arm measures the proxy's own work and not a network.
let corpus = ids_and_payloads(n, "proxy-crate");
let upstream = Arc::new(RustRepoZnippy::with_writable_archive(
"proxy-upstream".into(),
dir.join("upstream.znippy"),
));
for (id, data) in &corpus {
upstream
.put(id, data)
.with_context(|| format!("seed upstream with {}", id.name))?;
}
let primary = Arc::new(RustRepoZnippy::with_writable_archive(
"proxy-cache".into(),
dir.join("primary.znippy"),
));
let proxy = ProxyBackend::new(
primary.clone() as Arc<dyn RepositoryBackendTrait>,
vec![upstream.clone() as Arc<dyn RepositoryBackendTrait>],
);
let declared: u64 = corpus.iter().map(|(_, d)| d.len() as u64).sum();
// ── Pass 1: every coordinate is a MISS. ──
let t0 = Instant::now();
let mut miss_bytes = 0u64;
for (id, _) in &corpus {
let got = proxy.fetch(id)?;
miss_bytes += got.map(|b| b.len() as u64).unwrap_or(0);
}
let fill_elapsed = t0.elapsed();
anyhow::ensure!(
miss_bytes == declared,
"the proxy did not serve the whole corpus on the miss pass: {miss_bytes} of {declared} bytes"
);
// ANSWER floor: a miss must have WRITTEN THROUGH. If the primary does not now
// hold the coordinate, the arm measured a forward and not a cache-and-forward.
let (probe_id, probe_bytes) = &corpus[n - 1];
let cached = primary
.fetch(probe_id)
.context("probe the primary after the fill pass")?
.ok_or_else(|| {
anyhow::anyhow!(
"cache-fill did not write through: primary does not hold {}",
probe_id.name
)
})?;
anyhow::ensure!(
cached.as_ref() == probe_bytes.as_slice(),
"cached bytes differ from the upstream's for {}",
probe_id.name
);
// ── Pass 2: every coordinate is now a primary HIT. ──
let t1 = Instant::now();
let mut hit_bytes = 0u64;
for (id, _) in &corpus {
let got = proxy.fetch(id)?;
hit_bytes += got.map(|b| b.len() as u64).unwrap_or(0);
}
let hit_elapsed = t1.elapsed();
anyhow::ensure!(
hit_bytes == declared,
"the proxy did not serve the whole corpus on the hit pass: {hit_bytes} of {declared} bytes"
);
drop(proxy);
drop(primary);
drop(upstream);
std::fs::remove_dir_all(&dir).ok();
Ok(ProxyOutcome {
fill: throughput(n, declared, fill_elapsed),
hit: throughput(n, declared, hit_elapsed),
})
}
// ─────────────────── the upload gate: hashing the whole body ───────────────────
/// Bytes per upload in the checksum arm. Large enough that the digest is real
/// work (stinky-01 section 2.2 MEASURED sha256 at 0.61 GB/s on one core, so a
/// GiB is 1.75 s of one blocked thread), small enough that N of them fit on a
/// bench box.
const CHECKSUM_PAYLOAD_BYTES: usize = 4 * 1024 * 1024;
/// How many uploads each phase of the checksum arm performs. Fixed N,
/// `HOLGER_BENCH_CHECKSUM_UPLOADS` overrides.
fn checksum_upload_count(scale: &Scale) -> usize {
if let Ok(v) = std::env::var("HOLGER_BENCH_CHECKSUM_UPLOADS") {
if let Ok(n) = v.parse::<usize>() {
if n > 0 {
return n;
}
}
}
scale.entries.clamp(20, 60)
}
/// What the checksum arm measured.
pub struct ChecksumUploadOutcome {
/// Uploads with NO `X-Checksum-Sha256` header: the door collects the body
/// and dispatches it.
pub plain: Throughput,
/// The same uploads WITH the header: the door additionally runs the
/// integrity gate over the received bytes.
pub verified: Throughput,
/// `plain.mb_per_sec / verified.mb_per_sec - 1`, as a percentage. This is
/// the gate's cost, isolated inside one run, so a busy box moves both
/// numbers together and the ratio survives.
pub gate_cost_pct: f64,
pub payload_bytes: usize,
}
/// The upload integrity gate, measured against the same upload without it.
///
/// Two phases over one live Holger gateway and one writable znippy repository,
/// through the real `PUT /{repo}/api/v1/crates/new` door: N uploads with no
/// digest header, then N with `X-Checksum-Sha256` set to the sha256 of the body.
/// The DIFFERENCE is `verify_upload_checksum`.
///
/// This arm exists because nothing in the suite exercised that gate at all --
/// every write arm uses 16 KiB bodies and sends no digest header -- so the whole
/// cost stinky-01 section 2.2 measured had no number in the harness.
pub async fn checksum_upload(scale: &Scale) -> Result<ChecksumUploadOutcome> {
use sha2::{Digest, Sha256};
let n = checksum_upload_count(scale);
let dir = fresh_dir("checksum-arena")?;
let archive = dir.join("checksum.znippy");
let backend = Arc::new(RustRepoZnippy::with_writable_archive(
"chk".into(),
archive.clone(),
));
anyhow::ensure!(
backend.is_writable(),
"checksum arm: the repository at {} is not writable",
archive.display()
);
let as_trait: Arc<dyn RepositoryBackendTrait> = Arc::clone(&backend) as _;
let addr = served::start_holger_gateway(vec![("chk".to_string(), as_trait)]).await?;
let url = format!("http://{addr}/chk/api/v1/crates/new");
// Every body built BEFORE either clock starts: this arm measures the door,
// never the fixture.
let make = |tag: &str, i: usize| -> (Vec<u8>, String) {
let mut data = vec![0u8; CHECKSUM_PAYLOAD_BYTES];
let mut x = (i as u64)
.wrapping_mul(0x9E37_79B9_7F4A_7C15)
.wrapping_add(7);
for b in data.iter_mut() {
x ^= x << 13;
x ^= x >> 7;
x ^= x << 17;
*b = (x & 0xFF) as u8;
}
let name = format!("chk-{tag}-{i:05}");
let meta = format!(r#"{{"name":"{name}","vers":"1.0.0","deps":[],"features":{{}}}}"#);
let mut body = Vec::with_capacity(8 + meta.len() + data.len());
body.extend_from_slice(&(meta.len() as u32).to_le_bytes());
body.extend_from_slice(meta.as_bytes());
body.extend_from_slice(&(data.len() as u32).to_le_bytes());
body.extend_from_slice(&data);
let digest = hex_lower(&Sha256::digest(&body));
(body, digest)
};
let plain_bodies: Vec<(Vec<u8>, String)> = (0..n).map(|i| make("plain", i)).collect();
let verified_bodies: Vec<(Vec<u8>, String)> = (0..n).map(|i| make("verify", i)).collect();
let client = tls_roots_blocking(reqwest::blocking::Client::builder())
.tcp_keepalive(Duration::from_secs(30))
.timeout(Duration::from_secs(300))
.build()?;
let run = |bodies: &Vec<(Vec<u8>, String)>, with_digest: bool| -> Result<(u64, Duration)> {
let mut bytes = 0u64;
let t0 = Instant::now();
for (body, digest) in bodies {
let mut req = client.put(&url).body(body.clone());
if with_digest {
req = req.header("X-Checksum-Sha256", digest.as_str());
}
let resp = req.send()?;
let status = resp.status();
let _ = resp.bytes();
anyhow::ensure!(
status.is_success(),
"upload refused with {status} — the arm would be measuring a wall of errors"
);
bytes += body.len() as u64;
}
Ok((bytes, t0.elapsed()))
};
// Phase order matters only for the archive's size, which grows the same in
// both phases; the ratio is what is read, not either absolute.
let (plain_bytes, plain_elapsed) = run(&plain_bodies, false)?;
let (verified_bytes, verified_elapsed) = run(&verified_bodies, true)?;
// ANSWER floor: a gate that never refuses is not a gate. Send one body with a
// deliberately wrong digest and require a 400.
let (bad_body, _) = make("bad", 0);
let refused = client
.put(&url)
.header(
"X-Checksum-Sha256",
"0000000000000000000000000000000000000000000000000000000000000000",
)
.body(bad_body)
.send()?;
anyhow::ensure!(
refused.status().as_u16() == 400,
"the upload gate admitted a body whose advertised sha256 does not match it: {}",
refused.status()
);
// ANSWER floor: the verified uploads are actually IN the archive.
let last = ArtifactId {
namespace: None,
name: format!("chk-verify-{:05}", n - 1),
version: "1.0.0".into(),
};
anyhow::ensure!(
backend.fetch(&last)?.is_some(),
"the write path produced an archive that does not hold {}",
last.name
);
drop(backend);
std::fs::remove_dir_all(&dir).ok();
let plain = throughput(n, plain_bytes, plain_elapsed);
let verified = throughput(n, verified_bytes, verified_elapsed);
let gate_cost_pct = if verified.mb_per_sec > 0.0 {
(plain.mb_per_sec / verified.mb_per_sec - 1.0) * 100.0
} else {
0.0
};
Ok(ChecksumUploadOutcome {
plain,
verified,
gate_cost_pct,
payload_bytes: CHECKSUM_PAYLOAD_BYTES,
})
}
/// Lowercase hex. Local so this module needs no extra dependency for four lines.
fn hex_lower(bytes: &[u8]) -> String {
const HEX: &[u8; 16] = b"0123456789abcdef";
let mut out = String::with_capacity(bytes.len() * 2);
for b in bytes {
out.push(HEX[(b >> 4) as usize] as char);
out.push(HEX[(b & 0x0F) as usize] as char);
}
out
}
/// reqwest 0.13 with reqwest 0.12 `rustls-tls` behaviour (2026-10-02): the ring
/// provider and the bundled Mozilla roots (webpki), not 0.13's aws-lc-rs + OS
/// trust store. Roots added later on the builder are merged in, as in 0.12.
#[allow(dead_code)]
fn tls_roots_blocking(b: reqwest::blocking::ClientBuilder) -> reqwest::blocking::ClientBuilder {
let _ = rustls::crypto::ring::default_provider().install_default();
b.tls_certs_only(
webpki_root_certs::TLS_SERVER_ROOT_CERTS
.iter()
.filter_map(|c| reqwest::Certificate::from_der(c).ok()),
)
}