agent/cli/tests/integration_latency_tail.rs
//! Tail-latency comparison: Holger (znippy in-process) vs Nexus (HTTP) under GC pressure.
//!
//! This test uses the `testcontainers` crate to start a real Sonatype Nexus 3 (v3.72.0)
//! container configured with a constrained 512 MB JVM heap. This induces frequent G1GC
//! stop-the-world pauses, surfacing latency spikes in p99/p999 that don't show in
//! ops/s averages.
//!
//! ## Prerequisites
//!
//! - Podman socket enabled: `systemctl --user start podman.socket`
//! - Environment: `export DOCKER_HOST=unix:///run/user/1000/podman/podman.sock`
//! - Pre-cached znippy archives in `~/work/holger_tests/`
//! (auto-created by `integration_znippy_serve` test on first run)
//!
//! ## What happens
//!
//! 1. Starts Nexus 3 with `-Xms512m -Xmx512m -XX:+UseG1GC -XX:MaxGCPauseMillis=50`
//! 2. Uploads cached archives to fill heap and trigger GC
//! 3. 60-second bombardment on Nexus: records per-request latency, computes p50/p99/p999
//! 4. Same bombardment on Holger/znippy (in-process, no GC, no network)
//! 5. Prints side-by-side comparison table
//!
//! ## Run
//!
//! ```bash
//! DOCKER_HOST=unix:///run/user/1000/podman/podman.sock \
//! cargo test --release -p holger-agent-cli \
//! --test integration_latency_tail -- --ignored --nocapture
//! ```
use agent_lib::{operations, ConnectorTrait, NexusConnector};
use repository_znippy_maven::MavenRepoZnippy;
use repository_znippy_python::PipRepoZnippy;
use repository_znippy_rust::RustRepoZnippy;
use std::path::PathBuf;
use std::sync::Arc;
use std::time::{Duration, Instant};
use testcontainers::{
core::{ContainerPort, ExecCommand, WaitFor},
runners::AsyncRunner,
ContainerAsync, GenericImage, ImageExt,
};
/// `$HOME/work/holger_tests` — the pre-cached znippy archives.
fn holger_test_dir() -> String {
format!(
"{}/work/holger_tests",
std::env::var("HOME").unwrap_or_default()
)
}
const BENCH_SECS: u64 = 60;
// Small heap → frequent G1GC collections; MaxGCPauseMillis=50 means GC tries to keep
// pauses ≤50ms but will exceed that when heap is nearly full.
const NEXUS_JVM_ARGS: &str =
"-Xms512m -Xmx512m -XX:+UseG1GC -XX:MaxGCPauseMillis=50 -Xlog:gc:stderr";
// ── container helpers ────────────────────────────────────────────────────────
async fn read_nexus_admin_password(container: &ContainerAsync<GenericImage>) -> String {
let mut result = container
.exec(ExecCommand::new(["cat", "/nexus-data/admin.password"]))
.await
.expect("exec cat admin.password");
let stdout = result.stdout_to_vec().await.unwrap();
String::from_utf8(stdout).unwrap().trim().to_string()
}
async fn wait_for_nexus(base_url: &str, timeout: Duration) -> bool {
let client = tls_roots(reqwest::Client::builder())
.timeout(Duration::from_secs(5))
.build()
.unwrap();
let start = Instant::now();
while start.elapsed() < timeout {
if let Ok(resp) = client
.get(format!("{}/service/rest/v1/status", base_url))
.send()
.await
{
if resp.status().is_success() {
return true;
}
}
tokio::time::sleep(Duration::from_secs(2)).await;
}
false
}
async fn create_raw_repo(client: &reqwest::Client, base_url: &str, repo: &str) {
let body = serde_json::json!({
"name": repo,
"online": true,
"storage": {
"blobStoreName": "default",
"strictContentTypeValidation": false,
"writePolicy": "ALLOW"
}
});
let resp = client
.post(format!(
"{base_url}/service/rest/v1/repositories/raw/hosted"
))
.basic_auth("admin", Some("admin123"))
.json(&body)
.send()
.await
.unwrap();
let s = resp.status().as_u16();
assert!(s == 201 || s == 400, "create repo {repo}: status {s}");
}
// ── latency helpers ──────────────────────────────────────────────────────────
fn pct(sorted: &[Duration], p: f64) -> f64 {
if sorted.is_empty() {
return 0.0;
}
let idx = ((sorted.len() as f64 * p / 100.0) as usize).min(sorted.len() - 1);
sorted[idx].as_secs_f64() * 1000.0
}
fn print_row(label: &str, threads: usize, mut samples: Vec<Duration>) {
samples.sort_unstable();
let n = samples.len();
let ops_per_sec = n as f64 / BENCH_SECS as f64;
println!(
" {label:<24} {threads:>2}t {n:>8} ops {ops_per_sec:>8.0} ops/s \
| p50 {:.2}ms p95 {:.2}ms p99 {:.2}ms p999 {:.2}ms max {:.1}ms",
pct(&samples, 50.0),
pct(&samples, 95.0),
pct(&samples, 99.0),
pct(&samples, 99.9),
pct(&samples, 100.0),
);
}
/// Bombard `urls` with `threads` blocking HTTP clients for `BENCH_SECS`.
/// Returns a flat Vec of individual request latencies.
fn nexus_latencies(urls: &[String], threads: usize) -> Vec<Duration> {
std::thread::scope(|s| {
let handles: Vec<_> = (0..threads)
.map(|tid| {
s.spawn(move || {
let client = tls_roots_blocking(reqwest::blocking::Client::builder())
.tcp_keepalive(Duration::from_secs(30))
.build()
.unwrap();
let start = Instant::now();
let limit = Duration::from_secs(BENCH_SECS);
let mut ops = 0u64;
let mut lat = Vec::new();
while start.elapsed() < limit {
let idx = ((ops as usize) * 7 + tid * 1337) % urls.len();
let t0 = Instant::now();
let resp = client.get(&urls[idx]).send().unwrap();
let _body = resp.bytes().unwrap();
lat.push(t0.elapsed());
ops += 1;
}
lat
})
})
.collect();
handles
.into_iter()
.flat_map(|h| h.join().unwrap())
.collect()
})
}
/// Call `get_fn(op_index, thread_id)` from `threads` threads for `BENCH_SECS`.
/// Returns a flat Vec of individual call latencies.
fn holger_latencies<F>(threads: usize, get_fn: &F) -> Vec<Duration>
where
F: Fn(u64, usize) -> usize + Send + Sync,
{
std::thread::scope(|s| {
let handles: Vec<_> = (0..threads)
.map(|tid| {
s.spawn(move || {
let start = Instant::now();
let limit = Duration::from_secs(BENCH_SECS);
let mut ops = 0u64;
let mut lat = Vec::new();
while start.elapsed() < limit {
let t0 = Instant::now();
let _ = get_fn(ops, tid);
lat.push(t0.elapsed());
ops += 1;
}
lat
})
})
.collect();
handles
.into_iter()
.flat_map(|h| h.join().unwrap())
.collect()
})
}
// ── test ─────────────────────────────────────────────────────────────────────
#[tokio::test]
#[ignore = "heavy: needs Docker — boots a real Nexus 3 container (run with --ignored)"]
async fn test_tail_latency_gc_pressure() {
println!("\n🚀 Starting Nexus 3 with 512 MB heap (GC pressure)...");
println!(" JVM args: {NEXUS_JVM_ARGS}");
let image = GenericImage::new("sonatype/nexus3", "3.72.0")
.with_exposed_port(ContainerPort::Tcp(8081))
.with_wait_for(WaitFor::message_on_stdout("Started Sonatype Nexus OSS"))
.with_env_var("INSTALL4J_ADD_VM_PARAMS", NEXUS_JVM_ARGS);
let container = image.start().await.expect("start Nexus container");
let port = container
.get_host_port_ipv4(ContainerPort::Tcp(8081))
.await
.unwrap();
let base_url = format!("http://127.0.0.1:{port}");
println!("⏳ Waiting for Nexus on port {port} (may take 2-3 min with 512 MB heap)...");
assert!(
wait_for_nexus(&base_url, Duration::from_secs(360)).await,
"Nexus did not start within 6 minutes"
);
println!("✅ Nexus ready\n");
// Change password
let init_pw = read_nexus_admin_password(&container).await;
let client = tls_roots(reqwest::Client::builder())
.build()
.expect("reqwest client");
client
.put(format!(
"{base_url}/service/rest/v1/security/users/admin/change-password"
))
.basic_auth("admin", Some(&init_pw))
.header("Content-Type", "text/plain")
.body("admin123")
.send()
.await
.unwrap()
.error_for_status()
.unwrap();
// Create repos and upload
let archives: &[(&str, &str, &str)] = &[
("crates_archive.znippy", "crates-raw", "Rust crates"),
("jars_archive.znippy", "jars-raw", "Maven JARs"),
("pip_archive.znippy", "wheels-raw", "Python wheels"),
];
let connector = NexusConnector::new(
base_url.clone(),
Some("admin".into()),
Some("admin123".into()),
)
.unwrap();
struct RepoUrls {
label: &'static str,
urls: Vec<String>,
}
let mut repo_urls: Vec<RepoUrls> = Vec::new();
for (filename, repo, label) in archives {
let path = PathBuf::from(format!("{}/{filename}", holger_test_dir()));
if !path.exists() {
println!("⚠️ Skipping {label} — {path:?} not found");
continue;
}
create_raw_repo(&client, &base_url, repo).await;
println!("📤 Uploading {label}...");
operations::push_znippy_to_connector(&connector, path, repo)
.await
.unwrap_or_else(|e| panic!("upload {label}: {e}"));
let assets = connector.list_assets(repo).await.unwrap();
let urls: Vec<String> = assets
.iter()
.map(|a| format!("{base_url}/repository/{repo}/{}", a.path))
.collect();
println!(" → {} files ready", urls.len());
repo_urls.push(RepoUrls { label, urls });
}
// Load Holger repos into memory
let rust_repo = Arc::new(
RustRepoZnippy::with_archive(
"rust".into(),
PathBuf::from(format!("{}/crates_archive.znippy", holger_test_dir())),
)
.unwrap(),
);
let maven_repo = Arc::new(
MavenRepoZnippy::with_archive(
"maven".into(),
PathBuf::from(format!("{}/jars_archive.znippy", holger_test_dir())),
)
.unwrap(),
);
let pip_repo = Arc::new(
PipRepoZnippy::with_archive(
"pip".into(),
PathBuf::from(format!("{}/pip_archive.znippy", holger_test_dir())),
)
.unwrap(),
);
let rust_files = rust_repo.list_files();
let maven_files = maven_repo.list_files();
let pip_files = pip_repo.list_files();
println!(
"\n Holger files: Rust={}, Maven={}, Pip={}\n",
rust_files.len(),
maven_files.len(),
pip_files.len()
);
// ── benchmark loop ───────────────────────────────────────────────────────
for threads in [1usize, 4, 8, 16] {
println!("{}", "═".repeat(110));
println!(
" {} thread(s) × {}s (Nexus heap: 512 MB with G1GC)",
threads, BENCH_SECS
);
println!("{}", "─".repeat(110));
println!(
" {:<24} {:>2} {:>8} {:>9} {:>9} {:>9} {:>9} {:>9} {:>9}",
"backend", "t", "ops", "ops/s", "p50", "p95", "p99", "p999", "max"
);
println!("{}", "─".repeat(110));
// Nexus
for ru in &repo_urls {
let samples = nexus_latencies(&ru.urls, threads);
print_row(&format!("Nexus/{}", ru.label), threads, samples);
}
println!();
// Holger — Rust crates
{
let r = Arc::clone(&rust_repo);
let files = rust_files.clone();
let f = move |i: u64, tid: usize| -> usize {
let idx = ((i as usize) * 7 + tid * 1337) % files.len();
r.get_file(&files[idx]).unwrap().len()
};
print_row("Holger/Rust crates", threads, holger_latencies(threads, &f));
}
// Holger — Maven JARs
{
let r = Arc::clone(&maven_repo);
let files = maven_files.clone();
let f = move |i: u64, tid: usize| -> usize {
let idx = ((i as usize) * 11 + tid * 773) % files.len();
r.get_file(&files[idx]).unwrap().len()
};
print_row("Holger/Maven JARs", threads, holger_latencies(threads, &f));
}
// Holger — Python wheels
{
let r = Arc::clone(&pip_repo);
let files = pip_files.clone();
let f = move |i: u64, tid: usize| -> usize {
let idx = ((i as usize) * 13 + tid * 997) % files.len();
r.get_file(&files[idx]).unwrap().len()
};
print_row(
"Holger/Python wheels",
threads,
holger_latencies(threads, &f),
);
}
println!();
}
println!("{}", "═".repeat(110));
println!("=== Tail latency / GC pressure test complete ===");
}
/// 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(b: reqwest::ClientBuilder) -> reqwest::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()),
)
}
/// 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()),
)
}