diff --git a/README.md b/README.md index aa9eb8f..b8ad2f8 100644 --- a/README.md +++ b/README.md @@ -36,7 +36,7 @@ per-sample data (CI regression / reproducible runs). | Slot freshness / lag (slots behind the leading endpoint) | ✅ now | tick-aligned, fair same-moment comparison | | Transaction landing rate + slots-to-land | ✅ `--features send` | actual on-chain inclusion (not `sendTransaction`-returned-success) | | Yellowstone gRPC first-seen delta (concurrent two-endpoint race) | ✅ `--features grpc` | the metric that reflects co-located infra; never `blockTime` | -| Per-method matrix (`getAccountInfo`, `getMultipleAccounts`, …) | ⏳ roadmap | | +| Per-method matrix (`getSlot`/`getVersion`/`getLatestBlockhash`/`getAccountInfo`/`getMultipleAccounts`) | ✅ now | `solbench methods`; network-inclusive per-method round-trip | **Honesty first (it's the whole point):** `getSlot` round-trip is *read latency from the host running solbench*, dominated by network distance to the client. A globally-CDN'd public RPC will @@ -66,6 +66,7 @@ solbench probe # read latency + jitter + slot-lag, one table solbench probe --samples 50 # more samples for tighter percentiles solbench probe --interval-ms 150 # open-loop tick cadence (>= typical RTT) solbench probe --json # raw per-endpoint results as JSON +solbench methods # per-method read-latency matrix (add --json) solbench serve # live dashboard at http://127.0.0.1:8787 solbench report --region "…" # leaderboard-shaped JSON for a hosted board solbench demo # measurement pipeline over synthetic data @@ -138,7 +139,8 @@ Ordered by how much they close the gap to "what traders actually trade on": 4. **Landing-rate `send`** (on-chain inclusion) — ✅ done (`--features send`). 5. **Yellowstone gRPC first-seen** (concurrent two-endpoint race; never `blockTime`) — ✅ done (`--features grpc`). -6. **Per-method latency matrix**; HDR histograms; crates.io publish + prebuilt binaries (cargo-dist). +6. **Per-method latency matrix** (`solbench methods`) — ✅ done. +7. **HDR histograms**; crates.io publish + prebuilt binaries (cargo-dist). ## How it's built diff --git a/crates/solbench-cli/src/main.rs b/crates/solbench-cli/src/main.rs index 2f57cde..0cebd7e 100644 --- a/crates/solbench-cli/src/main.rs +++ b/crates/solbench-cli/src/main.rs @@ -1,17 +1,20 @@ //! `solbench` CLI. //! //! `probe` measures live RPC read-latency + slot-lag (open-loop, tick-aligned); -//! `serve` renders it as a local dashboard; `demo` exercises the measurement core. +//! `methods` measures a per-method read-latency matrix; `serve` renders a local +//! dashboard; `demo` exercises the measurement core. //! `grpc` (feature `grpc`) races Yellowstone first-seen and `send` (feature `send`) //! measures transaction landing — the metrics that reflect co-located infra. mod grpc; +mod methods; mod probe; mod report; mod send; mod server; use clap::{Parser, Subcommand}; +use methods::{method_specs, probe_methods, CLOCK}; use probe::{endpoints_from_env, probe_all}; use solbench_core::LatencyRecorder; @@ -40,6 +43,22 @@ enum Command { #[arg(long)] json: bool, }, + /// Per-method read-latency matrix across endpoints (getSlot, getVersion, + /// getLatestBlockhash, getAccountInfo, getMultipleAccounts). + Methods { + /// Samples per method per endpoint. + #[arg(long, default_value_t = 20)] + samples: usize, + /// Milliseconds between sample ticks (open-loop schedule). + #[arg(long, default_value_t = 100)] + interval_ms: u64, + /// getAccountInfo target account (default: Clock sysvar). + #[arg(long)] + account: Option, + /// Emit the matrix as JSON. + #[arg(long)] + json: bool, + }, /// Serve a live latency dashboard on localhost. Serve { #[arg(long, default_value_t = 8787)] @@ -153,6 +172,58 @@ fn main() { `solbench grpc` / `solbench send` (or run co-located) for the infra story." ); } + Command::Methods { + samples, + interval_ms, + account, + json, + } => { + let endpoints = endpoints_from_env(); + if !json && endpoints.iter().all(|e| e.label != "rpc edge") { + eprintln!("note: set SOLBENCH_RPCEDGE_URL to include rpc edge in the comparison."); + } + let account = account.unwrap_or_else(|| CLOCK.to_string()); + let specs = method_specs(&account); + let reports = probe_methods(&endpoints, &specs, samples, interval_ms); + + if json { + println!( + "{}", + serde_json::to_string_pretty(&reports).expect("reports serialize") + ); + return; + } + + let ms = |ns: u64| format!("{:.2}", ns as f64 / 1e6); + println!( + "{:<20} {:<16} {:<26} {:>8} {:>8} {:>8} {:>8}", + "method", "endpoint", "host", "p50ms", "p99ms", "jitter", "ok" + ); + for report in &reports { + for r in &report.results { + let (p50, p99, jitter) = match &r.latency { + Some(l) => (ms(l.p50_ns), ms(l.p99_ns), ms(l.stddev_ns)), + None => ("-".into(), "-".into(), "-".into()), + }; + println!( + "{:<20} {:<16} {:<26} {:>8} {:>8} {:>8} {:>8}", + report.method, + r.label, + r.host, + p50, + p99, + jitter, + format!("{}/{}", r.ok, samples), + ); + } + } + eprintln!( + "\nper-method latency is a network-inclusive round-trip from THIS host. Compare\n\ + endpoints WITHIN a method; read cross-method gaps as relative method cost, not\n\ + infra. Methods run in sequence, so a run may span changing network conditions.\n\ + getAccountInfo / getMultipleAccounts hit fixed Solana sysvar accounts." + ); + } Command::Serve { port, samples, diff --git a/crates/solbench-cli/src/methods.rs b/crates/solbench-cli/src/methods.rs new file mode 100644 index 0000000..968fb9d --- /dev/null +++ b/crates/solbench-cli/src/methods.rs @@ -0,0 +1,227 @@ +//! Per-method read-latency matrix: measure a set of read RPC methods across all +//! endpoints, reusing the fair open-loop sampler from `probe`. +//! +//! Latency here is a network-inclusive round-trip from THIS host. Methods differ +//! in server-side cost, so compare endpoints *within* a method and read +//! cross-method gaps as relative method cost, not infrastructure. + +use crate::probe::{redact_host, sample_endpoint, Endpoint}; +use serde::Serialize; +use serde_json::json; +use solbench_core::LatencySummary; +use std::thread; +use std::time::{Duration, Instant}; + +/// Canonical Solana sysvar accounts — present on any cluster, so the account-reading +/// methods stay provider-neutral and reproducible. +pub const CLOCK: &str = "SysvarC1ock11111111111111111111111111111111"; +pub const RENT: &str = "SysvarRent111111111111111111111111111111111"; +pub const RECENT_BLOCKHASHES: &str = "SysvarRecentB1ockHashes11111111111111111111"; + +/// A read method to probe: a display name and the exact JSON-RPC request body. +pub struct MethodSpec { + pub name: &'static str, + pub body: String, +} + +/// The default matrix. `account` is the `getAccountInfo` target (default: Clock sysvar). +pub fn method_specs(account: &str) -> Vec { + let mk = |name: &'static str, v: serde_json::Value| MethodSpec { + name, + body: v.to_string(), + }; + vec![ + mk( + "getSlot", + json!({"jsonrpc":"2.0","id":1,"method":"getSlot"}), + ), + mk( + "getVersion", + json!({"jsonrpc":"2.0","id":1,"method":"getVersion"}), + ), + mk( + "getLatestBlockhash", + json!({"jsonrpc":"2.0","id":1,"method":"getLatestBlockhash"}), + ), + mk( + "getAccountInfo", + json!({"jsonrpc":"2.0","id":1,"method":"getAccountInfo", + "params":[account, {"encoding":"base64"}]}), + ), + mk( + "getMultipleAccounts", + json!({"jsonrpc":"2.0","id":1,"method":"getMultipleAccounts", + "params":[[CLOCK, RENT, RECENT_BLOCKHASHES], {"encoding":"base64"}]}), + ), + ] +} + +/// Success = a JSON-RPC `result` is present and there is no `error`. The numeric +/// payload is unused for these methods, so return 0 on success. +fn methods_extract(v: &serde_json::Value) -> Option { + if v.get("error").is_some() { + None + } else { + v.get("result").is_some().then_some(0u64) + } +} + +#[derive(Serialize)] +pub struct MethodEndpointResult { + pub label: String, + pub host: String, + pub ok: usize, + pub errors: usize, + pub latency: Option, +} + +#[derive(Serialize)] +pub struct MethodReport { + pub method: String, + pub results: Vec, +} + +/// Probe each method across every endpoint. Methods run in sequence; within a +/// method, all endpoints share one fixed tick schedule (fair same-moment compare). +pub fn probe_methods( + endpoints: &[Endpoint], + specs: &[MethodSpec], + samples: usize, + interval_ms: u64, +) -> Vec { + let interval = Duration::from_millis(interval_ms.max(1)); + specs + .iter() + .map(|spec| { + let t0 = Instant::now() + Duration::from_millis(50); + let results: Vec = thread::scope(|scope| { + let handles: Vec<_> = endpoints + .iter() + .map(|ep| { + scope.spawn(move || { + let out = sample_endpoint( + &ep.url, + &spec.body, + samples, + interval, + t0, + methods_extract, + ); + MethodEndpointResult { + label: ep.label.clone(), + host: redact_host(&ep.url), + ok: out.rec.len(), + errors: out.errors, + latency: out.rec.summary(), + } + }) + }) + .collect(); + handles + .into_iter() + .map(|h| h.join().expect("methods probe thread")) + .collect() + }); + MethodReport { + method: spec.name.to_string(), + results, + } + }) + .collect() +} + +#[cfg(test)] +mod tests { + use super::*; + use tiny_http::{Response, Server}; + + fn serve_canned(body: &'static str, count: usize) -> (String, thread::JoinHandle<()>) { + let server = Server::http("127.0.0.1:0").unwrap(); + let port = server.server_addr().to_ip().unwrap().port(); + let url = format!("http://127.0.0.1:{port}/"); + let handle = thread::spawn(move || { + for _ in 0..count { + match server.recv() { + Ok(req) => { + let _ = req.respond(Response::from_string(body)); + } + Err(_) => break, + } + } + }); + (url, handle) + } + + #[test] + fn method_specs_are_well_formed() { + let specs = method_specs(CLOCK); + let names: Vec<_> = specs.iter().map(|s| s.name).collect(); + assert_eq!( + names, + [ + "getSlot", + "getVersion", + "getLatestBlockhash", + "getAccountInfo", + "getMultipleAccounts" + ] + ); + for s in &specs { + let v: serde_json::Value = serde_json::from_str(&s.body).unwrap(); + assert_eq!(v["method"], s.name); + } + let gai: serde_json::Value = serde_json::from_str( + &method_specs("ACCT") + .into_iter() + .find(|s| s.name == "getAccountInfo") + .unwrap() + .body, + ) + .unwrap(); + assert_eq!(gai["params"][0], "ACCT"); + let gma: serde_json::Value = serde_json::from_str( + &specs + .iter() + .find(|s| s.name == "getMultipleAccounts") + .unwrap() + .body, + ) + .unwrap(); + assert_eq!(gma["params"][0].as_array().unwrap().len(), 3); + } + + #[test] + fn methods_extract_distinguishes_success_and_error() { + assert_eq!(methods_extract(&json!({"result":{"x":1}})), Some(0)); + assert_eq!(methods_extract(&json!({"result":123})), Some(0)); + assert_eq!( + methods_extract(&json!({"error":{"code":-1,"message":"x"}})), + None + ); + assert_eq!(methods_extract(&json!({})), None); + } + + #[test] + fn probe_methods_reports_latency_per_method() { + let samples = 3; + let (url, handle) = + serve_canned(r#"{"jsonrpc":"2.0","id":1,"result":{"ok":true}}"#, samples); + let endpoints = vec![Endpoint { + label: "local".into(), + url, + }]; + let specs = vec![MethodSpec { + name: "getSlot", + body: r#"{"jsonrpc":"2.0","id":1,"method":"getSlot"}"#.to_string(), + }]; + let reports = probe_methods(&endpoints, &specs, samples, 10); + assert_eq!(reports.len(), 1); + assert_eq!(reports[0].method, "getSlot"); + assert_eq!(reports[0].results.len(), 1); + let r = &reports[0].results[0]; + assert_eq!(r.ok, samples); + assert_eq!(r.errors, 0); + assert!(r.latency.is_some()); + handle.join().unwrap(); + } +} diff --git a/crates/solbench-cli/src/probe.rs b/crates/solbench-cli/src/probe.rs index c2443e9..b871470 100644 --- a/crates/solbench-cli/src/probe.rs +++ b/crates/solbench-cli/src/probe.rs @@ -47,7 +47,7 @@ pub struct ProbeResult { /// Host portion of a URL with any credentials/query stripped /// (`https://rpc.rpcedge.com/?api-key=...` -> `rpc.rpcedge.com`). -fn redact_host(url: &str) -> String { +pub(crate) fn redact_host(url: &str) -> String { let no_scheme = url.split("://").nth(1).unwrap_or(url); no_scheme .split(['/', '?']) @@ -67,20 +67,40 @@ struct Raw { slots: Vec<(usize, u64)>, } +/// Output of one open-loop sampling run over a single endpoint. +pub(crate) struct SampleOutput { + pub rec: LatencyRecorder, + pub errors: usize, + /// (tick, extracted value) for each successful sample, tick-sorted. + pub extracted: Vec<(usize, u64)>, +} + /// Sample one endpoint `samples` times on the shared tick schedule starting at `t0`. /// /// True open-loop: each tick spawns its own request worker, so a slow reply never /// delays the next send (no coordinated omission). Latency is the actual send->reply -/// round-trip; the shared `agent` pools warm connections across workers. -fn run_endpoint(ep: &Endpoint, samples: usize, interval: Duration, t0: Instant) -> Raw { +/// round-trip; the shared `agent` pools warm connections across workers. `extract` +/// maps a parsed JSON-RPC response to `Some(value)` on success (value is an optional +/// numeric payload such as a slot; use 0 when unused) or `None` to count an error. +pub(crate) fn sample_endpoint( + url: &str, + body: &str, + samples: usize, + interval: Duration, + t0: Instant, + extract: F, +) -> SampleOutput +where + F: Fn(&serde_json::Value) -> Option + Sync, +{ let agent = ureq::AgentBuilder::new() .timeout(Duration::from_secs(6)) .build(); - let body = r#"{"jsonrpc":"2.0","id":1,"method":"getSlot"}"#; - // (tick, latency_ns, slot) for successful samples; separate error counter. + // (tick, latency_ns, extracted value) for successful samples; separate error counter. let ok: Arc>> = Arc::new(Mutex::new(Vec::new())); let errors = Arc::new(Mutex::new(0usize)); + let extract = &extract; thread::scope(|scope| { for i in 0..samples { @@ -90,7 +110,7 @@ fn run_endpoint(ep: &Endpoint, samples: usize, interval: Duration, t0: Instant) thread::sleep(intended - now); } let agent = agent.clone(); - let url = ep.url.clone(); + let url = url.to_string(); let ok = Arc::clone(&ok); let errors = Arc::clone(&errors); scope.spawn(move || { @@ -106,9 +126,11 @@ fn run_endpoint(ep: &Endpoint, samples: usize, interval: Duration, t0: Instant) .into_string() .ok() .and_then(|t| serde_json::from_str::(&t).ok()) - .and_then(|v| v.get("result").and_then(|r| r.as_u64())) { - Some(slot) => ok.lock().unwrap().push((i, latency_ns, slot)), + Some(v) => match extract(&v) { + Some(val) => ok.lock().unwrap().push((i, latency_ns, val)), + None => *errors.lock().unwrap() += 1, + }, None => *errors.lock().unwrap() += 1, } } @@ -123,21 +145,34 @@ fn run_endpoint(ep: &Endpoint, samples: usize, interval: Duration, t0: Instant) data.sort_by_key(|&(tick, _, _)| tick); let mut rec = LatencyRecorder::with_capacity(data.len()); - let mut slots = Vec::with_capacity(data.len()); - let mut last_slot = None; - for &(tick, latency_ns, slot) in &data { + let mut extracted = Vec::with_capacity(data.len()); + for &(tick, latency_ns, val) in &data { rec.record_ns(latency_ns); - slots.push((tick, slot)); - last_slot = Some(slot); // data is tick-sorted, so this ends on the latest + extracted.push((tick, val)); } + SampleOutput { + rec, + errors, + extracted, + } +} + +/// Sample one endpoint's `getSlot` read latency and capture per-tick slots for slot-lag. +fn run_endpoint(ep: &Endpoint, samples: usize, interval: Duration, t0: Instant) -> Raw { + const GET_SLOT: &str = r#"{"jsonrpc":"2.0","id":1,"method":"getSlot"}"#; + let out = sample_endpoint(&ep.url, GET_SLOT, samples, interval, t0, |v| { + v.get("result").and_then(|r| r.as_u64()) + }); + // `extracted` is tick-sorted, so the last entry is the latest slot. + let last_slot = out.extracted.last().map(|&(_, slot)| slot); Raw { label: ep.label.clone(), host: redact_host(&ep.url), - rec, - errors, + rec: out.rec, + errors: out.errors, last_slot, - slots, + slots: out.extracted, } } @@ -223,3 +258,67 @@ pub fn endpoints_from_env() -> Vec { }); endpoints } + +#[cfg(test)] +mod tests { + use super::*; + use tiny_http::{Response, Server}; + + /// Serve exactly `count` requests with a fixed body, then stop. Returns the URL. + fn serve_canned(body: &'static str, count: usize) -> (String, thread::JoinHandle<()>) { + let server = Server::http("127.0.0.1:0").unwrap(); + let port = server.server_addr().to_ip().unwrap().port(); + let url = format!("http://127.0.0.1:{port}/"); + let handle = thread::spawn(move || { + for _ in 0..count { + match server.recv() { + Ok(req) => { + let _ = req.respond(Response::from_string(body)); + } + Err(_) => break, + } + } + }); + (url, handle) + } + + #[test] + fn sample_endpoint_records_latency_and_extracts_value() { + let samples = 5; + let (url, handle) = serve_canned(r#"{"jsonrpc":"2.0","id":1,"result":123}"#, samples); + let t0 = Instant::now() + Duration::from_millis(50); + let out = sample_endpoint( + &url, + r#"{"jsonrpc":"2.0","id":1,"method":"getSlot"}"#, + samples, + Duration::from_millis(10), + t0, + |v| v.get("result").and_then(|r| r.as_u64()), + ); + assert_eq!(out.rec.len(), samples); + assert_eq!(out.errors, 0); + assert!(out.extracted.iter().all(|&(_, v)| v == 123)); + handle.join().unwrap(); + } + + #[test] + fn sample_endpoint_counts_malformed_as_errors() { + let samples = 4; + let (url, handle) = serve_canned("not json", samples); + let t0 = Instant::now() + Duration::from_millis(50); + let out = sample_endpoint(&url, "{}", samples, Duration::from_millis(10), t0, |v| { + v.get("result").and_then(|r| r.as_u64()) + }); + assert_eq!(out.errors, samples); + assert_eq!(out.rec.len(), 0); + handle.join().unwrap(); + } + + #[test] + fn redact_host_strips_credentials() { + assert_eq!( + redact_host("https://rpc.rpcedge.com/?api-key=secret"), + "rpc.rpcedge.com" + ); + } +}