diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 7a2d3d8..4c34b1b 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -16,8 +16,10 @@ jobs: steps: - uses: actions/checkout@v4 + # Pin to the version in rust-toolchain.toml so rustfmt/clippy components + # are installed for the toolchain cargo actually selects. - name: Install Rust toolchain - uses: dtolnay/rust-toolchain@stable + uses: dtolnay/rust-toolchain@1.90.0 with: components: rustfmt, clippy diff --git a/.github/workflows/release.yml b/.github/workflows/release.yml index a403690..7fe930b 100644 --- a/.github/workflows/release.yml +++ b/.github/workflows/release.yml @@ -30,8 +30,10 @@ jobs: steps: - uses: actions/checkout@v4 + # Pin to rust-toolchain.toml so the matrix target's std is installed for + # the toolchain cargo selects. - name: Install Rust toolchain - uses: dtolnay/rust-toolchain@stable + uses: dtolnay/rust-toolchain@1.90.0 with: targets: ${{ matrix.target }} diff --git a/ARCHITECTURE.md b/ARCHITECTURE.md new file mode 100644 index 0000000..27e7967 --- /dev/null +++ b/ARCHITECTURE.md @@ -0,0 +1,92 @@ +# Architecture + +`chainbench-grpc` is organized into four layers with dependencies pointing +**inward**. The domain is the stable core; everything else depends on it, and it +depends on nothing in the crate. + +``` + ┌─────────────────────────────────────────────┐ + │ presentation │ console · json · csv · html + └───────────────┬───────────────┬─────────────┘ + │ │ + ┌───────────────▼─────┐ │ + │ application │ │ use-case pipelines + │ run · slots · │ │ (orchestration) + │ throughput │ │ + └───────┬─────────────┘ │ + │ │ + ┌───────────▼───────────┐ │ + │ infrastructure │ │ I/O adapters + │ geyser · proto · │ │ (gRPC, NTP, TOML) + │ sntp · config_file │ │ + └───────────┬───────────┘ │ + │ │ + ┌───────▼───────────────────────▼─────┐ + │ domain │ pure model + logic + │ analysis · scoring · collector · │ + │ timing · clock · config · warmup │ + └──────────────────────────────────────┘ +``` + +**Dependency rule.** `domain` imports nothing from the other layers. +`infrastructure` imports only `domain` (it adapts protobuf/NTP/TOML *to* domain +types). `application` imports `domain` + `infrastructure`. `presentation` +imports `domain` + `application` result types. `main.rs` is the composition +root that wires them together and selects a renderer. + +## Module map + +| Layer | Module | Responsibility | +|---|---|---| +| domain | `analysis` | `RunSummary`/`EndpointSummary`, percentiles, win-rate, backfill + clock-skew handling | +| domain | `scoring` | composite 0–100 score (fixed-threshold weights) | +| domain | `collector` | `Comparator` + `TransactionAccumulator` (dedup, earliest-wins, emit-once) | +| domain | `timing` | `TransactionData`/`TimestampSource` value objects; `observe()` | +| domain | `clock` | `ClockOffset` value object + offset formula | +| domain | `config` | `BenchConfig`, `Endpoint`, `EndpointKind`, `ArgsCommitment` | +| domain | `warmup` | `WarmupGuard` | +| application | `run` | race/latency/full pipeline (spawn providers, shutdown, summarize) | +| application | `slots` | slot-lifecycle pipeline + computation | +| application | `throughput` | throughput pipeline | +| infrastructure | `geyser` | `GeyserProvider` trait, factory, `ProviderContext`, Yellowstone provider + gRPC client; protobuf↔domain conversions | +| infrastructure | `proto` | generated protobuf (build-time) | +| infrastructure | `sntp` | NTP probe over UDP → `ClockOffset` | +| infrastructure | `config_file` | TOML config loading | +| presentation | `output` | console tables, JSON, CSV | +| presentation | `html` | standalone HTML report | + +## Ubiquitous language + +- **Endpoint** — a gRPC provider under test. +- **Observation** — one endpoint receiving one transaction signature, carrying + server (`created_at`) and client timestamps (`domain::timing`). +- **Comparator** — aggregates observations per signature across endpoints + (`domain::collector`). +- **Run / Summary** — a completed benchmark and its computed metrics + (`domain::analysis`). +- **Provider** — an adapter that streams observations from an endpoint + (`infrastructure::geyser`). +- **Clock offset** — host-vs-UTC skew used to correct absolute latency + (`domain::clock`, probed by `infrastructure::sntp`). + +## How a `race`/`latency`/`full` run flows + +1. **`main`** parses the CLI, builds a `BenchConfig` + `Endpoint`s + (`domain::config`), and resolves the clock offset (probing + `infrastructure::sntp`). +2. **`application::run::run_comparison`** spawns one `infrastructure::geyser` + provider per endpoint. Each provider converts protobuf updates into + `domain::timing` observations and feeds the shared `domain::collector::Comparator`. +3. When the transaction target is hit (or Ctrl+C / max-duration), the pipeline + calls `domain::analysis::compute_run_summary`, which uses `domain::scoring`. +4. **`main`** hands the `RunSummary` to a `presentation` renderer. + +`slots` and `throughput` follow the same shape with their own pipelines. + +## Testing + +- **Unit tests** live beside the domain/infra logic they cover (percentiles, + scoring, comparator semantics, clock formula, NTP parsing, timestamp extraction). +- **`tests/pipeline.rs`** drives the public domain API end to end. +- **`tests/mock_server.rs`** runs the real Yellowstone provider against an + in-process mock gRPC server over plaintext HTTP/2. diff --git a/CHANGELOG.md b/CHANGELOG.md index e42225f..4742e65 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -4,6 +4,36 @@ All notable changes to this project are documented here. The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/) and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html). +## [0.4.0] - 2026-06-15 + +Layered (DDD-style) architecture refactor. **No behavioral change** — all 28 +tests pass unchanged — but the internal structure and the public module paths +moved, so this is a breaking change for library consumers. + +### Changed +- Reorganized `src/` into four layers with dependencies pointing inward: + `domain` (model + pure logic), `application` (use-case pipelines), + `infrastructure` (gRPC/NTP/TOML adapters), `presentation` (renderers). See + the new [ARCHITECTURE.md](ARCHITECTURE.md). +- **Public API paths moved** for library users, e.g. `chainbench_grpc::analysis` + → `chainbench_grpc::domain::analysis`, `chainbench_grpc::providers` → + `chainbench_grpc::infrastructure::geyser`, `output`/`html` → + `chainbench_grpc::presentation::*`. +- Split mixed modules onto layer boundaries: clock math (`domain::clock`) vs NTP + probe (`infrastructure::sntp`); config value objects (`domain::config`) vs TOML + loading (`infrastructure::config_file`); the protobuf `created_at` extraction + and commitment conversion moved into `infrastructure::geyser` so `domain` no + longer depends on generated protobuf; race/latency/full orchestration extracted + from `main` into `application::run`; slot/throughput console renderers moved to + `presentation::output`. + +### Removed (dead code) +- `BenchConfig.duration_secs` (written, never read). +- Unused dependencies `chrono` and a duplicate `tokio-stream` (kept as a + dev-dependency for tests). +- Dead `GeyserGrpcClient::subscribe_with_request` branch + the orphaned + `SubscribeSendError` error variant. + ## [0.3.0] - 2026-06-15 Clock-offset correction for absolute latency (TS1 clock-skew design, Tier 1). diff --git a/Cargo.lock b/Cargo.lock index 5787baa..57e557d 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -11,15 +11,6 @@ dependencies = [ "memchr", ] -[[package]] -name = "android_system_properties" -version = "0.1.5" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "819e7219dbd41043ac279b19830f2efc897156490d7fd6ea916720117ee66311" -dependencies = [ - "libc", -] - [[package]] name = "anstream" version = "1.0.0" @@ -256,12 +247,11 @@ checksum = "9330f8b2ff13f34540b44e946ef35111825727b38d33286ef986142615121801" [[package]] name = "chainbench-grpc" -version = "0.2.0" +version = "0.4.0" dependencies = [ "anyhow", "bs58", "bytes", - "chrono", "clap", "comfy-table", "dashmap", @@ -285,19 +275,6 @@ dependencies = [ "tracing-subscriber", ] -[[package]] -name = "chrono" -version = "0.4.44" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "c673075a2e0e5f4a1dde27ce9dee1ea4558c7ffe648f576438a20ca1d2acc4b0" -dependencies = [ - "iana-time-zone", - "js-sys", - "num-traits", - "wasm-bindgen", - "windows-link", -] - [[package]] name = "clang-sys" version = "1.8.1" @@ -849,30 +826,6 @@ dependencies = [ "tracing", ] -[[package]] -name = "iana-time-zone" -version = "0.1.65" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "e31bc9ad994ba00e440a8aa5c9ef0ec67d5cb5e5cb0cc7f8b744a35b389cc470" -dependencies = [ - "android_system_properties", - "core-foundation-sys", - "iana-time-zone-haiku", - "js-sys", - "log", - "wasm-bindgen", - "windows-core", -] - -[[package]] -name = "iana-time-zone-haiku" -version = "0.1.2" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "f31827a206f56af32e590ba56d5d2d085f558508192593743f16b2306495269f" -dependencies = [ - "cc", -] - [[package]] name = "id-arena" version = "2.3.0" @@ -2243,65 +2196,12 @@ version = "0.4.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "712e227841d057c1ee1cd2fb22fa7e5a5461ae8e48fa2ca79ec42cfc1931183f" -[[package]] -name = "windows-core" -version = "0.62.2" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "b8e83a14d34d0623b51dce9581199302a221863196a1dde71a7663a4c2be9deb" -dependencies = [ - "windows-implement", - "windows-interface", - "windows-link", - "windows-result", - "windows-strings", -] - -[[package]] -name = "windows-implement" -version = "0.60.2" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "053e2e040ab57b9dc951b72c264860db7eb3b0200ba345b4e4c3b14f67855ddf" -dependencies = [ - "proc-macro2", - "quote", - "syn", -] - -[[package]] -name = "windows-interface" -version = "0.59.3" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "3f316c4a2570ba26bbec722032c4099d8c8bc095efccdc15688708623367e358" -dependencies = [ - "proc-macro2", - "quote", - "syn", -] - [[package]] name = "windows-link" version = "0.2.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "f0805222e57f7521d6a62e36fa9163bc891acd422f971defe97d64e70d0a4fe5" -[[package]] -name = "windows-result" -version = "0.4.1" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "7781fa89eaf60850ac3d2da7af8e5242a5ea78d1a11c49bf2910bb5a73853eb5" -dependencies = [ - "windows-link", -] - -[[package]] -name = "windows-strings" -version = "0.5.1" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "7837d08f69c77cf6b07689544538e017c1bfcf57e34b4c0ff58e6c2cd3b37091" -dependencies = [ - "windows-link", -] - [[package]] name = "windows-sys" version = "0.52.0" diff --git a/Cargo.toml b/Cargo.toml index 41f53a2..5783e32 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "chainbench-grpc" -version = "0.3.0" +version = "0.4.0" edition = "2024" description = "Comprehensive Solana gRPC benchmarking tool" license = "Apache-2.0" @@ -9,7 +9,6 @@ authors = ["CSFeo"] [dependencies] tokio = { version = "1", features = ["full"] } -tokio-stream = "0.1" tonic = { version = "0.14", features = ["tls-native-roots"] } rustls = { version = "0.23", features = ["ring"] } tonic-prost = "0.14" @@ -30,7 +29,6 @@ tracing-subscriber = { version = "0.3", features = ["env-filter"] } bs58 = "0.5" bytes = "1" solana-pubkey = "2" -chrono = "0.4" [build-dependencies] tonic-prost-build = "0.14" diff --git a/README.md b/README.md index a94b40f..27a18f7 100644 --- a/README.md +++ b/README.md @@ -368,37 +368,47 @@ the benchmarking engine can be embedded without shelling out to the binary. chainbench-grpc = { git = "https://github.com/CSFeo/chainbench-grpc" } ``` +The code is organized into `domain` / `application` / `infrastructure` / +`presentation` layers (see [ARCHITECTURE.md](ARCHITECTURE.md)). + ```rust use std::collections::HashMap; -use chainbench_grpc::analysis::{compute_run_summary, RunMetadata}; -use chainbench_grpc::collector::Comparator; +use chainbench_grpc::domain::analysis::{compute_run_summary, RunMetadata}; +use chainbench_grpc::domain::collector::Comparator; +use chainbench_grpc::presentation::output; let comparator = Comparator::new(); // ... drive comparator.record_observation(...) from your own collection loop ... let summary = compute_run_summary( &comparator, &["endpoint-1".to_string()], - RunMetadata { duration_secs: 10.0, total_errors: 0, endpoint_runtime: HashMap::new() }, + RunMetadata { + duration_secs: 10.0, + total_errors: 0, + endpoint_runtime: HashMap::new(), + clock_offset_ms: 0.0, + }, ); -println!("{}", chainbench_grpc::output::output_json(&summary)); +println!("{}", output::output_json(&summary)); ``` -Key entry points: [`analysis::compute_run_summary`], [`providers::create_provider`], -[`slots::run_slot_benchmark`], [`throughput::run_throughput`], and the -`output` / `html` renderers. +Key entry points: `domain::analysis::compute_run_summary`, +`infrastructure::geyser::create_provider`, `application::run::run_comparison`, +`application::slots::run_slot_benchmark`, `application::throughput::run_throughput`, +and the `presentation::{output, html}` renderers. ## Development ```bash -cargo test # 23 tests (unit + integration) +cargo test # 28 tests (unit + integration) cargo fmt --all -- --check # formatting cargo clippy --all-targets -- -D warnings # lints (zero warnings) cargo build --release ``` -Tests include unit coverage for the analysis/collector/timing logic plus two -integration tests: a public-API pipeline test and a full **mock Yellowstone gRPC -server** test (`tests/mock_server.rs`) that drives the real provider end to end. +Tests include unit coverage for the domain/infra logic plus two integration +tests: a public-API pipeline test and a full **mock Yellowstone gRPC server** +test (`tests/mock_server.rs`) that drives the real provider end to end. CI (`.github/workflows/ci.yml`) runs fmt + clippy + test + release build on every push and PR. Tagging `vX.Y.Z` triggers `release.yml`, which builds binaries for @@ -406,26 +416,36 @@ linux x64/arm64 and macOS x64/arm64 and attaches them to a GitHub Release. ## Architecture +Layered, with dependencies pointing inward (`domain` depends on nothing). Full +detail — module map, ubiquitous language, run flow — in +[ARCHITECTURE.md](ARCHITECTURE.md). + ``` src/ -├── lib.rs # Library root — re-exports the modules below -├── main.rs # Thin CLI (clap), mode dispatch, orchestration -├── config.rs # TOML config + CLI flag parsing -├── timing.rs # Server-side timestamp extraction -├── warmup.rs # Warmup phase guard -├── collector.rs # DashMap comparator + transaction accumulator -├── analysis.rs # Win rate, latency, percentiles, composite score -├── output.rs # Console tables, JSON, CSV formatters -├── html.rs # Standalone HTML report generator -├── throughput.rs # Throughput measurement mode -├── slots.rs # Slot lifecycle tracking (6 stages, with reconnection) -├── proto.rs # Generated protobuf modules -└── providers/ - ├── mod.rs # GeyserProvider trait + factory + ProviderStats - ├── yellowstone.rs # Yellowstone provider (reconnection + backoff) - └── yellowstone_client.rs # gRPC client wrapper (TLS + x-token auth) +├── lib.rs # Library root (layer declarations + crate docs) +├── main.rs # Composition root: CLI + wiring + renderer selection +├── domain/ # Pure model + logic (no I/O) +│ ├── analysis.rs # summaries, percentiles, win-rate, skew/backfill +│ ├── scoring.rs # composite 0–100 score +│ ├── collector.rs # Comparator + accumulator +│ ├── timing.rs # observation value objects +│ ├── clock.rs # clock-offset value object + formula +│ ├── config.rs # BenchConfig / Endpoint value objects +│ └── warmup.rs +├── application/ # Use-case pipelines +│ ├── run.rs # race / latency / full +│ ├── slots.rs # slot lifecycle +│ └── throughput.rs +├── infrastructure/ # I/O adapters +│ ├── geyser/ # provider trait + Yellowstone provider + gRPC client +│ ├── proto.rs # generated protobuf +│ ├── sntp.rs # NTP clock probe (UDP) +│ └── config_file.rs# TOML loading +└── presentation/ # Renderers + ├── output.rs # console / JSON / CSV + └── html.rs # standalone HTML report tests/ -├── pipeline.rs # End-to-end test via the public library API +├── pipeline.rs # End-to-end test via the public domain API └── mock_server.rs # Real provider driven against an in-process mock server ``` diff --git a/src/application/mod.rs b/src/application/mod.rs new file mode 100644 index 0000000..115ede9 --- /dev/null +++ b/src/application/mod.rs @@ -0,0 +1,6 @@ +//! Application layer — use-case pipelines that orchestrate a benchmark run. +//! Depends on `domain` and `infrastructure`. + +pub mod run; +pub mod slots; +pub mod throughput; diff --git a/src/application/run.rs b/src/application/run.rs new file mode 100644 index 0000000..c7f7443 --- /dev/null +++ b/src/application/run.rs @@ -0,0 +1,142 @@ +//! Race / latency / full pipeline: subscribe to all endpoints concurrently, +//! collect observations into the comparator until the transaction target is hit +//! (or Ctrl+C / max-duration), then compute the run summary. + +use std::collections::HashMap; +use std::sync::Arc; +use std::sync::atomic::{AtomicBool, AtomicUsize}; +use std::time::Instant; + +use tokio::sync::broadcast; +use tracing::{error, info}; + +use crate::domain::analysis::{self, EndpointRuntime, RunMetadata, RunSummary}; +use crate::domain::collector::{Comparator, ProgressTracker}; +use crate::domain::config::{BenchConfig, Endpoint}; +use crate::domain::timing; +use crate::domain::warmup::WarmupGuard; +use crate::infrastructure::geyser::{ProviderContext, create_provider}; + +/// Inputs for a comparison run. +pub struct ComparisonRun { + pub endpoints: Vec, + pub config: BenchConfig, + pub max_duration_secs: u64, + /// Clock offset (ms, local-behind-UTC) to apply to absolute latency. + pub clock_offset_ms: f64, +} + +pub async fn run_comparison(run: ComparisonRun) -> RunSummary { + let ComparisonRun { + endpoints, + config, + max_duration_secs, + clock_offset_ms, + } = run; + + let (shutdown_tx, _) = broadcast::channel::<()>(1); + let comparator = Arc::new(Comparator::new()); + let warmup_guard = Arc::new(WarmupGuard::new(config.warmup_secs)); + let shared_counter = Arc::new(AtomicUsize::new(0)); + let shared_shutdown = Arc::new(AtomicBool::new(false)); + + let target = if config.transactions > 0 { + Some(config.transactions as usize) + } else { + None + }; + let progress = target.map(|t| Arc::new(ProgressTracker::new(t))); + let total_producers = endpoints.len(); + + let start_wallclock_ms = timing::get_current_timestamp_ms(); + let start_instant = Instant::now(); + + // Spawn provider tasks + let mut handles = Vec::new(); + let endpoint_names: Vec = endpoints.iter().map(|e| e.name.clone()).collect(); + + for ep in endpoints { + let provider = create_provider(&ep.kind); + let ctx = ProviderContext { + shutdown_tx: shutdown_tx.clone(), + shutdown_rx: shutdown_tx.subscribe(), + start_wallclock_ms, + start_instant, + comparator: Arc::clone(&comparator), + warmup: Arc::clone(&warmup_guard), + shared_counter: Arc::clone(&shared_counter), + shared_shutdown: Arc::clone(&shared_shutdown), + target_transactions: target, + total_producers, + progress: progress.clone(), + }; + handles.push(provider.process(ep, config.clone(), ctx)); + } + + // Ctrl+C handler + let ctrl_shutdown_tx = shutdown_tx.clone(); + let ctrl_shared_shutdown = Arc::clone(&shared_shutdown); + tokio::spawn(async move { + if tokio::signal::ctrl_c().await.is_ok() { + info!("Ctrl+C received, shutting down..."); + ctrl_shared_shutdown.store(true, std::sync::atomic::Ordering::SeqCst); + let _ = ctrl_shutdown_tx.send(()); + } + }); + + // Safety timeout + let timeout_shutdown_tx = shutdown_tx.clone(); + let timeout_shared_shutdown = Arc::clone(&shared_shutdown); + tokio::spawn(async move { + tokio::time::sleep(std::time::Duration::from_secs(max_duration_secs)).await; + if !timeout_shared_shutdown.load(std::sync::atomic::Ordering::SeqCst) { + info!( + "Max duration {}s reached, forcing shutdown", + max_duration_secs + ); + eprintln!( + "\n Warning: max duration {}s reached. Use --max-duration to increase.", + max_duration_secs + ); + timeout_shared_shutdown.store(true, std::sync::atomic::Ordering::SeqCst); + let _ = timeout_shutdown_tx.send(()); + } + }); + + // Wait for all providers, collect per-endpoint runtime stats, count errors + let mut total_errors = 0usize; + let mut endpoint_runtime: HashMap = HashMap::new(); + for handle in handles { + match handle.await { + Ok(Ok(stats)) => { + endpoint_runtime.insert( + stats.endpoint_name.clone(), + EndpointRuntime { + reconnect_count: stats.reconnect_count, + warmup_skipped: stats.warmup_skipped, + }, + ); + } + Ok(Err(e)) => { + error!("Provider error: {}", e); + total_errors += 1; + } + Err(e) => { + error!("Provider task panicked: {}", e); + total_errors += 1; + } + } + } + + let test_duration = start_instant.elapsed(); + let collection_duration = test_duration.as_secs_f64() - config.warmup_secs as f64; + + let metadata = RunMetadata { + duration_secs: collection_duration.max(0.0), + total_errors, + endpoint_runtime, + clock_offset_ms, + }; + + analysis::compute_run_summary(&comparator, &endpoint_names, metadata) +} diff --git a/src/slots.rs b/src/application/slots.rs similarity index 87% rename from src/slots.rs rename to src/application/slots.rs index ce13506..5233e39 100644 --- a/src/slots.rs +++ b/src/application/slots.rs @@ -13,13 +13,12 @@ use tokio::sync::broadcast; use tonic::transport::ClientTlsConfig; use tracing::{error, info, warn}; -use crate::{ - config::{BenchConfig, Endpoint}, - proto::geyser::{ - CommitmentLevel, SubscribeRequest, SubscribeRequestFilterSlots, SubscribeRequestPing, - subscribe_update::UpdateOneof, - }, - providers::yellowstone_client::GeyserGrpcClient, +use crate::domain::analysis::percentile; +use crate::domain::config::{BenchConfig, Endpoint}; +use crate::infrastructure::geyser::client::GeyserGrpcClient; +use crate::infrastructure::proto::geyser::{ + CommitmentLevel, SubscribeRequest, SubscribeRequestFilterSlots, SubscribeRequestPing, + subscribe_update::UpdateOneof, }; /// Slot status stages we track (from Thorofare) @@ -389,65 +388,3 @@ fn make_percentiles(data: &mut [f64]) -> PercentileSummary { p99_ms: Some(percentile(data, 0.99)), } } - -fn percentile(sorted: &[f64], p: f64) -> f64 { - if sorted.is_empty() { - return 0.0; - } - let idx = (p * (sorted.len() - 1) as f64).round() as usize; - sorted[idx] -} - -pub fn display_slot_console(result: &SlotBenchResult) { - use comfy_table::{ContentArrangement, Table}; - - println!("\n Slot Lifecycle Results"); - println!(" ============================================"); - println!( - " Common slots: {} | Duration: {:.1}s", - result.common_slots, result.duration_secs - ); - - let mut table = Table::new(); - #[cfg(not(target_os = "windows"))] - table.load_preset(comfy_table::presets::UTF8_FULL); - #[cfg(target_os = "windows")] - table.load_preset(comfy_table::presets::ASCII_FULL); - table.set_content_arrangement(ContentArrangement::Dynamic); - table.set_header(vec![ - "Endpoint", - "Slots", - "Complete", - "Download P50", - "Download P90", - "Replay P50", - "Replay P90", - "Confirm P50", - "Confirm P90", - "Finalize P50", - "Finalize P90", - ]); - - for ep in &result.endpoints { - table.add_row(vec![ - ep.endpoint.clone(), - ep.slots_collected.to_string(), - ep.slots_complete.to_string(), - f(ep.download.p50_ms), - f(ep.download.p90_ms), - f(ep.replay.p50_ms), - f(ep.replay.p90_ms), - f(ep.confirm.p50_ms), - f(ep.confirm.p90_ms), - f(ep.finalize.p50_ms), - f(ep.finalize.p90_ms), - ]); - } - - println!("{}", table); -} - -fn f(v: Option) -> String { - v.map(|x| format!("{:.0}ms", x)) - .unwrap_or_else(|| "-".into()) -} diff --git a/src/throughput.rs b/src/application/throughput.rs similarity index 79% rename from src/throughput.rs rename to src/application/throughput.rs index fdb5bf5..0f81500 100644 --- a/src/throughput.rs +++ b/src/application/throughput.rs @@ -13,13 +13,11 @@ use tokio::sync::broadcast; use tonic::transport::ClientTlsConfig; use tracing::{error, info}; -use crate::{ - config::{BenchConfig, Endpoint}, - proto::geyser::{ - CommitmentLevel, SubscribeRequest, SubscribeRequestFilterTransactions, - SubscribeRequestPing, subscribe_update::UpdateOneof, - }, - providers::yellowstone_client::GeyserGrpcClient, +use crate::domain::config::{BenchConfig, Endpoint}; +use crate::infrastructure::geyser::client::GeyserGrpcClient; +use crate::infrastructure::proto::geyser::{ + CommitmentLevel, SubscribeRequest, SubscribeRequestFilterTransactions, SubscribeRequestPing, + subscribe_update::UpdateOneof, }; #[derive(Debug, Clone, Serialize)] @@ -242,52 +240,3 @@ async fn measure_endpoint( Ok(result) } - -pub fn display_throughput_console(summary: &ThroughputSummary) { - use comfy_table::{ContentArrangement, Table}; - - println!("\n Throughput Results"); - println!(" ============================================"); - - let mut table = Table::new(); - #[cfg(not(target_os = "windows"))] - table.load_preset(comfy_table::presets::UTF8_FULL); - #[cfg(target_os = "windows")] - table.load_preset(comfy_table::presets::ASCII_FULL); - table.set_content_arrangement(ContentArrangement::Dynamic); - table.set_header(vec![ - "Endpoint", "Duration", "Messages", "Bytes", "Msgs/s", "KB/s", "Txs", "Slots", "Errors", - ]); - - for r in &summary.results { - table.add_row(vec![ - r.endpoint.clone(), - format!("{:.1}s", r.duration_secs), - r.total_messages.to_string(), - humanize_bytes(r.total_bytes), - format!("{:.1}", r.messages_per_sec), - format!("{:.1}", r.bytes_per_sec / 1024.0), - r.transactions.to_string(), - r.slots.to_string(), - r.errors.to_string(), - ]); - } - - println!("{}", table); -} - -pub fn output_throughput_json(summary: &ThroughputSummary) -> String { - serde_json::to_string_pretty(summary).unwrap_or_else(|_| "{}".to_string()) -} - -fn humanize_bytes(bytes: usize) -> String { - if bytes >= 1_073_741_824 { - format!("{:.1} GB", bytes as f64 / 1_073_741_824.0) - } else if bytes >= 1_048_576 { - format!("{:.1} MB", bytes as f64 / 1_048_576.0) - } else if bytes >= 1024 { - format!("{:.1} KB", bytes as f64 / 1024.0) - } else { - format!("{} B", bytes) - } -} diff --git a/src/analysis.rs b/src/domain/analysis.rs similarity index 84% rename from src/analysis.rs rename to src/domain/analysis.rs index 6659940..ccb568e 100644 --- a/src/analysis.rs +++ b/src/domain/analysis.rs @@ -4,8 +4,9 @@ use std::time::Duration; use serde::Serialize; -use crate::collector::Comparator; -use crate::timing::TimestampSource; +use crate::domain::collector::Comparator; +use crate::domain::scoring::compute_scores; +use crate::domain::timing::TimestampSource; /// Latency distribution buckets (from Shyft pattern) #[derive(Debug, Default, Clone, Serialize)] @@ -370,64 +371,6 @@ pub fn percentile(sorted_data: &[f64], p: f64) -> f64 { sorted_data[index] } -/// Composite score (0-100) combining win rate, latency, and reliability. -/// -/// Uses **fixed absolute thresholds**, not normalization relative to the best -/// peer in the run — so a given endpoint's score is stable regardless of which -/// competitors it is compared against (a relative-to-best score would shift as -/// the competitor set changes). The throughput component is the one exception: -/// it is relative to the busiest endpoint, measuring "did this endpoint keep up". -/// -/// Formula: -/// win_rate_component = first_share * 30 (30% weight) -/// latency_component = clamp((1000 - p50)/950) * 25 (25% weight, lower P50 = better) -/// reliability_component = coverage_pct / 100 * 25 (25% weight) -/// stability_component = clamp((1000 - (p99-p50))/950) * 10 (10% weight, low jitter) -/// throughput_component = min(observations / max_observations, 1.0) * 10 (10% weight) -fn compute_scores(endpoints: &mut [EndpointSummary], num_endpoints: usize) { - let max_observations = endpoints - .iter() - .map(|e| e.valid_transactions) - .max() - .unwrap_or(1) - .max(1); - - for ep in endpoints.iter_mut() { - let mut score = 0.0; - - // Win rate component (30 points) — only meaningful with 2+ endpoints - if num_endpoints > 1 { - score += ep.first_share * 30.0; - } else { - score += 30.0; // single endpoint gets full win rate score - } - - // Latency component (25 points) — based on absolute P50 - if let Some(p50) = ep.abs_p50_ms { - // Score: 25 if P50 <= 50ms, linearly to 0 at P50 >= 1000ms - let latency_score = ((1000.0 - p50) / 950.0).clamp(0.0, 1.0); - score += latency_score * 25.0; - } - - // Reliability component (25 points) — server timestamp coverage - score += (ep.timestamp_coverage_pct / 100.0) * 25.0; - - // Stability component (10 points) — low jitter (P99 - P50) - if let (Some(p50), Some(p99)) = (ep.abs_p50_ms, ep.abs_p99_ms) { - let jitter = p99 - p50; - // Score: 10 if jitter <= 50ms, linearly to 0 at jitter >= 1000ms - let stability = ((1000.0 - jitter) / 950.0).clamp(0.0, 1.0); - score += stability * 10.0; - } - - // Throughput component (10 points) — did this endpoint keep up? - let throughput_ratio = ep.valid_transactions as f64 / max_observations as f64; - score += throughput_ratio.min(1.0) * 10.0; - - ep.score = (score * 100.0).round() / 100.0; // round to 2 decimals - } -} - fn compare_latency(lhs: &EndpointSummary, rhs: &EndpointSummary) -> Ordering { match (lhs.rel_p50_ms, rhs.rel_p50_ms) { (Some(l), Some(r)) => l @@ -443,7 +386,7 @@ fn compare_latency(lhs: &EndpointSummary, rhs: &EndpointSummary) -> Ordering { #[cfg(test)] mod tests { use super::*; - use crate::timing::{TimestampSource, TransactionData}; + use crate::domain::timing::{TimestampSource, TransactionData}; use std::collections::HashMap; // --- percentile (nearest-rank with round) --- @@ -654,39 +597,4 @@ mod tests { assert_eq!(corrected.endpoints[0].abs_p50_ms, Some(25.0)); assert_eq!(corrected.clock_offset_ms, 84.0); } - - // --- compute_scores --- - - #[test] - fn single_endpoint_gets_full_win_component() { - let mut eps = vec![EndpointSummary { - name: "solo".into(), - abs_p50_ms: Some(50.0), - abs_p99_ms: Some(50.0), - timestamp_coverage_pct: 100.0, - valid_transactions: 100, - ..Default::default() - }]; - compute_scores(&mut eps, 1); - // win 30 + latency ~25 + reliability 25 + stability ~10 + throughput 10 ~= 100 - assert!(eps[0].score > 99.0, "score was {}", eps[0].score); - } - - #[test] - fn score_rewards_lower_latency() { - let base = |p50: f64| EndpointSummary { - name: "e".into(), - first_share: 0.5, - abs_p50_ms: Some(p50), - abs_p99_ms: Some(p50), - timestamp_coverage_pct: 100.0, - valid_transactions: 100, - ..Default::default() - }; - let mut fast = vec![base(50.0)]; - let mut slow = vec![base(900.0)]; - compute_scores(&mut fast, 2); - compute_scores(&mut slow, 2); - assert!(fast[0].score > slow[0].score); - } } diff --git a/src/domain/clock.rs b/src/domain/clock.rs new file mode 100644 index 0000000..44d73e7 --- /dev/null +++ b/src/domain/clock.rs @@ -0,0 +1,53 @@ +//! Clock-offset domain logic. +//! +//! Absolute one-way latency (`client_wallclock − server_created_at`) is only +//! valid when the two clocks are synchronized. This module owns the offset +//! *value object* and the *formula*; the NTP wire protocol and socket I/O that +//! feed it live in [`crate::infrastructure::sntp`]. +//! +//! Sign convention: **positive offset means the local clock is behind UTC**, so +//! the correction applied to a client timestamp is `client + offset`. + +#[derive(Debug, Clone)] +pub struct ClockOffset { + /// Estimated offset in milliseconds; positive = local clock is behind UTC. + pub offset_ms: f64, + /// Round-trip time to the reference server, milliseconds. + pub rtt_ms: f64, + /// Identifier of the server that produced this estimate. + pub server: String, +} + +/// NTP offset/delay from the four timestamps (all in Unix seconds): +/// T1 = client send, T2 = server recv, T3 = server send, T4 = client recv. +/// Returns `(offset_ms, rtt_ms)`; positive offset = local clock behind UTC. +pub fn compute_offset_ms(t1: f64, t2: f64, t3: f64, t4: f64) -> (f64, f64) { + let offset = ((t2 - t1) + (t3 - t4)) / 2.0; + let rtt = (t4 - t1) - (t3 - t2); + (offset * 1000.0, rtt * 1000.0) +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn offset_zero_when_clocks_agree_and_symmetric() { + // Client and server agree; symmetric 20ms each way. + let (offset, rtt) = compute_offset_ms(0.0, 0.020, 0.020, 0.040); + assert!(offset.abs() < 1e-6, "offset={offset}"); + assert!((rtt - 40.0).abs() < 1e-6, "rtt={rtt}"); + } + + #[test] + fn positive_offset_when_local_behind() { + // Server clock 84ms ahead of local; symmetric 25ms path. + let t1 = 0.0; + let t2 = 0.025 + 0.084; + let t3 = 0.025 + 0.084; + let t4 = 0.050; + let (offset, rtt) = compute_offset_ms(t1, t2, t3, t4); + assert!((offset - 84.0).abs() < 1.0, "offset={offset}"); + assert!((rtt - 50.0).abs() < 1.0, "rtt={rtt}"); + } +} diff --git a/src/collector.rs b/src/domain/collector.rs similarity index 98% rename from src/collector.rs rename to src/domain/collector.rs index fbb8521..1e86c87 100644 --- a/src/collector.rs +++ b/src/domain/collector.rs @@ -5,7 +5,7 @@ use std::{ }; use tracing::info; -use crate::timing::TransactionData; +use crate::domain::timing::TransactionData; /// Thread-safe concurrent comparator for multi-endpoint transaction tracking. /// Adapted from GeyserBench's Comparator with extended timestamp support. @@ -177,7 +177,7 @@ impl ProgressTracker { #[cfg(test)] mod tests { use super::*; - use crate::timing::{TimestampSource, TransactionData}; + use crate::domain::timing::{TimestampSource, TransactionData}; use std::time::Duration; fn tx(elapsed_ms: u64) -> TransactionData { diff --git a/src/config.rs b/src/domain/config.rs similarity index 64% rename from src/config.rs rename to src/domain/config.rs index 78e943a..46cefc2 100644 --- a/src/config.rs +++ b/src/domain/config.rs @@ -1,13 +1,9 @@ -use crate::proto::geyser::CommitmentLevel; -use anyhow::{Context, Result, anyhow}; -use serde::{Deserialize, Serialize}; -use std::fs; +//! Configuration value objects: what to benchmark and against which endpoints. +//! Pure domain types — TOML file loading lives in +//! [`crate::infrastructure::config_file`] and the gRPC commitment conversion in +//! [`crate::infrastructure::geyser`]. -#[derive(Debug, Deserialize, Serialize)] -pub struct ConfigToml { - pub config: BenchConfig, - pub endpoint: Vec, -} +use serde::{Deserialize, Serialize}; #[derive(Debug, Deserialize, Serialize, Clone)] pub struct BenchConfig { @@ -18,8 +14,6 @@ pub struct BenchConfig { pub commitment: ArgsCommitment, #[serde(default = "default_warmup")] pub warmup_secs: u64, - #[serde(default)] - pub duration_secs: Option, } fn default_transactions() -> i32 { @@ -59,16 +53,6 @@ pub enum ArgsCommitment { Finalized, } -impl From for CommitmentLevel { - fn from(commitment: ArgsCommitment) -> Self { - match commitment { - ArgsCommitment::Processed => CommitmentLevel::Processed, - ArgsCommitment::Confirmed => CommitmentLevel::Confirmed, - ArgsCommitment::Finalized => CommitmentLevel::Finalized, - } - } -} - impl ArgsCommitment { pub fn as_str(&self) -> &'static str { match self { @@ -91,12 +75,3 @@ impl EndpointKind { } } } - -impl ConfigToml { - pub fn load(path: &str) -> Result { - let content = - fs::read_to_string(path).with_context(|| format!("Failed to read config {}", path))?; - let config = toml::from_str(&content).map_err(|err| anyhow!(err))?; - Ok(config) - } -} diff --git a/src/domain/mod.rs b/src/domain/mod.rs new file mode 100644 index 0000000..684c43d --- /dev/null +++ b/src/domain/mod.rs @@ -0,0 +1,10 @@ +//! Domain layer — the benchmarking model and pure logic. Depends on nothing +//! else in the crate (no gRPC, no sockets, no rendering). + +pub mod analysis; +pub mod clock; +pub mod collector; +pub mod config; +pub mod scoring; +pub mod timing; +pub mod warmup; diff --git a/src/domain/scoring.rs b/src/domain/scoring.rs new file mode 100644 index 0000000..d01aabf --- /dev/null +++ b/src/domain/scoring.rs @@ -0,0 +1,100 @@ +//! Composite scoring — collapses an endpoint's metrics into a single 0–100 +//! number for ranking. + +use crate::domain::analysis::EndpointSummary; + +/// Composite score (0-100) combining win rate, latency, and reliability. +/// +/// Uses **fixed absolute thresholds**, not normalization relative to the best +/// peer in the run — so a given endpoint's score is stable regardless of which +/// competitors it is compared against (a relative-to-best score would shift as +/// the competitor set changes). The throughput component is the one exception: +/// it is relative to the busiest endpoint, measuring "did this endpoint keep up". +/// +/// Formula: +/// win_rate_component = first_share * 30 (30% weight) +/// latency_component = clamp((1000 - p50)/950) * 25 (25% weight, lower P50 = better) +/// reliability_component = coverage_pct / 100 * 25 (25% weight) +/// stability_component = clamp((1000 - (p99-p50))/950) * 10 (10% weight, low jitter) +/// throughput_component = min(observations / max_observations, 1.0) * 10 (10% weight) +pub fn compute_scores(endpoints: &mut [EndpointSummary], num_endpoints: usize) { + let max_observations = endpoints + .iter() + .map(|e| e.valid_transactions) + .max() + .unwrap_or(1) + .max(1); + + for ep in endpoints.iter_mut() { + let mut score = 0.0; + + // Win rate component (30 points) — only meaningful with 2+ endpoints + if num_endpoints > 1 { + score += ep.first_share * 30.0; + } else { + score += 30.0; // single endpoint gets full win rate score + } + + // Latency component (25 points) — based on absolute P50 + if let Some(p50) = ep.abs_p50_ms { + // Score: 25 if P50 <= 50ms, linearly to 0 at P50 >= 1000ms + let latency_score = ((1000.0 - p50) / 950.0).clamp(0.0, 1.0); + score += latency_score * 25.0; + } + + // Reliability component (25 points) — server timestamp coverage + score += (ep.timestamp_coverage_pct / 100.0) * 25.0; + + // Stability component (10 points) — low jitter (P99 - P50) + if let (Some(p50), Some(p99)) = (ep.abs_p50_ms, ep.abs_p99_ms) { + let jitter = p99 - p50; + // Score: 10 if jitter <= 50ms, linearly to 0 at jitter >= 1000ms + let stability = ((1000.0 - jitter) / 950.0).clamp(0.0, 1.0); + score += stability * 10.0; + } + + // Throughput component (10 points) — did this endpoint keep up? + let throughput_ratio = ep.valid_transactions as f64 / max_observations as f64; + score += throughput_ratio.min(1.0) * 10.0; + + ep.score = (score * 100.0).round() / 100.0; // round to 2 decimals + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn single_endpoint_gets_full_win_component() { + let mut eps = vec![EndpointSummary { + name: "solo".into(), + abs_p50_ms: Some(50.0), + abs_p99_ms: Some(50.0), + timestamp_coverage_pct: 100.0, + valid_transactions: 100, + ..Default::default() + }]; + compute_scores(&mut eps, 1); + // win 30 + latency ~25 + reliability 25 + stability ~10 + throughput 10 ~= 100 + assert!(eps[0].score > 99.0, "score was {}", eps[0].score); + } + + #[test] + fn score_rewards_lower_latency() { + let base = |p50: f64| EndpointSummary { + name: "e".into(), + first_share: 0.5, + abs_p50_ms: Some(p50), + abs_p99_ms: Some(p50), + timestamp_coverage_pct: 100.0, + valid_transactions: 100, + ..Default::default() + }; + let mut fast = vec![base(50.0)]; + let mut slow = vec![base(900.0)]; + compute_scores(&mut fast, 2); + compute_scores(&mut slow, 2); + assert!(fast[0].score > slow[0].score); + } +} diff --git a/src/timing.rs b/src/domain/timing.rs similarity index 59% rename from src/timing.rs rename to src/domain/timing.rs index 40a8c34..075ce73 100644 --- a/src/timing.rs +++ b/src/domain/timing.rs @@ -1,3 +1,10 @@ +//! Observation timing: the value object recorded each time an endpoint delivers +//! a transaction, plus the wall/monotonic clock reads used to build it. +//! +//! The domain works in plain Unix-millisecond `f64`s. Converting a provider's +//! protobuf `created_at` into milliseconds is an infrastructure concern (see +//! [`crate::infrastructure::geyser`]). + use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH}; use tracing::warn; @@ -21,10 +28,6 @@ pub struct TransactionData { pub start_wallclock_ms: f64, } -pub fn extract_server_timestamp(created_at: Option<&prost_types::Timestamp>) -> Option { - created_at.map(|ts| (ts.seconds as f64) * 1000.0 + (ts.nanos as f64) / 1_000_000.0) -} - pub fn get_current_timestamp_ms() -> f64 { let now = SystemTime::now(); match now.duration_since(UNIX_EPOCH) { @@ -36,15 +39,18 @@ pub fn get_current_timestamp_ms() -> f64 { } } -pub fn make_observation( - created_at: Option<&prost_types::Timestamp>, +/// Build an observation. `server_ms` is the server `created_at` in Unix +/// milliseconds when present; otherwise the client wallclock is used as the +/// timestamp (and flagged as such). +pub fn observe( + server_ms: Option, start_instant: Instant, start_wallclock_ms: f64, ) -> TransactionData { let client_wallclock_ms = get_current_timestamp_ms(); let elapsed = start_instant.elapsed(); - match extract_server_timestamp(created_at) { + match server_ms { Some(server_ms) => TransactionData { timestamp_ms: server_ms, timestamp_source: TimestampSource::ServerCreatedAt, @@ -67,46 +73,16 @@ mod tests { use super::*; #[test] - fn extract_server_timestamp_none_when_absent() { - assert_eq!(extract_server_timestamp(None), None); - } - - #[test] - fn extract_server_timestamp_combines_seconds_and_nanos() { - let ts = prost_types::Timestamp { - seconds: 1_700_000_000, - nanos: 500_000_000, // 0.5s -> 500ms - }; - let ms = extract_server_timestamp(Some(&ts)).unwrap(); - assert_eq!(ms, 1_700_000_000_000.0 + 500.0); - } - - #[test] - fn extract_server_timestamp_sub_millisecond_precision() { - // 1ns -> 1e-6 ms; verifies we don't lose sub-ms precision (unlike - // blocktime-based tools that only have second granularity). - let ts = prost_types::Timestamp { - seconds: 0, - nanos: 1, - }; - assert_eq!(extract_server_timestamp(Some(&ts)).unwrap(), 1e-6); - } - - #[test] - fn make_observation_uses_server_timestamp_when_present() { - let ts = prost_types::Timestamp { - seconds: 10, - nanos: 0, - }; - let obs = make_observation(Some(&ts), Instant::now(), 1_000.0); + fn observe_uses_server_timestamp_when_present() { + let obs = observe(Some(10_000.0), Instant::now(), 1_000.0); assert_eq!(obs.timestamp_source, TimestampSource::ServerCreatedAt); assert_eq!(obs.timestamp_ms, 10_000.0); assert_eq!(obs.start_wallclock_ms, 1_000.0); } #[test] - fn make_observation_falls_back_to_client_wallclock() { - let obs = make_observation(None, Instant::now(), 1_000.0); + fn observe_falls_back_to_client_wallclock() { + let obs = observe(None, Instant::now(), 1_000.0); assert_eq!(obs.timestamp_source, TimestampSource::ClientWallclock); // With no server stamp, timestamp_ms mirrors the client wallclock. assert_eq!(obs.timestamp_ms, obs.client_wallclock_ms); diff --git a/src/warmup.rs b/src/domain/warmup.rs similarity index 100% rename from src/warmup.rs rename to src/domain/warmup.rs diff --git a/src/infrastructure/config_file.rs b/src/infrastructure/config_file.rs new file mode 100644 index 0000000..a465f1f --- /dev/null +++ b/src/infrastructure/config_file.rs @@ -0,0 +1,22 @@ +//! TOML config-file loading — adapts an on-disk file to domain config types. + +use anyhow::{Context, Result, anyhow}; +use serde::{Deserialize, Serialize}; +use std::fs; + +use crate::domain::config::{BenchConfig, Endpoint}; + +#[derive(Debug, Deserialize, Serialize)] +pub struct ConfigToml { + pub config: BenchConfig, + pub endpoint: Vec, +} + +impl ConfigToml { + pub fn load(path: &str) -> Result { + let content = + fs::read_to_string(path).with_context(|| format!("Failed to read config {}", path))?; + let config = toml::from_str(&content).map_err(|err| anyhow!(err))?; + Ok(config) + } +} diff --git a/src/providers/yellowstone_client.rs b/src/infrastructure/geyser/client.rs similarity index 78% rename from src/providers/yellowstone_client.rs rename to src/infrastructure/geyser/client.rs index f7ffbd4..392d192 100644 --- a/src/providers/yellowstone_client.rs +++ b/src/infrastructure/geyser/client.rs @@ -1,10 +1,6 @@ use { bytes::Bytes, - futures::{ - channel::mpsc, - sink::{Sink, SinkExt}, - stream::Stream, - }, + futures::{channel::mpsc, sink::Sink, stream::Stream}, std::convert::TryInto, tonic::{ Request, Response, Status, @@ -15,7 +11,9 @@ use { }, }; -use crate::proto::geyser::{SubscribeRequest, SubscribeUpdate, geyser_client::GeyserClient}; +use crate::infrastructure::proto::geyser::{ + SubscribeRequest, SubscribeUpdate, geyser_client::GeyserClient, +}; #[derive(Clone, Debug)] pub struct InterceptorXToken { @@ -35,8 +33,6 @@ impl tonic::service::Interceptor for InterceptorXToken { pub enum GeyserGrpcClientError { #[error("gRPC status: {0}")] TonicStatus(#[from] Status), - #[error("Failed to send subscribe request: {0}")] - SubscribeSendError(#[from] mpsc::SendError), } pub type GeyserGrpcClientResult = Result; @@ -52,29 +48,15 @@ impl GeyserGrpcClient { Ok(GeyserGrpcBuilder::new(Endpoint::from_shared(endpoint)?)) } + /// Open a subscription. The caller drives the returned sink to send the + /// `SubscribeRequest` (filters, commitment) and reads updates from the stream. pub async fn subscribe( &mut self, ) -> GeyserGrpcClientResult<( impl Sink, impl Stream>, )> { - self.subscribe_with_request(None).await - } - - pub async fn subscribe_with_request( - &mut self, - request: Option, - ) -> GeyserGrpcClientResult<( - impl Sink, - impl Stream>, - )> { - let (mut subscribe_tx, subscribe_rx) = mpsc::unbounded(); - if let Some(request) = request { - subscribe_tx - .send(request) - .await - .map_err(GeyserGrpcClientError::SubscribeSendError)?; - } + let (subscribe_tx, subscribe_rx) = mpsc::unbounded(); let response: Response> = self.geyser.subscribe(subscribe_rx).await?; Ok((subscribe_tx, response.into_inner())) diff --git a/src/providers/mod.rs b/src/infrastructure/geyser/mod.rs similarity index 67% rename from src/providers/mod.rs rename to src/infrastructure/geyser/mod.rs index 666d6c2..96e283f 100644 --- a/src/providers/mod.rs +++ b/src/infrastructure/geyser/mod.rs @@ -8,14 +8,25 @@ use std::{ }; use tokio::sync::broadcast; -use crate::{ - collector::{Comparator, ProgressTracker}, - config::{BenchConfig, Endpoint, EndpointKind}, - warmup::WarmupGuard, -}; +use crate::domain::collector::{Comparator, ProgressTracker}; +use crate::domain::config::{ArgsCommitment, BenchConfig, Endpoint, EndpointKind}; +use crate::domain::warmup::WarmupGuard; +use crate::infrastructure::proto::geyser::CommitmentLevel; +pub mod client; pub mod yellowstone; -pub(crate) mod yellowstone_client; + +/// Adapt the domain commitment value object to the gRPC enum. Lives in the +/// infrastructure layer because `CommitmentLevel` is generated protobuf. +impl From for CommitmentLevel { + fn from(commitment: ArgsCommitment) -> Self { + match commitment { + ArgsCommitment::Processed => CommitmentLevel::Processed, + ArgsCommitment::Confirmed => CommitmentLevel::Confirmed, + ArgsCommitment::Finalized => CommitmentLevel::Finalized, + } + } +} /// Per-endpoint runtime stats returned by a provider task when it finishes. #[derive(Debug, Clone, Default)] diff --git a/src/providers/yellowstone.rs b/src/infrastructure/geyser/yellowstone.rs similarity index 89% rename from src/providers/yellowstone.rs rename to src/infrastructure/geyser/yellowstone.rs index 9caf2f0..3d2eb45 100644 --- a/src/providers/yellowstone.rs +++ b/src/infrastructure/geyser/yellowstone.rs @@ -6,18 +6,15 @@ use tokio::task; use tonic::transport::ClientTlsConfig; use tracing::{error, info, warn}; -use crate::proto::geyser::{ +use crate::domain::collector::TransactionAccumulator; +use crate::domain::config::{BenchConfig, Endpoint}; +use crate::domain::timing; +use crate::infrastructure::proto::geyser::{ CommitmentLevel, SubscribeRequest, SubscribeRequestFilterTransactions, SubscribeRequestPing, subscribe_update::UpdateOneof, }; -use crate::{ - collector::TransactionAccumulator, - config::{BenchConfig, Endpoint}, - timing, -}; - -use super::{GeyserProvider, ProviderContext, ProviderStats, yellowstone_client::GeyserGrpcClient}; +use super::{GeyserProvider, ProviderContext, ProviderStats, client::GeyserGrpcClient}; pub struct YellowstoneProvider; @@ -37,6 +34,12 @@ fn backoff_delay(reconnect_count: u32) -> std::time::Duration { std::time::Duration::from_secs(2u64.pow(reconnect_count.min(4))) } +/// Convert a protobuf `created_at` to Unix milliseconds (nanosecond precision). +/// This adapts the gRPC wire type to the plain `f64` the domain works in. +fn server_timestamp_ms(created_at: Option<&prost_types::Timestamp>) -> Option { + created_at.map(|ts| (ts.seconds as f64) * 1000.0 + (ts.nanos as f64) / 1_000_000.0) +} + async fn connect_client( endpoint_url: &str, endpoint_token: &Option, @@ -208,8 +211,8 @@ async fn process_yellowstone_endpoint( None => continue, }; - let tx_data = timing::make_observation( - msg.created_at.as_ref(), + let tx_data = timing::observe( + server_timestamp_ms(msg.created_at.as_ref()), start_instant, start_wallclock_ms, ); @@ -307,3 +310,35 @@ async fn process_yellowstone_endpoint( reconnect_count, }) } + +#[cfg(test)] +mod tests { + use super::server_timestamp_ms; + + #[test] + fn server_timestamp_none_when_absent() { + assert_eq!(server_timestamp_ms(None), None); + } + + #[test] + fn server_timestamp_combines_seconds_and_nanos() { + let ts = prost_types::Timestamp { + seconds: 1_700_000_000, + nanos: 500_000_000, // 0.5s -> 500ms + }; + assert_eq!( + server_timestamp_ms(Some(&ts)).unwrap(), + 1_700_000_000_000.0 + 500.0 + ); + } + + #[test] + fn server_timestamp_sub_millisecond_precision() { + // 1ns -> 1e-6 ms; verifies sub-ms precision is preserved. + let ts = prost_types::Timestamp { + seconds: 0, + nanos: 1, + }; + assert_eq!(server_timestamp_ms(Some(&ts)).unwrap(), 1e-6); + } +} diff --git a/src/infrastructure/mod.rs b/src/infrastructure/mod.rs new file mode 100644 index 0000000..4d35568 --- /dev/null +++ b/src/infrastructure/mod.rs @@ -0,0 +1,7 @@ +//! Infrastructure layer — I/O adapters. Depends on `domain`; converts external +//! representations (protobuf, NTP wire format, TOML) to/from domain types. + +pub mod config_file; +pub mod geyser; +pub mod proto; +pub mod sntp; diff --git a/src/proto.rs b/src/infrastructure/proto.rs similarity index 100% rename from src/proto.rs rename to src/infrastructure/proto.rs diff --git a/src/clock.rs b/src/infrastructure/sntp.rs similarity index 59% rename from src/clock.rs rename to src/infrastructure/sntp.rs index 52c9172..10df570 100644 --- a/src/clock.rs +++ b/src/infrastructure/sntp.rs @@ -1,16 +1,14 @@ -//! Minimal SNTP client for clock-offset calibration. +//! SNTP client — probes NTP servers over UDP and produces a domain +//! [`ClockOffset`]. The offset *formula* lives in [`crate::domain::clock`]; +//! this module owns the NTP wire format and socket I/O. //! -//! Absolute one-way latency (`client_wallclock − server_created_at`) is only -//! valid when the measurement host's clock is synchronized. This module probes -//! NTP servers at startup to estimate the host's offset vs UTC so that offset -//! can be corrected out of the absolute-latency measurement. -//! -//! Pure `std` (UDP) — no extra dependencies. Offset sign convention: **positive -//! means the local clock is behind UTC** (so the correction is `local + offset`). +//! Pure `std` (UDP) — no extra dependencies. use std::net::{ToSocketAddrs, UdpSocket}; use std::time::{Duration, SystemTime, UNIX_EPOCH}; +use crate::domain::clock::{ClockOffset, compute_offset_ms}; + /// Default public NTP servers probed when no manual offset is given. pub const DEFAULT_NTP_SERVERS: &[&str] = &["time.cloudflare.com", "time.google.com", "pool.ntp.org"]; @@ -18,16 +16,6 @@ pub const DEFAULT_NTP_SERVERS: &[&str] = /// Seconds between the NTP epoch (1900-01-01) and the Unix epoch (1970-01-01). const NTP_UNIX_DELTA: f64 = 2_208_988_800.0; -#[derive(Debug, Clone)] -pub struct ClockOffset { - /// Estimated offset in milliseconds; positive = local clock is behind UTC. - pub offset_ms: f64, - /// Round-trip time to the NTP server, milliseconds. - pub rtt_ms: f64, - /// Server that produced this estimate. - pub server: String, -} - fn now_unix_secs() -> f64 { SystemTime::now() .duration_since(UNIX_EPOCH) @@ -35,8 +23,8 @@ fn now_unix_secs() -> f64 { .unwrap_or(0.0) } -/// Convert an 8-byte NTP timestamp (big-endian: 32-bit seconds + 32-bit fraction) -/// to Unix seconds as f64. +/// Convert an 8-byte NTP timestamp (big-endian: 32-bit seconds + 32-bit +/// fraction) to Unix seconds as f64. fn ntp_to_unix(bytes: &[u8]) -> f64 { let secs = u32::from_be_bytes([bytes[0], bytes[1], bytes[2], bytes[3]]) as f64; let frac = @@ -44,15 +32,6 @@ fn ntp_to_unix(bytes: &[u8]) -> f64 { (secs + frac) - NTP_UNIX_DELTA } -/// NTP offset/delay from the four timestamps (all in Unix seconds): -/// T1 = client send, T2 = server recv, T3 = server send, T4 = client recv. -/// Returns (offset_ms, rtt_ms). Positive offset = local clock behind UTC. -pub fn compute_offset_ms(t1: f64, t2: f64, t3: f64, t4: f64) -> (f64, f64) { - let offset = ((t2 - t1) + (t3 - t4)) / 2.0; - let rtt = (t4 - t1) - (t3 - t2); - (offset * 1000.0, rtt * 1000.0) -} - /// Query a single NTP server. `timeout` bounds both send and receive. pub fn query_ntp_offset(server: &str, timeout: Duration) -> std::io::Result { let addr = format!("{server}:123") @@ -118,29 +97,6 @@ pub fn measure_clock_offset(servers: &[&str]) -> Option { mod tests { use super::*; - #[test] - fn offset_zero_when_clocks_agree_and_symmetric() { - // Client and server agree; symmetric 20ms each way. - // T1=0, T2=0.020, T3=0.020, T4=0.040 - let (offset, rtt) = compute_offset_ms(0.0, 0.020, 0.020, 0.040); - assert!((offset - 0.0).abs() < 1e-6, "offset={offset}"); - assert!((rtt - 40.0).abs() < 1e-6, "rtt={rtt}"); - } - - #[test] - fn positive_offset_when_local_behind() { - // Server clock is 84ms ahead of local; symmetric 25ms path. - // local sends at 0 -> arrives at server's 0.025+0.084; server replies - // instantly at 0.109; client receives at 0.050 (local). - let t1 = 0.0; - let t2 = 0.025 + 0.084; // server receive (server clock) - let t3 = 0.025 + 0.084; // server transmit (server clock, no processing) - let t4 = 0.050; // client receive (local clock) - let (offset, rtt) = compute_offset_ms(t1, t2, t3, t4); - assert!((offset - 84.0).abs() < 1.0, "offset={offset}"); - assert!((rtt - 50.0).abs() < 1.0, "rtt={rtt}"); - } - #[test] fn ntp_to_unix_known_value() { // NTP seconds for 1970-01-01 is exactly NTP_UNIX_DELTA -> unix 0. diff --git a/src/lib.rs b/src/lib.rs index 5f5ea30..0f60ab0 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -1,24 +1,35 @@ //! chainbench-grpc — Solana Yellowstone gRPC benchmarking library. //! -//! The CLI binary is a thin wrapper over these modules. They are exposed as a -//! library so the benchmarking engine can be embedded (e.g. inside ChainBench) -//! without shelling out to the binary. +//! The codebase is organized into layers with dependencies pointing **inward**: //! -//! Key entry points: -//! - [`analysis::compute_run_summary`] — turn collected observations into a [`analysis::RunSummary`] -//! - [`providers::create_provider`] — build a [`providers::GeyserProvider`] for an endpoint -//! - [`slots::run_slot_benchmark`] / [`throughput::run_throughput`] — the standalone mode pipelines -//! - [`output`] / [`html`] — render a summary to console/JSON/CSV/HTML +//! - [`domain`] — the benchmarking model and pure logic: observations, the +//! comparator, latency/percentile/scoring analysis, the clock-offset formula, +//! and configuration value objects. Depends on nothing else in the crate. +//! - [`application`] — use-case pipelines that orchestrate a run (race/latency/ +//! full, slots, throughput). Depends on `domain` and `infrastructure`. +//! - [`infrastructure`] — I/O adapters: the Yellowstone gRPC provider/client, +//! the generated protobuf, the SNTP clock probe, and TOML config loading. +//! Depends on `domain`. +//! - [`presentation`] — renderers (console/JSON/CSV/HTML). Depends on `domain` +//! and `application` result types. +//! +//! The binary (`main.rs`) is the composition root: it parses the CLI, wires the +//! layers together, and selects a renderer. +//! +//! ## Ubiquitous language +//! - **Endpoint** — a gRPC provider under test. +//! - **Observation** — one endpoint receiving one transaction signature, with +//! server (`created_at`) and client timestamps ([`domain::timing`]). +//! - **Comparator** — aggregates observations per signature across endpoints +//! ([`domain::collector`]). +//! - **Run / Summary** — a completed benchmark and its computed metrics +//! ([`domain::analysis`]). +//! - **Provider** — an adapter that streams observations from an endpoint +//! ([`infrastructure::geyser`]). +//! - **Clock offset** — host-vs-UTC skew used to correct absolute latency +//! ([`domain::clock`], probed by [`infrastructure::sntp`]). -pub mod analysis; -pub mod clock; -pub mod collector; -pub mod config; -pub mod html; -pub mod output; -pub mod proto; -pub mod providers; -pub mod slots; -pub mod throughput; -pub mod timing; -pub mod warmup; +pub mod application; +pub mod domain; +pub mod infrastructure; +pub mod presentation; diff --git a/src/main.rs b/src/main.rs index dbfbbd5..83bd593 100644 --- a/src/main.rs +++ b/src/main.rs @@ -1,22 +1,13 @@ -use std::{ - sync::{ - Arc, - atomic::{AtomicBool, AtomicUsize}, - }, - time::Instant, -}; - use clap::{Parser, Subcommand, ValueEnum}; -use tokio::sync::broadcast; use tracing::{error, info}; use tracing_subscriber::EnvFilter; -use chainbench_grpc::{analysis, clock, config, html, output, slots, throughput, timing}; - -use chainbench_grpc::collector::{Comparator, ProgressTracker}; -use chainbench_grpc::config::ConfigToml; -use chainbench_grpc::providers::{ProviderContext, create_provider}; -use chainbench_grpc::warmup::WarmupGuard; +use chainbench_grpc::application::run::{ComparisonRun, run_comparison}; +use chainbench_grpc::application::{slots, throughput}; +use chainbench_grpc::domain::config; +use chainbench_grpc::infrastructure::config_file::ConfigToml; +use chainbench_grpc::infrastructure::sntp; +use chainbench_grpc::presentation::{html, output}; const DEFAULT_CONFIG_PATH: &str = "config.toml"; @@ -176,7 +167,7 @@ async fn resolve_clock_offset(cli: &Cli) -> (f64, &'static str) { if let Some(off) = cli.clock_offset_ms { return (off, "manual"); } - match tokio::task::spawn_blocking(|| clock::measure_clock_offset(clock::DEFAULT_NTP_SERVERS)) + match tokio::task::spawn_blocking(|| sntp::measure_clock_offset(sntp::DEFAULT_NTP_SERVERS)) .await { Ok(Some(c)) => { @@ -256,7 +247,6 @@ async fn main() { account: cli.account.clone(), commitment, warmup_secs: cli.warmup, - duration_secs: None, }; (bench, endpoints) } else if let Some(config_path) = &cli.config { @@ -310,7 +300,7 @@ async fn main() { .await; match cli.output { - OutputFormat::Console => slots::display_slot_console(&result), + OutputFormat::Console => output::display_slot_console(&result), OutputFormat::Json => { println!( "{}", @@ -318,7 +308,7 @@ async fn main() { ); } OutputFormat::Csv => { - slots::display_slot_console(&result); + output::display_slot_console(&result); } OutputFormat::Html => { write_html_report(html::render_slots(&result)); @@ -342,8 +332,8 @@ async fn main() { let result = throughput::run_throughput(endpoints, bench_config, *duration).await; match cli.output { - OutputFormat::Console => throughput::display_throughput_console(&result), - OutputFormat::Json => println!("{}", throughput::output_throughput_json(&result)), + OutputFormat::Console => output::display_throughput_console(&result), + OutputFormat::Json => println!("{}", output::output_throughput_json(&result)), OutputFormat::Csv => println!("{}", output::throughput_to_csv(&result)), OutputFormat::Html => { write_html_report(html::render_throughput(&result)); @@ -398,127 +388,18 @@ async fn main() { } println!(); - // Setup shared state - let (shutdown_tx, _) = broadcast::channel::<()>(1); - let comparator = Arc::new(Comparator::new()); - let warmup_guard = Arc::new(WarmupGuard::new(bench_config.warmup_secs)); - let shared_counter = Arc::new(AtomicUsize::new(0)); - let shared_shutdown = Arc::new(AtomicBool::new(false)); - - let target = if bench_config.transactions > 0 { - Some(bench_config.transactions as usize) - } else { - None - }; - - let progress = target.map(|t| Arc::new(ProgressTracker::new(t))); - let total_producers = endpoints.len(); - - let start_wallclock_ms = timing::get_current_timestamp_ms(); - let start_instant = Instant::now(); - - // Spawn provider tasks - let mut handles = Vec::new(); - let endpoint_names: Vec = endpoints.iter().map(|e| e.name.clone()).collect(); - - for ep in endpoints { - let provider = create_provider(&ep.kind); - let ctx = ProviderContext { - shutdown_tx: shutdown_tx.clone(), - shutdown_rx: shutdown_tx.subscribe(), - start_wallclock_ms, - start_instant, - comparator: Arc::clone(&comparator), - warmup: Arc::clone(&warmup_guard), - shared_counter: Arc::clone(&shared_counter), - shared_shutdown: Arc::clone(&shared_shutdown), - target_transactions: target, - total_producers, - progress: progress.clone(), - }; - handles.push(provider.process(ep, bench_config.clone(), ctx)); - } - - // Ctrl+C handler - let ctrl_shutdown_tx = shutdown_tx.clone(); - let ctrl_shared_shutdown = Arc::clone(&shared_shutdown); - tokio::spawn(async move { - if tokio::signal::ctrl_c().await.is_ok() { - info!("Ctrl+C received, shutting down..."); - ctrl_shared_shutdown.store(true, std::sync::atomic::Ordering::SeqCst); - let _ = ctrl_shutdown_tx.send(()); - } - }); - - // Safety timeout - let timeout_shutdown_tx = shutdown_tx.clone(); - let timeout_shared_shutdown = Arc::clone(&shared_shutdown); - let max_duration = cli.max_duration; - tokio::spawn(async move { - tokio::time::sleep(std::time::Duration::from_secs(max_duration)).await; - if !timeout_shared_shutdown.load(std::sync::atomic::Ordering::SeqCst) { - info!("Max duration {}s reached, forcing shutdown", max_duration); - eprintln!( - "\n Warning: max duration {}s reached. Use --max-duration to increase.", - max_duration - ); - timeout_shared_shutdown.store(true, std::sync::atomic::Ordering::SeqCst); - let _ = timeout_shutdown_tx.send(()); - } - }); - - // Wait for all providers, collect per-endpoint runtime stats, count errors - let mut total_errors = 0usize; - let mut endpoint_runtime: std::collections::HashMap = - std::collections::HashMap::new(); - for handle in handles { - match handle.await { - Ok(Ok(stats)) => { - endpoint_runtime.insert( - stats.endpoint_name.clone(), - analysis::EndpointRuntime { - reconnect_count: stats.reconnect_count, - warmup_skipped: stats.warmup_skipped, - }, - ); - } - Ok(Err(e)) => { - error!("Provider error: {}", e); - total_errors += 1; - } - Err(e) => { - error!("Provider task panicked: {}", e); - total_errors += 1; - } - } - } - - let test_duration = start_instant.elapsed(); - let warmup_duration = bench_config.warmup_secs as f64; - let collection_duration = test_duration.as_secs_f64() - warmup_duration; - - // Compute and display results - let metadata = analysis::RunMetadata { - duration_secs: collection_duration.max(0.0), - total_errors, - endpoint_runtime, + let summary = run_comparison(ComparisonRun { + endpoints, + config: bench_config, + max_duration_secs: cli.max_duration, clock_offset_ms, - }; - - let summary = analysis::compute_run_summary(&comparator, &endpoint_names, metadata); + }) + .await; match cli.output { - OutputFormat::Console => { - output::display_console(&summary, show_race, show_latency); - } - OutputFormat::Json => { - println!("{}", output::output_json(&summary)); - } - OutputFormat::Csv => { - println!("{}", output::output_csv(&summary)); - } - OutputFormat::Html => { - write_html_report(html::render_run_summary(&summary)); - } + OutputFormat::Console => output::display_console(&summary, show_race, show_latency), + OutputFormat::Json => println!("{}", output::output_json(&summary)), + OutputFormat::Csv => println!("{}", output::output_csv(&summary)), + OutputFormat::Html => write_html_report(html::render_run_summary(&summary)), } } diff --git a/src/html.rs b/src/presentation/html.rs similarity index 98% rename from src/html.rs rename to src/presentation/html.rs index abc90dc..4bd5ac4 100644 --- a/src/html.rs +++ b/src/presentation/html.rs @@ -1,5 +1,5 @@ -use crate::analysis::{EndpointSummary, RunSummary}; -use crate::throughput::ThroughputSummary; +use crate::application::throughput::ThroughputSummary; +use crate::domain::analysis::{EndpointSummary, RunSummary}; pub fn render_run_summary(summary: &RunSummary) -> String { let mut endpoints_sorted: Vec<&EndpointSummary> = summary.endpoints.iter().collect(); @@ -265,7 +265,7 @@ pub fn render_throughput(summary: &ThroughputSummary) -> String { ) } -pub fn render_slots(result: &crate::slots::SlotBenchResult) -> String { +pub fn render_slots(result: &crate::application::slots::SlotBenchResult) -> String { let rows = result .endpoints .iter() diff --git a/src/presentation/mod.rs b/src/presentation/mod.rs new file mode 100644 index 0000000..56c31f5 --- /dev/null +++ b/src/presentation/mod.rs @@ -0,0 +1,5 @@ +//! Presentation layer — renderers for run results (console, JSON, CSV, HTML). +//! Depends on `domain` and `application` result types. + +pub mod html; +pub mod output; diff --git a/src/output.rs b/src/presentation/output.rs similarity index 80% rename from src/output.rs rename to src/presentation/output.rs index 1165f06..ba77ae4 100644 --- a/src/output.rs +++ b/src/presentation/output.rs @@ -2,7 +2,9 @@ use comfy_table::{ContentArrangement, Table}; use serde_json::{Map, json}; use std::cmp::Ordering; -use crate::analysis::{EndpointSummary, RunSummary}; +use crate::application::slots::SlotBenchResult; +use crate::application::throughput::ThroughputSummary; +use crate::domain::analysis::{EndpointSummary, RunSummary}; #[cfg(target_os = "windows")] fn table_preset() -> &'static str { @@ -363,7 +365,7 @@ pub fn output_csv(summary: &RunSummary) -> String { lines.join("\n") } -pub fn throughput_to_csv(summary: &crate::throughput::ThroughputSummary) -> String { +pub fn throughput_to_csv(summary: &crate::application::throughput::ThroughputSummary) -> String { let mut lines = Vec::new(); lines.push("endpoint,duration_s,total_msgs,total_bytes,msgs_per_sec,kb_per_sec,transactions,slots,errors".to_string()); @@ -384,3 +386,100 @@ pub fn throughput_to_csv(summary: &crate::throughput::ThroughputSummary) -> Stri lines.join("\n") } + +// --- slots (lifecycle) --- + +pub fn display_slot_console(result: &SlotBenchResult) { + println!("\n Slot Lifecycle Results"); + println!(" ============================================"); + println!( + " Common slots: {} | Duration: {:.1}s", + result.common_slots, result.duration_secs + ); + + let mut table = Table::new(); + table.load_preset(table_preset()); + table.set_content_arrangement(ContentArrangement::Dynamic); + table.set_header(vec![ + "Endpoint", + "Slots", + "Complete", + "Download P50", + "Download P90", + "Replay P50", + "Replay P90", + "Confirm P50", + "Confirm P90", + "Finalize P50", + "Finalize P90", + ]); + + for ep in &result.endpoints { + table.add_row(vec![ + ep.endpoint.clone(), + ep.slots_collected.to_string(), + ep.slots_complete.to_string(), + fmt_ms(ep.download.p50_ms), + fmt_ms(ep.download.p90_ms), + fmt_ms(ep.replay.p50_ms), + fmt_ms(ep.replay.p90_ms), + fmt_ms(ep.confirm.p50_ms), + fmt_ms(ep.confirm.p90_ms), + fmt_ms(ep.finalize.p50_ms), + fmt_ms(ep.finalize.p90_ms), + ]); + } + + println!("{}", table); +} + +fn fmt_ms(v: Option) -> String { + v.map(|x| format!("{:.0}ms", x)) + .unwrap_or_else(|| "-".into()) +} + +// --- throughput --- + +pub fn display_throughput_console(summary: &ThroughputSummary) { + println!("\n Throughput Results"); + println!(" ============================================"); + + let mut table = Table::new(); + table.load_preset(table_preset()); + table.set_content_arrangement(ContentArrangement::Dynamic); + table.set_header(vec![ + "Endpoint", "Duration", "Messages", "Bytes", "Msgs/s", "KB/s", "Txs", "Slots", "Errors", + ]); + + for r in &summary.results { + table.add_row(vec![ + r.endpoint.clone(), + format!("{:.1}s", r.duration_secs), + r.total_messages.to_string(), + humanize_bytes(r.total_bytes), + format!("{:.1}", r.messages_per_sec), + format!("{:.1}", r.bytes_per_sec / 1024.0), + r.transactions.to_string(), + r.slots.to_string(), + r.errors.to_string(), + ]); + } + + println!("{}", table); +} + +pub fn output_throughput_json(summary: &ThroughputSummary) -> String { + serde_json::to_string_pretty(summary).unwrap_or_else(|_| "{}".to_string()) +} + +fn humanize_bytes(bytes: usize) -> String { + if bytes >= 1_073_741_824 { + format!("{:.1} GB", bytes as f64 / 1_073_741_824.0) + } else if bytes >= 1_048_576 { + format!("{:.1} MB", bytes as f64 / 1_048_576.0) + } else if bytes >= 1024 { + format!("{:.1} KB", bytes as f64 / 1024.0) + } else { + format!("{} B", bytes) + } +} diff --git a/tests/mock_server.rs b/tests/mock_server.rs index 6c43bf1..75c2f30 100644 --- a/tests/mock_server.rs +++ b/tests/mock_server.rs @@ -16,19 +16,21 @@ use tokio_stream::wrappers::TcpListenerStream; use tokio_stream::{Stream, StreamExt}; use tonic::{Request, Response, Status, transport::Server}; -use chainbench_grpc::analysis::{RunMetadata, compute_run_summary}; -use chainbench_grpc::collector::Comparator; -use chainbench_grpc::config::{ArgsCommitment, BenchConfig, Endpoint, EndpointKind}; -use chainbench_grpc::proto::geyser::{ +use chainbench_grpc::domain::analysis::{RunMetadata, compute_run_summary}; +use chainbench_grpc::domain::collector::Comparator; +use chainbench_grpc::domain::config::{ArgsCommitment, BenchConfig, Endpoint, EndpointKind}; +use chainbench_grpc::domain::timing; +use chainbench_grpc::domain::warmup::WarmupGuard; +use chainbench_grpc::infrastructure::geyser::{ProviderContext, create_provider}; +use chainbench_grpc::infrastructure::proto::geyser::{ PingRequest, PongResponse, SubscribeUpdate, SubscribeUpdateTransaction, SubscribeUpdateTransactionInfo, geyser_server::{Geyser, GeyserServer}, subscribe_update::UpdateOneof, }; -use chainbench_grpc::proto::solana::storage::confirmed_block::{Message, Transaction}; -use chainbench_grpc::providers::{ProviderContext, create_provider}; -use chainbench_grpc::timing; -use chainbench_grpc::warmup::WarmupGuard; +use chainbench_grpc::infrastructure::proto::solana::storage::confirmed_block::{ + Message, Transaction, +}; /// "11111111111111111111111111111111" (System Program) decodes to 32 zero bytes, /// so the mock can emit that account without pulling in a base58 dependency. @@ -83,7 +85,9 @@ impl Geyser for MockGeyser { async fn subscribe( &self, - _request: Request>, + _request: Request< + tonic::Streaming, + >, ) -> Result, Status> { let updates: Vec> = (0..N_UPDATES).map(|i| Ok(make_update(i))).collect(); @@ -99,38 +103,52 @@ impl Geyser for MockGeyser { async fn subscribe_replay_info( &self, - _r: Request, - ) -> Result, Status> { + _r: Request, + ) -> Result< + Response, + Status, + > { unimplemented!() } async fn get_latest_blockhash( &self, - _r: Request, - ) -> Result, Status> { + _r: Request, + ) -> Result< + Response, + Status, + > { unimplemented!() } async fn get_block_height( &self, - _r: Request, - ) -> Result, Status> { + _r: Request, + ) -> Result< + Response, + Status, + > { unimplemented!() } async fn get_slot( &self, - _r: Request, - ) -> Result, Status> { + _r: Request, + ) -> Result, Status> + { unimplemented!() } async fn is_blockhash_valid( &self, - _r: Request, - ) -> Result, Status> { + _r: Request, + ) -> Result< + Response, + Status, + > { unimplemented!() } async fn get_version( &self, - _r: Request, - ) -> Result, Status> { + _r: Request, + ) -> Result, Status> + { unimplemented!() } } @@ -160,7 +178,6 @@ async fn provider_collects_from_mock_server() { account: ACCOUNT.to_string(), commitment: ArgsCommitment::Processed, warmup_secs: 0, - duration_secs: None, }; let (shutdown_tx, shutdown_rx) = broadcast::channel::<()>(1); diff --git a/tests/pipeline.rs b/tests/pipeline.rs index a6c5995..533e2cd 100644 --- a/tests/pipeline.rs +++ b/tests/pipeline.rs @@ -7,9 +7,9 @@ use std::collections::HashMap; use std::time::Duration; -use chainbench_grpc::analysis::{EndpointRuntime, RunMetadata, compute_run_summary}; -use chainbench_grpc::collector::Comparator; -use chainbench_grpc::timing::{TimestampSource, TransactionData}; +use chainbench_grpc::domain::analysis::{EndpointRuntime, RunMetadata, compute_run_summary}; +use chainbench_grpc::domain::collector::Comparator; +use chainbench_grpc::domain::timing::{TimestampSource, TransactionData}; const START: f64 = 1_000_000.0;