//! Benchmark tool for the broker.
//!
//! Measures throughput and latency under load.
//!
//! Supports different routing strategies to test worker selection:
//! - best_price: Route to cheapest worker
//! - best_latency: Route to fastest worker
//! - best_availability: Route to most available worker
//! - round_robin: Distribute evenly across workers
//! - random: Random worker selection
//! - weighted_capacity: Weighted by available resources
use std::sync::atomic::{AtomicU64, Ordering};
use std::sync::Arc;
use std::thread;
use std::time::{Duration, Instant};
use crate::async_exec;
use colored::Colorize;
use super::router::RoutingStrategy;
/// Percentage `num/denom*100`, guarding against a zero denominator
/// (returns 0.0 instead of NaN when there were no requests).
fn pct(num: u64, denom: u64) -> f64 {
if denom == 0 {
0.0
} else {
num as f64 / denom as f64 * 100.0
}
}
/// Concurrency clamped to at least 1 so request distribution never divides by zero.
fn effective_concurrency(c: usize) -> usize {
c.max(1)
}
/// How many completed requests must be seen before the failure rate is
/// allowed to end a run: small enough to bail out quickly, large enough that
/// a couple of unlucky requests can't.
const FAIL_FAST_MIN_SAMPLE: u64 = 20;
/// The share of failures, over that sample, that means the run is no longer
/// measuring anything.
const FAIL_FAST_RATE: f64 = 0.5;
/// Whether a run has failed enough to be worth stopping. `limit` is the
/// consecutive-failure trip, and 0 disables the guard entirely.
///
/// Two shapes of useless run, so two triggers: everything times out (a long
/// streak), or most things do while a minority succeed. The second is what a
/// broker that routes only some requests down a broken path produces, and a
/// streak counter alone almost never trips on it.
pub(crate) fn run_is_failing(completed: u64, failed: u64, consecutive: u64, limit: u32) -> bool {
if limit == 0 {
return false;
}
if consecutive >= limit as u64 {
return true;
}
completed >= FAIL_FAST_MIN_SAMPLE && (failed as f64) >= (completed as f64) * FAIL_FAST_RATE
}
/// The `p`th percentile of `latencies`, in microseconds. One rule, used by
/// both the live line and the summary: two implementations would eventually
/// disagree and leave a reader unsure which to believe.
pub(crate) fn percentile_us(latencies: &[u64], p: f64) -> u64 {
if latencies.is_empty() {
return 0;
}
let mut sorted = latencies.to_vec();
sorted.sort_unstable();
let idx = ((sorted.len() as f64 * p / 100.0) as usize).min(sorted.len() - 1);
sorted[idx]
}
/// p50 and p99 of the samples so far, for the live line.
pub(crate) fn percentiles_us(latencies: &[u64]) -> (u64, u64) {
(
percentile_us(latencies, 50.0),
percentile_us(latencies, 99.0),
)
}
/// Microseconds as something readable at a glance: `820µs`, `4.2ms`, `1.35s`.
pub(crate) fn fmt_us(us: u64) -> String {
match us {
0 => "—".to_string(),
us if us < 1_000 => format!("{us}µs"),
us if us < 1_000_000 => format!("{:.1}ms", us as f64 / 1_000.0),
us => format!("{:.2}s", us as f64 / 1_000_000.0),
}
}
/// A worker name short enough to sit on the live line.
pub(crate) fn short_worker(name: &str) -> String {
let trimmed = name.strip_prefix("worker-zc-worker-").unwrap_or(name);
match trimmed.chars().count() > 22 {
true => format!("{}…", trimmed.chars().take(21).collect::<String>()),
false => trimmed.to_string(),
}
}
/// The latency buckets, in milliseconds, shared by the live view and the
/// final histogram so the two can never tell different stories.
pub(crate) const LATENCY_BUCKETS_MS: [f64; 10] = [
1.0,
5.0,
10.0,
25.0,
50.0,
100.0,
250.0,
500.0,
1000.0,
f64::INFINITY,
];
/// How many samples fall in each bucket of [`LATENCY_BUCKETS_MS`].
pub(crate) fn histogram(latencies_us: &[u64]) -> Vec<u64> {
let mut counts = vec![0u64; LATENCY_BUCKETS_MS.len()];
for us in latencies_us {
let ms = *us as f64 / 1000.0;
if let Some(i) = LATENCY_BUCKETS_MS.iter().position(|&b| ms <= b) {
counts[i] += 1;
}
}
counts
}
/// A bucket's label, e.g. `≤25ms` or `>1000ms`.
pub(crate) fn bucket_label(i: usize) -> String {
match LATENCY_BUCKETS_MS[i].is_infinite() {
true => format!(">{}ms", LATENCY_BUCKETS_MS[i - 1]),
false => format!("≤{}ms", LATENCY_BUCKETS_MS[i]),
}
}
/// The node a worker belongs to, for the "where did the work go" view.
/// `worker-zc-worker-i9-1` is node `i9`; `dd715b7f-w0` is this machine's
/// broker, whose workers all share the node fingerprint before `-w`.
pub(crate) fn node_of(worker: &str) -> String {
if let Some(rest) = worker.strip_prefix("worker-zc-worker-") {
return rest
.rsplit_once('-')
.map_or(rest, |(node, _)| node)
.to_string();
}
if let Some((node, tail)) = worker.rsplit_once("-w") {
if tail.chars().all(|c| c.is_ascii_digit()) && !node.is_empty() {
return node.to_string();
}
}
worker.to_string()
}
/// Per-node counts, largest first, from per-worker counts.
pub(crate) fn by_node(
worker_counts: &std::collections::HashMap<String, u64>,
) -> Vec<(String, u64)> {
let mut nodes: std::collections::HashMap<String, u64> = std::collections::HashMap::new();
for (worker, n) in worker_counts {
*nodes.entry(node_of(worker)).or_insert(0) += n;
}
let mut v: Vec<(String, u64)> = nodes.into_iter().collect();
v.sort_by(|a, b| b.1.cmp(&a.1).then_with(|| a.0.cmp(&b.0)));
v
}
/// A proportional bar `width` cells wide.
pub(crate) fn bar(count: u64, max: u64, width: usize) -> String {
let filled = match max {
0 => 0,
max => ((count as f64 / max as f64) * width as f64).round() as usize,
};
format!(
"{}{}",
"█".repeat(filled.min(width)),
" ".repeat(width - filled.min(width))
)
}
/// The live view: one block, redrawn in place while the run goes on. It
/// answers the three questions a benchmark is actually asked — how fast, how
/// is it spread, and where is the work running — instead of only how far
/// along it is.
#[allow(clippy::too_many_arguments)]
pub(crate) fn live_block(
spin: char,
done: u64,
total: u64,
ok: u64,
failed: u64,
elapsed: Duration,
samples: &[u64],
worker_counts: &std::collections::HashMap<String, u64>,
) -> String {
use colored::Colorize;
let secs = elapsed.as_secs_f64().max(0.001);
let mut out = String::new();
out.push_str(&format!(
" {} {:>6}/{} {:>5.1}% {} ok {} failed {:>7.1} req/s {:.0}s\n",
spin.to_string().cyan(),
done,
total,
done as f64 / total.max(1) as f64 * 100.0,
ok.to_string().green(),
if failed > 0 {
failed.to_string().red().to_string()
} else {
failed.to_string()
},
done as f64 / secs,
secs,
));
let (p50, p99) = percentiles_us(samples);
out.push_str(&format!(
" {} p50 {} p90 {} p99 {}\n",
"latency".bold(),
fmt_us(p50),
fmt_us(percentile_us(samples, 90.0)),
fmt_us(p99),
));
let counts = histogram(samples);
let peak = counts.iter().copied().max().unwrap_or(0);
for (i, n) in counts.iter().enumerate() {
out.push_str(&format!(
" {:>8} {} {:>6}\n",
bucket_label(i).dimmed(),
bar(*n, peak, 24),
n
));
}
let nodes = by_node(worker_counts);
let busiest = nodes.first().map(|(_, n)| *n).unwrap_or(0);
out.push_str(&format!(" {}\n", "nodes".bold()));
if nodes.is_empty() {
out.push_str(" (nothing has been routed yet)\n");
}
for (node, n) in nodes.iter().take(6) {
out.push_str(&format!(
" {:>22} {} {:>6} ({:>4.1}%)\n",
short_worker(node).cyan(),
bar(*n, busiest, 24),
n,
*n as f64 / done.max(1) as f64 * 100.0
));
}
out
}
/// How wide a string looks on screen: colour escapes take no columns, and a
/// count of `char`s is what every padding decision here needs.
pub(crate) fn visible_width(s: &str) -> usize {
let mut n = 0;
let mut chars = s.chars();
while let Some(c) = chars.next() {
if c == '\u{1b}' {
for e in chars.by_ref() {
if e == 'm' {
break;
}
}
} else {
n += 1;
}
}
n
}
/// Pad `s` to `width` columns, measuring what is visible rather than what is
/// stored — the reason the old tables came out ragged once a name was longer
/// than its hard-coded column.
pub(crate) fn pad(s: &str, width: usize, right: bool) -> String {
let fill = " ".repeat(width.saturating_sub(visible_width(s)));
match right {
true => format!("{fill}{s}"),
false => format!("{s}{fill}"),
}
}
/// What distinguishes one worker from its siblings on the same node:
/// `dd715b7f-w2` → `w2`, `worker-zc-worker-i9-1` → `1`. The node is already
/// on the line above, so repeating it would only push the line wider.
pub(crate) fn worker_leaf(worker: &str) -> String {
let node = node_of(worker);
match worker.rsplit_once(&format!("{node}-")) {
Some((_, leaf)) if !leaf.is_empty() => leaf.to_string(),
_ => worker.to_string(),
}
}
/// A box sized to its own contents. The old one had a 43-column border around
/// 41 columns of text, which is why the title never lined up.
pub(crate) fn boxed(lines: &[String]) -> String {
use colored::Colorize;
let inner = lines.iter().map(|l| visible_width(l)).max().unwrap_or(0) + 4;
let mut out = String::new();
out.push_str(&format!(
" {}\n",
format!("╭{}╮", "─".repeat(inner)).cyan()
));
for line in lines {
out.push_str(&format!(
" {} {} {}\n",
"│".cyan(),
pad(line, inner - 4, false),
"│".cyan()
));
}
out.push_str(&format!(
" {}\n",
format!("╰{}╯", "─".repeat(inner)).cyan()
));
out
}
/// Named benchmark workloads
#[derive(Debug, Clone, PartialEq)]
pub enum BenchWorkload {
/// Trivial add(2, 3) — measures pure broker/routing overhead
Add,
/// Recursive fib(28) — CPU-intensive, ~50ms per call
Fib,
}
impl BenchWorkload {
pub fn name(&self) -> &'static str {
match self {
BenchWorkload::Add => "add",
BenchWorkload::Fib => "fib",
}
}
pub fn description(&self) -> &'static str {
match self {
BenchWorkload::Add => "add(2, 3) — routing overhead baseline",
BenchWorkload::Fib => "fib(28) — CPU-intensive recursive Fibonacci",
}
}
}
impl std::str::FromStr for BenchWorkload {
type Err = String;
fn from_str(s: &str) -> Result<Self, Self::Err> {
match s.to_lowercase().as_str() {
"add" => Ok(BenchWorkload::Add),
"fib" | "fibonacci" | "cpu" => Ok(BenchWorkload::Fib),
_ => Err(format!(
"Unknown benchmark workload: '{}'. Available: add, fib",
s
)),
}
}
}
/// Benchmark configuration
#[derive(Debug, Clone)]
pub struct BenchConfig {
/// Target broker URL
pub broker_url: String,
/// Number of concurrent workers
pub concurrency: usize,
/// Total number of requests to send
pub requests: u64,
/// Request timeout in seconds
pub timeout_secs: u64,
/// Stop the run after this many failures in a row (0 disables it).
pub fail_fast: u32,
/// User ID for requests
pub user_id: String,
/// API key for Bearer auth (read from ZAKURO_API_KEY env)
pub api_key: Option<String>,
/// CPU requirement per request
pub cpus: f64,
/// Memory requirement per request (bytes)
pub memory_bytes: u64,
/// Routing strategy to use
pub strategy: RoutingStrategy,
/// Run comparison across all strategies
pub compare_strategies: bool,
/// Named benchmark workload
pub workload: BenchWorkload,
/// Filter by worker type (e.g. "standard", "premium") — routes only to matching workers
pub worker_type: Option<String>,
/// Force cross-node dispatch: skip local workers, always offer to peer brokers.
pub remote_only: bool,
/// Per-request credit ceiling sent as `budget_credits`.
///
/// This bounds the remote reservation so it cannot reserve against an
/// unpriced (`f64::MAX`) peer worker before the offer resolves. It is a
/// CEILING, not a charge -- a bench job actually costs on the order of
/// 0.001 credits. It matters because the broker's precheck rejects the
/// request when the account balance is below the budget, so an
/// unnecessarily large value locks out accounts that can easily afford the
/// work.
pub budget_credits: f64,
}
impl Default for BenchConfig {
fn default() -> Self {
Self {
broker_url: "zc://localhost".to_string(),
concurrency: 10,
requests: 1000,
// 15 s, not 30: the slowest legitimate path measured on a mesh
// is a ~0.7 s round trip, and `fib(28)` is seconds at worst. A
// heavier workload can still pass --timeout.
timeout_secs: 15,
fail_fast: 10,
user_id: "bench-user".to_string(),
// Treat a set-but-empty ZAKURO_API_KEY as "no key" so we don't send
// an empty `Authorization: Bearer ` header (which the broker rejects).
// `zc login` writes the key to ~/.zakuro/credentials, and every
// other command reads it from there. Without this a normal shell
// benchmarks as `anonymous` with no credits, and every request
// that routes to a billed worker stalls until it times out.
api_key: {
crate::credentials::load_into_env();
std::env::var("ZAKURO_API_KEY")
}
.ok()
.filter(|k| !k.trim().is_empty()),
cpus: 0.1,
memory_bytes: 1024 * 1024 * 100, // 100 MiB
strategy: RoutingStrategy::BestPrice,
compare_strategies: false,
workload: BenchWorkload::Add,
worker_type: None,
remote_only: false,
// Generous next to a real bench job (~0.001 credits) while staying
// low enough that the balance precheck is not a gate. The previous
// hardcoded 1000.0 meant any account under 1000 credits failed
// EVERY request with a bare 402 that the summary did not surface --
// which reads as "dispatch is broken" rather than "fund the
// account". Raise it with --budget for genuinely long jobs.
budget_credits: 10.0,
}
}
}
/// Benchmark results
#[derive(Debug, Clone)]
pub struct BenchResults {
/// Total requests sent
pub total_requests: u64,
/// Successful requests
pub successful: u64,
/// Failed requests
pub failed: u64,
/// Total duration
pub duration: Duration,
/// Requests per second
pub rps: f64,
/// Latencies in microseconds
pub latencies_us: Vec<u64>,
/// Per-worker executed-request counts (worker name → count).
pub worker_counts: std::collections::HashMap<String, u64>,
/// Why requests failed (error text → count).
///
/// Always collected, not gated behind ZAKURO_BENCH_DEBUG. A bare
/// "Failed: 6 (100.0%)" is indistinguishable between "the account is out of
/// credits" (402) and "no worker could be reached" (503), and reads as the
/// latter -- which sends you debugging dispatch when the fix is to fund an
/// account.
pub failure_reasons: std::collections::HashMap<String, u64>,
}
impl BenchResults {
fn new() -> Self {
Self {
total_requests: 0,
successful: 0,
failed: 0,
worker_counts: std::collections::HashMap::new(),
failure_reasons: std::collections::HashMap::new(),
duration: Duration::ZERO,
rps: 0.0,
latencies_us: Vec::new(),
}
}
/// Failure reasons ordered by frequency, ties broken by text so the output
/// is deterministic across runs (HashMap iteration order is not).
fn ranked_failure_reasons(&self) -> Vec<(String, u64)> {
let mut reasons: Vec<(String, u64)> = self
.failure_reasons
.iter()
.map(|(k, v)| (k.clone(), *v))
.collect();
reasons.sort_by(|a, b| b.1.cmp(&a.1).then(a.0.cmp(&b.0)));
reasons
}
fn percentile(&self, p: f64) -> f64 {
percentile_us(&self.latencies_us, p) as f64 / 1000.0
}
fn avg_latency_ms(&self) -> f64 {
if self.latencies_us.is_empty() {
return 0.0;
}
let sum: u64 = self.latencies_us.iter().sum();
(sum as f64 / self.latencies_us.len() as f64) / 1000.0
}
fn min_latency_ms(&self) -> f64 {
self.latencies_us
.iter()
.min()
.map(|v| *v as f64 / 1000.0)
.unwrap_or(0.0)
}
fn max_latency_ms(&self) -> f64 {
self.latencies_us
.iter()
.max()
.map(|v| *v as f64 / 1000.0)
.unwrap_or(0.0)
}
pub fn print_report(&self) {
print!("{}", self.render_report());
}
/// The finished run, rendered. Every column is sized to what it holds, so
/// a long worker name cannot push a bar out of line, and the work is
/// grouped by node because that is the question being asked: where did it
/// run.
pub(crate) fn render_report(&self) -> String {
let mut out = String::new();
out.push('\n');
let ok = format!("{}", self.successful).green().to_string();
let bad = match self.failed {
0 => "0".to_string(),
n => n.to_string().red().to_string(),
};
out.push_str(&format!(
" {} {} req/s {} ok · {} failed in {:.1}s\n",
"result".bold(),
format!("{:.1}", self.rps).yellow().bold(),
ok,
bad,
self.duration.as_secs_f64(),
));
for (reason, count) in self.ranked_failure_reasons().iter().take(5) {
out.push_str(&format!(
" {} {}\n",
format!("{count}×").red(),
reason.dimmed()
));
}
// Latency, in two columns: the shape of the tail matters as much as
// the middle, and side by side they read in one glance.
let left = [
("min", self.min_latency_ms()),
("avg", self.avg_latency_ms()),
("max", self.max_latency_ms()),
];
let right = [
("p50", self.percentile(50.0)),
("p90", self.percentile(90.0)),
("p99", self.percentile(99.0)),
];
out.push_str(&format!("\n {}\n", "latency".bold()));
for (l, r) in left.iter().zip(right.iter()) {
out.push_str(&format!(
" {} {} {} {}\n",
l.0.dimmed(),
pad(&fmt_us((l.1 * 1000.0) as u64), 9, true),
r.0.dimmed(),
pad(&fmt_us((r.1 * 1000.0) as u64), 9, true),
));
}
// Where the time went, bucketed. Empty buckets stay: their emptiness
// is the point when everything lands past a second.
let counts = histogram(&self.latencies_us);
let peak = counts.iter().copied().max().unwrap_or(0);
let label_w = (0..counts.len())
.map(|i| visible_width(&bucket_label(i)))
.max()
.unwrap_or(0);
out.push_str(&format!("\n {}\n", "distribution".bold()));
for (i, n) in counts.iter().enumerate() {
out.push_str(&format!(
" {} {} {} {}\n",
pad(&bucket_label(i).dimmed().to_string(), label_w, true),
bar(*n, peak, 28),
pad(&n.to_string(), 5, true),
pad(&format!("{:.0}%", pct(*n, self.total_requests)), 4, true).dimmed(),
));
}
if !self.worker_counts.is_empty() {
out.push_str(&format!("\n {}\n", "where it ran".bold()));
let nodes = by_node(&self.worker_counts);
let busiest = nodes.first().map(|(_, n)| *n).unwrap_or(0);
let name_w = nodes
.iter()
.map(|(n, _)| visible_width(n))
.max()
.unwrap_or(0);
for (node, n) in &nodes {
out.push_str(&format!(
" {} {} {} {}\n",
pad(&node.cyan().to_string(), name_w, false),
bar(*n, busiest, 28),
pad(&n.to_string(), 5, true),
pad(&format!("{:.0}%", pct(*n, self.total_requests)), 4, true).dimmed(),
));
// The workers behind that node, compactly: useful when one of
// them is doing nothing while its siblings are busy.
let mut mine: Vec<(&String, &u64)> = self
.worker_counts
.iter()
.filter(|(w, _)| node_of(w) == *node)
.collect();
if mine.len() > 1 {
mine.sort_by(|a, b| b.1.cmp(a.1).then_with(|| a.0.cmp(b.0)));
let detail: Vec<String> = mine
.iter()
.map(|(w, c)| format!("{} {}", worker_leaf(w), c))
.collect();
out.push_str(&format!(
" {}\n",
format!("{:width$} {}", "", detail.join(" · "), width = name_w).dimmed()
));
}
}
}
out.push('\n');
out
}
}
/// The mesh-connectivity line `zc bench` prints. It states the fact and
/// nothing else: the broker's mesh address used to be appended here, and the
/// CLI never shows an IP to an end user (see `main.rs`'s IP-FREE RULE).
fn wireguard_line(connected: bool) -> String {
use colored::Colorize;
match connected {
true => "connected".green().to_string(),
false => "not connected (continuing)".yellow().to_string(),
}
}
/// Query the broker's /health endpoint and return (wireguard_connected, wireguard_ip, node_name).
fn check_wireguard_status(broker_url: &str) -> (bool, Option<String>, Option<String>) {
match crate::vpn::mesh_agent_for(broker_url, std::time::Duration::from_secs(30))
.get(&format!("{}/health", broker_url))
.config()
.timeout_global(Some(Duration::from_secs(5)))
.build()
.call()
{
Ok(resp) => {
if let Ok(body) = resp.into_body().read_to_string() {
if let Ok(v) = serde_json::from_str::<serde_json::Value>(&body) {
let connected = v["wireguard_connected"].as_bool().unwrap_or(false);
let ip = v["wireguard_ip"].as_str().map(|s| s.to_string());
let name = v["node_name"].as_str().map(|s| s.to_string());
return (connected, ip, name);
}
}
(false, None, None)
}
Err(_) => (false, None, None),
}
}
/// Run the benchmark
pub fn run_benchmark(config: BenchConfig) -> BenchResults {
println!();
print!(
"{}",
boxed(&[
format!("{}", "zc bench".bold().white()),
format!(
"{} · {} · {} concurrent · {} requests",
config.workload.name().yellow(),
config.strategy.as_str().yellow(),
config.concurrency,
config.requests
),
format!("{} {}", "broker".dimmed(), config.broker_url.cyan()),
])
);
println!();
println!(
" {} {}",
pad(
&config.workload.name().yellow().bold().to_string(),
14,
false
),
config.workload.description().dimmed()
);
println!(
" {} {}",
pad(
&config.strategy.as_str().yellow().bold().to_string(),
14,
false
),
config.strategy.description().dimmed()
);
if let Some(ref wt) = config.worker_type {
println!(" Worker Type: {}", wt.yellow().bold());
}
println!();
// First check if broker is reachable
print!(" Connecting to broker... ");
match crate::vpn::mesh_agent_for(&config.broker_url, std::time::Duration::from_secs(30))
.get(&format!("{}/health", config.broker_url))
.config()
.timeout_global(Some(Duration::from_secs(5)))
.build()
.call()
{
Ok(_) => println!("{}", "OK".green()),
Err(e) => {
println!("{}", "FAILED".red());
println!(" Error: {}", e);
return BenchResults::new();
}
}
// Check WireGuard connectivity — required for cross-node billing (skip for localhost)
if crate::vpn::is_loopback_url(config.broker_url.trim()) {
println!(" {} (skipping WireGuard check)", "Local broker".dimmed());
} else {
print!(" Checking WireGuard... ");
let (ts_connected, _, _) = check_wireguard_status(&config.broker_url);
if ts_connected {
println!("{}", wireguard_line(true));
} else {
// Not fatal: the broker is already reachable (its /health answered
// above), possibly over another transport such as WireGuard. Warn
// and continue — the workers check below validates the actual fleet.
println!("{}", wireguard_line(false));
}
}
// Check if there are workers
if config.api_key.is_none() {
println!(
" {} not signed in: requests run as {} and a billed worker will refuse them — run {}",
"Warning:".yellow(),
"anonymous".yellow(),
"zc login".cyan()
);
}
print!(" Checking workers... ");
match crate::vpn::mesh_agent_for(&config.broker_url, std::time::Duration::from_secs(30))
.get(&format!("{}/workers", config.broker_url))
.config()
.timeout_global(Some(Duration::from_secs(5)))
.build()
.call()
{
Ok(resp) => {
if let Ok(body) = resp.into_body().read_to_string() {
if let Ok(json) = serde_json::from_str::<serde_json::Value>(&body) {
let count = json["total"].as_u64().unwrap_or(0);
if count == 0 {
println!("{}", "NO WORKERS".yellow());
println!(" Warning: No workers connected. Benchmark may fail.");
} else {
println!("{} worker(s)", count.to_string().green());
}
}
}
}
Err(_) => println!("{}", "UNKNOWN".yellow()),
}
println!();
println!(" Starting benchmark...");
println!();
// Shared counters
let successful = Arc::new(AtomicU64::new(0));
let failed = Arc::new(AtomicU64::new(0));
let completed = Arc::new(AtomicU64::new(0));
let stop = Arc::new(std::sync::atomic::AtomicBool::new(false));
let consecutive_failures = Arc::new(AtomicU64::new(0));
let latencies = Arc::new(std::sync::Mutex::new(Vec::with_capacity(
config.requests as usize,
)));
let worker_counts = Arc::new(std::sync::Mutex::new(std::collections::HashMap::new()));
let failure_reasons = Arc::new(std::sync::Mutex::new(std::collections::HashMap::new()));
// Calculate requests per worker (clamp concurrency so `-c 0` cannot divide by zero)
let concurrency = effective_concurrency(config.concurrency);
let requests_per_worker = config.requests / concurrency as u64;
let extra_requests = config.requests % concurrency as u64;
let start = Instant::now();
// Spawn worker threads
let handles: Vec<_> = (0..concurrency)
.map(|i| {
let config = config.clone();
let successful = successful.clone();
let failed = failed.clone();
let completed = completed.clone();
let stop = stop.clone();
let consecutive_failures = consecutive_failures.clone();
let latencies = latencies.clone();
let worker_counts = worker_counts.clone();
let failure_reasons = failure_reasons.clone();
// First few workers get extra requests to handle remainder
let my_requests = if (i as u64) < extra_requests {
requests_per_worker + 1
} else {
requests_per_worker
};
async_exec::spawn_blocking(move || {
run_worker(
i,
my_requests,
&config,
&BenchCounters {
successful: &successful,
failed: &failed,
completed: &completed,
stop: &stop,
consecutive_failures: &consecutive_failures,
latencies: &latencies,
worker_counts: &worker_counts,
failure_reasons: &failure_reasons,
},
);
})
})
.collect();
// Progress reporter
let total = config.requests;
let completed_for_progress = completed.clone();
let successful_for_progress = successful.clone();
let failed_for_progress = failed.clone();
let latencies_for_progress = latencies.clone();
let workers_for_progress = worker_counts.clone();
let progress_handle = async_exec::spawn_blocking(move || {
let spinner = ['⠋', '⠙', '⠹', '⠸', '⠼', '⠴', '⠦', '⠧', '⠇', '⠏'];
let mut i = 0;
let started = Instant::now();
// Redrawing in place needs a terminal. Piped or redirected, print a
// plain line now and then instead of painting cursor moves into a file.
let tty = std::io::IsTerminal::is_terminal(&std::io::stdout());
let mut drawn = 0usize;
loop {
let done = completed_for_progress.load(Ordering::Relaxed);
let ok = successful_for_progress.load(Ordering::Relaxed);
let bad = failed_for_progress.load(Ordering::Relaxed);
let samples = latencies_for_progress
.lock()
.map(|l| l.clone())
.unwrap_or_default();
let workers = workers_for_progress
.lock()
.map(|w| w.clone())
.unwrap_or_default();
let block = live_block(
spinner[i % spinner.len()],
done,
total,
ok,
bad,
started.elapsed(),
&samples,
&workers,
);
if tty {
// Back to the top of the last block, then repaint it. Each
// line is cleared to its end so a shorter line can't leave the
// tail of a longer one behind.
let mut out = String::new();
if drawn > 0 {
out.push_str(&format!("\x1b[{drawn}A"));
}
for line in block.lines() {
out.push_str("\x1b[2K");
out.push_str(line);
out.push('\n');
}
print!("{out}");
drawn = block.lines().count();
} else if done >= total || i % 20 == 0 {
println!(
" {done}/{total} · {ok} ok · {bad} failed · {:.1} req/s",
done as f64 / started.elapsed().as_secs_f64().max(0.001)
);
}
let _ = std::io::Write::flush(&mut std::io::stdout());
if done >= total {
if tty && drawn > 0 {
// Wipe the live block; the summary below says all of it
// again, and leaving both on screen reads as a stutter.
print!("\x1b[{drawn}A");
for _ in 0..drawn {
println!("\x1b[2K");
}
print!("\x1b[{drawn}A");
let _ = std::io::Write::flush(&mut std::io::stdout());
}
break;
}
thread::sleep(Duration::from_millis(100));
i += 1;
}
});
// Wait for all workers
for handle in handles {
let _ = handle.join();
}
let _ = progress_handle.join();
let duration = start.elapsed();
// Collect results
let successful_count = successful.load(Ordering::Relaxed);
let failed_count = failed.load(Ordering::Relaxed);
let latencies_vec = latencies.lock().unwrap().clone();
let worker_counts_map = worker_counts.lock().unwrap().clone();
let failure_reasons_map = failure_reasons.lock().unwrap().clone();
let secs = duration.as_secs_f64();
let rps = if secs > 0.0 {
config.requests as f64 / secs
} else {
0.0
};
BenchResults {
total_requests: config.requests,
successful: successful_count,
failed: failed_count,
duration,
rps,
latencies_us: latencies_vec,
worker_counts: worker_counts_map,
failure_reasons: failure_reasons_map,
}
}
#[allow(clippy::too_many_arguments)]
/// Build the `X-Zakuro-Requirements` payload for a bench request.
///
/// Split out from `run_worker` so the budget wiring is testable without
/// standing up a broker.
fn build_requirements(config: &BenchConfig) -> serde_json::Value {
let mut requirements = serde_json::json!({
"cpus": config.cpus,
"memory_bytes": config.memory_bytes,
"gpus": 0,
"estimated_duration_secs": 0.1,
"strategy": config.strategy.as_str(),
"remote_only": config.remote_only,
// Bound the remote reservation so it doesn't reserve against an unpriced
// (f64::MAX) peer worker before the offer resolves. Configurable: the
// broker also rejects when balance < budget, so this must not be larger
// than it needs to be. See BenchConfig::budget_credits.
"budget_credits": config.budget_credits
});
if let Some(ref wt) = config.worker_type {
requirements["worker_type"] = serde_json::Value::String(wt.clone());
}
requirements
}
/// The accumulators every bench worker writes into.
///
/// Grouped rather than passed individually: they are one thing -- the shared
/// tally for a run -- and threading six references through each call site made
/// run_worker a nine-argument function, which clippy rejects. Adding an
/// argument to the tally now means adding a field here, not editing three
/// signatures.
struct BenchCounters<'a> {
successful: &'a AtomicU64,
failed: &'a AtomicU64,
completed: &'a AtomicU64,
/// Set once the run has failed enough times in a row to be pointless;
/// every thread checks it before sending the next request.
stop: &'a std::sync::atomic::AtomicBool,
/// Failures in a row across the whole run. Per-thread it would take
/// `limit` failures in one thread, which a concurrent run rarely reaches
/// even when almost everything is failing.
consecutive_failures: &'a AtomicU64,
latencies: &'a std::sync::Mutex<Vec<u64>>,
worker_counts: &'a std::sync::Mutex<std::collections::HashMap<String, u64>>,
failure_reasons: &'a std::sync::Mutex<std::collections::HashMap<String, u64>>,
}
fn run_worker(
_worker_id: usize,
requests: u64,
config: &BenchConfig,
counters: &BenchCounters<'_>,
) {
let BenchCounters {
successful,
failed,
completed,
stop,
consecutive_failures,
latencies,
worker_counts,
failure_reasons,
} = counters;
let payload = create_test_payload(&config.workload);
let requirements = build_requirements(config);
for _ in 0..requests {
if stop.load(Ordering::Relaxed) {
return;
}
let start = Instant::now();
let mut req =
// The agent's own timeout bounds the CONNECT too, which the
// per-request `timeout_global` below does not: a request that
// stalls reaching the broker used to sit here for 310 s whatever
// --timeout said.
crate::vpn::mesh_agent_for(
&config.broker_url,
Duration::from_secs(config.timeout_secs),
)
.post(&format!("{}/execute", config.broker_url))
.config()
.timeout_global(Some(Duration::from_secs(config.timeout_secs)))
.build()
.header("Content-Type", "application/octet-stream")
.header("X-Zakuro-User", &config.user_id)
.header("X-Zakuro-Requirements", &requirements.to_string());
if let Some(ref key) = config.api_key {
req = req.header("Authorization", &format!("Bearer {}", key));
}
let result = req.send(&payload);
let elapsed_us = start.elapsed().as_micros() as u64;
match result {
Ok(resp) => {
// Tally which worker the broker routed this request to.
let worker = resp
.headers()
.get("X-Zakuro-Worker")
.and_then(|v| v.to_str().ok())
.map(|w| w.to_string())
.unwrap_or_else(|| "unknown".to_string());
if let Ok(mut wc) = worker_counts.lock() {
*wc.entry(worker).or_insert(0) += 1;
}
successful.fetch_add(1, Ordering::Relaxed);
consecutive_failures.store(0, Ordering::Relaxed);
if let Ok(mut lats) = latencies.lock() {
lats.push(elapsed_us);
}
}
Err(e) => {
// Record ALWAYS, so the summary can say why. ZAKURO_BENCH_DEBUG
// now only controls the per-request echo.
if let Ok(mut fr) = failure_reasons.lock() {
*fr.entry(e.to_string()).or_insert(0) += 1;
}
if std::env::var("ZAKURO_BENCH_DEBUG").is_ok() {
eprintln!("[bench] request error: {}", e);
}
failed.fetch_add(1, Ordering::Relaxed);
let in_a_row = consecutive_failures.fetch_add(1, Ordering::Relaxed) + 1;
if run_is_failing(
completed.load(Ordering::Relaxed) + 1,
failed.load(Ordering::Relaxed),
in_a_row,
config.fail_fast,
) {
stop.store(true, Ordering::Relaxed);
}
}
}
completed.fetch_add(1, Ordering::Relaxed);
}
}
fn create_test_payload(workload: &BenchWorkload) -> Vec<u8> {
match workload {
BenchWorkload::Add => {
// {"func": add, "args": (2, 3), "kwargs": {}} → returns 5
// def add(a, b): return a + b
// cloudpickle.dumps({"func": add, "args": (2, 3), "kwargs": {}})
hex_decode("800595cd010000000000007d94288c0466756e63948c17636c6f75647069636b6c652e636c6f75647069636b6c65948c0e5f6d616b655f66756e6374696f6e9493942868028c0d5f6275696c74696e5f747970659493948c08436f6465547970659485945294284b024b004b004b024b024b03430c97007c007c017a0000005300944e8594298c0161948c01629486948c083c737472696e673e948c03616464948c03616464944b04430b8000d80b0c88718935804c944300942929749452947d94288c0b5f5f7061636b6167655f5f944e8c085f5f6e616d655f5f948c085f5f6d61696e5f5f94754e4e4e7494529468028c125f66756e6374696f6e5f7365747374617465949394681b7d947d942868188c03616464948c0c5f5f7175616c6e616d655f5f948c03616464948c0f5f5f616e6e6f746174696f6e735f5f947d948c0e5f5f6b7764656661756c74735f5f944e8c0c5f5f64656661756c74735f5f944e8c0a5f5f6d6f64756c655f5f9468198c075f5f646f635f5f944e8c0b5f5f636c6f737572655f5f944e8c175f636c6f75647069636b6c655f7375626d6f64756c6573945d948c0b5f5f676c6f62616c735f5f947d947586948652308c0461726773944b024b0386948c066b7761726773947d94752e")
}
BenchWorkload::Fib => {
// {"func": fib, "args": (28,), "kwargs": {}} → returns 317811
// def fib(n): return n if n <= 1 else fib(n-1) + fib(n-2)
// cloudpickle.dumps({"func": fib, "args": (28,), "kwargs": {}})
hex_decode("8005952f020000000000007d94288c0466756e63948c17636c6f75647069636b6c652e636c6f75647069636b6c65948c0e5f6d616b655f66756e6374696f6e9493942868028c0d5f6275696c74696e5f747970659493948c08436f6465547970659485945294284b014b004b004b014b054b03434a97007c0064016b1a000072027c005300740100000000000000007c0064017a0a0000ab01000000000000740100000000000000007c0064027a0a0000ab010000000000007a0000005300944e4b014b0287948c036669629485948c016e9485948c083c737472696e673e948c0366696294680c4b0543298000d8070888418276d80f108808dc0b0e8871903189758b3a9c039841a00199459b0ad10b22d00422944300942929749452947d94288c0b5f5f7061636b6167655f5f944e8c085f5f6e616d655f5f948c085f5f6d61696e5f5f94754e4e4e7494529468028c125f66756e6374696f6e5f7365747374617465949394681b7d947d942868188c03666962948c0c5f5f7175616c6e616d655f5f948c03666962948c0f5f5f616e6e6f746174696f6e735f5f947d948c0e5f5f6b7764656661756c74735f5f944e8c0c5f5f64656661756c74735f5f944e8c0a5f5f6d6f64756c655f5f9468198c075f5f646f635f5f944e8c0b5f5f636c6f737572655f5f944e8c175f636c6f75647069636b6c655f7375626d6f64756c6573945d948c0b5f5f676c6f62616c735f5f947d94680c681b737586948652308c0461726773944b1c85948c066b7761726773947d94752e")
}
}
}
/// Decode a hex string to bytes at compile time.
fn hex_decode(s: &str) -> Vec<u8> {
(0..s.len())
.step_by(2)
.map(|i| u8::from_str_radix(&s[i..i + 2], 16).unwrap())
.collect()
}
/// Parse benchmark arguments and run
pub fn run_from_args(args: &[String]) {
// `zc bench mesh ...` → mesh warmup/calibration (issue #47).
if args.first().map(|a| a == "mesh").unwrap_or(false) {
super::bench_mesh::run_from_args(&args[1..]);
return;
}
let mut config = BenchConfig::default();
let mut i = 0;
while i < args.len() {
match args[i].as_str() {
"-c" | "--concurrency" => {
if i + 1 < args.len() {
config.concurrency = args[i + 1].parse().unwrap_or(10);
i += 1;
}
}
"-n" | "--requests" => {
if i + 1 < args.len() {
config.requests = args[i + 1].parse().unwrap_or(1000);
i += 1;
}
}
"-u" | "--url" => {
if i + 1 < args.len() {
config.broker_url = match super::uri::resolve(&args[i + 1]) {
Ok(u) => u,
Err(e) => {
eprintln!("Error: {}", e);
std::process::exit(1);
}
};
i += 1;
}
}
"--user" => {
if i + 1 < args.len() {
config.user_id = args[i + 1].clone();
i += 1;
}
}
"--timeout" => {
if i + 1 < args.len() {
if let Ok(secs) = args[i + 1].parse::<u64>() {
if secs > 0 {
config.timeout_secs = secs;
}
}
i += 1;
}
}
"--fail-fast" => {
if i + 1 < args.len() {
if let Ok(n) = args[i + 1].parse::<u32>() {
config.fail_fast = n;
}
i += 1;
}
}
"-s" | "--strategy" => {
if i + 1 < args.len() {
match args[i + 1].parse::<RoutingStrategy>() {
Ok(s) => config.strategy = s,
Err(e) => {
eprintln!("Error: {}", e);
eprintln!("Available strategies: best_price, best_latency, best_availability, round_robin, random, weighted_capacity");
return;
}
}
i += 1;
}
}
"-b" | "--benchmark" | "--workload" => {
if i + 1 < args.len() {
match args[i + 1].parse::<BenchWorkload>() {
Ok(w) => config.workload = w,
Err(e) => {
eprintln!("Error: {}", e);
return;
}
}
i += 1;
}
}
"--worker-type" | "--type" => {
if i + 1 < args.len() {
config.worker_type = Some(args[i + 1].clone());
i += 1;
}
}
"--remote" | "--remote-only" => {
config.remote_only = true;
}
"--budget" | "--budget-credits" => {
if i + 1 < args.len() {
match args[i + 1].parse::<f64>() {
Ok(b) if b > 0.0 => config.budget_credits = b,
_ => {
eprintln!("Error: --budget requires a positive number of credits");
return;
}
}
i += 1;
}
}
"--api-key" | "--key" => {
if i + 1 < args.len() {
config.api_key = Some(args[i + 1].clone());
i += 1;
}
}
"--compare" | "--compare-strategies" => {
config.compare_strategies = true;
}
"--list-strategies" => {
print_strategies();
return;
}
"-h" | "--help" => {
print_bench_help();
return;
}
_ => {
if args[i].starts_with("zc://") {
config.broker_url = match super::uri::resolve(&args[i]) {
Ok(u) => u,
Err(e) => {
eprintln!("Error: {}", e);
std::process::exit(1);
}
};
} else if args[i].starts_with("http://") || args[i].starts_with("https://") {
eprintln!(
"Error: plain HTTP URLs are not allowed — use zc://node-name instead."
);
std::process::exit(1);
}
}
}
i += 1;
}
// Resolve default zc://localhost if it wasn't replaced by an explicit --url
if config.broker_url.starts_with("zc://") {
config.broker_url = match super::uri::resolve(&config.broker_url) {
Ok(u) => u,
Err(e) => {
eprintln!("Error: {}", e);
std::process::exit(1);
}
};
}
if config.compare_strategies {
run_strategy_comparison(config);
} else {
let results = run_benchmark(config);
results.print_report();
}
}
/// Print available routing strategies
fn print_strategies() {
println!();
println!("{}", "Available Routing Strategies:".bold());
println!();
let strategies = [
(
"best_price",
"Route to the cheapest worker",
"Minimizes cost",
),
(
"best_latency",
"Route to the fastest responding worker",
"Minimizes response time",
),
(
"best_availability",
"Route to the most available worker",
"Best for high load",
),
(
"round_robin",
"Distribute evenly across all workers",
"Even distribution",
),
("random", "Random worker selection", "Simple load balancing"),
(
"weighted_capacity",
"Weighted by available resources",
"Capacity-aware",
),
];
for (name, desc, note) in strategies {
println!(
" {:20} - {} {}",
name.cyan(),
desc,
format!("({})", note).dimmed()
);
}
println!();
}
/// Run benchmark across all strategies for comparison
fn run_strategy_comparison(base_config: BenchConfig) {
println!();
println!(
" {}",
"╔═══════════════════════════════════════════════════════════════╗".cyan()
);
println!(
" {} {} {}",
"║".cyan(),
"Strategy Comparison Benchmark".bold().white(),
"║".cyan()
);
println!(
" {}",
"╚═══════════════════════════════════════════════════════════════╝".cyan()
);
println!();
// Check WireGuard connectivity before running any strategy
print!(" Checking WireGuard... ");
let (ts_connected, ts_ip, _) = check_wireguard_status(&base_config.broker_url);
if ts_connected {
println!(
"{} {}",
"connected".green(),
ts_ip.as_deref().unwrap_or("").dimmed()
);
} else {
println!("{}", "NOT CONNECTED".red());
println!();
println!(" {} WireGuard is not connected on this broker.", "✗".red());
println!(" Remote workers won't be discovered and cross-node billing");
println!(" won't be active. Wait for WireGuard before benchmarking.");
println!();
println!(" Run {} to check mesh status.", "zc info".cyan());
return;
}
println!();
println!(
" Running {} requests per strategy with {} concurrent workers",
base_config.requests, base_config.concurrency
);
println!();
let strategies = [
RoutingStrategy::BestPrice,
RoutingStrategy::BestLatency,
RoutingStrategy::BestAvailability,
RoutingStrategy::RoundRobin,
RoutingStrategy::Random,
RoutingStrategy::WeightedCapacity,
];
let mut results: Vec<(RoutingStrategy, BenchResults)> = Vec::new();
for strategy in strategies {
let mut config = base_config.clone();
config.strategy = strategy;
println!(" {} Testing {}...", "▶".cyan(), strategy.as_str().yellow());
let result = run_benchmark_quiet(config);
results.push((strategy, result));
// Small delay between tests
thread::sleep(Duration::from_millis(500));
}
// Print comparison table
println!();
println!(" {}", "═".repeat(75).cyan());
println!(" {} Comparison Results", "◆".cyan());
println!(" {}", "═".repeat(75).cyan());
println!();
println!(
" {:<20} {:>10} {:>10} {:>10} {:>10} {:>10}",
"Strategy".bold(),
"RPS".bold(),
"Avg (ms)".bold(),
"p95 (ms)".bold(),
"p99 (ms)".bold(),
"Success%".bold()
);
println!(" {}", "─".repeat(75));
// Find best values for highlighting
let max_rps = results.iter().map(|(_, r)| r.rps).fold(0.0, f64::max);
let min_latency = results
.iter()
.map(|(_, r)| r.avg_latency_ms())
.filter(|l| *l > 0.0)
.fold(f64::INFINITY, f64::min);
for (strategy, result) in &results {
let success_pct = pct(result.successful, result.total_requests);
let avg_lat = result.avg_latency_ms();
let p95 = result.percentile(95.0);
let p99 = result.percentile(99.0);
// Highlight best values
let rps_str = if (result.rps - max_rps).abs() < 0.01 {
format!("{:.1}", result.rps).green().bold().to_string()
} else {
format!("{:.1}", result.rps)
};
let lat_str = if (avg_lat - min_latency).abs() < 0.01 && avg_lat > 0.0 {
format!("{:.1}", avg_lat).green().bold().to_string()
} else {
format!("{:.1}", avg_lat)
};
println!(
" {:<20} {:>10} {:>10} {:>10} {:>10} {:>9.1}%",
strategy.as_str(),
rps_str,
lat_str,
format!("{:.1}", p95),
format!("{:.1}", p99),
success_pct
);
}
println!(" {}", "─".repeat(75));
println!();
// Summary
let best_rps = results
.iter()
.max_by(|a, b| a.1.rps.partial_cmp(&b.1.rps).unwrap())
.unwrap();
let best_latency = results
.iter()
.filter(|(_, r)| r.avg_latency_ms() > 0.0)
.min_by(|a, b| {
a.1.avg_latency_ms()
.partial_cmp(&b.1.avg_latency_ms())
.unwrap()
});
println!(" {}", "Recommendations:".bold());
println!(
" Best throughput: {} ({:.1} RPS)",
best_rps.0.as_str().green().bold(),
best_rps.1.rps
);
if let Some((strat, result)) = best_latency {
println!(
" Lowest latency: {} ({:.1}ms avg)",
strat.as_str().green().bold(),
result.avg_latency_ms()
);
}
println!();
}
/// Run benchmark without verbose output (for comparison mode)
fn run_benchmark_quiet(config: BenchConfig) -> BenchResults {
// Check broker health first
if crate::vpn::mesh_agent_for(&config.broker_url, std::time::Duration::from_secs(30))
.get(&format!("{}/health", config.broker_url))
.config()
.timeout_global(Some(Duration::from_secs(5)))
.build()
.call()
.is_err()
{
return BenchResults::new();
}
let successful = Arc::new(AtomicU64::new(0));
let failed = Arc::new(AtomicU64::new(0));
let completed = Arc::new(AtomicU64::new(0));
let stop = Arc::new(std::sync::atomic::AtomicBool::new(false));
let consecutive_failures = Arc::new(AtomicU64::new(0));
let latencies = Arc::new(std::sync::Mutex::new(Vec::with_capacity(
config.requests as usize,
)));
let worker_counts = Arc::new(std::sync::Mutex::new(std::collections::HashMap::new()));
let failure_reasons = Arc::new(std::sync::Mutex::new(std::collections::HashMap::new()));
let concurrency = effective_concurrency(config.concurrency);
let requests_per_worker = config.requests / concurrency as u64;
let extra_requests = config.requests % concurrency as u64;
let start = Instant::now();
let handles: Vec<_> = (0..concurrency)
.map(|i| {
let config = config.clone();
let successful = successful.clone();
let failed = failed.clone();
let completed = completed.clone();
let stop = stop.clone();
let consecutive_failures = consecutive_failures.clone();
let latencies = latencies.clone();
let worker_counts = worker_counts.clone();
let failure_reasons = failure_reasons.clone();
let my_requests = if (i as u64) < extra_requests {
requests_per_worker + 1
} else {
requests_per_worker
};
async_exec::spawn_blocking(move || {
run_worker(
i,
my_requests,
&config,
&BenchCounters {
successful: &successful,
failed: &failed,
completed: &completed,
stop: &stop,
consecutive_failures: &consecutive_failures,
latencies: &latencies,
worker_counts: &worker_counts,
failure_reasons: &failure_reasons,
},
);
})
})
.collect();
for handle in handles {
let _ = handle.join();
}
let duration = start.elapsed();
let successful_count = successful.load(Ordering::Relaxed);
let failed_count = failed.load(Ordering::Relaxed);
let latencies_vec = latencies.lock().unwrap().clone();
let worker_counts_map = worker_counts.lock().unwrap().clone();
let failure_reasons_map = failure_reasons.lock().unwrap().clone();
BenchResults {
total_requests: config.requests,
successful: successful_count,
failed: failed_count,
duration,
rps: {
let s = duration.as_secs_f64();
if s > 0.0 {
config.requests as f64 / s
} else {
0.0
}
},
latencies_us: latencies_vec,
worker_counts: worker_counts_map,
failure_reasons: failure_reasons_map,
}
}
fn print_bench_help() {
println!();
println!("{}: zc bench [OPTIONS] [URL]", "Usage".bold());
println!();
println!("Benchmark the broker's request handling performance.");
println!();
println!("{}", "Options:".bold());
println!(" -c, --concurrency <N> Number of concurrent workers (default: 10)");
println!(" -n, --requests <N> Total requests to send (default: 1000)");
println!(" -u, --url <URL> Broker URL (default: http://127.0.0.1:9000)");
println!(" --user <ID> User ID for requests (default: bench-user)");
println!(" --timeout <SECS> Per-request budget (default: 15)");
println!(" --fail-fast <N> Stop after N failures in a row (default: 10, 0 = never)");
println!(" --api-key <KEY> API key for Bearer auth (required in P2P mode)");
println!(
" --budget <CREDITS> Per-request credit ceiling [default: 10]. The broker rejects"
);
println!(
" the request when your balance is below this, so raise it only"
);
println!(" for long jobs — a bench job costs ~0.001 credits.");
println!(" -b, --benchmark <NAME> Workload to run (default: add)");
println!(" -s, --strategy <NAME> Routing strategy (default: best_price)");
println!(" --compare Compare all routing strategies");
println!(" --list-strategies List available routing strategies");
println!(" -h, --help Show this help");
println!();
println!("{}", "Workloads:".bold());
println!(" add add(2, 3) — routing overhead baseline (default)");
println!(" fib fib(28) — CPU-intensive recursive Fibonacci");
println!();
println!("{}", "Routing Strategies:".bold());
println!(" best_price Route to cheapest worker");
println!(" best_latency Route to fastest worker");
println!(" best_availability Route to most available worker");
println!(" round_robin Distribute evenly across workers");
println!(" random Random worker selection");
println!(" weighted_capacity Weighted by available resources");
println!();
println!("{}", "Examples:".bold());
println!(" zc bench # Default settings (best_price)");
println!(" zc bench -s best_latency # Use fastest worker strategy");
println!(" zc bench --compare # Compare all strategies");
println!(" zc bench -c 50 -n 10000 # 50 concurrent, 10k requests");
println!(" zc bench http://broker:9000 # Custom broker URL");
println!(" zc bench -c 100 -n 5000 -s round_robin -u http://10.13.13.2:9000");
println!();
println!("{}", "Subcommands:".bold());
println!(
" zc bench mesh --workers a,b,c Warmup/calibrate a mesh (JSON; see `bench mesh -h`)"
);
println!();
}
#[cfg(test)]
mod panic_fix_tests {
use super::{build_requirements, effective_concurrency, pct, BenchConfig, BenchResults};
/// The old hardcoded 1000.0 was ~1,000,000x a real bench job's cost, and the
/// broker rejects when balance < budget, so any account under 1000 credits
/// failed every request. The default must stay well clear of that.
#[test]
fn default_budget_is_not_a_balance_gate() {
let b = BenchConfig::default().budget_credits;
assert!(b > 0.0, "budget must bound the reservation, got {b}");
assert!(
b <= 10.0,
"default budget {b} gates accounts that can easily afford a ~0.001 credit job"
);
}
#[test]
fn budget_is_sent_in_the_requirements_payload() {
let cfg = BenchConfig {
budget_credits: 2.5,
..Default::default()
};
let r = build_requirements(&cfg);
assert_eq!(r["budget_credits"], serde_json::json!(2.5));
}
#[test]
fn default_budget_reaches_the_payload() {
// Guards the wiring, not just the field: a default that never reaches
// the request would reintroduce the original bug silently.
let cfg = BenchConfig::default();
assert_eq!(
r_budget(&build_requirements(&cfg)),
cfg.budget_credits,
"the configured budget must be what is actually sent"
);
}
fn r_budget(v: &serde_json::Value) -> f64 {
v["budget_credits"]
.as_f64()
.expect("budget_credits is a number")
}
#[test]
fn failure_reasons_rank_by_count_then_text() {
let mut r = BenchResults::new();
r.failure_reasons.insert("http status: 503".into(), 2);
r.failure_reasons.insert("http status: 402".into(), 9);
r.failure_reasons.insert("timeout".into(), 2);
let ranked = r.ranked_failure_reasons();
assert_eq!(ranked[0], ("http status: 402".to_string(), 9));
// Equal counts fall back to text order, so the output is stable across
// runs -- HashMap iteration order is not.
assert_eq!(ranked[1], ("http status: 503".to_string(), 2));
assert_eq!(ranked[2], ("timeout".to_string(), 2));
}
#[test]
fn no_failures_means_no_reasons_to_print() {
assert!(BenchResults::new().ranked_failure_reasons().is_empty());
}
#[test]
fn pct_zero_denominator_is_zero_not_nan() {
assert_eq!(pct(0, 0), 0.0);
assert_eq!(pct(5, 0), 0.0);
}
#[test]
fn pct_normal() {
assert!((pct(1, 4) - 25.0).abs() < 1e-9);
assert!((pct(3, 3) - 100.0).abs() < 1e-9);
}
#[test]
fn effective_concurrency_never_zero() {
assert_eq!(effective_concurrency(0), 1);
assert_eq!(effective_concurrency(8), 8);
}
#[test]
fn requests_per_worker_no_divide_by_zero() {
// Reproduce the `-c 0` path: dividing by effective_concurrency must not panic.
let requests: u64 = 100;
let c = effective_concurrency(0);
let per = requests / c as u64;
assert_eq!(per, 100);
}
}
#[cfg(test)]
mod ip_free_tests {
use super::wireguard_line;
/// The broker's mesh address was printed right after "connected". zc
/// never shows an IP to an end user, and a benchmark against a mesh
/// broker is exactly where one used to appear.
#[test]
fn the_connectivity_line_carries_no_address() {
for connected in [true, false] {
let line = wireguard_line(connected);
assert!(
!line.chars().any(|c| c.is_ascii_digit()),
"no address, no octets: {line:?}"
);
}
assert!(wireguard_line(true).contains("connected"));
assert!(wireguard_line(false).contains("not connected"));
}
}
#[cfg(test)]
mod bench_run_tests {
use super::{run_is_failing, BenchConfig};
/// `zc bench` sent no Authorization header unless ZAKURO_API_KEY happened
/// to be exported, so a normal shell benchmarked as `anonymous` with no
/// credits: every request that routed to a billed worker stalled until it
/// timed out. `zc login` already wrote the key to ~/.zakuro/credentials.
#[test]
fn a_run_uses_the_key_zc_login_saved() {
let _env = crate::credentials::HOME_ENV_LOCK
.lock()
.unwrap_or_else(|e| e.into_inner());
let tmp = std::env::temp_dir().join(format!("zc-bench-auth-{}", std::process::id()));
std::fs::create_dir_all(&tmp).unwrap();
std::fs::write(tmp.join("credentials"), "api_key=zk_7_fromfile\n").unwrap();
let old_zh = std::env::var_os("ZAKURO_HOME");
let old_key = std::env::var_os("ZAKURO_API_KEY");
std::env::set_var("ZAKURO_HOME", &tmp);
std::env::remove_var("ZAKURO_API_KEY");
let from_file = BenchConfig::default().api_key;
// An exported key still wins: one-off runs override the saved one.
std::env::set_var("ZAKURO_API_KEY", "zk_7_fromenv");
let from_env = BenchConfig::default().api_key;
match old_zh {
Some(v) => std::env::set_var("ZAKURO_HOME", v),
None => std::env::remove_var("ZAKURO_HOME"),
}
match old_key {
Some(v) => std::env::set_var("ZAKURO_API_KEY", v),
None => std::env::remove_var("ZAKURO_API_KEY"),
}
std::fs::remove_dir_all(&tmp).ok();
assert_eq!(from_file.as_deref(), Some("zk_7_fromfile"));
assert_eq!(from_env.as_deref(), Some("zk_7_fromenv"));
}
/// A run where everything times out: the streak trips it immediately,
/// long before the sample size the rate check needs.
#[test]
fn a_run_that_only_fails_is_stopped_at_once() {
assert!(!run_is_failing(2, 2, 2, 3), "two failures is not a pattern");
assert!(run_is_failing(3, 3, 3, 3), "the third in a row stops it");
}
/// The shape a broken mesh route actually produced: about 15% succeed and
/// the rest time out, so the successes keep resetting the streak. With a
/// consecutive-failure counter alone, a 60-request run went to the end.
#[test]
fn a_run_that_mostly_fails_is_stopped_even_though_some_succeed() {
assert!(!run_is_failing(19, 16, 3, 10), "too early to judge");
assert!(
run_is_failing(20, 17, 3, 10),
"17 of 20 failing is not a measurement"
);
assert!(!run_is_failing(1000, 12, 2, 10), "12 of 1000 is fine");
assert!(run_is_failing(40, 20, 1, 10), "half of them gone");
}
#[test]
fn the_guard_can_be_turned_off() {
assert!(!run_is_failing(1000, 1000, 1000, 0), "0 disables the guard");
}
}
#[cfg(test)]
mod live_view_tests {
use super::{fmt_us, percentiles_us, short_worker};
/// The live line is read at a glance; raw microseconds are not.
#[test]
fn a_duration_reads_in_the_unit_that_suits_it() {
assert_eq!(fmt_us(0), "—", "nothing measured yet says so");
assert_eq!(fmt_us(820), "820µs");
assert_eq!(fmt_us(4_200), "4.2ms");
assert_eq!(fmt_us(1_350_000), "1.35s");
}
#[test]
fn percentiles_come_from_the_samples_so_far() {
assert_eq!(percentiles_us(&[]), (0, 0), "no samples, no percentiles");
// Nearest-rank on the sorted samples: index len*p/100, clamped.
let one_to_hundred: Vec<u64> = (1..=100).collect();
let (p50, p99) = percentiles_us(&one_to_hundred);
assert_eq!((p50, p99), (51, 100));
// Order of arrival must not matter.
let mut shuffled = one_to_hundred.clone();
shuffled.reverse();
assert_eq!(percentiles_us(&shuffled), (p50, p99));
}
/// The live line and the summary quote the same number for the same
/// samples: a reader comparing them must not be told two different things.
#[test]
fn the_live_line_and_the_summary_agree() {
let samples: Vec<u64> = vec![900, 1_200, 4_000, 4_200, 90_000];
let mut result = super::BenchResults::new();
result.latencies_us = samples.clone();
let (p50, p99) = percentiles_us(&samples);
assert_eq!(p50 as f64 / 1000.0, result.percentile(50.0));
assert_eq!(p99 as f64 / 1000.0, result.percentile(99.0));
}
#[test]
fn a_worker_name_is_shortened_to_fit_the_line() {
assert_eq!(short_worker("worker-zc-worker-i9-1"), "i9-1");
assert_eq!(short_worker("dd715b7f-w0"), "dd715b7f-w0");
let long = short_worker("worker-zc-worker-cloudcompute-with-a-long-suffix");
assert!(long.chars().count() <= 22, "{long:?}");
assert!(long.ends_with('…'));
}
}
#[cfg(test)]
mod live_block_tests {
use super::{bar, by_node, histogram, live_block, node_of, LATENCY_BUCKETS_MS};
use std::collections::HashMap;
use std::time::Duration;
#[test]
fn a_worker_is_attributed_to_its_node() {
assert_eq!(node_of("worker-zc-worker-i9-1"), "i9");
assert_eq!(node_of("worker-zc-worker-cloudcompute-2"), "cloudcompute");
// This machine's own workers share one broker fingerprint.
assert_eq!(node_of("dd715b7f-w0"), "dd715b7f");
assert_eq!(node_of("dd715b7f-w12"), "dd715b7f");
// Anything else stands for itself rather than being mangled.
assert_eq!(node_of("zc-broker-0"), "zc-broker-0");
}
#[test]
fn work_is_totalled_per_node_busiest_first() {
let counts = HashMap::from([
("dd715b7f-w0".to_string(), 7),
("dd715b7f-w1".to_string(), 9),
("worker-zc-worker-i9-1".to_string(), 8),
]);
assert_eq!(
by_node(&counts),
vec![("dd715b7f".to_string(), 16), ("i9".to_string(), 8)]
);
}
/// The live histogram and the final one count the same samples the same
/// way: a reader watching the run then reading the summary must not be
/// told two different things.
#[test]
fn the_histogram_buckets_every_sample_once() {
let samples = vec![500, 4_000, 9_500, 24_000, 700_000, 2_000_000];
let counts = histogram(&samples);
assert_eq!(counts.iter().sum::<u64>(), samples.len() as u64);
assert_eq!(counts[0], 1, "500µs is the ≤1ms bucket");
assert_eq!(counts[2], 1, "9.5ms is the ≤10ms bucket");
assert_eq!(
counts[LATENCY_BUCKETS_MS.len() - 1],
1,
"2s is the >1000ms bucket"
);
}
#[test]
fn a_bar_is_proportional_and_never_overflows_its_width() {
assert_eq!(bar(0, 10, 4), " ");
assert_eq!(bar(10, 10, 4), "████");
assert_eq!(bar(5, 10, 4), "██ ");
assert_eq!(bar(7, 0, 4), " ", "no maximum yet, nothing to scale to");
assert_eq!(bar(99, 10, 4).chars().count(), 4, "clamped to the width");
}
/// What a reader must be able to see at a glance, while it runs.
#[test]
fn the_live_block_shows_rate_spread_and_where_the_work_went() {
let samples = vec![3_000, 4_000, 800_000];
let workers = HashMap::from([
("dd715b7f-w0".to_string(), 2),
("worker-zc-worker-i9-1".to_string(), 1),
]);
let block = live_block(
'⠙',
3,
100,
2,
1,
Duration::from_secs(1),
&samples,
&workers,
);
let plain: String = block
.chars()
.filter(|c| *c != '\u{1b}')
.collect::<String>()
.replace(['[', ']'], "");
assert!(plain.contains("3/100"), "{block}");
assert!(plain.contains("req/s"), "{block}");
assert!(plain.contains("p50"), "{block}");
assert!(
plain.contains("≤5ms"),
"the buckets are visible live: {block}"
);
assert!(plain.contains("dd715b7f"), "where the work ran: {block}");
assert!(plain.contains("i9"), "{block}");
}
#[test]
fn the_live_block_says_so_before_anything_has_been_routed() {
let block = live_block(
'⠙',
0,
10,
0,
0,
Duration::from_millis(1),
&[],
&HashMap::new(),
);
assert!(block.contains("nothing has been routed yet"), "{block}");
}
}
#[cfg(test)]
mod report_layout_tests {
use super::{boxed, pad, visible_width, worker_leaf, BenchResults};
use colored::Colorize;
use std::collections::HashMap;
use std::time::Duration;
fn plain(s: &str) -> String {
let mut out = String::new();
let mut chars = s.chars();
while let Some(c) = chars.next() {
if c == '\u{1b}' {
for e in chars.by_ref() {
if e == 'm' {
break;
}
}
} else {
out.push(c);
}
}
out
}
#[test]
fn width_is_what_the_eye_sees_not_what_is_stored() {
assert_eq!(visible_width("abc"), 3);
assert_eq!(visible_width(&"abc".red().to_string()), 3, "colour is free");
assert_eq!(visible_width("≤1000ms"), 7, "multi-byte counts once");
}
#[test]
fn padding_lines_up_coloured_and_plain_alike() {
assert_eq!(plain(&pad(&"ok".green().to_string(), 5, false)), "ok ");
assert_eq!(plain(&pad("7", 4, true)), " 7");
assert_eq!(pad("toolong", 3, false), "toolong", "never truncates");
}
/// The old header drew a 43-column border around 41 columns of text.
#[test]
fn a_box_is_as_wide_as_its_widest_line() {
let out = boxed(&["short".to_string(), "a much longer line".to_string()]);
let widths: Vec<usize> = out.lines().map(|l| visible_width(l)).collect();
assert!(
widths.windows(2).all(|w| w[0] == w[1]),
"every border and row is the same width: {widths:?}\n{out}"
);
}
#[test]
fn a_worker_is_labelled_by_what_distinguishes_it() {
assert_eq!(worker_leaf("dd715b7f-w2"), "w2");
assert_eq!(worker_leaf("worker-zc-worker-i9-1"), "1");
}
/// The bug in the shipped report: `worker-zc-worker-cloudcompute-2` is far
/// wider than the column it was given, so its bar started further right
/// than everyone else's and the table stopped lining up.
#[test]
fn a_long_worker_name_cannot_push_the_bars_out_of_line() {
let mut r = BenchResults::new();
r.total_requests = 30;
r.successful = 30;
r.duration = Duration::from_secs(3);
r.rps = 10.0;
r.latencies_us = vec![3_000, 4_000, 900_000];
r.worker_counts = HashMap::from([
("dd715b7f-w0".to_string(), 10),
("worker-zc-worker-cloudcompute-2".to_string(), 12),
("worker-zc-worker-i9-1".to_string(), 8),
]);
let report = plain(&r.render_report());
// Bars inside one section start in one column. (Sections have their
// own label widths, so this is per-section by design.)
let where_it_ran = report.split("where it ran").nth(1).expect("a node section");
let bar_starts: Vec<usize> = where_it_ran
.lines()
.filter_map(|l| l.chars().position(|c| c == '█'))
.collect();
assert!(bar_starts.len() >= 3, "a bar per node: {report}");
assert!(
bar_starts.windows(2).all(|w| w[0] == w[1]),
"every node's bar starts in the same column: {bar_starts:?}\n{report}"
);
// Grouped by node, with the busiest first.
let cloud = report.find("cloudcompute").expect("a node line");
let i9 = report.find("\n i9").expect("the other node");
assert!(cloud < i9, "busiest node first:\n{report}");
}
}