diff --git a/rust/crates/truapi-host-cli/e2e/fleet.sh b/rust/crates/truapi-host-cli/e2e/fleet.sh index 18529526..24efec15 100755 --- a/rust/crates/truapi-host-cli/e2e/fleet.sh +++ b/rust/crates/truapi-host-cli/e2e/fleet.sh @@ -17,6 +17,9 @@ # SIGNER_BOT_SVC_TOKEN bearer token; sent only if set (a local dev bot needs none) # RUN_ID shared run id (default fleet-) # METRICS_JSONL shared metrics sink (default /tmp/fleet-metrics.jsonl) +# THINK_MS base pause (ms) between signer calls in the flow script (default 0) +# THINK_JITTER_MS extra uniform-random pause (ms), desyncs VUs (default 0) +# RAMP_JITTER extra random seconds (0..N) added to each VU's ramp delay (default 0) # # Each VU pairs against the bot, which auto-provisions an attested user and # signs, so scale is bounded by the bot's per-user attestation, not the host. @@ -29,12 +32,17 @@ BIN="$ROOT/target/debug/truapi-host" SCRIPT="${SCRIPT:-$ROOT/rust/crates/truapi-host-cli/js/scripts/battery.ts}" VUS="${VUS:-3}" RAMP="${RAMP:-3}" +RAMP_JITTER="${RAMP_JITTER:-0}" PRODUCT_ID="${PRODUCT_ID:-headless-playground.dot}" BASE_PORT="${BASE_PORT:-9955}" BOT="${SIGNER_BOT_BASE_URL:-http://localhost:3737}" NETWORK="${SIGNER_BOT_NETWORK:-paseo-next-v2}" export RUN_ID="${RUN_ID:-fleet-$(date +%s)}" export METRICS_JSONL="${METRICS_JSONL:-/tmp/fleet-metrics.jsonl}" +# Think-time knobs: exported so each VU's host process (and the script it +# spawns) inherits them; default 0 keeps flow scripts pause-free. +export THINK_MS="${THINK_MS:-0}" +export THINK_JITTER_MS="${THINK_JITTER_MS:-0}" [ -x "$BIN" ] || { echo "missing $BIN — build first (cargo build -p truapi-host-cli)" >&2; exit 2; } @@ -87,6 +95,7 @@ for i in $(seq 0 $((VUS - 1))); do run_vu "$i" "$((BASE_PORT + i))" & pids+=($!) sleep "$RAMP" + [ "$RAMP_JITTER" -gt 0 ] && sleep $((RANDOM % (RAMP_JITTER + 1))) done for p in "${pids[@]}"; do wait "$p" || true; done echo "fleet complete -> $METRICS_JSONL" diff --git a/rust/crates/truapi-host-cli/js/scripts/battery.ts b/rust/crates/truapi-host-cli/js/scripts/battery.ts index 97d00903..5ee77082 100644 --- a/rust/crates/truapi-host-cli/js/scripts/battery.ts +++ b/rust/crates/truapi-host-cli/js/scripts/battery.ts @@ -13,6 +13,19 @@ export {}; // module marker so top-level `await` is allowed const GENESIS_HASH = `0x${"11".repeat(32)}` as const; const account = host.productAccount(); +// Load-shaping knobs (both default 0 -> think() is a no-op, byte-identical run). +// THINK_MS base pause (ms) between signer-critical calls +// THINK_JITTER_MS extra uniform-random pause (ms), desyncs concurrent VUs +const THINK_MS = Number(process.env.THINK_MS) || 0; +const THINK_JITTER_MS = Number(process.env.THINK_JITTER_MS) || 0; +const THINK_ENABLED = THINK_MS + THINK_JITTER_MS > 0; + +async function think() { + if (!THINK_ENABLED) return; + const ms = THINK_MS + Math.random() * THINK_JITTER_MS; + await new Promise((r) => setTimeout(r, ms)); +} + interface Case { name: string; ok: boolean; @@ -40,6 +53,7 @@ if (!(login.isOk() && login.value === "Success")) { throw new Error("login did not succeed"); } +await think(); await record("account.getAccount", async () => { const result = await truapi.account.getAccount({ productAccountId: account }); return result.match( @@ -51,6 +65,7 @@ await record("account.getAccount", async () => { ); }); +await think(); await record("signing.signRaw(bytes)", async () => { const result = await truapi.signing.signRaw({ account, @@ -65,6 +80,7 @@ await record("signing.signRaw(bytes)", async () => { ); }); +await think(); await record("signing.signRaw(message)", async () => { const result = await truapi.signing.signRaw({ account, @@ -76,6 +92,7 @@ await record("signing.signRaw(message)", async () => { ); }); +await think(); await record("signing.signPayload", async () => { const result = await truapi.signing.signPayload({ account, @@ -103,6 +120,7 @@ await record("signing.signPayload", async () => { ); }); +await think(); await record("signing.createTransaction", async () => { const result = await truapi.signing.createTransaction({ signer: account, @@ -120,6 +138,7 @@ await record("signing.createTransaction", async () => { ); }); +await think(); await record("entropy.derive", async () => { const result = await truapi.entropy.derive({ context: "0x6d792d6b6579" }); return result.match( diff --git a/rust/crates/truapi-host-cli/src/frame_server.rs b/rust/crates/truapi-host-cli/src/frame_server.rs index 6ade4d2d..64858ad5 100644 --- a/rust/crates/truapi-host-cli/src/frame_server.rs +++ b/rust/crates/truapi-host-cli/src/frame_server.rs @@ -173,7 +173,11 @@ fn record_frame( let latency_ms = started.elapsed().as_secs_f64() * 1000.0; let (outcome, error_class) = match result { Err(err) => (Outcome::Error, Some(err.to_string())), - Ok(()) => match outcomes.lock().ok().and_then(|mut m| m.remove(&class.request_id)) { + Ok(()) => match outcomes + .lock() + .ok() + .and_then(|mut m| m.remove(&class.request_id)) + { Some(Outcome::Error) => (Outcome::Error, Some(RESPONSE_ERROR_CLASS.to_string())), Some(other) => (other, None), None => (Outcome::Success, None), diff --git a/rust/crates/truapi-host-cli/src/main.rs b/rust/crates/truapi-host-cli/src/main.rs index 466c6c0b..da09b767 100644 --- a/rust/crates/truapi-host-cli/src/main.rs +++ b/rust/crates/truapi-host-cli/src/main.rs @@ -17,6 +17,7 @@ mod frame_server; mod metrics; mod network; mod platform; +mod report; mod script_runner; use std::future::Future; @@ -67,6 +68,8 @@ enum Command { /// specified, and can accept pairing deeplinks. With `--script`, exits with /// the script's status; otherwise stays interactive. SigningHost(SigningHostArgs), + /// Aggregate a metrics JSONL into one comparable report. + MetricsReport(MetricsReportArgs), /// Probe the People chain for a mnemonic's registered identity/username. IdentityCheck { /// BIP-39 mnemonic to probe. @@ -159,6 +162,18 @@ struct SigningHostArgs { auto_accept: bool, } +#[derive(clap::Args)] +struct MetricsReportArgs { + /// Metrics JSONL produced by a host or fleet run (`METRICS_JSONL`). + file: PathBuf, + /// Older run to compare against; adds per-op delta columns. + #[arg(long)] + baseline: Option, + /// Emit the structured report as JSON instead of a table. + #[arg(long)] + json: bool, +} + #[tokio::main] async fn main() -> Result<()> { // Install a rustls crypto provider so `wss://` chain connections work; @@ -176,6 +191,28 @@ async fn main() -> Result<()> { match Cli::parse().command { Command::PairingHost(args) => run_pairing_host(args).await, Command::SigningHost(args) => run_signing_host(args).await, + Command::MetricsReport(args) => { + let current = report::load_report(&args.file)?; + let text = match args.baseline { + Some(baseline) => { + let cmp = report::compare(current, report::load_report(&baseline)?); + if args.json { + serde_json::to_string_pretty(&cmp)? + } else { + report::render_compare_table(&cmp) + } + } + None => { + if args.json { + serde_json::to_string_pretty(¤t)? + } else { + report::render_table(¤t) + } + } + }; + println!("{text}"); + Ok(()) + } Command::IdentityCheck { mnemonic, network } => { let entropy = bip39::Mnemonic::parse(mnemonic.trim()) .context("invalid BIP-39 mnemonic")? diff --git a/rust/crates/truapi-host-cli/src/metrics.rs b/rust/crates/truapi-host-cli/src/metrics.rs index 6ef8a874..55bcd85c 100644 --- a/rust/crates/truapi-host-cli/src/metrics.rs +++ b/rust/crates/truapi-host-cli/src/metrics.rs @@ -9,15 +9,14 @@ use std::io::Write; use std::path::{Path, PathBuf}; use std::sync::Arc; -use serde::Serialize; +use serde::{Deserialize, Serialize}; /// What kind of host operation a record measures. /// /// `Frame` is the coarse label for a whole product request frame at the /// WebSocket boundary; the fine-grained variants below are decoded from the -/// wire id. The `#[allow(dead_code)]` covers variants not yet emitted. -#[allow(dead_code)] -#[derive(Debug, Clone, Copy, Serialize)] +/// wire id. +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] #[serde(rename_all = "snake_case")] pub enum Category { Frame, @@ -33,8 +32,7 @@ pub enum Category { } /// Terminal outcome of a measured operation. -#[allow(dead_code)] -#[derive(Debug, Clone, Copy, Serialize)] +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] #[serde(rename_all = "snake_case")] pub enum Outcome { Success, @@ -43,7 +41,7 @@ pub enum Outcome { } /// One per-operation host metric event. Serialises to camelCase JSON. -#[derive(Debug, Clone, Serialize)] +#[derive(Debug, Clone, Serialize, Deserialize)] #[serde(rename_all = "camelCase")] pub struct HostMetricRecord { pub ts: String, @@ -174,10 +172,7 @@ pub fn classify_frame(frame: &[u8]) -> FrameClass { WireKind::Request(r) => (false, r.request_id == id || r.response_id == id), WireKind::Subscription(s) => ( true, - s.start_id == id - || s.stop_id == id - || s.interrupt_id == id - || s.receive_id == id, + s.start_id == id || s.stop_id == id || s.interrupt_id == id || s.receive_id == id, ), }; if matches { @@ -230,8 +225,8 @@ pub fn response_outcome(frame: &[u8]) -> Option<(String, Outcome)> { } /// Map a method name to a coarse metric category by its namespace prefix. -/// The buckets are provisional and unused in v1 (only `Frame` is emitted); -/// revisit them when the fine-grained category path goes live. +/// A few mappings are provisional (e.g. `statement` under Signing) and can be +/// refined later without touching the wire. fn category_for_method(method: &str) -> Category { match method.split('_').next().unwrap_or("") { "signing" | "entropy" | "statement" => Category::Signing, @@ -253,11 +248,26 @@ mod tests { #[test] fn category_maps_by_namespace() { - assert!(matches!(category_for_method("signing_sign_raw"), Category::Signing)); - assert!(matches!(category_for_method("chain_call"), Category::ChainRpc)); - assert!(matches!(category_for_method("local_storage_write"), Category::Storage)); - assert!(matches!(category_for_method("account_request_login"), Category::Pairing)); - assert!(matches!(category_for_method("mystery_method"), Category::Frame)); + assert!(matches!( + category_for_method("signing_sign_raw"), + Category::Signing + )); + assert!(matches!( + category_for_method("chain_call"), + Category::ChainRpc + )); + assert!(matches!( + category_for_method("local_storage_write"), + Category::Storage + )); + assert!(matches!( + category_for_method("account_request_login"), + Category::Pairing + )); + assert!(matches!( + category_for_method("mystery_method"), + Category::Frame + )); } fn sample() -> HostMetricRecord { @@ -319,7 +329,10 @@ mod tests { let frame = |request_id: &str, value: Vec| { ProtocolMessage { request_id: request_id.to_string(), - payload: Payload { id: response_id, value }, + payload: Payload { + id: response_id, + value, + }, } .encode() }; @@ -329,9 +342,26 @@ mod tests { assert_eq!(id, "req-ok"); assert!(matches!(outcome, Outcome::Success)); - let (id, outcome) = response_outcome(&frame("req-err", encode_versioned_err_payload(0u32, 1))) - .expect("response frame is classified"); + let (id, outcome) = + response_outcome(&frame("req-err", encode_versioned_err_payload(0u32, 1))) + .expect("response frame is classified"); assert_eq!(id, "req-err"); assert!(matches!(outcome, Outcome::Error)); } + + #[test] + fn record_round_trips_through_json() { + let line = r#"{"ts":"2026-07-20T10:00:00Z","runId":"fleet-1","vuIndex":2,"category":"signing","op":"signing_sign_raw","latencyMs":12.5,"outcome":"error","errorClass":"NoAllowance"}"#; + let rec: HostMetricRecord = serde_json::from_str(line).expect("deserializes"); + assert_eq!(rec.run_id, "fleet-1"); + assert_eq!(rec.vu_index, 2); + assert_eq!(rec.category, Category::Signing); + assert_eq!(rec.outcome, Outcome::Error); + assert_eq!(rec.error_class.as_deref(), Some("NoAllowance")); + let back = serde_json::to_string(&rec).expect("serializes"); + assert_eq!( + serde_json::from_str::(&back).unwrap().op, + rec.op + ); + } } diff --git a/rust/crates/truapi-host-cli/src/report.rs b/rust/crates/truapi-host-cli/src/report.rs new file mode 100644 index 00000000..396d772d --- /dev/null +++ b/rust/crates/truapi-host-cli/src/report.rs @@ -0,0 +1,616 @@ +//! Aggregates a fleet's `HostMetricRecord` JSONL into one comparable report. + +use std::collections::BTreeMap; +use std::io::BufRead; + +use anyhow::Context as _; +use serde::Serialize; + +use crate::metrics::{Category, HostMetricRecord, Outcome}; + +/// Width of the leading `category/op` (or `vu `) column in the rendered tables. +const OP_COL: usize = 44; + +/// Per-op latency/error aggregate; percentiles are nearest-rank over the run's latencies. +#[derive(Debug, Clone, Serialize)] +pub struct OpStats { + pub count: usize, + pub errors: usize, + pub error_rate: f64, + pub p50_ms: f64, + pub p95_ms: f64, + pub max_ms: f64, +} + +/// Per-VU latency/error aggregate; no per-op breakdown, p95 only (no p50/max). +#[derive(Debug, Serialize)] +pub struct VuStats { + pub count: usize, + pub errors: usize, + pub error_rate: f64, + pub p95_ms: f64, +} + +/// One run's full aggregate: per-op and per-VU stats, plus totals and skip count. +#[derive(Debug, Serialize)] +pub struct RunReport { + pub run_id: String, + pub started: String, + pub ended: String, + pub duration_secs: Option, + pub records: usize, + pub vus: usize, + pub skipped_lines: usize, + pub ops: BTreeMap, + pub per_vu: BTreeMap, + pub total: OpStats, +} + +/// Parses JSONL lines into records; malformed lines are counted, not fatal. +fn parse_lines(reader: R) -> (Vec, usize) { + let mut records = Vec::new(); + let mut skipped = 0usize; + for line in reader.lines() { + let Ok(line) = line else { + skipped += 1; + continue; + }; + if line.trim().is_empty() { + continue; + } + match serde_json::from_str::(&line) { + Ok(rec) => records.push(rec), + Err(_) => skipped += 1, + } + } + (records, skipped) +} + +/// Nearest-rank percentile over an ascending-sorted slice. +fn percentile(sorted: &[f64], q: f64) -> f64 { + if sorted.is_empty() { + return 0.0; + } + let rank = ((q / 100.0) * sorted.len() as f64).ceil() as usize; + sorted[rank.clamp(1, sorted.len()) - 1] +} + +fn op_stats(latencies: &mut [f64], errors: usize) -> OpStats { + latencies.sort_by(|a, b| a.total_cmp(b)); + let count = latencies.len(); + OpStats { + count, + errors, + error_rate: if count == 0 { + 0.0 + } else { + errors as f64 / count as f64 + }, + p50_ms: percentile(latencies, 50.0), + p95_ms: percentile(latencies, 95.0), + max_ms: latencies.last().copied().unwrap_or(0.0), + } +} + +/// Aggregates records into per-op, per-VU, and total stats; `run_id` collapses +/// to `multiple(n)` when the records span more than one run. +fn aggregate(records: &[HostMetricRecord], skipped_lines: usize) -> RunReport { + let mut by_op: BTreeMap, usize)> = BTreeMap::new(); + let mut by_vu: BTreeMap, usize)> = BTreeMap::new(); + let mut total: (Vec, usize) = (Vec::new(), 0); + let mut run_ids: std::collections::BTreeSet<&str> = std::collections::BTreeSet::new(); + let mut started: Option<&str> = None; + let mut ended: Option<&str> = None; + + for rec in records { + let is_error = rec.outcome == Outcome::Error; + let key = format!("{}/{}", category_key(rec.category), rec.op); + let op = by_op.entry(key).or_default(); + op.0.push(rec.latency_ms); + op.1 += is_error as usize; + let vu = by_vu.entry(rec.vu_index).or_default(); + vu.0.push(rec.latency_ms); + vu.1 += is_error as usize; + total.0.push(rec.latency_ms); + total.1 += is_error as usize; + run_ids.insert(&rec.run_id); + if started.is_none_or(|s| rec.ts.as_str() < s) { + started = Some(&rec.ts); + } + if ended.is_none_or(|e| rec.ts.as_str() > e) { + ended = Some(&rec.ts); + } + } + + let started = started.unwrap_or("").to_string(); + let ended = ended.unwrap_or("").to_string(); + let duration_secs = match ( + chrono::DateTime::parse_from_rfc3339(&started), + chrono::DateTime::parse_from_rfc3339(&ended), + ) { + (Ok(a), Ok(b)) => Some((b - a).num_milliseconds() as f64 / 1000.0), + _ => None, + }; + + RunReport { + run_id: match run_ids.len() { + 1 => run_ids.first().unwrap().to_string(), + n => format!("multiple({n})"), + }, + started, + ended, + duration_secs, + records: records.len(), + vus: by_vu.len(), + skipped_lines, + ops: by_op + .into_iter() + .map(|(k, (mut lat, err))| (k, op_stats(&mut lat, err))) + .collect(), + per_vu: by_vu + .into_iter() + .map(|(vu, (mut lat, err))| { + let s = op_stats(&mut lat, err); + ( + vu, + VuStats { + count: s.count, + errors: s.errors, + error_rate: s.error_rate, + p95_ms: s.p95_ms, + }, + ) + }) + .collect(), + total: op_stats(&mut total.0, total.1), + } +} + +fn category_key(c: Category) -> &'static str { + match c { + Category::Frame => "frame", + Category::Pairing => "pairing", + Category::Signing => "signing", + Category::Subscription => "subscription", + Category::HostCallback => "host_callback", + Category::ChainRpc => "chain_rpc", + Category::Storage => "storage", + Category::Permission => "permission", + Category::Memory => "memory", + Category::Session => "session", + } +} + +/// One op's current-vs-baseline comparison; fields are `None` when the op is +/// absent on one side, and `status` is `changed`/`new`/`gone` accordingly. +#[derive(Debug, Serialize)] +pub struct OpDelta { + pub current: Option, + pub baseline: Option, + pub delta_count: Option, + pub delta_error_rate_pts: Option, + pub delta_p50_ms: Option, + pub delta_p95_ms: Option, + pub status: &'static str, +} + +/// A current run paired with a baseline run and their per-op delta map. +#[derive(Debug, Serialize)] +pub struct CompareReport { + pub current: RunReport, + pub baseline: RunReport, + pub delta: BTreeMap, +} + +/// Builds the per-op delta over the union of both runs' op keys; an op present +/// on only one side gets a `new`/`gone` entry instead of a numeric delta. +pub fn compare(current: RunReport, baseline: RunReport) -> CompareReport { + let keys: std::collections::BTreeSet = current + .ops + .keys() + .chain(baseline.ops.keys()) + .cloned() + .collect(); + let delta = keys + .into_iter() + .map(|key| { + let cur = current.ops.get(&key).cloned(); + let base = baseline.ops.get(&key).cloned(); + let entry = match (&cur, &base) { + (Some(c), Some(b)) => OpDelta { + delta_count: Some(c.count as i64 - b.count as i64), + delta_error_rate_pts: Some((c.error_rate - b.error_rate) * 100.0), + delta_p50_ms: Some(c.p50_ms - b.p50_ms), + delta_p95_ms: Some(c.p95_ms - b.p95_ms), + status: "changed", + current: cur.clone(), + baseline: base.clone(), + }, + (Some(_), None) => OpDelta { + current: cur.clone(), + baseline: None, + delta_count: None, + delta_error_rate_pts: None, + delta_p50_ms: None, + delta_p95_ms: None, + status: "new", + }, + (None, Some(_)) => OpDelta { + current: None, + baseline: base.clone(), + delta_count: None, + delta_error_rate_pts: None, + delta_p50_ms: None, + delta_p95_ms: None, + status: "gone", + }, + (None, None) => unreachable!("key came from one of the two maps"), + }; + (key, entry) + }) + .collect(); + CompareReport { + current, + baseline, + delta, + } +} + +/// Renders a current-vs-baseline delta table: one row per op, plus a +/// changed/new/gone `status` column. +pub fn render_compare_table(cmp: &CompareReport) -> String { + use std::fmt::Write as _; + let mut out = String::new(); + let _ = writeln!( + out, + "current run {} ({} records, {} skipped) vs baseline run {} ({} records, {} skipped)", + cmp.current.run_id, + cmp.current.records, + cmp.current.skipped_lines, + cmp.baseline.run_id, + cmp.baseline.records, + cmp.baseline.skipped_lines, + ); + let _ = writeln!(out); + let _ = writeln!( + out, + "{:7} {:>8} {:>9} {:>9} {:>9} status", + "category/op", + "count", + "Δcount", + "Δerr pts", + "Δp50", + "Δp95", + w = OP_COL + ); + for (key, d) in &cmp.delta { + let count = d.current.as_ref().map_or(0, |s| s.count); + let fmt_i = |v: Option| v.map_or_else(|| "-".to_string(), |v| format!("{v:+}")); + let fmt_f = |v: Option| v.map_or_else(|| "-".to_string(), |v| format!("{v:+.1}")); + let _ = writeln!( + out, + "{:7} {:>8} {:>9} {:>9} {:>9} {}", + key, + count, + fmt_i(d.delta_count), + fmt_f(d.delta_error_rate_pts), + fmt_f(d.delta_p50_ms), + fmt_f(d.delta_p95_ms), + d.status, + w = OP_COL + ); + } + out +} + +/// Renders a run's header, per-op table, TOTAL row, and per-VU summary as +/// fixed-width text. +pub fn render_table(report: &RunReport) -> String { + use std::fmt::Write as _; + let mut out = String::new(); + let _ = writeln!( + out, + "run {} | {} -> {} ({}) | records {} | vus {} | skipped lines {}", + report.run_id, + report.started, + report.ended, + report + .duration_secs + .map_or_else(|| "n/a".to_string(), |s| format!("{s:.1}s")), + report.records, + report.vus, + report.skipped_lines, + ); + let _ = writeln!(out); + let _ = writeln!( + out, + "{:7} {:>7} {:>7} {:>9} {:>9} {:>9}", + "category/op", + "count", + "errors", + "err%", + "p50", + "p95", + "max", + w = OP_COL + ); + for (key, s) in &report.ops { + let _ = writeln!( + out, + "{:7} {:>7} {:>6.1}% {:>7.1}ms {:>7.1}ms {:>7.1}ms", + key, + s.count, + s.errors, + s.error_rate * 100.0, + s.p50_ms, + s.p95_ms, + s.max_ms, + w = OP_COL + ); + } + let t = &report.total; + let _ = writeln!( + out, + "{:7} {:>7} {:>6.1}% {:>7.1}ms {:>7.1}ms {:>7.1}ms", + "TOTAL", + t.count, + t.errors, + t.error_rate * 100.0, + t.p50_ms, + t.p95_ms, + t.max_ms, + w = OP_COL + ); + let _ = writeln!(out); + for (vu, s) in &report.per_vu { + let _ = writeln!( + out, + "vu {:7} {:>7} {:>6.1}% {:>17.1}ms", + vu, + s.count, + s.errors, + s.error_rate * 100.0, + s.p95_ms, + w = OP_COL - 3 + ); + } + out +} + +/// Loads and aggregates a JSONL report file; errors if it has zero valid records. +pub fn load_report(path: &std::path::Path) -> anyhow::Result { + let file = + std::fs::File::open(path).with_context(|| format!("cannot open {}", path.display()))?; + let (records, skipped) = parse_lines(std::io::BufReader::new(file)); + anyhow::ensure!( + !records.is_empty(), + "no valid records in {}", + path.display() + ); + Ok(aggregate(&records, skipped)) +} + +#[cfg(test)] +mod tests { + use super::*; + + fn rec(vu: u32, cat: Category, op: &str, ms: f64, outcome: Outcome) -> HostMetricRecord { + HostMetricRecord { + ts: "2026-07-20T10:00:00Z".into(), + run_id: "fleet-1".into(), + vu_index: vu, + category: cat, + op: op.into(), + latency_ms: ms, + outcome, + error_class: None, + } + } + + #[test] + fn parse_skips_malformed_lines_and_counts_them() { + let input = concat!( + r#"{"ts":"2026-07-20T10:00:00Z","runId":"r","vuIndex":0,"category":"frame","op":"a","latencyMs":1.0,"outcome":"success"}"#, + "\n", + "not json\n", + r#"{"ts":"2026-07-20T10:00:01Z","runId":"r","vuIndex":1,"category":"frame","op":"a","latencyMs":2.0,"outcome":"success"}"#, + "\n", + "{\"truncated\": \n", + ); + let (records, skipped) = parse_lines(input.as_bytes()); + assert_eq!(records.len(), 2); + assert_eq!(skipped, 2); + } + + #[test] + fn percentiles_single_record() { + let report = aggregate(&[rec(0, Category::Frame, "a", 7.0, Outcome::Success)], 0); + let stats = &report.ops["frame/a"]; + assert_eq!(stats.p50_ms, 7.0); + assert_eq!(stats.p95_ms, 7.0); + assert_eq!(stats.max_ms, 7.0); + } + + #[test] + fn percentiles_even_and_odd_counts() { + // odd: 1,2,3 -> p50 = ceil(0.5*3)=2nd -> 2.0 ; even: 1,2,3,4 -> p50 = ceil(0.5*4)=2nd -> 2.0, p95 = ceil(0.95*4)=4th -> 4.0 + let odd: Vec<_> = [1.0, 2.0, 3.0] + .iter() + .map(|ms| rec(0, Category::Frame, "a", *ms, Outcome::Success)) + .collect(); + assert_eq!(aggregate(&odd, 0).ops["frame/a"].p50_ms, 2.0); + let even: Vec<_> = [1.0, 2.0, 3.0, 4.0] + .iter() + .map(|ms| rec(0, Category::Frame, "a", *ms, Outcome::Success)) + .collect(); + let stats = aggregate(&even, 0).ops["frame/a"].clone(); + assert_eq!(stats.p50_ms, 2.0); + assert_eq!(stats.p95_ms, 4.0); + } + + #[test] + fn error_rate_counts_only_error_outcomes() { + let records = vec![ + rec(0, Category::Signing, "s", 1.0, Outcome::Success), + rec(0, Category::Signing, "s", 1.0, Outcome::Error), + rec(0, Category::Signing, "s", 1.0, Outcome::Skipped), + rec(0, Category::Signing, "s", 1.0, Outcome::Error), + ]; + let stats = aggregate(&records, 0).ops["signing/s"].clone(); + assert_eq!(stats.count, 4); + assert_eq!(stats.errors, 2); + assert!((stats.error_rate - 0.5).abs() < 1e-9); + } + + #[test] + fn header_covers_run_vus_span_and_skipped() { + let mut records = vec![ + rec(0, Category::Frame, "a", 1.0, Outcome::Success), + rec(1, Category::Frame, "a", 1.0, Outcome::Success), + rec(2, Category::Frame, "a", 1.0, Outcome::Success), + ]; + records[0].ts = "2026-07-20T10:00:00Z".into(); + records[2].ts = "2026-07-20T10:00:30Z".into(); + let report = aggregate(&records, 5); + assert_eq!(report.run_id, "fleet-1"); + assert_eq!(report.vus, 3); + assert_eq!(report.records, 3); + assert_eq!(report.skipped_lines, 5); + assert_eq!(report.started, "2026-07-20T10:00:00Z"); + assert_eq!(report.ended, "2026-07-20T10:00:30Z"); + assert_eq!(report.duration_secs, Some(30.0)); + } + + #[test] + fn mixed_run_ids_are_labelled() { + let mut records = vec![ + rec(0, Category::Frame, "a", 1.0, Outcome::Success), + rec(0, Category::Frame, "a", 1.0, Outcome::Success), + ]; + records[1].run_id = "fleet-2".into(); + assert_eq!(aggregate(&records, 0).run_id, "multiple(2)"); + } + + #[test] + fn table_contains_header_ops_and_totals() { + let records = vec![ + rec( + 0, + Category::Signing, + "signing_sign_raw", + 10.0, + Outcome::Success, + ), + rec( + 1, + Category::Signing, + "signing_sign_raw", + 30.0, + Outcome::Error, + ), + rec(0, Category::Storage, "storage_set", 5.0, Outcome::Success), + ]; + let table = render_table(&aggregate(&records, 1)); + for needle in [ + "run fleet-1", + "records 3", + "vus 2", + "skipped lines 1", + "signing/signing_sign_raw", + "storage/storage_set", + "TOTAL", + "p50", + "p95", + "err%", + "vu 0", + "vu 1", + ] { + assert!(table.contains(needle), "missing {needle:?} in:\n{table}"); + } + } + + #[test] + fn compare_marks_changed_new_and_gone_ops() { + let current = aggregate( + &[ + rec(0, Category::Signing, "s", 20.0, Outcome::Error), + rec(0, Category::Storage, "new_op", 1.0, Outcome::Success), + ], + 0, + ); + let baseline = aggregate( + &[ + rec(0, Category::Signing, "s", 10.0, Outcome::Success), + rec(0, Category::Frame, "old_op", 1.0, Outcome::Success), + ], + 0, + ); + let cmp = compare(current, baseline); + let s = &cmp.delta["signing/s"]; + assert_eq!(s.status, "changed"); + assert_eq!(s.delta_count, Some(0)); + assert!((s.delta_error_rate_pts.unwrap() - 100.0).abs() < 1e-9); + assert!((s.delta_p95_ms.unwrap() - 10.0).abs() < 1e-9); + assert_eq!(cmp.delta["storage/new_op"].status, "new"); + assert_eq!(cmp.delta["frame/old_op"].status, "gone"); + } + + #[test] + fn compare_table_shows_deltas_and_status() { + let current = aggregate(&[rec(0, Category::Signing, "s", 20.0, Outcome::Error)], 0); + let baseline = aggregate(&[rec(0, Category::Signing, "s", 10.0, Outcome::Success)], 0); + let table = render_compare_table(&compare(current, baseline)); + for needle in ["signing/s", "Δp95", "+10.0", "+100.0", "changed", "skipped"] { + assert!(table.contains(needle), "missing {needle:?} in:\n{table}"); + } + } + + #[test] + fn load_report_errors_on_missing_and_empty_files() { + assert!(load_report(std::path::Path::new("/nonexistent/x.jsonl")).is_err()); + let dir = std::env::temp_dir().join("metrics-report-test"); + std::fs::create_dir_all(&dir).unwrap(); + let empty = dir.join("empty.jsonl"); + std::fs::write(&empty, "not json\n").unwrap(); + let err = load_report(&empty).unwrap_err().to_string(); + assert!(err.contains("no valid records"), "got: {err}"); + } + + #[test] + fn json_output_is_deterministic() { + let records = vec![ + rec(1, Category::Storage, "b", 2.0, Outcome::Success), + rec(0, Category::Signing, "a", 1.0, Outcome::Error), + ]; + let a = serde_json::to_string_pretty(&aggregate(&records, 0)).unwrap(); + let b = serde_json::to_string_pretty(&aggregate(&records, 0)).unwrap(); + assert_eq!(a, b); + assert!( + a.find("signing/a").unwrap() < a.find("storage/b").unwrap(), + "ops must be key-sorted" + ); + } + + #[test] + fn category_key_matches_serde_names() { + for c in [ + Category::Frame, + Category::Pairing, + Category::Signing, + Category::Subscription, + Category::HostCallback, + Category::ChainRpc, + Category::Storage, + Category::Permission, + Category::Memory, + Category::Session, + ] { + let serde_name = serde_json::to_value(c) + .unwrap() + .as_str() + .unwrap() + .to_string(); + assert_eq!(category_key(c), serde_name); + } + } +}