bench-scenarios/src/grpc_door.rs
//! The **gRPC door** — the transport every native client, the console, the
//! mirror's far leg and `MigrationService` actually speak.
//!
//! ## Why this module exists
//!
//! holger's bench suite measured the engines (`hops`, `promote`, `retention`,
//! `audit`) and the HTTP door (`served`, `stress`), and **nothing measured the
//! gRPC door at all**. Lane R15 named it as the remaining hole: "gRPC door arms
//! are still not built." Every number in the suite therefore described a path
//! that `holger`, `holger-ui` and the replication/migration RPCs do
//! not take.
//!
//! ## What is being measured, precisely
//!
//! A **real holger**, booted from a real RON config through the same
//! `read_ron_config` → `instantiate_backends` → `wire_holger` → `Holger::start`
//! sequence `holger start --config` runs, serving a real writable znippy
//! archive on disk, over a real loopback socket. The clients are the
//! checked-in GENERATED clients (`RepositoryServiceClient`) over h2c and — for
//! the gRPC-web arm — raw `application/grpc-web+proto` frames over HTTP/1.1,
//! which is byte for byte what the browser console sends.
//!
//! Four arms:
//!
//! * [`grpc_fetch`] — `RepositoryService.FetchArtifact`, N ops at concurrency C.
//! * [`grpc_fetch_stream`] — `RepositoryService.StreamArtifact` over artifacts
//! of at least 32 MiB, so the 64 KiB chunked path is what is timed, **plus
//! the process's resident-set growth while a stream is in flight**, which is
//! the number the arm exists for (see "the streaming finding" below).
//! * [`grpc_store`] — `RepositoryService.PutArtifact`, N puts at concurrency C.
//! * [`grpc_web_fetch`] — the same Fetch over gRPC-web, measured in the SAME
//! run as a native Fetch so the cost of the browser transport is a ratio
//! taken inside one window and survives a busy box.
//!
//! ## Honesty notes that must travel with these numbers
//!
//! * **The audit log is ON** (lane HOL-NCSC: no Holger starts without it), so a gRPC
//! arm pays its chained, flushed row per mutating call while the HTTP arms in
//! [`crate::served`] hand the gateway a `NoopAuditLog`. Read the write arms
//! against `holger.audit_append`, which measures that row alone; the read arms
//! (Fetch) write no row.
//! * **No rival column.** Nexus and Artifactory speak no gRPC; these are
//! holger-against-holger measurements.
//! * **The floors are checked outside the clock.** Verifying every response by
//! sha256 inside the timed loop would measure the client's hashing —
//! stinky-01 §2.2 measured sha256 at 0.61 GB/s on one core, the same order as
//! a loopback fetch — so each arm runs its timed pass with a cheap length
//! check and then a separate, untimed pass that sha256-verifies EVERY
//! artifact. `verified_ops` reports how many; an arm whose verification pass
//! does not cover the whole corpus fails rather than reporting a number.
//! * **The load average is recorded beside every number** (`load_avg_1m`), read
//! from `/proc/loadavg` in Rust. oden runs several lanes at once; an absolute
//! throughput from a loaded window is not "the number" — the ratios taken
//! inside one run are.
//!
//! ## The streaming finding, and why it is REPORTED rather than a red floor
//!
//! `grpc/services.rs::stream_artifact` says it in its own comment: *"the
//! chunking below is a WIRE-FRAMING detail, not a memory bound — the whole
//! artifact is materialised here first and then handed out 64 KiB at a time"*.
//! [`grpc_fetch_stream`] measures exactly that, as resident-set growth while a
//! stream is in flight, at two artifact sizes so the SCALING is the
//! measurement and not one allocator's noise.
//!
//! It is a reported metric (`rss_bytes_per_artifact_byte`, `rss_growth_ratio`,
//! `buffers_whole_body`) and not an ANSWER floor, and the reason is structural
//! rather than convenient: `RepositoryBackendTrait::fetch` returns the whole
//! artifact as one `Bytes`. There is no streaming seam in the backend trait for
//! the door to stream FROM, so a floor here would not be a regression guard, it
//! would be an arm that is red on every run against a design decision one layer
//! down. The floors this arm does carry — the reassembled bytes are identical
//! and no frame exceeds the declared 64 KiB chunk — are the ones a change to
//! the door can actually break.
use std::io::Write as _;
use std::net::SocketAddr;
use std::path::{Path, PathBuf};
use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
use std::sync::Arc;
use std::time::{Duration, Instant};
use anyhow::{Context, Result};
use prost::Message as _;
use sha2::{Digest, Sha256};
use server_lib::grpc::holger_proto::repository_service_client::RepositoryServiceClient;
use server_lib::grpc::holger_proto::{
ArtifactId as ProtoArtifactId, FetchArtifactRequest, FetchArtifactResponse, PutArtifactRequest,
};
use server_lib::{read_ron_config, wire_holger, Holger};
use crate::{bench_root, cores, served, Scale, Throughput};
/// The repository every arm serves out of. One name, so a metric never depends
/// on which arm wrote the config.
const REPO: &str = "bench";
/// Bytes per artifact in the unary arms. Mirrors `hops::PAYLOAD_BYTES` and
/// `stress::PAYLOAD_BYTES` so a gRPC number and an HTTP number are over the
/// same unit of work.
const PAYLOAD_BYTES: usize = 16 * 1024;
/// The chunk size `stream_artifact` declares. No frame may exceed it — that is
/// one of the streaming arm's floors.
const DECLARED_CHUNK_BYTES: usize = 64 * 1024;
/// Client-side message-size caps. tonic's generated client defaults to a 4 MiB
/// DECODE limit, which a large unary fetch trips before the door is ever
/// measured. The server's own cap is 1 GiB (`ron_max_body_bytes`'s default);
/// match it, so the client is never the thing that refuses.
const CLIENT_MSG_CAP: usize = 1024 * 1024 * 1024;
// ─────────────────────────────── knobs ───────────────────────────────
fn env_usize(key: &str) -> Option<usize> {
std::env::var(key)
.ok()?
.parse::<usize>()
.ok()
.filter(|n| *n > 0)
}
/// How many distinct artifacts the read arms serve out of. Small on purpose:
/// this is the corpus, not the op count — the reads loop over it.
fn corpus_count(scale: &Scale) -> usize {
env_usize("HOLGER_BENCH_GRPC_CORPUS").unwrap_or_else(|| scale.entries.clamp(256, 1_024))
}
/// Total unary fetches an arm performs, across all workers. Fixed N, never a
/// time-boxed loop, for the same reason `rust_crate_put`'s N is fixed: a cost
/// that grows with the store hides inside a fixed duration by completing fewer
/// operations.
fn fetch_ops(scale: &Scale) -> usize {
env_usize("HOLGER_BENCH_GRPC_OPS").unwrap_or_else(|| scale.entries.clamp(200_000, 1_000_000))
}
/// Total puts the write arm performs. Lower than the read count: every one of
/// them appends to a growing archive behind one write lock.
fn store_ops(scale: &Scale) -> usize {
env_usize("HOLGER_BENCH_GRPC_PUTS").unwrap_or_else(|| scale.entries.clamp(20_000, 50_000))
}
/// Client concurrency — one connection per worker.
fn concurrency() -> usize {
env_usize("HOLGER_BENCH_GRPC_CONC").unwrap_or_else(|| cores().clamp(2, 32))
}
/// Bytes per artifact in the streaming arm. At least 32 MiB by the lane's
/// brief, so the chunked path and not the framing overhead is measured.
fn stream_artifact_bytes() -> usize {
env_usize("HOLGER_BENCH_GRPC_STREAM_MB").unwrap_or(32) * 1024 * 1024
}
/// How many large artifacts the streaming arm moves.
fn stream_artifact_count() -> usize {
env_usize("HOLGER_BENCH_GRPC_STREAMS").unwrap_or(8)
}
// ────────────────────────────── fixtures ──────────────────────────────
/// Distinct, not-compressible-to-nothing payload for artifact `i`. Same
/// generator as `hops::payload`, so nothing dedups and the content ids differ.
fn payload(i: usize, len: usize) -> Vec<u8> {
let mut v = vec![0u8; len];
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 proto_id(prefix: &str, i: usize) -> ProtoArtifactId {
ProtoArtifactId {
namespace: String::new(),
name: format!("{prefix}-{i:06}"),
version: "1.0.0".into(),
}
}
/// A scratch directory under the bench root, carrying this lane's branch name
/// and 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(format!("grpc-door-arms-{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)
}
/// 1-minute load average, read in Rust from `/proc/loadavg`. Never `uptime`,
/// never a shell. `0.0` off Linux — a missing number reads as "not recorded",
/// which is what it is.
pub fn load_avg_1m() -> f64 {
std::fs::read_to_string("/proc/loadavg")
.ok()
.and_then(|s| s.split_whitespace().next().and_then(|v| v.parse().ok()))
.unwrap_or(0.0)
}
/// Resident set size of THIS process, in bytes, from `/proc/self/statm` field 2
/// (resident pages). Pure Rust, no shelling `ps`.
fn rss_bytes() -> u64 {
let Ok(s) = std::fs::read_to_string("/proc/self/statm") else {
return 0;
};
s.split_whitespace()
.nth(1)
.and_then(|v| v.parse::<u64>().ok())
.map(|pages| pages * 4096)
.unwrap_or(0)
}
/// A booted holger, its runtime, and the scratch directory it serves out of.
///
/// The runtime is a MULTI-THREAD one and it outlives the `block_on` that boots
/// the server: `Holger::start` `tokio::spawn`s its accept loop onto it, and
/// that task keeps being driven by the runtime's own worker threads after
/// `block_on` returns. That is what lets every client in this module run on
/// plain OS threads outside any runtime — the same shape
/// `served::bombard_http_get` uses against the HTTP door, and the reason no
/// client here can starve the server's reactor.
struct Running {
rt: tokio::runtime::Runtime,
grpc: SocketAddr,
dir: PathBuf,
holger: Holger,
}
impl Running {
fn endpoint(&self) -> String {
format!("http://{}", self.grpc)
}
/// Stop the server, then remove the fixture. Explicit rather than `Drop`, so
/// a failing floor leaves the directory behind for a human to look at.
fn shutdown(self) {
let Running {
rt, dir, holger, ..
} = self;
rt.shutdown_timeout(Duration::from_secs(2));
drop(holger);
std::fs::remove_dir_all(&dir).ok();
}
}
/// The RON a bench holger boots from: one writable znippy `rust` repository,
/// one gRPC endpoint, no HTTP gateway, audit off.
fn bench_ron(dir: &Path, port: u16) -> String {
let d = dir.display();
format!(
r#"(
operation_mode: ReadWrite,
audit: ( dir: "{d}/audit", enabled: true ),
exposed_endpoints: [
( ron_name: "bench", ron_url: "127.0.0.1:{port}", ron_http_url: None ),
],
storage_endpoints: [
( ron_name: "bench-store", ron_storage_type: "znippy", ron_path: "{d}/store/" ),
],
repositories: [
( ron_name: "{REPO}", ron_repo_type: "rust", ron_writable_archive: Some("{d}/bench.znippy"), ron_upstreams: [], ron_in: None, ron_out: Some(( ron_storage_endpoint: "bench-store", ron_exposed_endpoint: "bench" )) ),
],
)
"#
)
}
/// Boot a real holger over a fresh writable archive and wait until its gRPC
/// door answers.
fn start_bench_holger(label: &str) -> Result<Running> {
let dir = fresh_dir(label)?;
std::fs::create_dir_all(dir.join("store")).ok();
let port = traits::test_support::hold_port();
let cfg_path = dir.join("bench.ron");
{
let mut f = std::fs::File::create(&cfg_path)
.with_context(|| format!("write {}", cfg_path.display()))?;
f.write_all(bench_ron(&dir, port).as_bytes())?;
}
let mut holger: Holger =
read_ron_config(&cfg_path).context("the bench RON must parse as a holger config")?;
holger
.instantiate_backends()
.context("the bench repository's backend must build")?;
wire_holger(&mut holger).context("the bench config must wire")?;
let rt = tokio::runtime::Builder::new_multi_thread()
.enable_all()
.thread_name("grpc-door-bench")
.build()
.context("build the server runtime")?;
rt.block_on(async { holger.start() })
.context("start the bench holger's gRPC endpoint")?;
let grpc: SocketAddr = format!("127.0.0.1:{port}").parse().unwrap();
let url = format!("http://{grpc}");
// Readiness: the accept loop binds inside a spawned task, so a client that
// dials too early gets a connection refused rather than a slow answer.
let ready = rt.block_on(async {
for _ in 0..500 {
if RepositoryServiceClient::connect(url.clone()).await.is_ok() {
return true;
}
tokio::time::sleep(Duration::from_millis(20)).await;
}
false
});
anyhow::ensure!(ready, "the bench holger's gRPC door never came up at {url}");
Ok(Running {
rt,
grpc,
dir,
holger,
})
}
/// Build a native client against `endpoint`, with both message caps raised.
async fn client(endpoint: &str) -> Result<RepositoryServiceClient<tonic::transport::Channel>> {
let c = RepositoryServiceClient::connect(endpoint.to_string())
.await
.with_context(|| format!("dial {endpoint}"))?;
Ok(c.max_decoding_message_size(CLIENT_MSG_CAP)
.max_encoding_message_size(CLIENT_MSG_CAP))
}
/// Seed `n` artifacts of `len` bytes through the REAL Store door, serially,
/// OUTSIDE any clock. Returns the expected sha256 of each, by index.
fn seed(running: &Running, prefix: &str, n: usize, len: usize) -> Result<Vec<[u8; 32]>> {
let endpoint = running.endpoint();
let prefix = prefix.to_string();
running.rt.block_on(async move {
let mut c = client(&endpoint).await?;
let mut digests = Vec::with_capacity(n);
for i in 0..n {
let data = payload(i, len);
digests.push(<[u8; 32]>::from(Sha256::digest(&data)));
let resp = c
.put_artifact(PutArtifactRequest {
repository: REPO.into(),
id: Some(proto_id(&prefix, i)),
data: prost::bytes::Bytes::from(data),
async_job: false,
})
.await
.with_context(|| format!("seed {prefix}-{i:06}"))?
.into_inner();
anyhow::ensure!(
resp.success,
"seeding {prefix}-{i:06} was refused: {}",
resp.message
);
}
Ok(digests)
})
}
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,
}
}
/// What one unary arm measured.
pub struct UnaryOutcome {
pub t: Throughput,
pub p50_ms: f64,
pub p99_ms: f64,
pub p999_ms: f64,
pub mean_ms: f64,
pub concurrency: usize,
pub payload_bytes: usize,
/// ANSWER floor: how many artifacts were read back and sha256-compared with
/// what was stored, in the untimed verification pass.
pub verified_ops: usize,
pub load_avg_1m: f64,
}
// ─────────────────────────── arm 1: grpc_fetch ───────────────────────────
/// `RepositoryService.FetchArtifact` over loopback h2c: N fetches spread over C
/// independent connections, round-robin over a corpus that was stored through
/// the same door.
///
/// ANSWER floors: every fetch returns exactly `PAYLOAD_BYTES` bytes with a
/// success status (a wall of NOT_FOUND would otherwise read as a very fast
/// door), and afterwards every artifact in the corpus is fetched once more and
/// its bytes compared with what was stored, by sha256.
pub fn grpc_fetch(scale: &Scale) -> Result<UnaryOutcome> {
let running = start_bench_holger("fetch")?;
let corpus = corpus_count(scale);
let outcome = (|| -> Result<UnaryOutcome> {
let digests = seed(&running, "fetch", corpus, PAYLOAD_BYTES)?;
let mut o = fetch_native(&running, "fetch", corpus, fetch_ops(scale), concurrency())?;
o.verified_ops = verify_corpus(&running, "fetch", &digests)?;
Ok(o)
})();
if outcome.is_ok() {
running.shutdown();
}
outcome
}
/// The timed native-fetch pass. Each worker owns one connection and one
/// current-thread runtime; the server is on its own multi-thread runtime, so
/// nothing here can starve its reactor.
///
/// NOT `gatling`: `znippy_zoomies::gatling` is the house engine for fanning a
/// CPU work-slice over cores, and these workers are long-lived async
/// connections that spend their time blocked on a socket rather than on a
/// slice. This is the same `std::thread::scope` pool `served::bombard_http_get`
/// drives the HTTP door with, which keeps the two doors' client side identical
/// — the whole point of putting the numbers beside each other. No rayon, here
/// or anywhere (ROOT LAW #0).
fn fetch_native(
running: &Running,
prefix: &str,
corpus: usize,
ops: usize,
conc: usize,
) -> Result<UnaryOutcome> {
let endpoint = running.endpoint();
let conc = conc.max(1);
let per = ops.div_ceil(conc);
let load = load_avg_1m();
let t0 = Instant::now();
let per_worker: Vec<Result<(u64, u64, Vec<f32>)>> = std::thread::scope(|s| {
let handles: Vec<_> = (0..conc)
.map(|tid| {
let endpoint = endpoint.clone();
let prefix = prefix.to_string();
s.spawn(move || -> Result<(u64, u64, Vec<f32>)> {
let rt = tokio::runtime::Builder::new_current_thread()
.enable_all()
.build()
.context("worker runtime")?;
rt.block_on(async move {
let mut c = client(&endpoint).await?;
let mut ok = 0u64;
let mut bytes = 0u64;
let mut lat: Vec<f32> = Vec::with_capacity(per);
for i in 0..per {
let idx = (i * 7 + tid * 1337) % corpus;
let t = Instant::now();
let resp: FetchArtifactResponse = c
.fetch_artifact(FetchArtifactRequest {
repository: REPO.into(),
id: Some(proto_id(&prefix, idx)),
})
.await
.with_context(|| format!("fetch {prefix}-{idx:06}"))?
.into_inner();
lat.push(t.elapsed().as_secs_f64() as f32 * 1000.0);
// Cheap per-response floor. The sha256 one runs
// afterwards, outside the clock.
anyhow::ensure!(
resp.data.len() == PAYLOAD_BYTES,
"the door answered {} bytes for {prefix}-{idx:06}, not {PAYLOAD_BYTES}",
resp.data.len()
);
bytes += resp.data.len() as u64;
ok += 1;
}
Ok((ok, bytes, lat))
})
})
})
.collect();
handles
.into_iter()
.map(|h| {
h.join()
.unwrap_or_else(|_| Err(anyhow::anyhow!("a fetch worker panicked")))
})
.collect()
});
let elapsed = t0.elapsed();
let mut ok = 0u64;
let mut bytes = 0u64;
let mut lat: Vec<f32> = Vec::with_capacity(ops);
for r in per_worker {
let (o, b, mut l) = r?;
ok += o;
bytes += b;
lat.append(&mut l);
}
anyhow::ensure!(
ok as usize == per * conc,
"the fetch pool completed {ok} of {} ops",
per * conc
);
let (p50, p99, p999, mean) = served::latency_percentiles(&mut lat);
Ok(UnaryOutcome {
t: throughput(ok as usize, bytes, elapsed),
p50_ms: p50,
p99_ms: p99,
p999_ms: p999,
mean_ms: mean,
concurrency: conc,
payload_bytes: PAYLOAD_BYTES,
verified_ops: 0,
load_avg_1m: load,
})
}
/// The ANSWER floor, untimed: fetch every artifact once more and compare its
/// bytes with what was stored, by sha256. Returns how many were verified.
fn verify_corpus(running: &Running, prefix: &str, digests: &[[u8; 32]]) -> Result<usize> {
let endpoint = running.endpoint();
let prefix = prefix.to_string();
let digests = digests.to_vec();
running.rt.block_on(async move {
let mut c = client(&endpoint).await?;
for (i, want) in digests.iter().enumerate() {
let resp = c
.fetch_artifact(FetchArtifactRequest {
repository: REPO.into(),
id: Some(proto_id(&prefix, i)),
})
.await
.with_context(|| format!("verify {prefix}-{i:06}"))?
.into_inner();
let got = <[u8; 32]>::from(Sha256::digest(&resp.data));
anyhow::ensure!(
&got == want,
"the door returned bytes that are not what was stored for {prefix}-{i:06} \
({} bytes back)",
resp.data.len()
);
}
Ok(digests.len())
})
}
// ─────────────────────── arm 2: grpc_fetch_stream ───────────────────────
/// What the streaming arm measured.
pub struct StreamOutcome {
/// The large-artifact pass: bytes off the wire per second, one "op" per
/// artifact streamed end to end.
pub t: Throughput,
pub artifact_bytes: usize,
pub artifacts: usize,
pub chunks: u64,
pub max_chunk_bytes: usize,
pub p50_ms: f64,
pub p99_ms: f64,
pub p999_ms: f64,
pub mean_ms: f64,
/// The small-artifact control pass, for the scaling comparison.
pub small_artifact_bytes: usize,
/// Peak resident-set growth of this process while a SMALL artifact was in
/// flight, and while the LARGE ones were.
pub rss_growth_small: u64,
pub rss_growth_large: u64,
/// `rss_growth_large / artifact_bytes`. A door that streams straight off
/// the backend holds one chunk; a door that materialises the artifact first
/// holds all of it, and this reads ≈ 1.0 (or ≈ N for N concurrent streams).
pub rss_bytes_per_artifact_byte: f64,
/// `rss_growth_large / rss_growth_small`, to be read beside `size_ratio`.
/// Equal ⇒ the memory scales with the artifact.
pub rss_growth_ratio: f64,
pub size_ratio: f64,
/// The finding, as a flag: does the resident set grow by at least half of
/// one artifact while a single stream is open?
pub buffers_whole_body: bool,
pub verified_ops: usize,
pub load_avg_1m: f64,
}
/// One streamed artifact: wall time, byte count, chunk count, largest frame,
/// and (optionally) the sha256 of the reassembled body.
struct StreamRead {
elapsed: Duration,
bytes: u64,
chunks: u64,
max_chunk: usize,
digest: Option<[u8; 32]>,
}
/// Stream one artifact to completion. `hash` is off in the timed pass (the
/// client's sha256 is the same order of cost as the door's work) and on in the
/// untimed floor pass.
async fn stream_one(
c: &mut RepositoryServiceClient<tonic::transport::Channel>,
prefix: &str,
i: usize,
hash: bool,
) -> Result<StreamRead> {
let t = Instant::now();
let mut s = c
.stream_artifact(FetchArtifactRequest {
repository: REPO.into(),
id: Some(proto_id(prefix, i)),
})
.await
.with_context(|| format!("open stream for {prefix}-{i:06}"))?
.into_inner();
let mut bytes = 0u64;
let mut chunks = 0u64;
let mut max_chunk = 0usize;
let mut hasher = if hash { Some(Sha256::new()) } else { None };
while let Some(chunk) = s
.message()
.await
.with_context(|| format!("chunk of {prefix}-{i:06}"))?
{
bytes += chunk.data.len() as u64;
chunks += 1;
max_chunk = max_chunk.max(chunk.data.len());
if let Some(h) = hasher.as_mut() {
h.update(&chunk.data);
}
}
Ok(StreamRead {
elapsed: t.elapsed(),
bytes,
chunks,
max_chunk,
digest: hasher.map(|h| <[u8; 32]>::from(h.finalize())),
})
}
/// Peak resident-set growth of this process while `f` runs, sampled every
/// 500 µs by a watcher thread. Returns `(result, peak_growth_bytes)`.
fn with_rss_peak<T>(f: impl FnOnce() -> T) -> (T, u64) {
let base = rss_bytes();
let stop = Arc::new(AtomicBool::new(false));
let peak = Arc::new(AtomicU64::new(base));
let (s, p) = (stop.clone(), peak.clone());
let watcher = std::thread::spawn(move || {
while !s.load(Ordering::Relaxed) {
p.fetch_max(rss_bytes(), Ordering::Relaxed);
std::thread::sleep(Duration::from_micros(500));
}
});
let out = f();
stop.store(true, Ordering::Relaxed);
let _ = watcher.join();
(out, peak.load(Ordering::Relaxed).saturating_sub(base))
}
/// `RepositoryService.StreamArtifact` over artifacts of at least 32 MiB, so the
/// 64 KiB chunked path is what the number describes — plus the resident-set
/// growth while a stream is in flight, at two artifact sizes.
///
/// ANSWER floors: every stream delivers exactly `artifact_bytes`, no frame is
/// larger than the door's declared 64 KiB chunk, and in an untimed pass every
/// artifact's reassembled body is sha256-identical to what was stored. The RSS
/// numbers are REPORTED, not a floor — the module docs say why.
pub fn grpc_fetch_stream(_scale: &Scale) -> Result<StreamOutcome> {
let running = start_bench_holger("stream")?;
let outcome = (|| -> Result<StreamOutcome> {
let big = stream_artifact_bytes();
let small = (big / 16).max(1024 * 1024);
let n = stream_artifact_count();
anyhow::ensure!(
big >= 32 * 1024 * 1024,
"the streaming arm is about the chunked path; {big} bytes is below the 32 MiB floor"
);
let big_digests = seed(&running, "big", n, big)?;
let small_digests = seed(&running, "small", 1, small)?;
let load = load_avg_1m();
// ── The control: one SMALL artifact, with the resident set watched. ──
let endpoint = running.endpoint();
let (small_read, rss_small) = with_rss_peak(|| {
running.rt.block_on(async {
let mut c = client(&endpoint).await?;
stream_one(&mut c, "small", 0, false).await
})
});
let small_read = small_read?;
anyhow::ensure!(
small_read.bytes == small as u64,
"the control stream delivered {} of {small} bytes",
small_read.bytes
);
// ── The measurement: N LARGE artifacts, one at a time, watched. ──
let endpoint = running.endpoint();
let (reads, rss_large) = with_rss_peak(|| {
running.rt.block_on(async {
let mut c = client(&endpoint).await?;
let mut reads = Vec::with_capacity(n);
let t0 = Instant::now();
for i in 0..n {
reads.push(stream_one(&mut c, "big", i, false).await?);
}
Ok::<_, anyhow::Error>((reads, t0.elapsed()))
})
});
let (reads, elapsed) = reads?;
// ANSWER floors over the timed pass.
let mut bytes = 0u64;
let mut chunks = 0u64;
let mut max_chunk = 0usize;
let mut lat: Vec<f32> = Vec::with_capacity(n);
for (i, r) in reads.iter().enumerate() {
anyhow::ensure!(
r.bytes == big as u64,
"stream {i} delivered {} of {big} bytes",
r.bytes
);
bytes += r.bytes;
chunks += r.chunks;
max_chunk = max_chunk.max(r.max_chunk);
lat.push(r.elapsed.as_secs_f64() as f32 * 1000.0);
}
anyhow::ensure!(
max_chunk <= DECLARED_CHUNK_BYTES,
"a stream frame carried {max_chunk} bytes, more than the door's declared \
{DECLARED_CHUNK_BYTES}-byte chunk — the chunking is not what it says it is"
);
// ── The untimed floor pass: reassemble and compare, by sha256. ──
let endpoint = running.endpoint();
let verified = running.rt.block_on(async {
let mut c = client(&endpoint).await?;
let mut verified = 0usize;
for (i, want) in big_digests.iter().enumerate() {
let r = stream_one(&mut c, "big", i, true).await?;
anyhow::ensure!(
r.digest.as_ref() == Some(want),
"the reassembled stream of big-{i:06} is not the artifact that was stored"
);
verified += 1;
}
let r = stream_one(&mut c, "small", 0, true).await?;
anyhow::ensure!(
r.digest.as_ref() == Some(&small_digests[0]),
"the reassembled control stream is not the artifact that was stored"
);
verified += 1;
Ok::<_, anyhow::Error>(verified)
})?;
let (p50, p99, p999, mean) = served::latency_percentiles(&mut lat);
let rss_bytes_per_artifact_byte = rss_large as f64 / big as f64;
Ok(StreamOutcome {
t: throughput(n, bytes, elapsed),
artifact_bytes: big,
artifacts: n,
chunks,
max_chunk_bytes: max_chunk,
p50_ms: p50,
p99_ms: p99,
p999_ms: p999,
mean_ms: mean,
small_artifact_bytes: small,
rss_growth_small: rss_small,
rss_growth_large: rss_large,
rss_bytes_per_artifact_byte,
rss_growth_ratio: rss_large as f64 / (rss_small as f64).max(1.0),
size_ratio: big as f64 / small as f64,
buffers_whole_body: rss_bytes_per_artifact_byte >= 0.5,
verified_ops: verified,
load_avg_1m: load,
})
})();
if outcome.is_ok() {
running.shutdown();
}
outcome
}
// ─────────────────────────── arm 3: grpc_store ───────────────────────────
/// `RepositoryService.PutArtifact` over loopback h2c: N puts spread over C
/// independent connections into one growing writable znippy archive.
///
/// Fixed N, never a time-boxed loop, for the same reason `rust_crate_put`'s N
/// is fixed: a cost that grows with the store hides inside a fixed duration by
/// completing fewer operations.
///
/// ANSWER floor: after the run, every artifact that was stored is fetched back
/// and compared with what was sent, by sha256 — a write door that returns
/// `success: true` and keeps nothing is the failure this arm exists to catch.
pub fn grpc_store(scale: &Scale) -> Result<UnaryOutcome> {
let running = start_bench_holger("store")?;
let outcome = (|| -> Result<UnaryOutcome> {
let ops = store_ops(scale);
let conc = concurrency().max(1);
let per = ops.div_ceil(conc);
let endpoint = running.endpoint();
let load = load_avg_1m();
// Every body built BEFORE the clock starts: this arm measures the door,
// never the fixture. `tid * per + i` keeps the ids disjoint per worker.
let digests: Vec<[u8; 32]> = (0..per * conc)
.map(|i| <[u8; 32]>::from(Sha256::digest(payload(i, PAYLOAD_BYTES))))
.collect();
let t0 = Instant::now();
let per_worker: Vec<Result<(u64, u64, Vec<f32>)>> = std::thread::scope(|s| {
let handles: Vec<_> = (0..conc)
.map(|tid| {
let endpoint = endpoint.clone();
s.spawn(move || -> Result<(u64, u64, Vec<f32>)> {
let rt = tokio::runtime::Builder::new_current_thread()
.enable_all()
.build()
.context("worker runtime")?;
rt.block_on(async move {
let mut c = client(&endpoint).await?;
let mut ok = 0u64;
let mut bytes = 0u64;
let mut lat: Vec<f32> = Vec::with_capacity(per);
for i in 0..per {
let idx = tid * per + i;
let data = payload(idx, PAYLOAD_BYTES);
let len = data.len() as u64;
let t = Instant::now();
let resp = c
.put_artifact(PutArtifactRequest {
repository: REPO.into(),
id: Some(proto_id("put", idx)),
data: prost::bytes::Bytes::from(data),
async_job: false,
})
.await
.with_context(|| format!("store put-{idx:06}"))?
.into_inner();
lat.push(t.elapsed().as_secs_f64() as f32 * 1000.0);
anyhow::ensure!(
resp.success,
"the door refused put-{idx:06}: {}",
resp.message
);
bytes += len;
ok += 1;
}
Ok((ok, bytes, lat))
})
})
})
.collect();
handles
.into_iter()
.map(|h| {
h.join()
.unwrap_or_else(|_| Err(anyhow::anyhow!("a store worker panicked")))
})
.collect()
});
let elapsed = t0.elapsed();
let mut ok = 0u64;
let mut bytes = 0u64;
let mut lat: Vec<f32> = Vec::with_capacity(per * conc);
for r in per_worker {
let (o, b, mut l) = r?;
ok += o;
bytes += b;
lat.append(&mut l);
}
anyhow::ensure!(
ok as usize == per * conc,
"the store pool completed {ok} of {} puts",
per * conc
);
let verified = verify_corpus(&running, "put", &digests)?;
let (p50, p99, p999, mean) = served::latency_percentiles(&mut lat);
Ok(UnaryOutcome {
t: throughput(ok as usize, bytes, elapsed),
p50_ms: p50,
p99_ms: p99,
p999_ms: p999,
mean_ms: mean,
concurrency: conc,
payload_bytes: PAYLOAD_BYTES,
verified_ops: verified,
load_avg_1m: load,
})
})();
if outcome.is_ok() {
running.shutdown();
}
outcome
}
// ───────────────────── arm 4: grpc_web_fetch ─────────────────────
/// gRPC-web request framing: one length-prefixed message, `0x00` = data frame.
/// Spec-exact and four lines, so this module carries no extra dependency for a
/// wire format it only ever writes one frame of.
fn grpc_web_frame(msg: &[u8]) -> Vec<u8> {
let mut out = Vec::with_capacity(5 + msg.len());
out.push(0x00);
out.extend_from_slice(&(msg.len() as u32).to_be_bytes());
out.extend_from_slice(msg);
out
}
/// Parse a gRPC-web response body into `(message_bytes, trailers)`. The body is
/// a sequence of frames: `0x00` carries a protobuf message, `0x80` carries the
/// HTTP/2 trailers as text, which over HTTP/1.1 is the only place `grpc-status`
/// can live.
fn grpc_web_parse(body: &[u8]) -> Result<(Vec<u8>, String)> {
let mut msg = Vec::new();
let mut trailers = String::new();
let mut at = 0usize;
while at + 5 <= body.len() {
let flag = body[at];
let len =
u32::from_be_bytes([body[at + 1], body[at + 2], body[at + 3], body[at + 4]]) as usize;
let start = at + 5;
let end = start
.checked_add(len)
.filter(|e| *e <= body.len())
.ok_or_else(|| {
anyhow::anyhow!("gRPC-web frame claims {len} bytes past the body end")
})?;
if flag & 0x80 == 0 {
msg.extend_from_slice(&body[start..end]);
} else {
trailers.push_str(&String::from_utf8_lossy(&body[start..end]));
}
at = end;
}
anyhow::ensure!(
at == body.len(),
"gRPC-web body has {} trailing bytes",
body.len() - at
);
Ok((msg, trailers))
}
/// The `grpc-status` a gRPC-web answer carries, from the trailer frame or the
/// response headers. `None` when neither states one.
fn grpc_web_status(trailers: &str, headers: &reqwest::header::HeaderMap) -> Option<i32> {
for line in trailers.split(['\r', '\n']) {
if let Some(v) = line.trim().strip_prefix("grpc-status:") {
if let Ok(n) = v.trim().parse::<i32>() {
return Some(n);
}
}
}
headers
.get("grpc-status")
.and_then(|v| v.to_str().ok())
.and_then(|v| v.trim().parse::<i32>().ok())
}
/// What the gRPC-web arm measured: the browser transport and, in the SAME
/// window, the native one, so the cost is a ratio and not two absolutes from
/// two different minutes on a shared box.
pub struct WebFetchOutcome {
pub web: UnaryOutcome,
/// The FIRST native leg, with its own percentiles.
pub native: UnaryOutcome,
/// The SECOND native leg's rate, run after the web one. The arm is
/// interleaved native → web → native precisely so this exists: two
/// measurements of the same thing, before and after, on a box that is
/// doing other work.
pub native_after_ops_sec: f64,
/// The mean of the two native legs — what the ratio is taken against.
pub native_mean_ops_sec: f64,
/// `|after - before| / mean`, as a percentage: how far the box moved DURING
/// this arm. A ratio quoted beside a large drift is worth less, and saying
/// so is cheaper than pretending it did not happen.
pub native_drift_pct: f64,
/// `native_mean_ops_sec / web.ops_sec - 1`, as a percentage: what the
/// browser transport costs over the native one.
pub web_cost_pct: f64,
/// `web.ops_sec / native_mean_ops_sec`: the same fact the other way up, and
/// the number to quote on a loaded box.
pub web_over_native: f64,
}
/// `RepositoryService.FetchArtifact` over **gRPC-web** — HTTP/1.1,
/// `application/grpc-web+proto`, the exact transport the browser console pays —
/// measured beside native h2c passes over the same corpus, in the same run,
/// INTERLEAVED native → web → native so the load's drift during the arm is a
/// reported number rather than a silent bias on the ratio.
///
/// ANSWER floors: every gRPC-web answer carries `grpc-status: 0` and exactly
/// `PAYLOAD_BYTES` bytes of artifact, and the corpus is sha256-verified
/// afterwards through the same web transport, so the floor covers the door that
/// was actually measured rather than the native one beside it.
pub fn grpc_web_fetch(scale: &Scale) -> Result<WebFetchOutcome> {
let running = start_bench_holger("web")?;
let outcome = (|| -> Result<WebFetchOutcome> {
let corpus = corpus_count(scale);
let digests = seed(&running, "web", corpus, PAYLOAD_BYTES)?;
// The SAME op count as the native leg, deliberately. The ratio these
// two produce is the whole point of the arm, and a shorter web leg
// would make it a comparison between two different window lengths on a
// box whose load moves inside a second.
let ops = fetch_ops(scale);
let conc = concurrency();
// INTERLEAVED: native → web → native. oden is not a quiet box, and two
// passes back to back would hand the second one whatever the load did
// in between. Measuring the native leg on both sides of the web one
// turns that into a number (`native_drift_pct`) instead of a bias.
let native = fetch_native(&running, "web", corpus, ops, conc)?;
let mut web = fetch_web(&running, "web", corpus, ops, conc)?;
let native_after = fetch_native(&running, "web", corpus, ops, conc)?;
web.verified_ops = verify_corpus_web(&running, "web", &digests)?;
let native_after_ops_sec = native_after.t.ops_sec;
let native_mean_ops_sec = (native.t.ops_sec + native_after_ops_sec) / 2.0;
let native_drift_pct = if native_mean_ops_sec > 0.0 {
(native_after_ops_sec - native.t.ops_sec).abs() / native_mean_ops_sec * 100.0
} else {
0.0
};
let web_cost_pct = if web.t.ops_sec > 0.0 {
(native_mean_ops_sec / web.t.ops_sec - 1.0) * 100.0
} else {
0.0
};
let web_over_native = if native_mean_ops_sec > 0.0 {
web.t.ops_sec / native_mean_ops_sec
} else {
0.0
};
Ok(WebFetchOutcome {
web,
native,
native_after_ops_sec,
native_mean_ops_sec,
native_drift_pct,
web_cost_pct,
web_over_native,
})
})();
if outcome.is_ok() {
running.shutdown();
}
outcome
}
/// The timed gRPC-web pass. Blocking `reqwest` over HTTP/1.1 — no h2 upgrade,
/// no ALPN — with one keep-alive client per worker, which is the browser's
/// shape.
fn fetch_web(
running: &Running,
prefix: &str,
corpus: usize,
ops: usize,
conc: usize,
) -> Result<UnaryOutcome> {
let base = running.endpoint();
let conc = conc.max(1);
let per = ops.div_ceil(conc);
let load = load_avg_1m();
let url = format!("{base}/holger.v1.RepositoryService/FetchArtifact");
let t0 = Instant::now();
let per_worker: Vec<Result<(u64, u64, Vec<f32>)>> = std::thread::scope(|s| {
let handles: Vec<_> = (0..conc)
.map(|tid| {
let url = url.clone();
let prefix = prefix.to_string();
s.spawn(move || -> Result<(u64, u64, Vec<f32>)> {
let http = tls_roots_blocking(reqwest::blocking::Client::builder())
.tcp_keepalive(Duration::from_secs(30))
.timeout(Duration::from_secs(120))
.build()
.context("build the gRPC-web client")?;
let mut ok = 0u64;
let mut bytes = 0u64;
let mut lat: Vec<f32> = Vec::with_capacity(per);
for i in 0..per {
let idx = (i * 7 + tid * 1337) % corpus;
let t = Instant::now();
let data = grpc_web_call(&http, &url, &prefix, idx)?;
lat.push(t.elapsed().as_secs_f64() as f32 * 1000.0);
anyhow::ensure!(
data.len() == PAYLOAD_BYTES,
"the gRPC-web door answered {} bytes for {prefix}-{idx:06}, not \
{PAYLOAD_BYTES}",
data.len()
);
bytes += data.len() as u64;
ok += 1;
}
Ok((ok, bytes, lat))
})
})
.collect();
handles
.into_iter()
.map(|h| {
h.join()
.unwrap_or_else(|_| Err(anyhow::anyhow!("a gRPC-web worker panicked")))
})
.collect()
});
let elapsed = t0.elapsed();
let mut ok = 0u64;
let mut bytes = 0u64;
let mut lat: Vec<f32> = Vec::with_capacity(ops);
for r in per_worker {
let (o, b, mut l) = r?;
ok += o;
bytes += b;
lat.append(&mut l);
}
anyhow::ensure!(
ok as usize == per * conc,
"the gRPC-web pool completed {ok} of {} ops",
per * conc
);
let (p50, p99, p999, mean) = served::latency_percentiles(&mut lat);
Ok(UnaryOutcome {
t: throughput(ok as usize, bytes, elapsed),
p50_ms: p50,
p99_ms: p99,
p999_ms: p999,
mean_ms: mean,
concurrency: conc,
payload_bytes: PAYLOAD_BYTES,
verified_ops: 0,
load_avg_1m: load,
})
}
/// One gRPC-web `FetchArtifact`, returning the artifact bytes.
fn grpc_web_call(
http: &reqwest::blocking::Client,
url: &str,
prefix: &str,
idx: usize,
) -> Result<Vec<u8>> {
let req = FetchArtifactRequest {
repository: REPO.into(),
id: Some(proto_id(prefix, idx)),
};
let mut buf = Vec::with_capacity(req.encoded_len());
req.encode(&mut buf)?;
let resp = http
.post(url)
.header("content-type", "application/grpc-web+proto")
.header("accept", "application/grpc-web+proto")
.header("x-grpc-web", "1")
.body(grpc_web_frame(&buf))
.send()
.with_context(|| format!("gRPC-web POST for {prefix}-{idx:06}"))?;
let status = resp.status();
let headers = resp.headers().clone();
let body = resp.bytes()?;
anyhow::ensure!(
status.is_success(),
"the gRPC-web door answered HTTP {status} for {prefix}-{idx:06}"
);
let (msg, trailers) = grpc_web_parse(&body)?;
match grpc_web_status(&trailers, &headers) {
Some(0) | None => {}
Some(code) => anyhow::bail!(
"the gRPC-web door answered grpc-status {code} for {prefix}-{idx:06} ({trailers})"
),
}
let decoded = FetchArtifactResponse::decode(msg.as_slice())
.with_context(|| format!("decode the gRPC-web answer for {prefix}-{idx:06}"))?;
Ok(decoded.data.to_vec())
}
/// The gRPC-web ANSWER floor, untimed: read every artifact back THROUGH THE WEB
/// TRANSPORT and compare it with what was stored, by sha256.
fn verify_corpus_web(running: &Running, prefix: &str, digests: &[[u8; 32]]) -> Result<usize> {
let url = format!(
"{}/holger.v1.RepositoryService/FetchArtifact",
running.endpoint()
);
let http = tls_roots_blocking(reqwest::blocking::Client::builder())
.tcp_keepalive(Duration::from_secs(30))
.timeout(Duration::from_secs(120))
.build()
.context("build the gRPC-web verification client")?;
for (i, want) in digests.iter().enumerate() {
let data = grpc_web_call(&http, &url, prefix, i)?;
let got = <[u8; 32]>::from(Sha256::digest(&data));
anyhow::ensure!(
&got == want,
"the gRPC-web door returned bytes that are not what was stored for {prefix}-{i:06} \
({} bytes back)",
data.len()
);
}
Ok(digests.len())
}
#[cfg(test)]
mod door_smoke {
use super::*;
/// The whole module in one cheap pass: boot a real holger, store through the
/// gRPC door, fetch it back natively AND over gRPC-web, and stream it. Tiny
/// N — this proves the door answers and the framing is right, never a
/// throughput.
#[test]
fn every_grpc_door_answers_and_round_trips() {
let running = start_bench_holger("smoke").expect("a bench holger must boot");
let digests =
seed(&running, "smoke", 4, PAYLOAD_BYTES).expect("store through the gRPC door");
let out = fetch_native(&running, "smoke", 4, 8, 2).expect("native fetch");
assert_eq!(out.t.ops, 8, "every native fetch completes");
assert_eq!(
verify_corpus(&running, "smoke", &digests).expect("native verify"),
4
);
let web = fetch_web(&running, "smoke", 4, 4, 2).expect("gRPC-web fetch");
assert_eq!(web.t.ops, 4, "every gRPC-web fetch completes");
assert_eq!(
verify_corpus_web(&running, "smoke", &digests).expect("gRPC-web verify"),
4,
"the browser transport returns the stored bytes"
);
let endpoint = running.endpoint();
let read = running
.rt
.block_on(async {
let mut c = client(&endpoint).await?;
stream_one(&mut c, "smoke", 0, true).await
})
.expect("stream");
assert_eq!(read.bytes, PAYLOAD_BYTES as u64);
assert_eq!(read.digest, Some(digests[0]));
assert!(read.max_chunk <= DECLARED_CHUNK_BYTES);
running.shutdown();
}
/// The gRPC-web framing, without a server: a data frame round-trips and a
/// trailer frame is read as trailers rather than as message bytes.
#[test]
fn grpc_web_framing_round_trips() {
let body = {
let mut b = grpc_web_frame(b"hello");
b.push(0x80);
let t = "grpc-status:0\r\n";
b.extend_from_slice(&(t.len() as u32).to_be_bytes());
b.extend_from_slice(t.as_bytes());
b
};
let (msg, trailers) = grpc_web_parse(&body).expect("a well-formed body parses");
assert_eq!(msg, b"hello");
assert_eq!(
grpc_web_status(&trailers, &reqwest::header::HeaderMap::new()),
Some(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()),
)
}