agent/cli/tests/integration_nexus_load.rs
//! Load test: push cached znippy archives to a real Sonatype Nexus 3 container.
//!
//! This test uses the `testcontainers` crate to automatically start a real
//! Sonatype Nexus 3 (v3.72.0) container via Podman (or Docker). The container
//! is started, configured, used for benchmarking, and torn down — all within
//! the test.
//!
//! ## 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. Pulls and starts `sonatype/nexus3:3.72.0` container (port 8081 → random host port)
//! 2. Waits up to 180s for Nexus REST API to become ready
//! 3. Reads initial admin password from container filesystem
//! 4. Changes admin password to `admin123`, creates raw hosted repositories
//! 5. Pushes all cached znippy archives (crates, JARs, wheels) → measures upload MB/s
//! 6. Downloads all artifacts back → measures download MB/s
//! 7. Multi-threaded retrieval bombardment (1, 2, 4, 8, N threads) on Nexus
//! 8. Same bombardment directly on Holger/znippy for comparison
//! 9. Container is automatically stopped and removed
//!
//! ## Run
//!
//! ```bash
//! DOCKER_HOST=unix:///run/user/1000/podman/podman.sock \
//! cargo test --release -p holger-agent-cli \
//! --test integration_nexus_load -- --ignored --nocapture
//! ```
use agent_lib::operations;
use agent_lib::{ConnectorTrait, NexusConnector};
use std::path::PathBuf;
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()
)
}
/// Read the initial admin password from the Nexus container.
async fn read_nexus_admin_password(container: &ContainerAsync<GenericImage>) -> String {
let mut result = container
.exec(ExecCommand::new(["cat", "/nexus-data/admin.password"]))
.await
.expect("Failed to exec cat admin.password");
let stdout = result.stdout_to_vec().await.unwrap();
String::from_utf8(stdout).unwrap().trim().to_string()
}
/// Wait for Nexus to be fully started (it takes 60-120s on first boot).
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
}
/// Create a raw hosted repository in Nexus via the scripting API or REST API.
async fn create_raw_repo(base_url: &str, repo_name: &str) {
let client = tls_roots(reqwest::Client::builder())
.build()
.expect("reqwest client");
// Nexus 3 REST API to create a raw hosted repository
let body = serde_json::json!({
"name": repo_name,
"online": true,
"storage": {
"blobStoreName": "default",
"strictContentTypeValidation": false,
"writePolicy": "ALLOW"
}
});
let resp = client
.post(format!(
"{}/service/rest/v1/repositories/raw/hosted",
base_url
))
.basic_auth("admin", Some("admin123"))
.json(&body)
.send()
.await
.expect("Failed to create repository");
// 201 Created or 400 if already exists
let status = resp.status().as_u16();
assert!(
status == 201 || status == 400,
"Unexpected status creating repo {}: {}",
repo_name,
status
);
}
/// Push a znippy archive to Nexus, return (file_count, total_bytes, duration).
async fn push_archive_to_nexus(
base_url: &str,
archive_path: &str,
repo_name: &str,
) -> (usize, u64, Duration) {
let path = PathBuf::from(archive_path);
if !path.exists() {
panic!(
"Archive not found: {} — run integration_znippy_serve first to build caches",
archive_path
);
}
let connector = NexusConnector::new(
base_url.to_string(),
Some("admin".into()),
Some("admin123".into()),
)
.unwrap();
let archive_size = std::fs::metadata(&path).unwrap().len();
let start = Instant::now();
operations::push_znippy_to_connector(&connector, path, repo_name)
.await
.expect("push_znippy_to_connector failed");
let elapsed = start.elapsed();
// Count files by listing what we pushed
let assets = connector.list_assets(repo_name).await.unwrap();
let file_count = assets.len();
(file_count, archive_size, elapsed)
}
/// Download all assets from a Nexus repo, return (file_count, total_bytes, duration).
async fn download_all_from_nexus(base_url: &str, repo_name: &str) -> (usize, u64, Duration) {
let connector = NexusConnector::new(
base_url.to_string(),
Some("admin".into()),
Some("admin123".into()),
)
.unwrap();
let assets = connector.list_assets(repo_name).await.unwrap();
let file_count = assets.len();
let start = Instant::now();
let mut total_bytes = 0u64;
for asset in &assets {
let data = connector.download_asset(asset).await.unwrap();
total_bytes += data.len() as u64;
}
let elapsed = start.elapsed();
(file_count, total_bytes, elapsed)
}
/// Compute JVM args using Sonatype's actual Nexus-3 sizing guidance, not raw
/// "give it 80% of the box". Nexus is **not heap-bound**:
///
/// - **Heap** holds metadata + Orient/H2 DB caches. Sonatype's own docs cap
/// their "very large" recommendation at ~16 GB; default is 2703 MB. More
/// heap = bigger GC pauses, no throughput win because blob bytes never
/// live on the heap. We scale heap = `min(cores * 0.5 GB, 16 GB)`,
/// floor 4 GB.
/// - **Direct memory** is where blob I/O buffers + NIO sockets live.
/// Sonatype recommends >= heap; for high-throughput servers 2-4× heap.
/// We use `heap * 4`, capped at 64 GB.
/// - **Kernel page cache** does the real lifting for repeated blob reads.
/// By leaving most RAM unallocated, the OS caches the blobstore tree,
/// which on a 499 GB host can mean every artifact lives in RAM after
/// first touch. That's the win on a monster host — not a 400 GB heap.
/// - **Jetty threads**: blocking I/O work. ~10-20 per core is the sweet
/// spot; >~512 trashes throughput via context switching. Floor 200
/// (Nexus default), ceiling 512.
/// - **G1GC** with a 200 ms pause target — throughput-leaning but still
/// responsive.
fn nexus_jvm_args_for_host() -> String {
let cores = std::thread::available_parallelism()
.map(|n| n.get())
.unwrap_or(8);
let mem_kb = std::fs::read_to_string("/proc/meminfo")
.ok()
.and_then(|s| {
s.lines()
.find(|l| l.starts_with("MemTotal:"))
.and_then(|l| l.split_whitespace().nth(1))
.and_then(|n| n.parse::<u64>().ok())
})
.unwrap_or(8 * 1024 * 1024);
let mem_gb = mem_kb / 1024 / 1024;
// Heap: cores × 0.5 GB, clamped [4, 16]. Avoid >32 GB compressed-oops
// cliff entirely — Sonatype's own "very large" sizing tops out at 16 GB.
let heap_gb = ((cores as u64) / 2).clamp(4, 16);
// Direct memory: 4× heap, capped at 64 GB or 25% of host RAM (whichever
// smaller). Nexus uses direct buffers for blob streaming + NIO; this is
// where to spend RAM, not heap.
let direct_gb = (heap_gb * 4).min(64).min(mem_gb / 4);
// Jetty threads: blocking I/O, 16/core is the sweet spot. Clamp [200, 512].
let max_threads = ((cores * 16) as u64).clamp(200, 512);
let min_threads = (max_threads / 4).max(50);
// Code cache: Nexus's plugin system + Groovy scripts generate a lot of
// JIT'd code; default 240 MB can fill on big installs.
let reserved_code_mb = 512;
format!(
"-Xms{heap}g -Xmx{heap}g \
-XX:MaxDirectMemorySize={direct}g \
-XX:ReservedCodeCacheSize={code}m \
-XX:+UseG1GC -XX:MaxGCPauseMillis=200 \
-XX:+ParallelRefProcEnabled \
-XX:+AlwaysPreTouch \
-XX:+ExplicitGCInvokesConcurrent \
-Djava.util.prefs.userRoot=/nexus-data/javaprefs \
-Djetty.threadPool.maxThreads={max_t} \
-Djetty.threadPool.minThreads={min_t} \
-Djetty.threadPool.idleTimeout=60000",
heap = heap_gb,
direct = direct_gb,
code = reserved_code_mb,
max_t = max_threads,
min_t = min_threads,
)
}
/// Full Nexus load test: upload znippy archives then download and benchmark.
#[tokio::test]
#[ignore = "heavy: needs Docker — boots a real Nexus 3 container (run with --ignored)"]
async fn test_nexus_load_with_real_artifacts() {
println!("🚀 Starting Sonatype Nexus 3 container...");
// Dynamic tuning: 80% of host cores + memory. Defaults (~2.7g heap, 200
// Jetty threads) would bottleneck on a beefy host; on a tiny CI host we
// still want a reasonable floor.
let nexus_jvm_args = nexus_jvm_args_for_host();
println!(" JVM args: {nexus_jvm_args}");
let nexus_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 = nexus_image
.start()
.await
.expect("Failed to start Nexus container");
let port = container
.get_host_port_ipv4(ContainerPort::Tcp(8081))
.await
.expect("Failed to get port");
let base_url = format!("http://127.0.0.1:{}", port);
println!("⏳ Waiting for Nexus to start on port {}...", port);
assert!(
wait_for_nexus(&base_url, Duration::from_secs(180)).await,
"Nexus did not start within 180 seconds"
);
println!("✅ Nexus is ready!");
// Read initial admin password from container
let admin_password = read_nexus_admin_password(&container).await;
println!("🔑 Got admin password ({} chars)", admin_password.len());
// Change admin password to something known, and disable anonymous access setup wizard
let client = tls_roots(reqwest::Client::builder())
.build()
.expect("reqwest client");
// First, change password via Nexus API
let resp = client
.put(format!(
"{}/service/rest/v1/security/users/admin/change-password",
base_url
))
.basic_auth("admin", Some(&admin_password))
.header("Content-Type", "text/plain")
.body("admin123")
.send()
.await
.unwrap();
assert!(
resp.status().is_success(),
"Failed to change admin password: {}",
resp.status()
);
println!("🔐 Admin password set to 'admin123'");
// Create repositories for each artifact type
for repo in &["crates-raw", "jars-raw", "wheels-raw"] {
create_raw_repo(&base_url, repo).await;
println!("📁 Created repository: {}", repo);
}
// Define archives to push
let archives = [
(
format!("{}/crates_archive.znippy", holger_test_dir()),
"crates-raw",
"Rust crates",
),
(
format!("{}/jars_archive.znippy", holger_test_dir()),
"jars-raw",
"Maven JARs",
),
(
format!("{}/pip_archive.znippy", holger_test_dir()),
"wheels-raw",
"Python wheels",
),
];
println!("\n{}", "=".repeat(60));
println!("=== UPLOAD BENCHMARK ===");
println!("{}\n", "=".repeat(60));
let mut total_files = 0usize;
let mut total_bytes = 0u64;
let mut total_upload_time = Duration::ZERO;
for (path, repo, label) in &archives {
if !PathBuf::from(path).exists() {
println!("⚠️ Skipping {} — archive not found: {}", label, path);
continue;
}
println!("📤 Uploading {} → {}...", label, repo);
let (files, bytes, elapsed) = push_archive_to_nexus(&base_url, path, repo).await;
let mb = bytes as f64 / (1024.0 * 1024.0);
let mb_per_sec = mb / elapsed.as_secs_f64();
let files_per_sec = files as f64 / elapsed.as_secs_f64();
println!(
" {} — {} files, {:.1} MB archive, {:.1}s → {:.1} MB/s, {:.0} files/s",
label,
files,
mb,
elapsed.as_secs_f64(),
mb_per_sec,
files_per_sec
);
total_files += files;
total_bytes += bytes;
total_upload_time += elapsed;
}
let total_mb = total_bytes as f64 / (1024.0 * 1024.0);
println!(
"\n📊 Upload total: {} files, {:.1} MB in {:.1}s → {:.1} MB/s",
total_files,
total_mb,
total_upload_time.as_secs_f64(),
total_mb / total_upload_time.as_secs_f64()
);
println!("\n{}", "=".repeat(60));
println!("=== DOWNLOAD BENCHMARK ===");
println!("{}\n", "=".repeat(60));
let mut total_download_bytes = 0u64;
let mut total_download_time = Duration::ZERO;
let mut total_download_files = 0usize;
for (_path, repo, label) in &archives {
if !PathBuf::from(_path).exists() {
continue;
}
println!("📥 Downloading all from {}...", repo);
let (files, bytes, elapsed) = download_all_from_nexus(&base_url, repo).await;
let mb = bytes as f64 / (1024.0 * 1024.0);
let mb_per_sec = mb / elapsed.as_secs_f64();
let files_per_sec = files as f64 / elapsed.as_secs_f64();
println!(
" {} — {} files, {:.1} MB, {:.1}s → {:.1} MB/s, {:.0} files/s",
label,
files,
mb,
elapsed.as_secs_f64(),
mb_per_sec,
files_per_sec
);
total_download_files += files;
total_download_bytes += bytes;
total_download_time += elapsed;
}
let total_dl_mb = total_download_bytes as f64 / (1024.0 * 1024.0);
println!(
"\n📊 Download total: {} files, {:.1} MB in {:.1}s → {:.1} MB/s",
total_download_files,
total_dl_mb,
total_download_time.as_secs_f64(),
total_dl_mb / total_download_time.as_secs_f64()
);
// ── Phase 3: Bombardment benchmark (random GETs) ─────────────────────
println!("\n{}", "=".repeat(60));
println!("=== RETRIEVAL BOMBARDMENT (Nexus) ===");
println!("{}\n", "=".repeat(60));
let num_cpus = std::thread::available_parallelism()
.map(|n| n.get())
.unwrap_or(4);
let bench_secs = 10u64;
println!(" CPUs: {}, duration: {}s per test\n", num_cpus, bench_secs);
// Collect asset download URLs per repo
let connector = NexusConnector::new(
base_url.clone(),
Some("admin".into()),
Some("admin123".into()),
)
.unwrap();
struct RepoAssets {
label: &'static str,
urls: Vec<String>,
}
let mut repo_assets: Vec<RepoAssets> = Vec::new();
for (_path, repo, label) in &archives {
if !PathBuf::from(_path).exists() {
continue;
}
let assets = connector.list_assets(repo).await.unwrap();
let urls: Vec<String> = assets
.iter()
.map(|a| format!("{}/repository/{}/{}", base_url, repo, a.path))
.collect();
println!(" {} — {} URLs for bombardment", label, urls.len());
repo_assets.push(RepoAssets { label, urls });
}
// Bombardment helper: spawn threads, each doing random GETs for `duration`
fn bombard_nexus(
label: &str,
urls: &[String],
num_threads: usize,
duration: Duration,
) -> (u64, Duration) {
use std::sync::atomic::{AtomicU64, Ordering};
let total_ops = AtomicU64::new(0);
let total_ops_ref = &total_ops;
let start = Instant::now();
std::thread::scope(|s| {
for tid in 0..num_threads {
s.spawn(move || {
let client = tls_roots_blocking(reqwest::blocking::Client::builder())
.build()
.expect("reqwest client");
let mut ops = 0u64;
while start.elapsed() < duration {
let idx = ((ops as usize) * 7 + tid * 1337) % urls.len();
let resp = client.get(&urls[idx]).send().unwrap();
let _bytes = resp.bytes().unwrap();
ops += 1;
}
total_ops_ref.fetch_add(ops, Ordering::Relaxed);
});
}
});
let elapsed = start.elapsed();
let ops = total_ops.load(std::sync::atomic::Ordering::Relaxed);
let rate = ops as f64 / elapsed.as_secs_f64();
println!(
" {} ({} threads): {:>10.0} ops/sec ({} ops in {:.1}s)",
label,
num_threads,
rate,
ops,
elapsed.as_secs_f64()
);
(ops, elapsed)
}
let bench_duration = Duration::from_secs(bench_secs);
// Single-threaded
println!("── Single-threaded ──────────────────────────────────────────────");
for ra in &repo_assets {
bombard_nexus(ra.label, &ra.urls, 1, bench_duration);
}
// Multi-threaded (sweep: 2, 4, 8, num_cpus)
for threads in [2, 4, 8, num_cpus]
.iter()
.copied()
.filter(|&t| t <= num_cpus)
{
println!(
"\n── {} threads ────────────────────────────────────────────────",
threads
);
for ra in &repo_assets {
bombard_nexus(ra.label, &ra.urls, threads, bench_duration);
}
}
// ── Tear down Nexus before Holger benchmark for clean isolation ────
println!("\n⏹ Stopping Nexus container...");
drop(container);
tokio::time::sleep(Duration::from_secs(2)).await;
println!("✅ Nexus stopped — running Holger benchmark in isolation\n");
// ── Phase 4: Holger (znippy direct) comparison ───────────────────────
println!("{}", "=".repeat(60));
println!("=== RETRIEVAL BOMBARDMENT (Holger/znippy direct) ===");
println!("{}\n", "=".repeat(60));
use repository_znippy_maven::MavenRepoZnippy;
use repository_znippy_python::PipRepoZnippy;
use repository_znippy_rust::RustRepoZnippy;
use std::sync::Arc;
let crate_archive = PathBuf::from(format!("{}/crates_archive.znippy", holger_test_dir()));
let jar_archive = PathBuf::from(format!("{}/jars_archive.znippy", holger_test_dir()));
let pip_archive = PathBuf::from(format!("{}/pip_archive.znippy", holger_test_dir()));
let rust_repo =
Arc::new(RustRepoZnippy::with_archive("rust".to_string(), crate_archive).unwrap());
let maven_repo =
Arc::new(MavenRepoZnippy::with_archive("maven".to_string(), jar_archive).unwrap());
let pip_repo = Arc::new(PipRepoZnippy::with_archive("pip".to_string(), pip_archive).unwrap());
let rust_files: Vec<String> = rust_repo.list_files();
let maven_files: Vec<String> = maven_repo.list_files();
let pip_files: Vec<String> = pip_repo.list_files();
println!(
" Rust: {} files, Maven: {} files, Pip: {} files\n",
rust_files.len(),
maven_files.len(),
pip_files.len()
);
fn bench_znippy<F: Fn(u64, usize) -> usize + Send + Sync>(
label: &str,
num_threads: usize,
duration: Duration,
get_fn: &F,
) -> (u64, Duration) {
use std::sync::atomic::{AtomicU64, Ordering};
let total_ops = AtomicU64::new(0);
let total_ops_ref = &total_ops;
let start = Instant::now();
std::thread::scope(|s| {
for tid in 0..num_threads {
s.spawn(move || {
let mut ops = 0u64;
while start.elapsed() < duration {
let _ = get_fn(ops, tid);
ops += 1;
}
total_ops_ref.fetch_add(ops, Ordering::Relaxed);
});
}
});
let elapsed = start.elapsed();
let ops = total_ops.load(Ordering::Relaxed);
let rate = ops as f64 / elapsed.as_secs_f64();
println!(
" {} ({} threads): {:>10.0} ops/sec ({} ops in {:.1}s)",
label,
num_threads,
rate,
ops,
elapsed.as_secs_f64()
);
(ops, elapsed)
}
// Single-threaded holger
println!("── Single-threaded ──────────────────────────────────────────────");
{
let r = Arc::clone(&rust_repo);
let files = rust_files.clone();
let f = move |i: u64, _tid: usize| -> usize {
r.get_file(&files[(i as usize) % files.len()])
.unwrap()
.len()
};
bench_znippy("Rust crates", 1, bench_duration, &f);
}
{
let r = Arc::clone(&maven_repo);
let files = maven_files.clone();
let f = move |i: u64, _tid: usize| -> usize {
r.get_file(&files[(i as usize) % files.len()])
.unwrap()
.len()
};
bench_znippy("Maven JARs", 1, bench_duration, &f);
}
{
let r = Arc::clone(&pip_repo);
let files = pip_files.clone();
let f = move |i: u64, _tid: usize| -> usize {
r.get_file(&files[(i as usize) % files.len()])
.unwrap()
.len()
};
bench_znippy("Python wheels", 1, bench_duration, &f);
}
// Multi-threaded holger
println!(
"\n── {} threads ────────────────────────────────────────────────",
num_cpus
);
{
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()
};
bench_znippy("Rust crates", num_cpus, bench_duration, &f);
}
{
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()
};
bench_znippy("Maven JARs", num_cpus, bench_duration, &f);
}
{
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()
};
bench_znippy("Python wheels", num_cpus, bench_duration, &f);
}
println!("\n=== Holger vs Nexus load 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()),
)
}