//! 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)
}
/// 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,
/// 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,
timeout_secs: 30,
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).
api_key: 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 {
if self.latencies_us.is_empty() {
return 0.0;
}
let mut sorted = self.latencies_us.clone();
sorted.sort();
let idx = ((sorted.len() as f64 * p / 100.0) as usize).min(sorted.len() - 1);
sorted[idx] as f64 / 1000.0 // Convert to ms
}
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) {
println!();
println!(" {}", "═".repeat(60).cyan());
println!(" {} Benchmark Results", "◆".cyan());
println!(" {}", "═".repeat(60).cyan());
println!();
// Summary
println!(" {}", "Summary".bold());
println!(" {}", "─".repeat(40));
println!(" Total Requests: {}", self.total_requests);
println!(
" Successful: {} {}",
self.successful.to_string().green(),
format!("({:.1}%)", pct(self.successful, self.total_requests)).dimmed()
);
println!(
" Failed: {} {}",
if self.failed > 0 {
self.failed.to_string().red()
} else {
"0".to_string().green()
},
format!("({:.1}%)", pct(self.failed, self.total_requests)).dimmed()
);
// Print WHY, right under the count. Without this a total failure looks
// identical whether the account is out of credits or no worker was
// reachable, and the reader reasonably guesses the latter.
let ranked = self.ranked_failure_reasons();
for (reason, count) in ranked.iter().take(5) {
println!(" {} {}", format!("{}x", count).red(), reason.dimmed());
}
if ranked.len() > 5 {
println!(
" {}",
format!("... and {} more distinct errors", ranked.len() - 5).dimmed()
);
}
println!(" Duration: {:.2}s", self.duration.as_secs_f64());
println!(
" Throughput: {} req/s",
format!("{:.2}", self.rps).yellow().bold()
);
println!();
// Latency
println!(" {}", "Latency (ms)".bold());
println!(" {}", "─".repeat(40));
println!(" Min: {:.2}", self.min_latency_ms());
println!(" Avg: {:.2}", self.avg_latency_ms());
println!(" Max: {:.2}", self.max_latency_ms());
println!(" p50: {:.2}", self.percentile(50.0));
println!(" p75: {:.2}", self.percentile(75.0));
println!(" p90: {:.2}", self.percentile(90.0));
println!(" p95: {:.2}", self.percentile(95.0));
println!(" p99: {:.2}", self.percentile(99.0));
println!();
// Histograms
self.print_histogram();
self.print_worker_histogram();
}
/// Per-worker executed-request distribution (which nodes the broker routed to).
fn print_worker_histogram(&self) {
if self.worker_counts.is_empty() {
return;
}
println!(" {}", "Worker Distribution".bold());
println!(" {}", "─".repeat(40));
// Sort by count (desc), then name for stable ties.
let mut entries: Vec<(&String, &u64)> = self.worker_counts.iter().collect();
entries.sort_by(|a, b| b.1.cmp(a.1).then(a.0.cmp(b.0)));
let max = entries.iter().map(|(_, c)| **c).max().unwrap_or(0);
let total: u64 = self.worker_counts.values().sum();
let bar_width = 30usize;
for (name, count) in entries {
let c = *count;
let bar_len = if max > 0 {
(c as f64 / max as f64 * bar_width as f64).round() as usize
} else {
0
};
let bar = "█".repeat(bar_len);
let pct = if total > 0 {
c as f64 / total as f64 * 100.0
} else {
0.0
};
println!(
" {:>14} [{:<30}] {:>5} ({:>5.1}%)",
name,
bar.green(),
c,
pct
);
}
println!();
}
fn print_histogram(&self) {
if self.latencies_us.is_empty() {
return;
}
println!(" {}", "Latency Histogram".bold());
println!(" {}", "─".repeat(40));
let mut sorted = self.latencies_us.clone();
sorted.sort();
// Define buckets (in ms)
let buckets = [
1.0,
5.0,
10.0,
25.0,
50.0,
100.0,
250.0,
500.0,
1000.0,
f64::INFINITY,
];
let mut counts = vec![0u64; buckets.len()];
for lat_us in &sorted {
let lat_ms = *lat_us as f64 / 1000.0;
for (i, &bucket) in buckets.iter().enumerate() {
if lat_ms <= bucket {
counts[i] += 1;
break;
}
}
}
let max_count = *counts.iter().max().unwrap_or(&1);
let bar_width = 30;
for (i, &bucket) in buckets.iter().enumerate() {
let label = if bucket == f64::INFINITY {
">1000ms".to_string()
} else {
format!("≤{:.0}ms", bucket)
};
let count = counts[i];
let bar_len = if max_count > 0 {
(count as f64 / max_count as f64 * bar_width as f64) as usize
} else {
0
};
let bar: String = "█".repeat(bar_len);
let pct = count as f64 / self.latencies_us.len() as f64 * 100.0;
println!(
" {:>8} [{:<30}] {:>5} ({:>5.1}%)",
label,
bar.cyan(),
count,
pct
);
}
println!();
}
}
/// 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(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!();
println!(
" {}",
"╔═══════════════════════════════════════════╗".cyan()
);
println!(
" {} {} {}",
"║".cyan(),
"Broker Benchmark".bold().white(),
"║".cyan()
);
println!(
" {}",
"╚═══════════════════════════════════════════╝".cyan()
);
println!();
println!(" Target: {}", config.broker_url.cyan());
println!(" Concurrency: {}", config.concurrency);
println!(" Requests: {}", config.requests);
println!(
" Workload: {} {}",
config.workload.name().yellow().bold(),
format!("({})", config.workload.description()).dimmed()
);
println!(
" Strategy: {} {}",
config.strategy.as_str().yellow().bold(),
format!("({})", 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(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)
let url_trimmed = config.broker_url.trim();
let is_localhost =
url_trimmed.contains("127.0.0.1") || url_trimmed.to_lowercase().contains("localhost");
if is_localhost {
println!(" {} (skipping WireGuard check)", "Local broker".dimmed());
} else {
print!(" Checking WireGuard... ");
let (ts_connected, ts_ip, _) = check_wireguard_status(&config.broker_url);
if ts_connected {
println!(
"{} {}",
"connected".green(),
ts_ip.as_deref().unwrap_or("").dimmed()
);
} 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!("{}", "not connected (continuing)".yellow());
}
}
// Check if there are workers
print!(" Checking workers... ");
match crate::vpn::mesh_agent(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 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 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,
latencies: &latencies,
worker_counts: &worker_counts,
failure_reasons: &failure_reasons,
},
);
})
})
.collect();
// Progress reporter
let total = config.requests;
let completed_for_progress = completed.clone();
let progress_handle = async_exec::spawn_blocking(move || {
let spinner = ['⠋', '⠙', '⠹', '⠸', '⠼', '⠴', '⠦', '⠧', '⠇', '⠏'];
let mut i = 0;
loop {
let done = completed_for_progress.load(Ordering::Relaxed);
let pct = done as f64 / total as f64 * 100.0;
print!(
"\r {} Progress: {:>6}/{} ({:>5.1}%) ",
spinner[i % spinner.len()].to_string().cyan(),
done,
total,
pct
);
let _ = std::io::Write::flush(&mut std::io::stdout());
if done >= total {
break;
}
thread::sleep(Duration::from_millis(100));
i += 1;
}
println!();
});
// 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,
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,
latencies,
worker_counts,
failure_reasons,
} = counters;
let payload = create_test_payload(&config.workload);
let requirements = build_requirements(config);
for _ in 0..requests {
let start = Instant::now();
let mut req = crate::vpn::mesh_agent(std::time::Duration::from_secs(310))
.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);
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);
}
}
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;
}
}
"-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(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 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 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,
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!(" --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);
}
}