diff --git a/.containerignore b/.containerignore new file mode 100644 index 0000000..3ea0852 --- /dev/null +++ b/.containerignore @@ -0,0 +1,2 @@ +target/ +.git/ diff --git a/Cargo.lock b/Cargo.lock index 81a7b7b..1ed5e20 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2423,6 +2423,20 @@ version = "1.0.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "11d3d7f243d5c5a8b9bb5d6dd2b1602c0cb0b9db1621bafc7ed66e35ff9fe092" +[[package]] +name = "loadgen" +version = "0.0.0" +dependencies = [ + "anyhow", + "clap", + "serde", + "serde_json", + "thiserror 1.0.69", + "tokio", + "tracing", + "ztest", +] + [[package]] name = "lock_api" version = "0.4.14" diff --git a/Cargo.toml b/Cargo.toml index 79625f2..900180f 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -1,5 +1,5 @@ [workspace] -members = ["macros"] +members = ["macros", "crates/loadgen"] resolver = "2" # Post-ironwood dependency line. The whole zcash_* wallet stack now resolves diff --git a/crates/loadgen/Cargo.toml b/crates/loadgen/Cargo.toml new file mode 100644 index 0000000..9c2060c --- /dev/null +++ b/crates/loadgen/Cargo.toml @@ -0,0 +1,27 @@ +[package] +name = "loadgen" +version.workspace = true +edition.workspace = true +license.workspace = true +authors.workspace = true +repository.workspace = true +homepage.workspace = true + +[[bin]] +name = "loadgen" +path = "src/main.rs" + +[dependencies] +# The whole point: reuse ztest's load-test library (LwdClient / LoadDriver / +# ChainLinkOracle / LoadReport). default-features off drops the wallet backends +# (librustzcash/zingo) we don't need; kube/k8s-openapi remain non-optional but +# we never touch them — we only speak gRPC to an already-running zainod. +ztest = { path = "../..", default-features = false } + +tokio = { workspace = true, features = ["rt-multi-thread", "macros"] } +clap = { version = "4", features = ["derive"] } +serde = { version = "1", features = ["derive"] } +serde_json = { workspace = true } +tracing = { workspace = true } +thiserror = { workspace = true } +anyhow = "1" diff --git a/crates/loadgen/Containerfile b/crates/loadgen/Containerfile new file mode 100644 index 0000000..f754846 --- /dev/null +++ b/crates/loadgen/Containerfile @@ -0,0 +1,25 @@ +# Multi-stage build for the `loadgen` zainod load generator. +# +# Build context MUST be the ztest repo root (loadgen depends on the ztest lib +# via `path = "../.."` and inherits the workspace `[patch.crates-io]`): +# +# podman build -f crates/loadgen/Containerfile -t loadgen:dev . +# +# `ztest` is pulled with default-features off (see crates/loadgen/Cargo.toml), +# so the optional zingolib/pepper-sync/test-vector git deps are NOT fetched — +# only crates.io + the public lightwallet-protocol patch. No build creds needed. + +FROM docker.io/library/rust:1.95-slim-bookworm AS builder +RUN apt-get update \ + && apt-get install -y --no-install-recommends protobuf-compiler pkg-config git \ + && rm -rf /var/lib/apt/lists/* +WORKDIR /src +COPY . . +RUN cargo build --release -p loadgen + +FROM docker.io/library/debian:bookworm-slim AS runtime +RUN apt-get update \ + && apt-get install -y --no-install-recommends ca-certificates \ + && rm -rf /var/lib/apt/lists/* +COPY --from=builder /src/target/release/loadgen /usr/local/bin/loadgen +ENTRYPOINT ["/usr/local/bin/loadgen"] diff --git a/crates/loadgen/README.md b/crates/loadgen/README.md new file mode 100644 index 0000000..b2bce4e --- /dev/null +++ b/crates/loadgen/README.md @@ -0,0 +1,103 @@ +# loadgen + +A concurrency/load generator for a **running** zainod. It attaches to a zainod's +gRPC `CompactTxStreamer` endpoint, fans out N concurrent clients driving a chosen +RPC, and reports real latency percentiles, throughput, error counts, and +correctness violations. + +Unlike the rest of ztest it does **not** spawn a topology — it points at an +endpoint you give it (`--target`). The same binary loads a regtest node on a +laptop or a mainnet node in-cluster, and runs as a plain Kubernetes Job. + +## Architecture + +```mermaid +flowchart LR + subgraph job["k8s Job (runs IN-cluster)"] + lg["loadgen
N per-task gRPC connections"] + end + lg -->|"CompactTxStreamer RPC
GetBlockRange · GetBlock · GetLatestBlock"| z["zainod service
zaino.<ns>.svc:8137"] + z -->|"fetch backend: JSON-RPC
state backend: direct DB read"| zeb["zebra"] + lg -.->|"hdrhistogram + chain-link oracle"| rep["LoadReport
human table (stderr) + JSON (stdout)"] +``` + +It reuses `ztest::loadtest` (the `LwdClient` / `LoadDriver` / `ChainLinkOracle` / +`LoadReport` stack); this crate is the runnable front-end. Three seams are kept +independent: + +- **driver** — the swarm + metrics (`ztest::loadtest`), speaks only the gRPC wire + protocol, so it is version- and target-agnostic. +- **target** — supplied as a URI; how the zainod got there (regtest spawn, + ephemeral deploy, mainnet) is not loadgen's concern. +- **reporting** — a human table on stderr and a machine-readable JSON summary on + stdout (`--json`). + +## Running + +### Locally (against any reachable endpoint) + +``` +cargo run -p loadgen -- \ + --target http://127.0.0.1:8137 \ + --rpc block-range --connections 64 --duration 30 --json +``` + +### In-cluster (the representative path) + +Latency is only meaningful measured **inside** the cluster — a port-forward from a +remote node would dominate the numbers. Run it as a Job in the target namespace; +the target is derived from the pod's own namespace: + +``` +kubectl -n apply -f crates/loadgen/k8s/job.yaml +kubectl -n logs -f job/loadgen +``` + +### Connection sweep (find the knee) + +`scripts/sweep.sh` runs a sequential connection sweep (one Job per level — never +parallel, which would confound the measurement) and collects one JSON line per +level: + +``` +scripts/sweep.sh "1 4 8 16 32 64 128" 20 +``` + +## Flags + +| flag | default | meaning | +|---|---|---| +| `--target` | (required) | zainod gRPC endpoint, e.g. `http://zaino.ns.svc:8137` | +| `--rpc` | `block-range` | `block-range` \| `latest-block` \| `block` | +| `--connections` | 64 | concurrent connections (each a spawned task) | +| `--conn-mode` | `per-task` | `per-task` (socket per client) or `shared` (one multiplexed channel) | +| `--range` | — | `START..END`; either side may be empty (`a..` = to tip, `..b` = from genesis) | +| `--tip-window` | 50000 | when `--range` is absent, sweep the last N blocks below the discovered tip | +| `--blocks` | 100 | blocks per `GetBlockRange` window | +| `--dist` | `even` | how windows spread across the pool (`even` \| `scatter`) | +| `--duration` / `--count` | 30s | run for D seconds, or N ops per connection | +| `--no-oracle` | off | disable the chain-link correctness oracle | +| `--json` | off | emit the JSON summary to stdout | + +## Output + +- **Human table → stderr** (via `ztest::loadtest::LoadReport::print`). +- **JSON summary → stdout** with `--json`: per-op `p50/p90/p99/p99.9/max` (ms), + throughput, total ops, error count, and any correctness violations. +- **Logs** — structured `tracing` (set `ZTEST_LOG=loadgen=info,ztest=info`). + +The **chain-link oracle** validates every response under load: blocks must link +(`prev_hash == prior.hash`), heights strictly increase, genesis is well-formed. +It adapts per RPC — it validates single blocks for `GetBlock` and is skipped for +the block-less `GetLatestBlock`. `0 violations` means every block served under +load was a correctly-linked chain segment, not just that the server was fast. + +## Reproducibility & caveats + +- The container image (`zingodevops/loadgen`) pins the exact binary; the committed + `k8s/job.yaml` pins the exact args. A run is `kubectl apply` + read the logs. +- Results are a **measurement, not a calibrated SLO**: absolute latency is only + trustworthy on a CPU-pinned, I/O-calibrated node. Trust the *shapes* (saturation, + tail knee, collapse) and the differential (backend/version A/B), not raw absolutes. +- Target zainod should run with a **real finalised state** (not + `ZAINO_EPHEMERAL_FINALISED_STATE`, which inflates read performance). diff --git a/crates/loadgen/k8s/job.yaml b/crates/loadgen/k8s/job.yaml new file mode 100644 index 0000000..673c5de --- /dev/null +++ b/crates/loadgen/k8s/job.yaml @@ -0,0 +1,52 @@ +# In-cluster load run: a plain Job that hits the LOCAL zaino service directly. +# +# Nothing here is target-specific — apply it into whichever namespace's zaino +# you want to test; the target is derived from the pod's own namespace, and the +# height window is auto-discovered from the chain tip: +# +# kubectl --context zingo-infra -n preview-070-rc1b apply -f job.yaml +# kubectl --context zingo-infra -n preview-070-rc1b logs -f job/loadgen +# +# The Job runs in the target namespace so in-cluster DNS resolves and no +# cross-namespace NetworkPolicy is in the way. +apiVersion: batch/v1 +kind: Job +metadata: + name: loadgen + labels: + app: loadgen +spec: + backoffLimit: 0 + ttlSecondsAfterFinished: 3600 + template: + metadata: + labels: + app: loadgen + spec: + restartPolicy: Never + containers: + - name: loadgen + image: zingodevops/loadgen:dev + imagePullPolicy: Always + env: + - name: POD_NAMESPACE + valueFrom: + fieldRef: + fieldPath: metadata.namespace + - name: ZTEST_LOG + value: loadgen=info,ztest=info + args: + - --target=http://zaino.$(POD_NAMESPACE).svc.cluster.local:8137 + - --connections=64 + - --tip-window=50000 + - --blocks=100 + - --conn-mode=per-task + - --duration=30 + - --json + resources: + requests: + cpu: "1" + memory: 512Mi + limits: + cpu: "2" + memory: 1Gi diff --git a/crates/loadgen/scripts/sweep.sh b/crates/loadgen/scripts/sweep.sh new file mode 100755 index 0000000..b0d2227 --- /dev/null +++ b/crates/loadgen/scripts/sweep.sh @@ -0,0 +1,66 @@ +#!/usr/bin/env bash +# Connection-count sweep for loadgen: run one Job per concurrency level, +# SEQUENTIALLY (parallel levels would load the server at once and confound the +# measurement), and collect one JSON line per level. +# +# scripts/sweep.sh [connections] [duration_s] [rpc] +# scripts/sweep.sh preview-070-rc1c "1 4 8 16 32 64 128" 20 block-range +# +# Requires: kubectl context already pointed at the target cluster, and a zaino +# service reachable at zaino..svc.cluster.local:8137. +set -euo pipefail + +NS="${1:?usage: sweep.sh [connections] [duration_s] [rpc]}" +CONNS="${2:-1 4 8 16 32 64 128}" +DUR="${3:-20}" +RPC="${4:-block-range}" +IMG="${LOADGEN_IMAGE:-docker.io/zingodevops/loadgen:dev}" +OUT="${OUT:-sweep-${NS}-${RPC}.jsonl}" + +: > "$OUT" +for c in $CONNS; do + name="loadgen-c${c}" + kubectl -n "$NS" delete job "$name" --ignore-not-found >/dev/null 2>&1 || true + cat </dev/null +apiVersion: batch/v1 +kind: Job +metadata: { name: $name, labels: { app: loadgen } } +spec: + backoffLimit: 0 + ttlSecondsAfterFinished: 1800 + template: + metadata: { labels: { app: loadgen } } + spec: + restartPolicy: Never + containers: + - name: loadgen + image: $IMG + imagePullPolicy: Always + env: + - { name: POD_NAMESPACE, valueFrom: { fieldRef: { fieldPath: metadata.namespace } } } + - { name: ZTEST_LOG, value: "loadgen=info,ztest=info" } + args: + - --target=http://zaino.$NS.svc.cluster.local:8137 + - --rpc=$RPC + - --connections=$c + - --tip-window=50000 + - --blocks=100 + - --conn-mode=per-task + - --duration=$DUR + - --json +EOF + kubectl -n "$NS" wait --for=condition=complete "job/$name" --timeout=300s >/dev/null 2>&1 \ + || kubectl -n "$NS" wait --for=condition=failed "job/$name" --timeout=5s >/dev/null 2>&1 || true + json=$(kubectl -n "$NS" logs "job/$name" 2>/dev/null | grep '^{' | head -1 || true) + if [ -n "$json" ]; then + echo "$json" >> "$OUT" + echo "conns=$c OK" + else + echo "conns=$c FAILED" + kubectl -n "$NS" logs "job/$name" 2>/dev/null | tail -3 || true + fi + kubectl -n "$NS" delete job "$name" --ignore-not-found >/dev/null 2>&1 || true +done + +echo "=== results in $OUT ===" +cat "$OUT" diff --git a/crates/loadgen/src/main.rs b/crates/loadgen/src/main.rs new file mode 100644 index 0000000..2f3a60d --- /dev/null +++ b/crates/loadgen/src/main.rs @@ -0,0 +1,337 @@ +//! Load generator for a *running* zainod: attaches to a gRPC endpoint rather +//! than spawning a topology, so one binary loads a regtest node on a laptop or +//! a mainnet node in-cluster and runs as a plain k8s Job. +//! +//! It emits a measurement, not a pass/fail SLO: absolute latency gating is only +//! trustworthy on a CPU-pinned, I/O-calibrated cluster, so treat the numbers as +//! observations and gate via the A/B differential path instead. + +use std::time::Duration; + +use anyhow::{Result, bail}; +use clap::{Parser, ValueEnum}; +use tracing::{info, warn}; +use ztest::loadtest::{ + ChainLinkOracle, ConnMode, Distribution, LoadDriver, LoadReport, LwdClient, Scenario, Until, +}; +use ztest::observ::{self, Sink}; + +/// Concurrency/load generator for zainod (gRPC CompactTxStreamer). +#[derive(Parser, Debug)] +#[command(name = "loadgen", version, about)] +struct Args { + /// Target zainod gRPC endpoint, e.g. `http://zaino.preview-070-rc1b.svc:8137`. + #[arg(long)] + target: String, + + /// Number of concurrent connections (each a spawned task). + #[arg(long, default_value_t = 64)] + connections: usize, + + /// Which RPC to load. `block-range` sweeps windows; `latest-block` polls the + /// tip (ignores --range/--blocks); `block` fetches single blocks (ignores --blocks). + #[arg(long, value_enum, default_value_t = RpcArg::BlockRange)] + rpc: RpcArg, + + /// Height range `START..END`; either side may be empty (`a..` = to tip, + /// `..b` = from genesis). Overrides tip auto-discovery. + #[arg(long)] + range: Option, + + /// Auto-discover the chain tip (GetLatestBlock) and sweep the last N blocks. + /// Used when --range is absent. + #[arg(long, default_value_t = 50_000)] + tip_window: u64, + + /// Blocks per `GetBlockRange` window each connection fetches. + #[arg(long, default_value_t = 100)] + blocks: u64, + + /// How windows are spread across the pool. + #[arg(long, value_enum, default_value_t = DistArg::Even)] + dist: DistArg, + + /// Connection model: one shared multiplexed channel, or a real socket per task. + #[arg(long, value_enum, default_value_t = ConnModeArg::PerTask)] + conn_mode: ConnModeArg, + + /// Run for this many seconds. Mutually exclusive with --count. + #[arg(long, group = "budget")] + duration: Option, + + /// Each connection performs this many ops, then stops. Mutually exclusive with --duration. + #[arg(long, group = "budget")] + count: Option, + + /// Stagger spawns by this many milliseconds to avoid a SYN burst. + #[arg(long, default_value_t = 1)] + spawn_stagger_ms: u64, + + /// Disable the chain-link correctness oracle (enabled by default). + #[arg(long)] + no_oracle: bool, + + /// Emit a machine-readable JSON summary to stdout (the human table always goes to stderr). + #[arg(long)] + json: bool, + + /// Label for the run, shown in the report. + #[arg(long, default_value = "loadgen")] + label: String, +} + +#[derive(Clone, Copy, Debug, ValueEnum)] +enum DistArg { + Even, + Scatter, +} + +#[derive(Clone, Copy, Debug, ValueEnum)] +enum ConnModeArg { + Shared, + PerTask, +} + +#[derive(Clone, Copy, Debug, ValueEnum)] +enum RpcArg { + BlockRange, + LatestBlock, + Block, +} + +/// A `--range` spec with Rust-style open bounds: `a..b`, `a..` (end = tip), +/// `..b` (start = genesis), `..` (whole chain). Open ends are filled by +/// [`RangeSpec::resolve`] once the tip is known. +#[derive(Clone, Copy, Debug)] +struct RangeSpec { + start: Option, + end: Option, +} + +impl RangeSpec { + fn needs_tip(&self) -> bool { + self.end.is_none() + } + + fn resolve(&self, tip: Option) -> Result { + let end = match self.end { + Some(e) => e, + None => tip.ok_or(RangeError::TipUnknown)?, + }; + HeightRange::new(self.start.unwrap_or(0), end) + } +} + +impl std::str::FromStr for RangeSpec { + type Err = RangeError; + + fn from_str(s: &str) -> Result { + let (start, end) = s.split_once("..").ok_or(RangeError::Format)?; + let bound = |x: &str| -> Result, RangeError> { + let x = x.trim(); + Ok(if x.is_empty() { None } else { Some(x.parse()?) }) + }; + Ok(Self { start: bound(start)?, end: bound(end)? }) + } +} + +/// A non-empty half-open height range `[start, end)`. Invalid ranges are +/// unrepresentable: the only constructor enforces `start < end`. +#[derive(Clone, Copy, Debug)] +struct HeightRange { + start: u64, + end: u64, +} + +impl HeightRange { + fn new(start: u64, end: u64) -> Result { + if end <= start { + return Err(RangeError::Empty { start, end }); + } + Ok(Self { start, end }) + } + + fn as_range(&self) -> std::ops::Range { + self.start..self.end + } +} + +#[derive(Debug, thiserror::Error)] +enum RangeError { + #[error("range must be `START..END` (either side may be empty)")] + Format, + #[error("invalid height: {0}")] + Height(#[from] std::num::ParseIntError), + #[error("empty range: {start}..{end}")] + Empty { start: u64, end: u64 }, + #[error("open-ended range needs the chain tip")] + TipUnknown, +} + +#[tokio::main(flavor = "multi_thread")] +async fn main() -> Result<()> { + observ::init(Sink::Stderr); + let args = Args::parse(); + + if args.blocks == 0 { + bail!("--blocks must be > 0"); + } + + let until = match (args.duration, args.count) { + (Some(_), Some(_)) => bail!("--duration and --count are mutually exclusive"), + (Some(secs), None) => Until::Duration(Duration::from_secs(secs)), + (None, Some(n)) => Until::CountPerConn(n), + (None, None) => Until::Duration(Duration::from_secs(30)), + }; + + let dist = match args.dist { + DistArg::Even => Distribution::Even, + DistArg::Scatter => Distribution::Scatter, + }; + let conn_mode = match args.conn_mode { + ConnModeArg::Shared => ConnMode::Shared, + ConnModeArg::PerTask => ConnMode::PerTask, + }; + + let client = LwdClient::connect(args.target.clone()) + .await + .map_err(|e| anyhow::anyhow!("connect to {}: {e}", args.target))?; + + let scenario = match args.rpc { + RpcArg::LatestBlock => Scenario::LatestBlockPoll, + RpcArg::Block => { + let range = resolve_range(&client, args.range, args.tip_window).await?; + info!(start = range.start, end = range.end, "resolved block pool"); + Scenario::BlockPoll { pool: range.as_range(), dist } + } + RpcArg::BlockRange => { + let range = resolve_range(&client, args.range, args.tip_window).await?; + info!(start = range.start, end = range.end, blocks = args.blocks, "resolved sweep window"); + Scenario::BlockRangeSweep { pool: range.as_range(), blocks: args.blocks, dist } + } + }; + + info!( + endpoint = %args.target, + rpc = ?args.rpc, + connections = args.connections, + ?conn_mode, + ?until, + oracle = !args.no_oracle, + "starting load run", + ); + + let mut driver = LoadDriver::new(client) + .label(args.label.clone()) + .connections(args.connections) + .conn_mode(conn_mode) + .spawn_stagger(Duration::from_millis(args.spawn_stagger_ms)) + .scenario(scenario) + .until(until); + if !args.no_oracle { + driver = driver.oracle(ChainLinkOracle); + } + + let report = driver + .run() + .await + .map_err(|e| anyhow::anyhow!("load run failed: {e}"))?; + + report.print(); + + if args.json { + println!("{}", report_to_json(&report)); + } + + // Measurement, not a gate: always exit 0. Correctness issues surface via the + // warning and the report's violation_count. + if report.errors > 0 || !report.violations.is_empty() { + warn!( + errors = report.errors, + violations = report.violations.len(), + "rpc errors / correctness violations", + ); + } + + Ok(()) +} + +async fn discover_tip(client: &LwdClient) -> Result { + let tip = client + .latest_height() + .await + .map_err(|e| anyhow::anyhow!("discover tip via GetLatestBlock: {e}"))?; + info!(tip, "discovered chain tip"); + Ok(tip) +} + +async fn resolve_range( + client: &LwdClient, + spec: Option, + tip_window: u64, +) -> Result { + match spec { + Some(spec) => { + let tip = if spec.needs_tip() { + Some(discover_tip(client).await?) + } else { + None + }; + Ok(spec.resolve(tip)?) + } + None => { + let tip = discover_tip(client).await?; + Ok(HeightRange::new(tip.saturating_sub(tip_window), tip)?) + } + } +} + +fn dur_ms(d: Duration) -> f64 { + d.as_secs_f64() * 1000.0 +} + +/// `LoadReport` doesn't derive `Serialize`, so hand-build the JSON. Latencies in ms. +fn report_to_json(r: &LoadReport) -> String { + let by_op: serde_json::Map = r + .by_op + .iter() + .map(|(op, s)| { + ( + op.to_string(), + serde_json::json!({ + "count": s.count, + "p50_ms": dur_ms(s.p50), + "p90_ms": dur_ms(s.p90), + "p99_ms": dur_ms(s.p99), + "p999_ms": dur_ms(s.p999), + "max_ms": dur_ms(s.max), + }), + ) + }) + .collect(); + + let violations: Vec = r + .violations + .iter() + .map(|v| { + serde_json::json!({ + "height": v.height, + "field": v.field, + "detail": v.detail, + }) + }) + .collect(); + + serde_json::json!({ + "label": r.label, + "connections": r.connections, + "total_ops": r.total_ops, + "errors": r.errors, + "throughput_ops_per_sec": r.throughput, + "wall_seconds": r.wall.as_secs_f64(), + "by_op": by_op, + "violation_count": r.violations.len(), + "violations": violations, + }) + .to_string() +} diff --git a/src/loadtest/driver.rs b/src/loadtest/driver.rs index bc7aa76..32ab7b4 100644 --- a/src/loadtest/driver.rs +++ b/src/loadtest/driver.rs @@ -18,7 +18,7 @@ use crate::EnvError; use crate::loadtest::client::LwdClient; use crate::loadtest::oracle::{FieldDiff, Observed, Oracle, Violation, diff_compact_block}; use crate::loadtest::report::{LatencyStats, LoadReport, OpKind, ParityRecord}; -use crate::loadtest::scenario::Scenario; +use crate::loadtest::scenario::{Op, Scenario}; use crate::proto::CompactBlock; /// Whether every connection shares one multiplexed channel, or dials its own. @@ -151,7 +151,6 @@ impl LoadDriver { pub async fn run(self) -> Result { let scenario = Arc::new(self.scenario); - let op_kind = scenario.op_kind(); let stop = StopCond::from(self.until); let wall = Instant::now(); @@ -161,7 +160,7 @@ impl LoadDriver { ConnMode::Shared => self.client.clone(), ConnMode::PerTask => self.client.dial().await?, }; - let (start, end) = scenario.range_for(i, self.connections); + let op = scenario.op(i, self.connections); let oracle = self.oracle.clone(); handles.push(tokio::spawn(async move { let mut tally = Tally::new(); @@ -169,10 +168,17 @@ impl LoadDriver { let mut done = stop.starter(); while !done.reached() { let t = Instant::now(); - match client.block_range(start, end).await { + let (result, window) = match op { + Op::LatestBlock => (client.latest_height().await.map(|_| Vec::new()), None), + Op::Block(h) => (client.block_at(h).await.map(|b| vec![b]), Some((h, h))), + Op::BlockRange(start, end) => { + (client.block_range(start, end).await, Some((start, end))) + } + }; + match result { Ok(blocks) => { - tally.record(op_kind, t.elapsed()); - if let Some(o) = &oracle { + tally.record(op.kind(), t.elapsed()); + if let (Some(o), Some((start, end))) = (&oracle, window) { for v in o.observe(&Observed { start, end, diff --git a/src/loadtest/scenario.rs b/src/loadtest/scenario.rs index 9f06a32..c348fa7 100644 --- a/src/loadtest/scenario.rs +++ b/src/loadtest/scenario.rs @@ -1,4 +1,4 @@ -//! L1 — what each virtual connection does. Kept deterministic: the range a +//! L1 — what each virtual connection does. Kept deterministic: the work a //! connection targets is a pure function of its index, so a run is reproducible //! without an RNG (a load test that can't be re-run doesn't help a developer //! bisect a regression). @@ -18,8 +18,7 @@ pub enum Distribution { Scatter, } -/// What a connection fetches. Extensible; today the block-range sweep covers the -/// three `zaino-admin` load modes. +/// What a connection fetches, repeatedly, for the life of the run. #[derive(Debug, Clone)] pub enum Scenario { /// Each connection repeatedly fetches a `blocks`-sized window somewhere in @@ -29,44 +28,91 @@ pub enum Scenario { blocks: u64, dist: Distribution, }, + /// Each connection repeatedly polls the chain tip (`GetLatestBlock`) — the + /// cheap unary path every wallet hammers. + LatestBlockPoll, + /// Each connection repeatedly fetches a single block (`GetBlock`) at a fixed + /// height in `pool`, positioned per [`Distribution`]. + BlockPoll { pool: Range, dist: Distribution }, +} + +/// The concrete RPC a connection issues, resolved from the [`Scenario`] and the +/// connection's index. Fixed per connection so the run stays reproducible. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub(crate) enum Op { + LatestBlock, + Block(u64), + BlockRange(u64, u64), +} + +impl Op { + pub(crate) fn kind(&self) -> OpKind { + match self { + Op::LatestBlock => OpKind::GetLatestBlock, + Op::Block(_) => OpKind::GetBlock, + Op::BlockRange(..) => OpKind::GetBlockRange, + } + } } impl Scenario { pub(crate) fn op_kind(&self) -> OpKind { match self { Scenario::BlockRangeSweep { .. } => OpKind::GetBlockRange, + Scenario::LatestBlockPoll => OpKind::GetLatestBlock, + Scenario::BlockPoll { .. } => OpKind::GetBlock, + } + } + + /// The op connection `index` of `count` issues. Windows may overlap when work + /// exceeds the pool — fine for load; the point is concurrent readers, not + /// disjoint coverage. + pub(crate) fn op(&self, index: usize, count: usize) -> Op { + match self { + Scenario::BlockRangeSweep { .. } => { + let (start, end) = self.range_for(index, count); + Op::BlockRange(start, end) + } + Scenario::LatestBlockPoll => Op::LatestBlock, + Scenario::BlockPoll { pool, dist } => { + let span = pool.end.saturating_sub(pool.start).saturating_sub(1); + Op::Block(pool.start + offset(span, *dist, index, count)) + } } } - /// The inclusive `(start, end)` range connection `index` of `count` targets. - /// Windows may overlap when `blocks * count` exceeds the pool — fine for - /// load; the point is concurrent readers, not disjoint coverage. + /// The inclusive `(start, end)` window for a [`Scenario::BlockRangeSweep`]. pub(crate) fn range_for(&self, index: usize, count: usize) -> (u64, u64) { match self { Scenario::BlockRangeSweep { pool, blocks, dist } => { let pool_size = pool.end.saturating_sub(pool.start); let span = pool_size.saturating_sub(*blocks); - let offset = match dist { - Distribution::Even => { - if count <= 1 || span == 0 { - 0 - } else { - let step = span as f64 / (count - 1) as f64; - (step * index as f64).round() as u64 - } - } - Distribution::Scatter => { - if span == 0 { - 0 - } else { - splitmix64(index as u64) % (span + 1) - } - } - }; - let start = pool.start + offset; + let start = pool.start + offset(span, *dist, index, count); let end = (start + blocks.saturating_sub(1)).min(pool.end.saturating_sub(1)); (start, end) } + _ => unreachable!("range_for is only defined for BlockRangeSweep"), + } + } +} + +/// The pool offset connection `index` of `count` targets, per [`Distribution`]. +fn offset(span: u64, dist: Distribution, index: usize, count: usize) -> u64 { + match dist { + Distribution::Even => { + if count <= 1 || span == 0 { + 0 + } else { + let step = span as f64 / (count - 1) as f64; + (step * index as f64).round() as u64 + } + } + Distribution::Scatter => { + if span == 0 { + 0 + } else { + splitmix64(index as u64) % (span + 1) + } } } } @@ -118,4 +164,19 @@ mod tests { let s = sweep(Distribution::Scatter); assert_eq!(s.range_for(7, 100), s.range_for(7, 100)); } + + #[test] + fn block_poll_picks_a_fixed_height_in_pool() { + let s = Scenario::BlockPoll { pool: 100..200, dist: Distribution::Even }; + assert_eq!(s.op(0, 10), Op::Block(100)); + match s.op(9, 10) { + Op::Block(h) => assert!((100..200).contains(&h)), + other => panic!("expected Op::Block, got {other:?}"), + } + } + + #[test] + fn latest_block_poll_is_paramless() { + assert_eq!(Scenario::LatestBlockPoll.op(3, 10), Op::LatestBlock); + } }