bench-scenarios/src/stress.rs
//! Write-side stress scenarios — the arms the suite did not have.
//!
//! ## Why this module exists
//!
//! Every pre-existing arm in `holger-bench-scenarios` is **read-heavy**. The
//! rival head-to-heads bombard GETs; `rust_crate_*` / `java_jar_*` /
//! `pip_wheel_*` read archives; `pip_index_serve` serves an index. The only
//! write numbers in the whole suite were `promote_ledger` and
//! `audit_append`, and both are deliberately **single-writer**.
//!
//! Two measurements from the 2026-09-20 baseline say where the pain is:
//!
//! * `promote_ledger` did **606 seals/s on loki's 12 cores and 349 on oden's
//! 32** — twenty more cores made sealing **42 % slower**, because the seal
//! path is strictly serial and therefore tracks single-core clock. The same
//! hardware step took `pip_wheel_mt32` from 1918 to 4224 MB/s.
//! * `secure_update_prepare` went 10.84 → 7.69 MB/s while pinning 32 of 32
//! cores — busy cores, no throughput.
//!
//! So the write path is where a customer meets the ceiling, and nothing
//! measured it under **concurrency**, under **sustained mixed load**, or at the
//! **edges**. These three arms do.
//!
//! ## What is being measured, precisely
//!
//! The door is `PUT /{repo}/api/v1/crates/new` — the real `cargo publish`
//! endpoint — against a real Holger HTTP gateway
//! ([`crate::served::start_holger_gateway`]) over a **writable** znippy
//! repository. That path is
//! `RustRepoZnippy::put` → one process-wide `write_lock` mutex →
//! `znippy_common::meta_sink_append::append_files` → index-cache invalidation.
//! Two properties of that path are what these arms exist to quantify:
//!
//! 1. the mutex **serialises every publisher**, and
//! 2. the append is a **read-modify-write of the whole archive**, so the cost of
//! one upload grows with how much is already stored.
//!
//! A bad number here is a GOOD result: it is the "before" half of a
//! before/after, and it is measured rather than asserted.
//!
//! ## Honesty notes that must travel with these numbers
//!
//! * There is **no rival column**. Nexus and Artifactory are not driven here.
//! These are holger-against-holger measurements of holger's own write path.
//! * The disk-filling edge is an **analogue, not a true ENOSPC**: it makes the
//! store un-writable (permissions) rather than filling a real filesystem. A
//! genuine ENOSPC arm needs a bounded filesystem, which needs a mount, which
//! needs root — an owner decision, not something a bench should do to a shared
//! box. What the analogue does answer is the part that matters: does a failed
//! write get a clean refusal, and does the archive survive it.
use std::io::{Read, Write as _};
use std::net::{SocketAddr, TcpStream};
use std::path::Path;
use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
use std::sync::Arc;
use std::time::{Duration, Instant};
use anyhow::{Context, Result};
use repository_znippy_rust::RustRepoZnippy;
use std::os::unix::fs::PermissionsExt;
use traits::RepositoryBackendTrait;
use crate::{build_crate_archive, served, Scale};
/// Bytes per uploaded crate. Large enough that the append is real work, small
/// enough that thousands of them stay on a bench box's disk.
const PAYLOAD_BYTES: usize = 16 * 1024;
/// Minimum seconds per measured phase. A `quick` scale's 2 s is too short to
/// separate an upload rate from its own startup noise.
fn phase_secs(scale: &Scale) -> u64 {
scale.bench_secs.max(3)
}
/// Distinct payload for upload `i` — deterministic but unique per artifact, so
/// no two uploads share a content address and nothing short-circuits on dedup.
fn payload(i: u64) -> Vec<u8> {
let mut v = vec![0u8; PAYLOAD_BYTES];
let seed = i.wrapping_mul(0x9E37_79B9_7F4A_7C15).wrapping_add(1);
for (j, b) in v.iter_mut().enumerate() {
*b = ((seed ^ (j as u64).wrapping_mul(131)) & 0xFF) as u8;
}
v
}
/// Build a cargo publish body: `[u32 LE json_len][json metadata][u32 LE
/// crate_len][crate bytes]` — the exact wire shape `cargo publish` sends to
/// `PUT /api/v1/crates/new`.
fn publish_body(name: &str, vers: &str, crate_bytes: &[u8]) -> Vec<u8> {
let meta = format!(r#"{{"name":"{name}","vers":"{vers}","deps":[],"features":{{}}}}"#);
let mut body = Vec::with_capacity(8 + meta.len() + crate_bytes.len());
body.extend_from_slice(&(meta.len() as u32).to_le_bytes());
body.extend_from_slice(meta.as_bytes());
body.extend_from_slice(&(crate_bytes.len() as u32).to_le_bytes());
body.extend_from_slice(crate_bytes);
body
}
/// What one bombardment of publishes produced.
#[derive(Debug, Clone, Default)]
struct PublishPhase {
ok: u64,
failed: u64,
bytes: u64,
per_sec: f64,
mbs: f64,
p50_ms: f64,
p99_ms: f64,
p999_ms: f64,
mean_ms: f64,
}
fn percentiles(lat: &mut [f32]) -> (f64, f64, f64, f64) {
if lat.is_empty() {
return (0.0, 0.0, 0.0, 0.0);
}
lat.sort_unstable_by(|a, b| a.partial_cmp(b).unwrap_or(std::cmp::Ordering::Equal));
let pick = |p: f64| -> f64 {
let idx = (((lat.len() as f64) * p).floor() as usize).min(lat.len() - 1);
lat[idx] as f64
};
let mean = lat.iter().map(|v| *v as f64).sum::<f64>() / lat.len() as f64;
(pick(0.50), pick(0.99), pick(0.999), mean)
}
/// **Pure-Rust parallel PUBLISH pool** — the write-side twin of
/// [`crate::served::bombard_http_get`]. `threads` worker threads each hold a
/// keep-alive blocking client and publish distinct crates until `dur` elapses.
///
/// `next_id` is shared across every phase of a run so no two uploads ever collide
/// on a crate name: `RustRepoZnippy::put` treats a re-published version as an
/// idempotent no-op, and a no-op is not an upload. Without a shared counter the
/// second phase would "measure" thousands of instant nothings.
fn bombard_publish(
base: &str,
repo: &str,
threads: usize,
dur: Duration,
next_id: &AtomicU64,
) -> PublishPhase {
let url = format!("{base}/{repo}/api/v1/crates/new");
let start = Instant::now();
let per_thread: Vec<(u64, u64, u64, Vec<f32>)> = std::thread::scope(|s| {
let handles: Vec<_> = (0..threads.max(1))
.map(|_| {
let url = &url;
s.spawn(move || {
let client = match tls_roots_blocking(reqwest::blocking::Client::builder())
.tcp_keepalive(Duration::from_secs(30))
.timeout(Duration::from_secs(60))
.build()
{
Ok(c) => c,
Err(_) => return (0u64, 0u64, 0u64, Vec::new()),
};
let (mut ok, mut failed, mut bytes) = (0u64, 0u64, 0u64);
let mut lat: Vec<f32> = Vec::new();
while start.elapsed() < dur {
let id = next_id.fetch_add(1, Ordering::Relaxed);
let data = payload(id);
let body = publish_body(&format!("stress-crate-{id}"), "0.1.0", &data);
let t0 = Instant::now();
match client.put(url).body(body).send() {
Ok(resp) => {
let success = resp.status().is_success();
// Drain the body so the connection is reusable and
// the timing covers the full round trip.
let _ = resp.bytes();
if success {
ok += 1;
bytes += data.len() as u64;
lat.push(t0.elapsed().as_secs_f64() as f32 * 1000.0);
} else {
failed += 1;
}
}
Err(_) => failed += 1,
}
}
(ok, failed, bytes, lat)
})
})
.collect();
handles
.into_iter()
.map(|h| h.join().unwrap_or((0, 0, 0, Vec::new())))
.collect()
});
let elapsed = start.elapsed().as_secs_f64().max(1e-6);
let (mut ok, mut failed, mut bytes) = (0u64, 0u64, 0u64);
let mut lat: Vec<f32> = Vec::new();
for (o, f, b, mut l) in per_thread {
ok += o;
failed += f;
bytes += b;
lat.append(&mut l);
}
let (p50_ms, p99_ms, p999_ms, mean_ms) = percentiles(&mut lat);
PublishPhase {
ok,
failed,
bytes,
per_sec: ok as f64 / elapsed,
mbs: (bytes as f64 / 1_048_576.0) / elapsed,
p50_ms,
p99_ms,
p999_ms,
mean_ms,
}
}
/// Stand up a real Holger HTTP gateway over a **writable** znippy repository at
/// `archive`, routed at `/{repo}`. Returns the bound loopback address.
async fn writable_gateway(repo: &str, archive: &Path) -> Result<(SocketAddr, Arc<RustRepoZnippy>)> {
let backend = Arc::new(RustRepoZnippy::with_writable_archive(
repo.to_string(),
archive.to_path_buf(),
));
anyhow::ensure!(
backend.is_writable(),
"stress bench: repository at {} is not writable — the upload arms would \
measure a wall of 403s",
archive.display()
);
let as_trait: Arc<dyn RepositoryBackendTrait> = Arc::clone(&backend) as _;
let addr = served::start_holger_gateway(vec![(repo.to_string(), as_trait)]).await?;
Ok((addr, backend))
}
// ── arm 1: concurrent uploads ───────────────────────────────────────────────
/// Outcome of [`concurrent_upload`].
#[derive(Debug, Clone)]
pub struct ConcurrentUploadOutcome {
pub concurrency: usize,
pub payload_bytes: usize,
/// One publisher, measured BEFORE the concurrent phase.
pub serial_per_sec_before: f64,
/// `concurrency` publishers at once.
pub concurrent_per_sec: f64,
pub concurrent_mbs: f64,
/// One publisher again, measured AFTER the concurrent phase. This is the
/// control: it separates "concurrency bought nothing" from "the archive grew
/// and every append got more expensive".
pub serial_per_sec_after: f64,
/// `concurrent_per_sec` ÷ the mean of the two serial phases. 1.0 means
/// `concurrency` publishers achieved exactly what one publisher achieves.
pub scaling_x: f64,
/// `serial_per_sec_before` ÷ `serial_per_sec_after`. Above 1.0 means a single
/// upload got slower purely because the archive holds more — the O(n)
/// read-modify-write cost, isolated.
pub append_cost_growth_x: f64,
pub uploads_ok: u64,
pub uploads_failed: u64,
pub p50_ms: f64,
pub p99_ms: f64,
pub p999_ms: f64,
pub mean_ms: f64,
pub archive_files_after: usize,
pub elapsed_ms: f64,
}
/// **Many clients publishing at once.** Three phases against one writable
/// repository, in order: one publisher, then `scale.threads_mt` publishers, then
/// one publisher again.
///
/// The three-phase shape is the point. A two-phase serial-then-concurrent
/// measurement cannot tell a serialising lock apart from an append that got more
/// expensive as the archive grew, because both make the second phase look bad.
/// Running the serial phase again afterwards separates them: `scaling_x` answers
/// "did more clients help", `append_cost_growth_x` answers "did storing more make
/// each upload dearer".
pub async fn concurrent_upload(scale: &Scale) -> Result<ConcurrentUploadOutcome> {
let started = Instant::now();
let dur = Duration::from_secs(phase_secs(scale));
let concurrency = scale.threads_mt;
let dir = tempfile::tempdir().context("stress bench: tempdir (upload archive)")?;
let archive = dir.path().join("uploads.znippy");
let (addr, backend) = writable_gateway("sparring", &archive).await?;
let base = format!("http://{addr}");
let ids = AtomicU64::new(1);
// The publish pool blocks OS threads; run it off the async runtime so the
// gateway's accept loop keeps its workers.
let serial_before =
tokio::task::block_in_place(|| bombard_publish(&base, "sparring", 1, dur, &ids));
let concurrent =
tokio::task::block_in_place(|| bombard_publish(&base, "sparring", concurrency, dur, &ids));
let serial_after =
tokio::task::block_in_place(|| bombard_publish(&base, "sparring", 1, dur, &ids));
// A run where nothing was stored is a broken harness, not a slow product.
anyhow::ensure!(
concurrent.ok > 0,
"stress bench: 0 successful publishes out of {} attempts — the upload door \
did not accept anything, refusing to report a 0 MB/s write path",
concurrent.ok + concurrent.failed
);
let serial_mean = (serial_before.per_sec + serial_after.per_sec) / 2.0;
let archive_files_after = backend.list_files().len();
Ok(ConcurrentUploadOutcome {
concurrency,
payload_bytes: PAYLOAD_BYTES,
serial_per_sec_before: serial_before.per_sec,
concurrent_per_sec: concurrent.per_sec,
concurrent_mbs: concurrent.mbs,
serial_per_sec_after: serial_after.per_sec,
scaling_x: if serial_mean > 0.0 {
concurrent.per_sec / serial_mean
} else {
0.0
},
append_cost_growth_x: if serial_after.per_sec > 0.0 {
serial_before.per_sec / serial_after.per_sec
} else {
0.0
},
uploads_ok: serial_before.ok + concurrent.ok + serial_after.ok,
uploads_failed: serial_before.failed + concurrent.failed + serial_after.failed,
p50_ms: concurrent.p50_ms,
p99_ms: concurrent.p99_ms,
p999_ms: concurrent.p999_ms,
mean_ms: concurrent.mean_ms,
archive_files_after,
elapsed_ms: started.elapsed().as_secs_f64() * 1000.0,
})
}
// ── arm 2: sustained mixed read/write ───────────────────────────────────────
/// Outcome of [`mixed_read_write`].
#[derive(Debug, Clone)]
pub struct MixedOutcome {
pub corpus_files: usize,
pub read_threads: usize,
pub write_threads: usize,
/// Phase 1 — reads with no writer running. The control.
pub read_ops_per_sec_quiet: f64,
pub read_mbs_quiet: f64,
pub read_p99_ms_quiet: f64,
pub read_p999_ms_quiet: f64,
/// Phase 2 — the same read pool, with publishers running against the same
/// repository at the same time.
pub read_ops_per_sec_mixed: f64,
pub read_mbs_mixed: f64,
pub read_p99_ms_mixed: f64,
pub read_p999_ms_mixed: f64,
/// Share of quiet read throughput still delivered while writes run.
pub read_throughput_retained_pct: f64,
/// How many times worse the read p999 became once writers appeared.
pub read_p999_inflation_x: f64,
/// Reads that did NOT return 2xx during the mixed phase. Non-zero means
/// concurrent appends are visible to readers as failures, which is a
/// correctness finding, not a performance one.
pub read_failed_mixed: u64,
pub write_ops_per_sec: f64,
pub writes_ok: u64,
pub writes_failed: u64,
pub elapsed_ms: f64,
}
/// **Sustained mixed read/write against a repository holding many thousands of
/// artifacts.** The corpus is the same synthetic crate archive the rival
/// head-to-heads use (`scale.entries` crates), copied to a writable path so the
/// same repository both serves and accepts.
///
/// Phase 1 reads alone; phase 2 runs the identical read pool while publishers
/// write to the same archive. The contested number is not the write rate — it is
/// what the readers lose, because in a registry the readers are every build in
/// the company and the writers are a handful of CI jobs.
pub async fn mixed_read_write(scale: &Scale) -> Result<MixedOutcome> {
let started = Instant::now();
let dur = Duration::from_secs(phase_secs(scale));
let read_threads = scale.threads_mt;
// A realistic registry is read-dominated. A handful of publishers is the
// honest shape; flooding writes would measure a workload nobody runs.
let write_threads = (scale.threads_mt / 8).max(2);
// Seed from the shared fixture so the repository really does hold thousands
// of artifacts before a single byte is written.
let fixture = build_crate_archive(scale.entries)?;
let dir = tempfile::tempdir().context("stress bench: tempdir (mixed archive)")?;
let archive = dir.path().join("mixed.znippy");
std::fs::copy(&fixture, &archive).with_context(|| {
format!(
"stress bench: seed writable archive from {}",
fixture.display()
)
})?;
let (addr, backend) = writable_gateway("rust", &archive).await?;
let base = format!("http://{addr}");
let files = backend.list_files();
anyhow::ensure!(
!files.is_empty(),
"stress bench: seeded archive {} lists no files",
archive.display()
);
let corpus_files = files.len();
let urls = served::crate_download_urls(&base, "rust", &files);
anyhow::ensure!(
!urls.is_empty(),
"stress bench: no crate-download URLs from the seeded corpus"
);
// Phase 1: the control — reads with nothing else touching the archive.
let quiet = tokio::task::block_in_place(|| served::bombard_http_get(&urls, read_threads, dur));
anyhow::ensure!(
quiet.ok_ops > 0,
"stress bench: the seeded corpus served 0/{} GETs — refusing to report a \
read/write ratio against an empty baseline",
quiet.ops
);
// Phase 2: the same read pool, plus publishers, at the same time.
let ids = AtomicU64::new(1);
let (mixed, writes) = tokio::task::block_in_place(|| {
std::thread::scope(|s| {
let r = s.spawn(|| served::bombard_http_get(&urls, read_threads, dur));
let w = s.spawn(|| bombard_publish(&base, "rust", write_threads, dur, &ids));
(r.join().unwrap_or_default(), w.join().unwrap_or_default())
})
});
let retained = if quiet.ops_per_sec > 0.0 {
mixed.ops_per_sec / quiet.ops_per_sec * 100.0
} else {
0.0
};
let inflation = if quiet.p999_ms > 0.0 {
mixed.p999_ms / quiet.p999_ms
} else {
0.0
};
Ok(MixedOutcome {
corpus_files,
read_threads,
write_threads,
read_ops_per_sec_quiet: quiet.ops_per_sec,
read_mbs_quiet: quiet.mbs,
read_p99_ms_quiet: quiet.p99_ms,
read_p999_ms_quiet: quiet.p999_ms,
read_ops_per_sec_mixed: mixed.ops_per_sec,
read_mbs_mixed: mixed.mbs,
read_p99_ms_mixed: mixed.p99_ms,
read_p999_ms_mixed: mixed.p999_ms,
read_throughput_retained_pct: retained,
read_p999_inflation_x: inflation,
read_failed_mixed: mixed.ops.saturating_sub(mixed.ok_ops),
write_ops_per_sec: writes.per_sec,
writes_ok: writes.ok,
writes_failed: writes.failed,
elapsed_ms: started.elapsed().as_secs_f64() * 1000.0,
})
}
// ── arm 3: the edges ────────────────────────────────────────────────────────
/// Outcome of [`upload_edges`].
#[derive(Debug, Clone)]
pub struct EdgeOutcome {
/// How many trickle-reading connections were held open.
pub slow_clients: usize,
pub baseline_mbs: f64,
pub mbs_with_slow_clients: f64,
/// Share of baseline throughput healthy clients still got while the slow
/// clients held their connections. 100 % means slow readers are isolated;
/// a low number means one bad client degrades everyone.
pub slow_client_retained_pct: f64,
/// Uploads opened with a declared `Content-Length`, partially written, then
/// abandoned by closing the socket.
pub abandoned_uploads: u64,
pub mbs_after_abandon: f64,
pub abandon_retained_pct: f64,
/// Did `/healthz` still answer after the abandoned uploads?
pub server_alive_after_abandon: bool,
/// The ENOSPC ANALOGUE (permissions, not a full disk — see the module docs):
/// publishes attempted against a store that cannot be written.
pub write_refused_attempts: u64,
/// Of those, how many got a clean HTTP status back rather than hanging or
/// killing the connection.
pub write_refused_clean_answers: u64,
/// Could the repository still serve reads after the failed writes? A `false`
/// here means a failed append damaged the archive, which is a correctness
/// defect and outranks every throughput number in this file.
pub reads_ok_after_write_refused: bool,
pub elapsed_ms: f64,
}
/// Hold a connection open reading the response body one byte at a time until
/// `stop` is set. Returns when the server closes, the socket errors, or `stop`.
fn slow_reader(addr: SocketAddr, path: &str, stop: &AtomicBool) {
let Ok(mut sock) = TcpStream::connect(addr) else {
return;
};
let _ = sock.set_read_timeout(Some(Duration::from_millis(500)));
let req = format!(
"GET {path} HTTP/1.1\r\nHost: {addr}\r\nUser-Agent: holger-bench-slow-client\r\n\r\n"
);
if sock.write_all(req.as_bytes()).is_err() {
return;
}
let mut byte = [0u8; 1];
while !stop.load(Ordering::Relaxed) {
match sock.read(&mut byte) {
Ok(0) => break,
Ok(_) => {}
Err(ref e) if e.kind() == std::io::ErrorKind::WouldBlock => {}
Err(ref e) if e.kind() == std::io::ErrorKind::TimedOut => {}
Err(_) => break,
}
std::thread::sleep(Duration::from_millis(100));
}
}
/// Open a publish, declare a body far larger than what is sent, send a fragment,
/// then drop the socket — a client that disappears mid-upload.
fn abandon_upload(addr: SocketAddr, repo: &str, declared: usize, sent: usize) -> bool {
let Ok(mut sock) = TcpStream::connect(addr) else {
return false;
};
let _ = sock.set_write_timeout(Some(Duration::from_secs(2)));
let head = format!(
"PUT /{repo}/api/v1/crates/new HTTP/1.1\r\nHost: {addr}\r\n\
Content-Type: application/octet-stream\r\nContent-Length: {declared}\r\n\r\n"
);
if sock.write_all(head.as_bytes()).is_err() {
return false;
}
let chunk = vec![0x5Au8; sent.min(declared)];
let _ = sock.write_all(&chunk);
// No shutdown, no rest of the body: the peer simply vanishes.
drop(sock);
true
}
/// **The edges: a slow client, a client that disappears mid-upload, and a store
/// that cannot be written.**
///
/// These are not throughput questions. Each one asks whether a single badly
/// behaved or unlucky client can damage the service for everyone else, and each
/// answer is a number rather than an opinion.
pub async fn upload_edges(scale: &Scale) -> Result<EdgeOutcome> {
let started = Instant::now();
let dur = Duration::from_secs(phase_secs(scale));
let concurrency = scale.threads_mt;
let fixture = build_crate_archive(scale.entries)?;
let dir = tempfile::tempdir().context("stress bench: tempdir (edges archive)")?;
let archive = dir.path().join("edges.znippy");
std::fs::copy(&fixture, &archive).context("stress bench: seed edges archive")?;
let (addr, backend) = writable_gateway("rust", &archive).await?;
let base = format!("http://{addr}");
let files = backend.list_files();
let urls = served::crate_download_urls(&base, "rust", &files);
anyhow::ensure!(
!urls.is_empty(),
"stress bench: no crate-download URLs for the edge arms"
);
// ── Edge 1: the slow client. Baseline first, then the same pool while a
// batch of connections trickle-read their responses and never finish. ──
let baseline =
tokio::task::block_in_place(|| served::bombard_http_get(&urls, concurrency, dur));
anyhow::ensure!(
baseline.ok_ops > 0,
"stress bench: 0/{} baseline GETs succeeded — no basis to measure an edge against",
baseline.ops
);
let slow_clients = concurrency.max(8);
let slow_path = urls[0]
.strip_prefix(&base)
.unwrap_or("/rust/crates/stress/0.1.0/download")
.to_string();
let stop = AtomicBool::new(false);
let stop_ref = &stop;
let with_slow = tokio::task::block_in_place(|| {
std::thread::scope(|s| {
for _ in 0..slow_clients {
let p = slow_path.clone();
s.spawn(move || slow_reader(addr, &p, stop_ref));
}
// Let the slow connections establish before measuring.
std::thread::sleep(Duration::from_millis(500));
let out = served::bombard_http_get(&urls, concurrency, dur);
stop.store(true, Ordering::Relaxed);
out
})
});
// ── Edge 2: the client that disappears mid-upload. ──
let abandon_target = 64u64;
let mut abandoned = 0u64;
for _ in 0..abandon_target {
if abandon_upload(addr, "rust", 8 * 1024 * 1024, 64 * 1024) {
abandoned += 1;
}
}
// `reqwest::blocking` must not be constructed on a Tokio runtime thread —
// it panics there. Every blocking client in this crate is therefore built
// inside a plain OS thread, the same rule `bombard_http_get` follows.
let health_url = format!("{base}/healthz");
let alive = std::thread::scope(|s| {
s.spawn(|| {
tls_roots_blocking(reqwest::blocking::Client::builder())
.timeout(Duration::from_secs(5))
.build()
.ok()
.and_then(|c| c.get(&health_url).send().ok())
.map(|r| r.status().is_success())
.unwrap_or(false)
})
.join()
.unwrap_or(false)
});
let after_abandon =
tokio::task::block_in_place(|| served::bombard_http_get(&urls, concurrency, dur));
// ── Edge 3: the ENOSPC analogue — a store that refuses writes. ──
//
// Read the module docs before quoting this: it takes the write permission
// away rather than filling a real filesystem, because a true ENOSPC needs a
// bounded mount and root on a shared bench box. What it does establish is the
// part that matters — a failed append must produce a clean refusal and must
// not damage what is already stored.
std::fs::set_permissions(&archive, std::fs::Permissions::from_mode(0o444))
.context("stress bench: make archive read-only")?;
std::fs::set_permissions(dir.path(), std::fs::Permissions::from_mode(0o555))
.context("stress bench: make archive directory read-only")?;
let attempts = 16u64;
let publish_url = format!("{base}/rust/api/v1/crates/new");
let clean = std::thread::scope(|s| {
s.spawn(|| {
let Ok(client) = tls_roots_blocking(reqwest::blocking::Client::builder())
.timeout(Duration::from_secs(30))
.build()
else {
return 0u64;
};
let mut clean = 0u64;
for i in 0..attempts {
let data = payload(1_000_000 + i);
let body = publish_body(&format!("refused-crate-{i}"), "0.1.0", &data);
// A clean answer is ANY HTTP status that comes back inside the
// timeout. A hang, a reset or a panic is the failure mode this
// edge exists to catch; which status code a refusal carries is
// the backend's business, not this bench's.
if client.put(&publish_url).body(body).send().is_ok() {
clean += 1;
}
}
clean
})
.join()
.unwrap_or(0)
});
// Restore before the temp dir is dropped, or its cleanup fails.
std::fs::set_permissions(dir.path(), std::fs::Permissions::from_mode(0o755)).ok();
std::fs::set_permissions(&archive, std::fs::Permissions::from_mode(0o644)).ok();
// Does the archive still serve what it held before the refused writes?
let after_refused =
tokio::task::block_in_place(|| served::bombard_http_get(&urls, 4, Duration::from_secs(2)));
let reads_ok_after_write_refused = after_refused.ok_ops > 0;
let retained_slow = if baseline.mbs > 0.0 {
with_slow.mbs / baseline.mbs * 100.0
} else {
0.0
};
let retained_abandon = if baseline.mbs > 0.0 {
after_abandon.mbs / baseline.mbs * 100.0
} else {
0.0
};
Ok(EdgeOutcome {
slow_clients,
baseline_mbs: baseline.mbs,
mbs_with_slow_clients: with_slow.mbs,
slow_client_retained_pct: retained_slow,
abandoned_uploads: abandoned,
mbs_after_abandon: after_abandon.mbs,
abandon_retained_pct: retained_abandon,
server_alive_after_abandon: alive,
write_refused_attempts: attempts,
write_refused_clean_answers: clean,
reads_ok_after_write_refused,
elapsed_ms: started.elapsed().as_secs_f64() * 1000.0,
})
}
/// 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()),
)
}