diff --git a/.gitignore b/.gitignore index ca1f766..6ded5d1 100644 --- a/.gitignore +++ b/.gitignore @@ -247,3 +247,5 @@ results/*/callgrind-cache.out targets/cjson/candidate-*/ targets/cjson/cand-*/ targets/cjson/asan*/ +targets/cjson/svc-baseline/ +targets/cjson/svc-candidate/ diff --git a/Cargo.lock b/Cargo.lock index 2b820fd..0658d93 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -77,6 +77,7 @@ dependencies = [ "anyhow", "clap", "ledger", + "libc", "serde", "serde_json", "toml", @@ -373,6 +374,7 @@ name = "report" version = "0.1.0" dependencies = [ "anyhow", + "clap", "ledger", "serde", "serde_json", diff --git a/crates/bench-runner/Cargo.toml b/crates/bench-runner/Cargo.toml index 89fec89..2281413 100644 --- a/crates/bench-runner/Cargo.toml +++ b/crates/bench-runner/Cargo.toml @@ -10,6 +10,7 @@ repository.workspace = true anyhow.workspace = true clap.workspace = true ledger = { path = "../ledger" } +libc = "0.2" serde.workspace = true serde_json.workspace = true toml.workspace = true diff --git a/crates/bench-runner/src/lib.rs b/crates/bench-runner/src/lib.rs index bb789a4..55cfbf1 100644 --- a/crates/bench-runner/src/lib.rs +++ b/crates/bench-runner/src/lib.rs @@ -9,6 +9,7 @@ pub mod config; pub mod exec; pub mod fingerprint; pub mod schedule; +pub mod service; pub mod stats; use ledger::Verdict; diff --git a/crates/bench-runner/src/main.rs b/crates/bench-runner/src/main.rs index e79f95d..1c56ee2 100644 --- a/crates/bench-runner/src/main.rs +++ b/crates/bench-runner/src/main.rs @@ -47,6 +47,297 @@ enum Cmd { #[arg(long)] out: PathBuf, }, + /// Service mode (SPEC §3.1 mode c): interleaved A/B of two HTTP + /// server binaries under a coordinated-omission-correct open-loop + /// load. Reports p50 and p99 latency speedup (baseline/candidate) + /// with bootstrap CIs. `--aa` runs the same binary both sides. + Service { + #[arg(long)] + baseline_bin: String, + #[arg(long)] + candidate_bin: String, + #[arg(long)] + doc: String, + #[arg(long, default_value_t = 1)] + iters: u64, + /// Open-loop arrival rate (requests/sec). + #[arg(long, default_value_t = 200.0)] + rate: f64, + /// Timed requests per session. + #[arg(long, default_value_t = 1000)] + count: usize, + /// Interleaved sessions per side. + #[arg(long, default_value_t = 12)] + sessions: usize, + #[arg(long, default_value_t = 200)] + warmup: usize, + #[arg(long, default_value_t = 32)] + workers: usize, + /// Prefix pinning the server to a core (e.g. "taskset -c 2"). + #[arg(long, default_value = "")] + pin: String, + /// Same binary both sides — must yield a null verdict on p99. + #[arg(long, default_value_t = false)] + aa: bool, + /// Optional JSON output path (latency percentiles + CIs). + #[arg(long)] + out: Option, + }, + /// Service-mode calibration: A/A false-positive rate + injected + /// latency-regression detection rate on p50 (SPEC §3.1). + ServiceCalibrate { + #[arg(long)] + server_bin: String, + #[arg(long)] + doc: String, + #[arg(long, default_value_t = 1)] + iters: u64, + #[arg(long, default_value_t = 150.0)] + rate: f64, + #[arg(long, default_value_t = 500)] + count: usize, + /// Interleaved rounds per compare. + #[arg(long, default_value_t = 6)] + rounds: usize, + /// Calibration compares (A/A + injection each). + #[arg(long, default_value_t = 10)] + sessions: usize, + #[arg(long, default_value_t = 100)] + warmup: usize, + #[arg(long, default_value_t = 32)] + workers: usize, + #[arg(long, default_value = "")] + pin: String, + #[arg(long, default_value_t = 0.05)] + slowdown: f64, + #[arg(long)] + out: PathBuf, + }, +} + +#[allow(clippy::too_many_arguments)] +fn service_bench( + cfg: &AcceptConfig, + baseline_bin: &str, + candidate_bin: &str, + doc: &str, + iters: u64, + rate: f64, + count: usize, + sessions: usize, + warmup: usize, + workers: usize, + pin: &str, + aa: bool, + out: &Option, +) -> Result<()> { + use bench_runner::service::{run_compare, ServiceCfg}; + + let mk = |bin: &str| ServiceCfg { + server_bin: bin.to_string(), + doc: doc.to_string(), + iters, + rate, + count, + warmup, + workers, + pin_prefix: pin.to_string(), + inject_us: None, + }; + let base_cfg = mk(baseline_bin); + let cand_cfg = mk(candidate_bin); + + eprintln!( + "service: {sessions} interleaved sessions/side, {count} req @ {rate} rps, \ + CO-correct open loop, server pin='{pin}'" + ); + + let r = run_compare( + &base_cfg, + &cand_cfg, + sessions, + cfg.bootstrap_iters, + cfg.confidence, + cfg.bootstrap_seed, + |k, bp50, bp99, cp50, cp99| { + eprintln!( + " round {}/{sessions}: base p50={:.3}ms p99={:.3}ms | cand p50={:.3}ms p99={:.3}ms", + k + 1, + bp50 * 1e3, + bp99 * 1e3, + cp50 * 1e3, + cp99 * 1e3 + ); + }, + )?; + let (ci50, ci99, drop_rate) = (r.ci50, r.ci99, r.drop_rate); + let (b_p50, c_p50, b_p99, c_p99) = (&r.base_p50, &r.cand_p50, &r.base_p99, &r.cand_p99); + + // Drop rate must be negligible or the load was above capacity and the + // percentiles are unstable — refuse to report a number we don't trust. + let total_req = sessions * count * 2; + let total_dropped = (drop_rate * total_req as f64).round() as usize; + anyhow::ensure!( + drop_rate < 0.005, + "drop rate {:.2}% exceeds 0.5% — load is above server capacity; lower --rate", + drop_rate * 100.0 + ); + + println!( + "p50 latency speedup (baseline/candidate): median {:.4}, {:.0}% CI [{:.4}, {:.4}]", + ci50.median, + cfg.confidence * 100.0, + ci50.lo, + ci50.hi + ); + println!( + "p99 latency speedup (baseline/candidate): median {:.4}, {:.0}% CI [{:.4}, {:.4}]", + ci99.median, + cfg.confidence * 100.0, + ci99.lo, + ci99.hi + ); + println!( + "drop rate: {:.3}% ({total_dropped}/{total_req})", + drop_rate * 100.0 + ); + + if let Some(path) = out { + let ev = serde_json::json!({ + "mode": "service-latency", + "workload": format!("{doc}, {iters} parse+print/req, {rate} rps open-loop, CO-correct"), + "sessions": sessions, "count_per_session": count, "rate_rps": rate, + "p50_speedup_median": ci50.median, "p50_speedup_ci": [ci50.lo, ci50.hi], + "p99_speedup_median": ci99.median, "p99_speedup_ci": [ci99.lo, ci99.hi], + "baseline_p50_ms_median": stats::median(b_p50) * 1e3, + "candidate_p50_ms_median": stats::median(c_p50) * 1e3, + "baseline_p99_ms_median": stats::median(b_p99) * 1e3, + "candidate_p99_ms_median": stats::median(c_p99) * 1e3, + "drop_rate": drop_rate, + }); + std::fs::write(path, serde_json::to_string_pretty(&ev)?)?; + } + + if aa { + // Null verdict required on BOTH percentiles. + for (label, ci) in [("p50", ci50), ("p99", ci99)] { + if decide(ci, cfg.threshold) == Verdict::Accepted { + anyhow::bail!( + "service A/A FAILED on {label}: speedup claimed from identical servers" + ); + } + } + println!("service A/A self-test passed (null verdict on p50 and p99)"); + } else { + println!("p50 verdict: {}", decide(ci50, cfg.threshold).as_str()); + println!("p99 verdict: {}", decide(ci99, cfg.threshold).as_str()); + } + Ok(()) +} + +/// Service-mode calibration (SPEC §3.1): N A/A compares (false-positive +/// rate on p50 must be <5%) + N injection compares (a per-request +/// busy-wait ≈ `slowdown` of measured service time in the candidate; the +/// harness must resolve the regression — p50 CI entirely < 1 — ≥95%). +#[allow(clippy::too_many_arguments)] +fn service_calibrate( + cfg: &AcceptConfig, + server_bin: &str, + doc: &str, + iters: u64, + rate: f64, + count: usize, + rounds: usize, + n: usize, + warmup: usize, + workers: usize, + pin: &str, + slowdown: f64, + out: &PathBuf, +) -> Result<()> { + use bench_runner::service::{run_compare, run_session, ServiceCfg}; + + let mk = |inject: Option| ServiceCfg { + server_bin: server_bin.to_string(), + doc: doc.to_string(), + iters, + rate, + count, + warmup, + workers, + pin_prefix: pin.to_string(), + inject_us: inject, + }; + + // Size the injected busy-wait from a probe of the service's p50. + let probe = run_session(&mk(None), bench_runner::service::free_port()?)?; + let p50 = bench_runner::service::percentile(&probe.latencies, 0.50); + let inject_us = (p50 * slowdown * 1e6).round() as u64; + eprintln!( + "service-calibrate: probe p50={:.3}ms, injecting {inject_us}µs (~{:.0}%) for detection", + p50 * 1e3, + slowdown * 100.0 + ); + + let (mut fp, mut det) = (0usize, 0usize); + let (mut aa_rows, mut inj_rows) = (Vec::new(), Vec::new()); + for i in 0..n { + let seed = cfg.bootstrap_seed.wrapping_add(i as u64); + let aa = run_compare( + &mk(None), + &mk(None), + rounds, + cfg.bootstrap_iters, + cfg.confidence, + seed, + |_, _, _, _, _| {}, + )?; + if decide(aa.ci50, cfg.threshold) == Verdict::Accepted { + fp += 1; + } + let inj = run_compare( + &mk(None), + &mk(Some(inject_us)), + rounds, + cfg.bootstrap_iters, + cfg.confidence, + seed, + |_, _, _, _, _| {}, + )?; + if inj.ci50.hi < 1.0 { + det += 1; + } + aa_rows.push(serde_json::json!({"lo": aa.ci50.lo, "hi": aa.ci50.hi})); + inj_rows.push(serde_json::json!({"lo": inj.ci50.lo, "hi": inj.ci50.hi})); + eprintln!( + "session {}/{n}: A/A p50 [{:.4},{:.4}] fp={fp}; inj p50 [{:.4},{:.4}] det={det}", + i + 1, + aa.ci50.lo, + aa.ci50.hi, + inj.ci50.lo, + inj.ci50.hi + ); + } + let fp_rate = fp as f64 / n as f64; + let det_rate = det as f64 / n as f64; + let pass = fp_rate < 0.05 && det_rate >= 0.95; + let ev = serde_json::json!({ + "mode": "service-latency-calibration", + "server_bin": server_bin, "workload": doc, "rate_rps": rate, + "count_per_session": count, "rounds_per_compare": rounds, "sessions": n, + "injected_slowdown": slowdown, "injected_us": inject_us, + "aa_false_positive_rate": fp_rate, "aa_sessions": aa_rows, + "injection_detection_rate": det_rate, "injection_sessions": inj_rows, + "acceptance": {"fp_lt": 0.05, "detection_ge": 0.95, "pass": pass}, + }); + std::fs::write(out, serde_json::to_string_pretty(&ev)?)?; + println!( + "service calibration: A/A false-positive {fp_rate:.3} (<0.05), \ + injection detection {det_rate:.3} (>=0.95) -> {}", + if pass { "PASS" } else { "FAIL" } + ); + anyhow::ensure!(pass, "service calibration acceptance criteria not met"); + Ok(()) } fn calibrate( @@ -152,6 +443,56 @@ fn main() -> Result<()> { } => { return calibrate(&cfg, cmd, *sessions, *slowdown, out); } + Cmd::Service { + baseline_bin, + candidate_bin, + doc, + iters, + rate, + count, + sessions, + warmup, + workers, + pin, + aa, + out, + } => { + #[allow(clippy::used_underscore_items)] + return service_bench( + &cfg, + baseline_bin, + candidate_bin, + doc, + *iters, + *rate, + *count, + *sessions, + *warmup, + *workers, + pin, + *aa, + out, + ); + } + Cmd::ServiceCalibrate { + server_bin, + doc, + iters, + rate, + count, + rounds, + sessions, + warmup, + workers, + pin, + slowdown, + out, + } => { + return service_calibrate( + &cfg, server_bin, doc, *iters, *rate, *count, *rounds, *sessions, *warmup, + *workers, pin, *slowdown, out, + ); + } }; let fp = EnvFingerprint::collect("none", "system-default"); diff --git a/crates/bench-runner/src/service.rs b/crates/bench-runner/src/service.rs new file mode 100644 index 0000000..cae2e57 --- /dev/null +++ b/crates/bench-runner/src/service.rs @@ -0,0 +1,248 @@ +//! Service-mode latency bench (SPEC §3.1 mode (c)). A coordinated- +//! omission-CORRECT, open-loop, fixed-rate load generator: requests are +//! scheduled at `start + i/rate` and each latency is measured from its +//! *intended* send time, not its actual send time. So when the server +//! falls behind, the requests that pile up behind it each carry the full +//! queueing delay a real client would see — the exact error (coordinated +//! omission) that closed-loop tools like `ab` hide. We store every +//! latency and take exact percentiles (no HdrHistogram bucketing error; +//! the sample is bounded). +//! +//! Interleaved A/B, like the wall-time bench: each session yields one +//! percentile (p50/p99) per side; the existing bootstrap ratio-CI machine +//! then works unchanged, treating "p99 latency" the way it treats "wall +//! time" — lower is better, speedup = baseline / candidate. + +use anyhow::{anyhow, ensure, Result}; +use std::io::{BufRead, BufReader, Read, Write}; +use std::net::{Ipv4Addr, TcpStream}; +use std::process::{Child, Command, Stdio}; +use std::sync::atomic::{AtomicUsize, Ordering}; +use std::sync::{Arc, Mutex}; +use std::time::{Duration, Instant}; + +use crate::stats::{self, RatioCi}; + +pub struct ServiceCfg { + pub server_bin: String, + pub doc: String, + pub iters: u64, + pub rate: f64, + pub count: usize, + pub warmup: usize, + pub workers: usize, + /// Prefix that pins the SERVER to a core (e.g. `taskset -c 2`); the + /// load generator runs on the other cores. + pub pin_prefix: String, + /// Calibration only: inject this many µs of busy-wait per request + /// (via HOTPATH_INJECT_US) to validate the harness detects a known + /// latency regression. None in every real run. + pub inject_us: Option, +} + +/// Result of one interleaved A/B service compare. +pub struct CompareResult { + pub ci50: RatioCi, + pub ci99: RatioCi, + pub base_p50: Vec, + pub cand_p50: Vec, + pub base_p99: Vec, + pub cand_p99: Vec, + pub drop_rate: f64, +} + +/// Interleaved A/B: `rounds` alternating baseline/candidate sessions, +/// each yielding one p50 and one p99; bootstrap ratio-CI over the rounds. +#[allow(clippy::too_many_arguments)] +pub fn run_compare( + base: &ServiceCfg, + cand: &ServiceCfg, + rounds: usize, + bootstrap_iters: usize, + confidence: f64, + seed: u64, + mut on_round: impl FnMut(usize, f64, f64, f64, f64), +) -> Result { + let (mut b50, mut b99, mut c50, mut c99) = (Vec::new(), Vec::new(), Vec::new(), Vec::new()); + let mut dropped = 0usize; + for k in 0..rounds { + let bs = run_session(base, free_port()?)?; + let cs = run_session(cand, free_port()?)?; + dropped += bs.dropped + cs.dropped; + let (bp50, bp99) = ( + percentile(&bs.latencies, 0.50), + percentile(&bs.latencies, 0.99), + ); + let (cp50, cp99) = ( + percentile(&cs.latencies, 0.50), + percentile(&cs.latencies, 0.99), + ); + b50.push(bp50); + b99.push(bp99); + c50.push(cp50); + c99.push(cp99); + on_round(k, bp50, bp99, cp50, cp99); + } + let total = rounds * base.count * 2; + let drop_rate = dropped as f64 / total as f64; + let ci50 = stats::bootstrap_ratio_ci(&b50, &c50, bootstrap_iters, confidence, seed); + let ci99 = stats::bootstrap_ratio_ci(&b99, &c99, bootstrap_iters, confidence, seed); + Ok(CompareResult { + ci50, + ci99, + base_p50: b50, + cand_p50: c50, + base_p99: b99, + cand_p99: c99, + drop_rate, + }) +} + +pub struct SessionResult { + pub latencies: Vec, + pub dropped: usize, +} + +/// Exact percentile (0.0–1.0) of a sorted-in-place copy. +pub fn percentile(latencies: &[f64], q: f64) -> f64 { + if latencies.is_empty() { + return f64::NAN; + } + let mut xs = latencies.to_vec(); + xs.sort_by(|a, b| a.partial_cmp(b).unwrap()); + let idx = ((xs.len() as f64 - 1.0) * q).round() as usize; + xs[idx.min(xs.len() - 1)] +} + +fn spawn_server(cfg: &ServiceCfg, port: u16) -> Result { + let inject = match cfg.inject_us { + Some(us) => format!("HOTPATH_INJECT_US={us} "), + None => String::new(), + }; + let cmd = format!( + "{inject}{pin} {bin} {port} {doc} {iters}", + pin = cfg.pin_prefix, + bin = cfg.server_bin, + doc = cfg.doc, + iters = cfg.iters, + ); + let mut child = Command::new("sh") + .arg("-c") + .arg(cmd.trim()) + .stdout(Stdio::piped()) + .stderr(Stdio::null()) + .spawn()?; + // Block until the server prints "READY ". + let stdout = child + .stdout + .take() + .ok_or_else(|| anyhow!("no server stdout"))?; + let mut reader = BufReader::new(stdout); + let mut line = String::new(); + reader.read_line(&mut line)?; + ensure!( + line.starts_with("READY"), + "server did not signal READY (got {line:?})" + ); + Ok(child) +} + +fn set_linger_zero(s: &TcpStream) { + // SO_LINGER with a 0 timeout → RST on close instead of FIN, so the + // tens of thousands of short client connections a load session opens + // don't pile up in TIME_WAIT and exhaust the ephemeral port range. + // std::net::TcpStream::set_linger is still unstable, so go through + // setsockopt directly. The response is fully read before close, so the + // RST is harmless. + use std::os::unix::io::AsRawFd; + let l = libc::linger { + l_onoff: 1, + l_linger: 0, + }; + unsafe { + libc::setsockopt( + s.as_raw_fd(), + libc::SOL_SOCKET, + libc::SO_LINGER, + &l as *const libc::linger as *const libc::c_void, + std::mem::size_of::() as libc::socklen_t, + ); + } +} + +fn one_request(port: u16) -> std::io::Result<()> { + let mut s = TcpStream::connect((Ipv4Addr::LOCALHOST, port))?; + s.set_nodelay(true)?; + set_linger_zero(&s); + s.write_all(b"GET / HTTP/1.1\r\nHost: x\r\n\r\n")?; + let mut buf = [0u8; 512]; + // Read the full response (server sends Content-Length + Connection: close). + loop { + match s.read(&mut buf) { + Ok(0) => break, + Ok(_) => {} + Err(e) => return Err(e), + } + } + Ok(()) +} + +/// One open-loop load session against a freshly spawned server. +pub fn run_session(cfg: &ServiceCfg, port: u16) -> Result { + let mut child = spawn_server(cfg, port)?; + // Warm up closed-loop (untimed) so the first timed request doesn't eat + // page faults / cold caches. + for _ in 0..cfg.warmup { + let _ = one_request(port); + } + + let period = Duration::from_secs_f64(1.0 / cfg.rate); + let start = Instant::now(); + let idx = Arc::new(AtomicUsize::new(0)); + let out = Arc::new(Mutex::new((Vec::::with_capacity(cfg.count), 0usize))); + let count = cfg.count; + + let mut handles = Vec::new(); + for _ in 0..cfg.workers { + let idx = idx.clone(); + let out = out.clone(); + handles.push(std::thread::spawn(move || { + let mut local = Vec::new(); + let mut dropped = 0usize; + loop { + let i = idx.fetch_add(1, Ordering::Relaxed); + if i >= count { + break; + } + let target = start + period * (i as u32); + let now = Instant::now(); + if now < target { + std::thread::sleep(target - now); + } + // Latency from the INTENDED send time — the CO correction. + match one_request(port) { + Ok(()) => local.push(Instant::now().duration_since(target).as_secs_f64()), + Err(_) => dropped += 1, + } + } + let mut g = out.lock().unwrap(); + g.0.extend(local); + g.1 += dropped; + })); + } + for h in handles { + let _ = h.join(); + } + let _ = child.kill(); + let _ = child.wait(); + + let (latencies, dropped) = Arc::try_unwrap(out).unwrap().into_inner().unwrap(); + Ok(SessionResult { latencies, dropped }) +} + +/// Pick a free loopback TCP port (small race between probe-close and the +/// server bind; harmless on loopback with SO_REUSEADDR). +pub fn free_port() -> Result { + let l = std::net::TcpListener::bind((Ipv4Addr::LOCALHOST, 0))?; + Ok(l.local_addr()?.port()) +} diff --git a/crates/ledger/src/lib.rs b/crates/ledger/src/lib.rs index b718a97..3e0793e 100644 --- a/crates/ledger/src/lib.rs +++ b/crates/ledger/src/lib.rs @@ -190,6 +190,50 @@ impl Ledger { /// onto a completed attempt (harnessd `read_verdict`). Returns the /// verdict plus the bench CIs; never enough to game, since the row /// is already written and immutable. + /// Full row data the ROI report generator needs (SPEC §9). Includes + /// the workload string pulled out of the bench evidence. + pub fn report_row(&self, run_id: &str) -> Result, LedgerError> { + let row = self.conn.query_row( + "SELECT target, playbook_class, hypothesis, verdict, bench, timestamp \ + FROM attempts WHERE run_id = ?1", + [run_id], + |r| { + Ok(( + r.get::<_, String>(0)?, + r.get::<_, u8>(1)?, + r.get::<_, String>(2)?, + r.get::<_, String>(3)?, + r.get::<_, Option>(4)?, + r.get::<_, String>(5)?, + )) + }, + ); + match row { + Ok((target, class, hypothesis, verdict, bench, ts)) => { + let bench: Option = + bench.map(|b| serde_json::from_str(&b)).transpose()?; + let workload = bench + .as_ref() + .and_then(|b| b.get("env_fingerprint")) + .and_then(|f| f.get("workload")) + .cloned(); + Ok(Some(serde_json::json!({ + "run_id": run_id, + "target": target, + "playbook_class": class, + "hypothesis": hypothesis, + "verdict": verdict, + "timestamp": ts, + "speedup_median": bench.as_ref().and_then(|b| b.get("speedup_median").cloned()), + "speedup_ci": bench.as_ref().and_then(|b| b.get("speedup_ci").cloned()), + "workload": workload, + }))) + } + Err(rusqlite::Error::QueryReturnedNoRows) => Ok(None), + Err(e) => Err(e.into()), + } + } + pub fn verdict_summary(&self, run_id: &str) -> Result, LedgerError> { let row = self.conn.query_row( "SELECT verdict, bench FROM attempts WHERE run_id = ?1", diff --git a/crates/report/Cargo.toml b/crates/report/Cargo.toml index d5a06da..3583d21 100644 --- a/crates/report/Cargo.toml +++ b/crates/report/Cargo.toml @@ -8,6 +8,7 @@ repository.workspace = true [dependencies] anyhow.workspace = true +clap.workspace = true ledger = { path = "../ledger" } serde.workspace = true serde_json.workspace = true diff --git a/crates/report/src/main.rs b/crates/report/src/main.rs new file mode 100644 index 0000000..fc4a023 --- /dev/null +++ b/crates/report/src/main.rs @@ -0,0 +1,198 @@ +//! `just report ` — the pitch artifact (SPEC §9). Reads a ledger +//! row (throughput speedup + CI) and, optionally, a service-latency JSON +//! (percentile speedups + CIs), and renders ONE mechanical report: +//! throughput → cores → dollars/year and/or latency percentile deltas, +//! every figure with its 95% CI and workload, the methodology printed +//! inline so the number survives hostile review. No hand-editing: the +//! numbers come only from the ledger and the calibrated bench. + +use anyhow::{anyhow, Context, Result}; +use clap::Parser; +use report::{pricing::Pricing, roi_from_speedup_ci}; +use std::path::PathBuf; + +#[derive(Parser)] +#[command( + name = "report", + about = "Mechanical ROI report from a ledger row (SPEC §9)" +)] +struct Cli { + /// Ledger run_id to report on (throughput ROI). + #[arg(long)] + run_id: String, + /// Fleet size the workload runs on, in cores (engagement input). + #[arg(long, default_value_t = 1000.0)] + fleet_cores: f64, + /// Optional service-latency JSON (from `bench-runner service --out`) + /// to add the latency-percentile ROI section. + #[arg(long)] + service_json: Option, + #[arg(long, default_value = "results/ledger.sqlite")] + db: PathBuf, + #[arg(long, default_value = "config/pricing.toml")] + pricing: PathBuf, + /// Output markdown path; stdout if omitted. + #[arg(long)] + out: Option, +} + +fn fmt_usd(x: f64) -> String { + if x.abs() >= 1000.0 { + format!("${:.0}", x) + } else { + format!("${:.2}", x) + } +} + +fn main() -> Result<()> { + let cli = Cli::parse(); + let ledger = ledger::Ledger::open(&cli.db)?; + let row = ledger + .report_row(&cli.run_id)? + .ok_or_else(|| anyhow!("no ledger row for run_id {}", cli.run_id))?; + let pricing = Pricing::load(&cli.pricing).context("load pricing.toml")?; + + let target = row["target"].as_str().unwrap_or("?"); + let verdict = row["verdict"].as_str().unwrap_or("?"); + let class = row["playbook_class"].as_u64().unwrap_or(0); + let workload = row["workload"] + .as_str() + .unwrap_or("(workload not recorded)"); + let speedup = row["speedup_median"].as_f64(); + let ci = row["speedup_ci"] + .as_array() + .and_then(|a| Some((a.first()?.as_f64()?, a.get(1)?.as_f64()?))); + + let mut md = String::new(); + md.push_str(&format!("# ROI report — `{}`\n\n", cli.run_id)); + md.push_str(&format!( + "**Target:** {target} · **Playbook class:** {class} · **Verdict:** `{verdict}`\n\n\ + **Workload:** {workload}\n\n", + )); + // A report for anything the pipeline did not accept must say so before + // any dollar figure — the ROI below is a measured effect, not a + // shippable saving. + if verdict != "accepted" { + md.push_str(&format!( + "> ⚠️ **Not shipped.** This attempt's verdict is `{verdict}`, not `accepted`. Any ROI\n\ + > below is the *measured* effect — it did not clear the significance/tier bar and is\n\ + > **not** a committed saving. Shown for completeness; the ledger records it as-is.\n\n", + )); + } + + match (speedup, ci) { + (Some(sp), Some((lo, hi))) => { + md.push_str("## Throughput → cores → dollars\n\n"); + md.push_str(&format!( + "Measured speedup (baseline/candidate): **{:.4}**, 95% bootstrap CI **[{:.4}, {:.4}]**.\n\n", + sp, lo, hi + )); + let roi = roi_from_speedup_ci( + cli.fleet_cores, + sp, + (lo, hi), + pricing.dollars_per_core_hour, + pricing.hours_per_year, + ); + md.push_str(&format!( + "On a **{:.0}-core** fleet running this workload:\n\n\ + | metric | median | 95% CI |\n|---|---|---|\n\ + | cores returned | {:.1} | [{:.1}, {:.1}] |\n\ + | annualized saving | {} | [{}, {}] |\n\n", + cli.fleet_cores, + roi.cores_median, + roi.cores_lo, + roi.cores_hi, + fmt_usd(roi.dollars_median), + fmt_usd(roi.dollars_lo), + fmt_usd(roi.dollars_hi), + )); + md.push_str(&format!( + "Pricing: **{}/core-hour**, **{:.0} h/year**. Source: _{}_\n\n\ + The **CI lower bound is the number to quote** — {:.1} cores / {} per year is\n\ + the saving that survives the 95% interval; the point estimate is not a promise.\n\n", + fmt_usd(pricing.dollars_per_core_hour), pricing.hours_per_year, pricing.rate_source, + roi.cores_lo, fmt_usd(roi.dollars_lo), + )); + } + _ => { + md.push_str(&format!( + "## Throughput\n\nNo bench speedup recorded (verdict `{verdict}`); this attempt\n\ + produced no throughput ROI. The row is in the ledger as a complete outcome.\n\n" + )); + } + } + + if let Some(path) = &cli.service_json { + let svc: serde_json::Value = + serde_json::from_str(&std::fs::read_to_string(path).context("read service json")?)?; + let g = |k: &str| svc.get(k).and_then(serde_json::Value::as_f64); + let ga = |k: &str| { + svc.get(k) + .and_then(|v| v.as_array()) + .and_then(|a| Some((a.first()?.as_f64()?, a.get(1)?.as_f64()?))) + }; + md.push_str("## Latency under replayed load\n\n"); + md.push_str(&format!( + "Workload: {}. Coordinated-omission-correct open-loop replay.\n\n", + svc.get("workload") + .and_then(|v| v.as_str()) + .unwrap_or("(service)") + )); + md.push_str( + "| percentile | baseline | candidate | speedup | 95% CI |\n|---|---|---|---|---|\n", + ); + if let (Some(bp50), Some(cp50), Some(s50), Some((l50, h50))) = ( + g("baseline_p50_ms_median"), + g("candidate_p50_ms_median"), + g("p50_speedup_median"), + ga("p50_speedup_ci"), + ) { + md.push_str(&format!( + "| p50 | {:.3} ms | {:.3} ms | {:.4} | [{:.4}, {:.4}] |\n", + bp50, cp50, s50, l50, h50 + )); + } + if let (Some(bp99), Some(cp99), Some(s99), Some((l99, h99))) = ( + g("baseline_p99_ms_median"), + g("candidate_p99_ms_median"), + g("p99_speedup_median"), + ga("p99_speedup_ci"), + ) { + md.push_str(&format!( + "| p99 | {:.3} ms | {:.3} ms | {:.4} | [{:.4}, {:.4}] |\n", + bp99, cp99, s99, l99, h99 + )); + } + md.push('\n'); + } + + // Methodology ships INSIDE the report (SPEC §9) — the differentiation. + md.push_str( + "## Methodology (ships with the number)\n\n\ + - **Interleaved A/B**, baseline rebuilt from a pristine checkout, never the agent's\n\ + workspace. Candidate and baseline alternate to control thermal/background drift.\n\ + - **Bootstrap 95% CI** of the ratio-of-medians; an effect is accepted only if the CI\n\ + lower bound clears the threshold in `config/accept.toml` (2%).\n\ + - **Equivalence gates** passed before any number was trusted: upstream test suite,\n\ + byte-identical golden replay, differential fuzzing, and ASan/UBSan sanitizers.\n\ + - **Calibrated hardware**: the workload passed A/A (false-positive <5%) and injected-\n\ + regression (≥95% detection) self-tests, recorded in `results/calibration/`.\n\ + - **Coordinated-omission correct** latency (service mode): each request's latency is\n\ + measured from its intended send time, so queueing is counted, not hidden.\n\n\ + ## Caveats\n\n\ + Every figure is specific to the workload and hardware named above. Throughput→cores\n\ + assumes the fleet is CPU-bound on this path and scales linearly; latency figures are\n\ + for the stated arrival rate. The saving to commit to is the **CI lower bound**, not\n\ + the median.\n", + ); + + match &cli.out { + Some(p) => { + std::fs::write(p, &md)?; + eprintln!("wrote {}", p.display()); + } + None => print!("{md}"), + } + Ok(()) +} diff --git a/justfile b/justfile index 5f86829..20d252c 100644 --- a/justfile +++ b/justfile @@ -65,3 +65,22 @@ install-coz: coz target iters="200": sh scripts/coz-profile.sh {{target}} {{iters}} python3 scripts/coz-summary.py results/{{target}}/coz/profile.coz + +# Service mode (Phase 4): interleaved A/B latency bench of two server +# binaries under coordinated-omission-correct open-loop load. p50/p99 CIs. +service baseline candidate doc *flags: + cargo run -p bench-runner -- --config config/accept.toml service \ + --baseline-bin "{{baseline}}" --candidate-bin "{{candidate}}" \ + --doc "{{doc}}" --pin "taskset -c 2" {{flags}} + +# Service-mode calibration: A/A false-positive + injected latency-regression +# detection (SPEC §3.1); writes JSON evidence. +service-calibrate server doc out *flags: + cargo run -p bench-runner -- --config config/accept.toml service-calibrate \ + --server-bin "{{server}}" --doc "{{doc}}" --pin "taskset -c 2" \ + --out "{{out}}" {{flags}} + +# Mechanical ROI report from a ledger row (SPEC §9): throughput→cores→ +# dollars and/or latency percentiles, CIs + methodology printed inline. +report run_id *flags: + cargo run -p report -- --run-id "{{run_id}}" {{flags}} diff --git a/targets/cjson/service.c b/targets/cjson/service.c new file mode 100644 index 0000000..cca5d90 --- /dev/null +++ b/targets/cjson/service.c @@ -0,0 +1,124 @@ +/* hotpath service harness for the cjson target (SPEC §3.1 service mode). + * + * TRUST-LAYER CODE — outside targets/cjson/workspace, compiled with the + * (patched) workspace cJSON.c, so a patch changes the parser under HTTP + * load but never this server. A minimal single-threaded HTTP/1.1 server: + * each request parses+serializes a fixed JSON document `iters` times + * (deterministic CPU service time) and replies with the checksum. Single + * worker by design — the cleanest, least-noisy service-time signal; the + * coordinated-omission-correct load generator captures the queueing when + * arrival rate approaches the worker's capacity. + * + * cjson-service + */ +#include +#include +#include +#include +#include +#include +#include +#include +#include "cJSON.h" + +#include + +static char *g_doc; +static long g_inject_ns = 0; /* calibration: busy-wait per request (HOTPATH_INJECT_US) */ + +static void busy_wait_ns(long ns) { + if (ns <= 0) return; + struct timespec s; + clock_gettime(CLOCK_MONOTONIC, &s); + long long target = (long long)s.tv_sec * 1000000000LL + s.tv_nsec + ns; + for (;;) { + struct timespec n; + clock_gettime(CLOCK_MONOTONIC, &n); + if ((long long)n.tv_sec * 1000000000LL + n.tv_nsec >= target) break; + } +} + +static unsigned long do_work(long iters) { + unsigned long acc = 1469598103934665603UL; + busy_wait_ns(g_inject_ns); + for (long i = 0; i < iters; i++) { + cJSON *root = cJSON_Parse(g_doc); + if (!root) return 0; + char *out = cJSON_PrintUnformatted(root); + if (out) { + for (char *p = out; *p; p++) acc = (acc ^ (unsigned char)*p) * 1099511628211UL; + free(out); + } + cJSON_Delete(root); + } + return acc; +} + +int main(int argc, char **argv) { + if (argc < 4) { + fprintf(stderr, "usage: %s \n", argv[0]); + return 2; + } + int port = atoi(argv[1]); + long iters = atol(argv[3]); + if (iters < 1) iters = 1; + /* Calibration-only: inject a fixed per-request busy-wait to validate + * the load generator detects a known latency regression. Off in every + * real run (unset). */ + const char *inj = getenv("HOTPATH_INJECT_US"); + if (inj) g_inject_ns = atol(inj) * 1000L; + + FILE *f = fopen(argv[2], "rb"); + if (!f) { perror("fopen"); return 2; } + fseek(f, 0, SEEK_END); + long n = ftell(f); + fseek(f, 0, SEEK_SET); + g_doc = (char *)malloc((size_t)n + 1); + if (fread(g_doc, 1, (size_t)n, f) != (size_t)n) { return 2; } + g_doc[n] = '\0'; + fclose(f); + + int srv = socket(AF_INET, SOCK_STREAM, 0); + if (srv < 0) { perror("socket"); return 2; } + int one = 1; + setsockopt(srv, SOL_SOCKET, SO_REUSEADDR, &one, sizeof(one)); + struct sockaddr_in addr; + memset(&addr, 0, sizeof(addr)); + addr.sin_family = AF_INET; + addr.sin_addr.s_addr = htonl(INADDR_LOOPBACK); + addr.sin_port = htons((unsigned short)port); + if (bind(srv, (struct sockaddr *)&addr, sizeof(addr)) < 0) { perror("bind"); return 2; } + if (listen(srv, 1024) < 0) { perror("listen"); return 2; } + + /* Signal readiness on stdout so the harness knows the port is up. */ + printf("READY %d\n", port); + fflush(stdout); + + char reqbuf[8192]; + for (;;) { + int c = accept(srv, NULL, NULL); + if (c < 0) continue; + setsockopt(c, IPPROTO_TCP, TCP_NODELAY, &one, sizeof(one)); + /* Drain the request line + headers (we ignore them). */ + ssize_t r = recv(c, reqbuf, sizeof(reqbuf) - 1, 0); + if (r <= 0) { close(c); continue; } + + unsigned long acc = do_work(iters); + + char body[64]; + int blen = snprintf(body, sizeof(body), "checksum=%lu\n", acc); + char resp[256]; + int rlen = snprintf(resp, sizeof(resp), + "HTTP/1.1 200 OK\r\nContent-Type: text/plain\r\n" + "Content-Length: %d\r\nConnection: close\r\n\r\n%s", + blen, body); + ssize_t off = 0; + while (off < rlen) { + ssize_t w = send(c, resp + off, (size_t)(rlen - off), 0); + if (w <= 0) break; + off += w; + } + close(c); + } + return 0; +}