From aed4be8592bf512f3d345e28d3c76fe6b55e3134 Mon Sep 17 00:00:00 2001 From: Luke Tornquist Date: Thu, 29 Oct 2026 13:49:37 -0400 Subject: [PATCH] Add a scenario-driven relay load generator (#8636) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ## Summary Buzz has no way to load-test the relay as a whole. `wamp_bench` sends kind:9 messages from one identity to one channel, and `perf/relay_bus_scaling.py` only measures Redis fan-out. Neither can show what happens to a relay with many more users, channels, or deep threads. This PR adds `buzz-loadgen`, a load generator in `crates/buzz-test-client`: - `buzz-loadgen seed ` creates communities (through `POST /operator/communities`), channels, channel history, and deep threads. - `buzz-loadgen run ` drives simulated humans and agents through ramp steps (1x, 2x, ...) and prints p50/p90/p99 latency, errors, and rate-limit rejections per operation for each step. - Humans hold a WebSocket with live subscriptions, fetch history on connect, and send presence, typing, messages, thread replies, and reactions. Agents authenticate with a NIP-OA tag, poll HTTP `/query`, post through HTTP `/events`, and send encrypted observer frames to their owner. - Per-client rates come from the scenario. Set them to model a deployment, and a step of `users = 10` means 10x that deployment's population. - Every identity, host, and channel id comes from the scenario's `run_id`, so `--shard i/n` splits a population across machines with no coordination. - Each step measures from ramp start to hold end, so connect failures during the ramp count. A step aborts the run when an operation's error rate or p99 (delivery included) crosses the scenario's `[abort]` limits, when a relay `CLOSED`s a live subscription often enough, or when a client exits unexpectedly. The report names the limit that tripped. - Each connect plus authentication has a deadline (`target.connect_timeout_secs`), and shutdown is bounded, so a relay that stalls connections still produces a report and a non-zero exit. - `seed` exits non-zero unless the relay accepted every planned post. - `NostrWsConnection::connect_to_addr` opens the TCP connection to a given address while the URL still names the host. This lets a client inside the relay's cluster reach a community host through the relay's service address. - `wamp_bench` is deleted. Its paced single-channel load is a subset of a scenario. What got simpler: one load tool instead of two partial ones. The change touches only test tooling, the unit-test lane, and one additive method in `buzz-ws-client`. Relay code is unchanged. ### Related issue None found. ### Testing Smoke run on a Linux workstation against a local relay built from this branch, with its own Postgres, Redis, and MinIO containers: ```bash cargo run --release -p buzz-test-client --bin buzz-loadgen -- seed perf/scenarios/local-smoke.toml cargo run --release -p buzz-test-client --bin buzz-loadgen -- run perf/scenarios/local-smoke.toml ``` - `seed`: 710 of 710 posts accepted (10 channels, 200 history messages, 10 thread roots, 500 thread replies). - `run`: both steps (1x = 50 humans + 5 agents, then 2x) completed with 0 errors and 0 rate-limit rejections. Every operation recorded traffic: 55/55 connect/auth, 205/205 live subscriptions, 150 history fetches, presence/typing/observer frames, messages and reactions, HTTP `/query`, HTTP `/events`, and publish→delivery. Delivery p99 was 45 ms at 1x and 42 ms at 2x. - Failure paths are covered by unit tests: a listener that accepts TCP but never answers (connect deadline, then `run` aborts and returns), a fake `/events` server that rejects posts (`seed` fails), `CLOSED` subscriptions, and invalid scenarios. `just test-unit` (and its `scripts/run-tests.sh` fallback) now runs the new `loadgen::` unit tests. Before this change, no CI lane ran `buzz-test-client` lib tests. 🤖 Generated with [Claude Code](https://claude.com/claude-code) --------- Signed-off-by: Luke Tornquist Co-authored-by: Elrond <28d6302a099e5225b02c4155ac4236e4912603df2ab08dbfc2f4fef08ce598c8@buzz.block.builderlab.xyz> Co-authored-by: Claude Opus 5.5 --- Cargo.lock | 2 + Justfile | 3 + crates/buzz-test-client/Cargo.toml | 10 + crates/buzz-test-client/src/bin/loadgen.rs | 85 + crates/buzz-test-client/src/bin/wamp_bench.rs | 118 -- crates/buzz-test-client/src/lib.rs | 5 + .../buzz-test-client/src/loadgen/identity.rs | 113 ++ crates/buzz-test-client/src/loadgen/mod.rs | 18 + crates/buzz-test-client/src/loadgen/net.rs | 200 ++ crates/buzz-test-client/src/loadgen/run.rs | 1654 +++++++++++++++++ .../buzz-test-client/src/loadgen/scenario.rs | 499 +++++ crates/buzz-test-client/src/loadgen/seed.rs | 690 +++++++ crates/buzz-test-client/src/loadgen/stats.rs | 593 ++++++ crates/buzz-ws-client/src/connection.rs | 35 +- perf/LOADGEN.md | 125 ++ perf/scenarios/local-smoke.toml | 41 + scripts/run-tests.sh | 4 + 17 files changed, 4074 insertions(+), 121 deletions(-) create mode 100644 crates/buzz-test-client/src/bin/loadgen.rs delete mode 100644 crates/buzz-test-client/src/bin/wamp_bench.rs create mode 100644 crates/buzz-test-client/src/loadgen/identity.rs create mode 100644 crates/buzz-test-client/src/loadgen/mod.rs create mode 100644 crates/buzz-test-client/src/loadgen/net.rs create mode 100644 crates/buzz-test-client/src/loadgen/run.rs create mode 100644 crates/buzz-test-client/src/loadgen/scenario.rs create mode 100644 crates/buzz-test-client/src/loadgen/seed.rs create mode 100644 crates/buzz-test-client/src/loadgen/stats.rs create mode 100644 perf/LOADGEN.md create mode 100644 perf/scenarios/local-smoke.toml diff --git a/Cargo.lock b/Cargo.lock index 16ae718739..8af7b132d3 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1558,6 +1558,7 @@ dependencies = [ "buzz-sdk", "buzz-ws-client", "chrono", + "clap", "futures-util", "hex", "nostr 0.44.7", @@ -1572,6 +1573,7 @@ dependencies = [ "thiserror 2.0.18", "tokio", "tokio-tungstenite 0.29.0", + "toml", "tracing", "tracing-subscriber", "url", diff --git a/Justfile b/Justfile index 9ca2a64b2f..4c55a0116e 100644 --- a/Justfile +++ b/Justfile @@ -638,6 +638,9 @@ test-unit: # relay events and agent prompts. They are infra-free; ignored lifecycle # tests remain excluded and run in their dedicated integration lanes. cargo nextest run -p buzz-acp --lib + # Load generator scenario, identity, shard and stats unit tests; all + # infra-free. Keep scripts/run-tests.sh in step. + cargo nextest run -p buzz-test-client --lib -E 'test(/^loadgen::/)' else ./scripts/run-tests.sh unit fi diff --git a/crates/buzz-test-client/Cargo.toml b/crates/buzz-test-client/Cargo.toml index e495c16300..b5972249ca 100644 --- a/crates/buzz-test-client/Cargo.toml +++ b/crates/buzz-test-client/Cargo.toml @@ -23,6 +23,12 @@ thiserror = { workspace = true } uuid = { workspace = true } url = { workspace = true } rustls = "0.23" +base64 = { workspace = true } +clap = { version = "4", features = ["derive"] } +hex = { workspace = true } +reqwest = { workspace = true } +sha2 = { workspace = true } +toml = "1.0" [dev-dependencies] tracing-subscriber = { workspace = true } @@ -42,3 +48,7 @@ buzz-sdk = { workspace = true } [[bin]] name = "buzz-test-cli" path = "src/main.rs" + +[[bin]] +name = "buzz-loadgen" +path = "src/bin/loadgen.rs" diff --git a/crates/buzz-test-client/src/bin/loadgen.rs b/crates/buzz-test-client/src/bin/loadgen.rs new file mode 100644 index 0000000000..aa4e302f17 --- /dev/null +++ b/crates/buzz-test-client/src/bin/loadgen.rs @@ -0,0 +1,85 @@ +//! Scenario-driven relay load generator. See `perf/LOADGEN.md`. +//! +//! Usage: +//! buzz-loadgen seed +//! buzz-loadgen run [--shard i/n] [--report ] + +use std::path::PathBuf; +use std::sync::Arc; + +use buzz_test_client::loadgen::{net::Net, run, scenario::Scenario, seed, stats}; +use clap::{Parser, Subcommand}; + +#[derive(Parser)] +#[command( + name = "buzz-loadgen", + about = "Scenario-driven Buzz relay load generator" +)] +struct Cli { + #[command(subcommand)] + command: Command, +} + +#[derive(Subcommand)] +enum Command { + /// Create the scenario's communities, channels, history and deep threads. + Seed { scenario: PathBuf }, + /// Drive the simulated population through the scenario's ramp. + Run { + scenario: PathBuf, + /// Run only clients with global index g where g % n == i. + #[arg(long, default_value = "0/1", value_parser = run::Shard::parse)] + shard: run::Shard, + /// Also write the JSON report here. + #[arg(long)] + report: Option, + }, +} + +#[tokio::main] +async fn main() -> anyhow::Result<()> { + let _ = rustls::crypto::CryptoProvider::install_default( + rustls::crypto::aws_lc_rs::default_provider(), + ); + tracing_subscriber::fmt() + .with_env_filter(tracing_subscriber::EnvFilter::from_default_env()) + .with_writer(std::io::stderr) + .init(); + + let cli = Cli::parse(); + let path = match &cli.command { + Command::Seed { scenario } | Command::Run { scenario, .. } => scenario, + }; + let scenario = Scenario::parse(&std::fs::read_to_string(path)?) + .map_err(|e| anyhow::anyhow!("{}: {e}", path.display()))?; + let net = Net::new(&scenario).await?; + + match cli.command { + Command::Seed { .. } => { + let stats = Arc::new(stats::Stats::default()); + let started = std::time::Instant::now(); + let outcome = seed::seed(&scenario, &net, &stats).await; + let summary = stats.snapshot(); + let base = stats::Snapshot::default(); + let report = serde_json::json!({ + "run_id": scenario.run_id, + "ops": stats::ops_json(&summary.since(&base), started.elapsed().as_secs_f64()), + "failures": summary.reasons_since(&base, 10), + }); + println!("{report}"); + outcome + } + Command::Run { shard, report, .. } => { + let (outcome, value) = match run::run(scenario, net, shard).await { + Ok(value) => (Ok(()), value), + Err(value) => (Err(anyhow::anyhow!("ramp aborted; see report")), value), + }; + let text = serde_json::to_string_pretty(&value)?; + println!("{text}"); + if let Some(path) = report { + std::fs::write(path, text)?; + } + outcome + } + } +} diff --git a/crates/buzz-test-client/src/bin/wamp_bench.rs b/crates/buzz-test-client/src/bin/wamp_bench.rs deleted file mode 100644 index b1bcad2104..0000000000 --- a/crates/buzz-test-client/src/bin/wamp_bench.rs +++ /dev/null @@ -1,118 +0,0 @@ -//! Paced kind:9 load generator for relay write-amplification benchmarking. -//! -//! Opens `conns` authenticated WebSocket connections (one shared identity) -//! and sends kind:9 text events to `channel_uuid` at a total rate of `qps` -//! for `duration_secs`. Each connection is synchronous (send -> await OK), -//! paced by a per-connection tokio interval, so OK latency is measured -//! end-to-end. Emits a JSON summary on stdout and one raw latency sample -//! (milliseconds, f64) per line to `latency_out`. -//! -//! Usage: wamp-bench -//! Env: BUZZ_RELAY_URL (default ws://localhost:3000), BENCH_PRIVATE_KEY (hex) - -use std::time::{Duration, Instant}; - -use buzz_test_client::BuzzTestClient; -use nostr::Keys; -use tokio::time::MissedTickBehavior; - -#[tokio::main] -async fn main() -> anyhow::Result<()> { - let _ = rustls::crypto::CryptoProvider::install_default( - rustls::crypto::aws_lc_rs::default_provider(), - ); - let args: Vec = std::env::args().collect(); - if args.len() != 6 { - eprintln!("Usage: wamp-bench "); - std::process::exit(1); - } - let channel_id = args[1].clone(); - let qps: f64 = args[2].parse()?; - let duration_secs: u64 = args[3].parse()?; - let conns: usize = args[4].parse()?; - let latency_out = args[5].clone(); - - let url = std::env::var("BUZZ_RELAY_URL").unwrap_or_else(|_| "ws://localhost:3000".into()); - let keys = match std::env::var("BENCH_PRIVATE_KEY") { - Ok(hex) => Keys::parse(&hex)?, - Err(_) => anyhow::bail!("BENCH_PRIVATE_KEY is required (channel member secret key)"), - }; - - let per_conn_interval = Duration::from_secs_f64(conns as f64 / qps); - let deadline = Instant::now() + Duration::from_secs(duration_secs); - - let mut tasks = Vec::new(); - for conn_idx in 0..conns { - let url = url.clone(); - let keys = keys.clone(); - let channel_id = channel_id.clone(); - tasks.push(tokio::spawn(async move { - let mut client = BuzzTestClient::connect(&url, &keys).await?; - let mut interval = tokio::time::interval(per_conn_interval); - interval.set_missed_tick_behavior(MissedTickBehavior::Delay); - let mut latencies: Vec = Vec::new(); - let mut sent: u64 = 0; - let mut rejected: u64 = 0; - let mut seq: u64 = 0; - while Instant::now() < deadline { - interval.tick().await; - seq += 1; - let content = format!( - "wamp-bench c{conn_idx} m{seq} payload: the quick brown fox jumps over the lazy dog 0123456789" - ); - let start = Instant::now(); - let ok = client - .send_text_message(&keys, &channel_id, &content, 9) - .await?; - let elapsed_ms = start.elapsed().as_secs_f64() * 1e3; - sent += 1; - if ok.accepted { - latencies.push(elapsed_ms); - } else { - rejected += 1; - eprintln!("REJECTED conn={conn_idx} seq={seq}: {}", ok.message); - } - } - client.disconnect().await?; - Ok::<_, anyhow::Error>((sent, rejected, latencies)) - })); - } - - let mut sent = 0u64; - let mut rejected = 0u64; - let mut latencies: Vec = Vec::new(); - for task in tasks { - let (s, r, l) = task.await??; - sent += s; - rejected += r; - latencies.extend(l); - } - latencies.sort_by(|a, b| a.partial_cmp(b).expect("finite latencies")); - let pct = |p: f64| -> f64 { - if latencies.is_empty() { - return f64::NAN; - } - let idx = ((latencies.len() as f64 - 1.0) * p).round() as usize; - latencies[idx] - }; - let raw: String = latencies.iter().map(|l| format!("{l:.3}\n")).collect(); - std::fs::write(&latency_out, raw)?; - println!( - "{}", - serde_json::json!({ - "sent": sent, - "accepted": sent - rejected, - "rejected": rejected, - "qps_target": qps, - "duration_secs": duration_secs, - "conns": conns, - "ok_latency_ms": { - "p50": pct(0.50), - "p95": pct(0.95), - "p99": pct(0.99), - "max": latencies.last().copied().unwrap_or(f64::NAN), - }, - }) - ); - Ok(()) -} diff --git a/crates/buzz-test-client/src/lib.rs b/crates/buzz-test-client/src/lib.rs index 89bff3020d..eb64663b9d 100644 --- a/crates/buzz-test-client/src/lib.rs +++ b/crates/buzz-test-client/src/lib.rs @@ -13,6 +13,11 @@ use tracing::debug; use buzz_ws_client::NostrWsConnection; pub use buzz_ws_client::{parse_relay_message, OkResponse, RelayMessage, WsClientError}; +/// Scenario-driven load generator behind the `buzz-loadgen` binary. +/// The scenario schema is documented in `perf/LOADGEN.md`. +#[allow(missing_docs)] +pub mod loadgen; + /// Errors returned by [`BuzzTestClient`] operations. #[derive(Debug, Error)] pub enum TestClientError { diff --git a/crates/buzz-test-client/src/loadgen/identity.rs b/crates/buzz-test-client/src/loadgen/identity.rs new file mode 100644 index 0000000000..8dfbc1ced8 --- /dev/null +++ b/crates/buzz-test-client/src/loadgen/identity.rs @@ -0,0 +1,113 @@ +//! Deterministic identities and fixture ids derived from the run id. + +use base64::engine::general_purpose::STANDARD as B64; +use base64::Engine as _; +use nostr::secp256k1::Message; +use nostr::{EventBuilder, JsonUtil, Keys, Kind, SecretKey, Tag}; +use sha2::{Digest, Sha256}; + +/// Role of a derived identity; part of the derivation so roles never collide. +#[derive(Clone, Copy, Debug)] +pub enum Role { + Owner, + Human, + Agent, +} + +impl Role { + fn label(self) -> &'static str { + match self { + Role::Owner => "owner", + Role::Human => "human", + Role::Agent => "agent", + } + } +} + +fn digest(parts: &[&str]) -> [u8; 32] { + let mut hasher = Sha256::new(); + for part in parts { + hasher.update(part.as_bytes()); + hasher.update([0u8]); + } + hasher.finalize().into() +} + +/// Keys for identity `index` of `role` in run `run_id`. +pub fn keys(run_id: &str, role: Role, index: usize) -> Keys { + let mut counter = 0u32; + loop { + let seed = digest(&[ + "buzz-loadgen", + run_id, + role.label(), + &index.to_string(), + &counter.to_string(), + ]); + // A digest outside the curve order is astronomically unlikely; retry anyway. + if let Ok(secret) = SecretKey::from_slice(&seed) { + return Keys::new(secret); + } + counter += 1; + } +} + +/// Channel UUID for channel `index` of community `community`. +pub fn channel_id(run_id: &str, community: usize, index: usize) -> String { + let seed = digest(&[ + "buzz-loadgen-channel", + run_id, + &community.to_string(), + &index.to_string(), + ]); + let mut bytes = [0u8; 16]; + bytes.copy_from_slice(&seed[..16]); + uuid::Builder::from_random_bytes(bytes) + .into_uuid() + .to_string() +} + +/// NIP-OA `auth` tag in which `owner` authorizes `agent`. +pub fn nip_oa_tag(owner: &Keys, agent: &Keys) -> Tag { + let preimage = format!("nostr:agent-auth:{}:", agent.public_key().to_hex()); + let digest: [u8; 32] = Sha256::digest(preimage.as_bytes()).into(); + let signature = owner.sign_schnorr(&Message::from_digest(digest)); + Tag::parse([ + "auth".to_string(), + owner.public_key().to_hex(), + String::new(), + signature.to_string(), + ]) + .expect("auth tag is well-formed") +} + +/// NIP-98 `Authorization` header value for one request. +pub fn nip98_header(keys: &Keys, method: &str, url: &str, body: &[u8]) -> String { + let tags = [ + Tag::parse(["u", url]).expect("u tag"), + Tag::parse(["method", method]).expect("method tag"), + Tag::parse(["nonce", &uuid::Uuid::new_v4().to_string()]).expect("nonce tag"), + Tag::parse(["payload", &hex::encode(Sha256::digest(body))]).expect("payload tag"), + ]; + let event = EventBuilder::new(Kind::Custom(27235), "") + .tags(tags) + .sign_with_keys(keys) + .expect("sign NIP-98 event"); + format!("Nostr {}", B64.encode(event.as_json())) +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn derivation_is_deterministic_and_role_scoped() { + let a = keys("run", Role::Human, 7); + assert_eq!(a.public_key(), keys("run", Role::Human, 7).public_key()); + assert_ne!(a.public_key(), keys("run", Role::Agent, 7).public_key()); + assert_ne!(a.public_key(), keys("other", Role::Human, 7).public_key()); + assert_eq!(channel_id("run", 0, 3), channel_id("run", 0, 3)); + assert_ne!(channel_id("run", 0, 3), channel_id("run", 1, 3)); + assert!(uuid::Uuid::parse_str(&channel_id("run", 0, 0)).is_ok()); + } +} diff --git a/crates/buzz-test-client/src/loadgen/mod.rs b/crates/buzz-test-client/src/loadgen/mod.rs new file mode 100644 index 0000000000..4e8bf8c17a --- /dev/null +++ b/crates/buzz-test-client/src/loadgen/mod.rs @@ -0,0 +1,18 @@ +//! Scenario-driven relay load generator. +//! +//! A scenario (TOML, see `perf/LOADGEN.md`) describes the data shape to seed +//! and a population of simulated humans and agents with per-connection rates. +//! `seed` creates the communities, +//! channels, history and deep threads; `run` drives the population through a +//! ramp of user multipliers and reports latency percentiles per operation. +//! +//! Every identity, community host and channel id is derived from the +//! scenario's `run_id`, so independent shards (`--shard i/n`) agree on the +//! fixture without sharing state. + +pub mod identity; +pub mod net; +pub mod run; +pub mod scenario; +pub mod seed; +pub mod stats; diff --git a/crates/buzz-test-client/src/loadgen/net.rs b/crates/buzz-test-client/src/loadgen/net.rs new file mode 100644 index 0000000000..0c0008674c --- /dev/null +++ b/crates/buzz-test-client/src/loadgen/net.rs @@ -0,0 +1,200 @@ +//! Connection helpers that honor the scenario's `connect_addr` override. + +use std::net::SocketAddr; +use std::time::Duration; + +use buzz_ws_client::NostrWsConnection; +use nostr::{Event, JsonUtil, Keys, Tag}; + +use super::identity; +use super::scenario::{Community, Scenario}; + +/// Shared endpoints for one shard: an HTTP client whose community hosts +/// resolve to `connect_addr` when it is set. +#[derive(Clone)] +pub struct Net { + pub http: reqwest::Client, + connect_addr: Option, + connect_timeout: Duration, +} + +impl Net { + pub async fn new(scenario: &Scenario) -> anyhow::Result { + let mut builder = reqwest::Client::builder() + .timeout(Duration::from_secs(60)) + .pool_max_idle_per_host(1024); + if let Some(addr) = &scenario.target.connect_addr { + let resolved: SocketAddr = tokio::net::lookup_host(addr) + .await? + .next() + .ok_or_else(|| anyhow::anyhow!("connect_addr {addr} did not resolve"))?; + for community in scenario.communities() { + if let Some(host) = &community.host { + builder = builder.resolve(host, resolved); + } + } + } + Ok(Self { + http: builder.build()?, + connect_addr: scenario.target.connect_addr.clone(), + connect_timeout: Duration::from_secs_f64(scenario.target.connect_timeout_secs), + }) + } + + /// Opens a WebSocket to `community` and authenticates as `keys`, failing + /// once the whole attempt exceeds `target.connect_timeout_secs`. + pub async fn connect( + &self, + community: &Community, + keys: &Keys, + auth_tag: Option<&Tag>, + ) -> anyhow::Result { + let attempt = async { + let mut conn = match (&self.connect_addr, &community.host) { + (Some(addr), Some(_)) => { + NostrWsConnection::connect_to_addr(&community.ws_url, addr).await? + } + _ => NostrWsConnection::connect(&community.ws_url).await?, + }; + conn.authenticate(keys, auth_tag).await?; + anyhow::Ok(conn) + }; + tokio::time::timeout(self.connect_timeout, attempt) + .await + .map_err(|_| anyhow::anyhow!("connect timeout after {:?}", self.connect_timeout))? + } + + /// `POST {base}{path}` with a NIP-98 header; returns status and body. + pub async fn post( + &self, + keys: &Keys, + base: &str, + path: &str, + body: Vec, + auth_tag: Option<&Tag>, + ) -> anyhow::Result<(reqwest::StatusCode, String)> { + let url = format!("{base}{path}"); + let mut request = self + .http + .post(&url) + .header( + "Authorization", + identity::nip98_header(keys, "POST", &url, &body), + ) + .header("Content-Type", "application/json"); + if let Some(tag) = auth_tag { + request = request.header("x-auth-tag", serde_json::to_string(tag.as_slice())?); + } + let response = request.body(body).send().await?; + let status = response.status(); + Ok((status, response.text().await?)) + } + + /// `GET {url}` with a NIP-98 header; returns status and body. + pub async fn get( + &self, + keys: &Keys, + url: &str, + ) -> anyhow::Result<(reqwest::StatusCode, String)> { + let response = self + .http + .get(url) + .header( + "Authorization", + identity::nip98_header(keys, "GET", url, &[]), + ) + .send() + .await?; + let status = response.status(); + Ok((status, response.text().await?)) + } + + /// Submits `event` through HTTP `/events`; errors unless the relay accepted it. + pub async fn submit( + &self, + keys: &Keys, + community: &Community, + event: &Event, + auth_tag: Option<&Tag>, + ) -> anyhow::Result<()> { + let (status, body) = self + .post( + keys, + &community.http_url, + "/events", + event.as_json().into_bytes(), + auth_tag, + ) + .await?; + if !status.is_success() { + anyhow::bail!("{}", http_reason(status, &body)); + } + let value: serde_json::Value = serde_json::from_str(&body)?; + if value["accepted"].as_bool() != Some(true) { + anyhow::bail!("rejected: {}", value["message"].as_str().unwrap_or(&body)); + } + Ok(()) + } +} + +pub fn truncate(s: &str) -> String { + s.chars().take(200).collect() +} + +/// Failure text for a non-success HTTP response. It starts `HTTP `, +/// which `stats` reads to tell a 429 from every other failure. +pub fn http_reason(status: reqwest::StatusCode, body: &str) -> String { + format!("HTTP {status}: {}", truncate(body)) +} + +#[cfg(test)] +mod tests { + use super::*; + + #[tokio::test] + async fn stalled_upgrade_fails_at_the_connect_deadline() { + // A peer that accepts TCP but never answers the WebSocket upgrade. + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let addr = listener.local_addr().unwrap().to_string(); + let held = tokio::spawn(async move { + let mut open = Vec::new(); + while let Ok((socket, _)) = listener.accept().await { + open.push(socket); + } + }); + let scenario = Scenario::parse(&format!( + r#" +run_id = "t" +[target] +ws_url = "ws://{addr}" +http_url = "http://{addr}" +connect_timeout_secs = 0.3 +[shape] +communities = 1 +channels_per_community = 1 +[population] +humans = 1 +agents = 0 +channels_per_member = 1 +[[ramp]] +users = 1.0 +hold_secs = 1 +"# + )) + .unwrap(); + let net = Net::new(&scenario).await.unwrap(); + let community = &scenario.communities()[0]; + let started = std::time::Instant::now(); + let err = tokio::time::timeout( + Duration::from_secs(5), + net.connect(community, &Keys::generate(), None), + ) + .await + .expect("connect returns at its own deadline") + .err() + .expect("a silent peer cannot complete the upgrade"); + assert!(err.to_string().contains("connect timeout"), "{err}"); + assert!(started.elapsed() < Duration::from_secs(2)); + held.abort(); + } +} diff --git a/crates/buzz-test-client/src/loadgen/run.rs b/crates/buzz-test-client/src/loadgen/run.rs new file mode 100644 index 0000000000..ac10d3d845 --- /dev/null +++ b/crates/buzz-test-client/src/loadgen/run.rs @@ -0,0 +1,1654 @@ +//! Drives the simulated population through the scenario's ramp. + +use std::collections::HashMap; +use std::sync::Arc; +use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH}; + +use buzz_core::kind::{ + KIND_AGENT_OBSERVER_FRAME, KIND_MEMBER_ADDED_NOTIFICATION, KIND_MEMBER_REMOVED_NOTIFICATION, + KIND_PRESENCE_UPDATE, KIND_REACTION, KIND_STREAM_MESSAGE, KIND_TYPING_INDICATOR, +}; +use buzz_core::observer::{ + encrypt_observer_payload, OBSERVER_AGENT_TAG, OBSERVER_FRAME_TAG, OBSERVER_FRAME_TELEMETRY, +}; +use buzz_ws_client::{NostrWsConnection, RelayMessage, WsClientError}; +use nostr::{Event, EventBuilder, Filter, Keys, Kind, PublicKey, Tag, Timestamp}; +use serde_json::{json, Value}; +use tokio::sync::{watch, Mutex}; + +use super::identity::{self, Role}; +use super::net::{http_reason, Net}; +use super::scenario::{Community, Scenario}; +use super::seed::find_roots; +use super::stats::{abort_reason, ops_json, Op, Stats}; + +/// Content prefix of timed messages: `lg:`. +const TIMED_PREFIX: &str = "lg:"; +/// Longest an idle client blocks on its socket before rechecking the stop flag. +const MAX_IDLE: Duration = Duration::from_secs(1); +/// How long stopped clients get to settle what they have in flight and +/// disconnect before they are cancelled. +const SHUTDOWN_GRACE: Duration = Duration::from_secs(15); +/// Live subscriptions every human opens after connecting. +const HUMAN_SUBS: [&str; 4] = ["live", "mentions", "typing", "presence"]; +/// Prefix of the finite history subscriptions a human opens and closes. +const HISTORY_PREFIX: &str = "history-"; + +/// `i/n`: this process runs every global client index `g` with `g % n == i`. +#[derive(Clone, Copy, Debug)] +pub struct Shard { + pub index: usize, + pub count: usize, +} + +impl Shard { + pub fn parse(text: &str) -> Result { + let (i, n) = text.split_once('/').ok_or("shard must be i/n")?; + let index: usize = i.parse().map_err(|_| "shard index must be a number")?; + let count: usize = n.parse().map_err(|_| "shard count must be a number")?; + if count == 0 || index >= count { + return Err("shard needs 0 <= i < n".into()); + } + Ok(Self { index, count }) + } + + fn owns(&self, global: usize) -> bool { + global % self.count == self.index + } +} + +struct Shared { + scenario: Scenario, + communities: Vec, + net: Net, + stats: Arc, + stop: watch::Receiver, + /// Seeded thread roots per channel id, resolved lazily by whichever client needs them first. + roots: Mutex>>>, +} + +/// Runs every ramp step and returns the JSON report; `Err` when a step aborted. +pub async fn run(scenario: Scenario, net: Net, shard: Shard) -> Result { + run_with(scenario, net, shard, SHUTDOWN_GRACE).await +} + +/// [`run`] with the shutdown grace period given, so tests need not wait out +/// [`SHUTDOWN_GRACE`] for a client that never settles. +async fn run_with( + scenario: Scenario, + net: Net, + shard: Shard, + grace: Duration, +) -> Result { + let stats = Arc::new(Stats::default()); + let (stop_tx, stop) = watch::channel(false); + let shared = Arc::new(Shared { + communities: scenario.communities(), + scenario, + net, + stats: stats.clone(), + stop, + roots: Mutex::new(HashMap::new()), + }); + let scenario = &shared.scenario; + let pop = &scenario.population; + let connect_gap = Duration::from_secs_f64(1.0 / pop.connect_rate_per_sec); + + let mut tasks = Vec::new(); + let (mut humans, mut agents) = (0usize, 0usize); + let mut steps = Vec::new(); + let mut aborted = false; + // End of the last reported window and start of the last judged one. + let mut reported = stats.snapshot(); + let mut last_window = reported.clone(); + + for (n, step) in scenario.ramp.iter().enumerate() { + let want_humans = (pop.humans as f64 * step.users).ceil() as usize; + let want_agents = (pop.agents as f64 * step.users).ceil() as usize; + // The window covers the ramp too, so connect/auth and history + // failures from new clients count toward the abort check. + let before = stats.snapshot(); + last_window = before.clone(); + let ramp_start = Instant::now(); + while humans < want_humans || agents < want_agents { + // Interleave so agents and humans arrive in proportion. + let spawn_agent = agents < want_agents + && (humans >= want_humans || agents * want_humans <= humans * want_agents); + let global = if spawn_agent { agents } else { humans }; + if shard.owns(global) { + let shared = shared.clone(); + tasks.push(tokio::spawn(async move { + if spawn_agent { + agent(shared, global).await + } else { + human(shared, global).await + } + })); + tokio::time::sleep(connect_gap).await; + } + if spawn_agent { + agents += 1; + } else { + humans += 1; + } + } + let ramp_secs = ramp_start.elapsed().as_secs_f64(); + + let hold_start = Instant::now(); + tokio::time::sleep(Duration::from_secs(step.hold_secs)).await; + let hold_secs = hold_start.elapsed().as_secs_f64(); + let after = stats.snapshot(); + let summaries = after.since(&before); + + // Clients only return once stopped, so a finished one panicked. + let lost = tasks.iter().filter(|t| t.is_finished()).count(); + let reason = if lost > 0 { + Some(format!("{lost} clients exited unexpectedly")) + } else { + abort_reason( + &summaries, + scenario.abort.max_error_rate, + scenario.abort.max_p99_ms, + ) + }; + let report = json!({ + "step": n, + "users": step.users, + "humans": want_humans, + "agents": want_agents, + "shard": format!("{}/{}", shard.index, shard.count), + "ramp_secs": (ramp_secs * 10.0).round() / 10.0, + "hold_secs": (hold_secs * 10.0).round() / 10.0, + "ops": ops_json(&summaries, ramp_secs + hold_secs), + "failures": after + .reasons_since(&before, 10) + .into_iter() + .map(|(reason, count)| json!({"reason": reason, "count": count})) + .collect::>(), + "aborted": reason.is_some(), + "abort_reason": reason, + }); + eprintln!("{report}"); + steps.push(report); + reported = after; + if reason.is_some() { + aborted = true; + break; + } + } + + let _ = stop_tx.send(true); + let shutdown_start = Instant::now(); + let (panicked, cancelled) = shutdown(tasks, grace).await; + // Operations still in flight at the stop settle now, after the last step + // was judged: a stalled connect as a connect timeout, a cancelled client + // as `unfinished at stop`. They are judged together with the last step's + // window, not alone: work in flight at an instant is skewed toward slow + // operations, so on its own it would fail healthy runs on p99. + let closing = stats.snapshot(); + let closed = closing.since(&reported); + let shutdown_reason = abort_reason( + &closing.since(&last_window), + shared.scenario.abort.max_error_rate, + shared.scenario.abort.max_p99_ms, + ); + let report = json!({ + "run_id": shared.scenario.run_id, + "steps": steps, + "shutdown": { + "secs": (shutdown_start.elapsed().as_secs_f64() * 10.0).round() / 10.0, + "ops": ops_json(&closed, shutdown_start.elapsed().as_secs_f64()), + "failures": closing + .reasons_since(&reported, 10) + .into_iter() + .map(|(reason, count)| json!({"reason": reason, "count": count})) + .collect::>(), + "abort_reason": shutdown_reason, + }, + "clients_panicked": panicked, + "clients_cancelled": cancelled, + }); + if failed(aborted, panicked, cancelled, shutdown_reason.as_deref()) { + Err(report) + } else { + Ok(report) + } +} + +/// A run fails when a step aborted, a client panicked, shutdown had to cancel +/// a client, or the operations that settled during shutdown crossed the abort +/// limits. +fn failed(aborted: bool, panicked: usize, cancelled: usize, shutdown: Option<&str>) -> bool { + aborted || panicked > 0 || cancelled > 0 || shutdown.is_some() +} + +/// Waits up to `grace` for stopped clients, then cancels the rest and waits +/// for their futures to drop. Returns how many panicked and how many were +/// cancelled. +async fn shutdown(tasks: Vec>, grace: Duration) -> (usize, usize) { + let deadline = tokio::time::Instant::now() + grace; + let (mut panicked, mut cancelled) = (0, 0); + for mut task in tasks { + match tokio::time::timeout_at(deadline, &mut task).await { + Ok(Ok(())) => {} + Ok(Err(e)) => panicked += usize::from(e.is_panic()), + Err(_) => { + task.abort(); + // Dropping the future records what the client left in flight. + let _ = task.await; + cancelled += 1; + } + } + } + (panicked, cancelled) +} + +/// Small deterministic PRNG so every client draws an independent schedule. +struct Rng(u64); + +impl Rng { + fn new(seed: u64) -> Self { + Self(seed ^ 0x9E37_79B9_7F4A_7C15) + } + + fn next_u64(&mut self) -> u64 { + self.0 = self.0.wrapping_add(0x9E37_79B9_7F4A_7C15); + let mut z = self.0; + z = (z ^ (z >> 30)).wrapping_mul(0xBF58_476D_1CE4_E5B9); + z = (z ^ (z >> 27)).wrapping_mul(0x94D0_49BB_1331_11EB); + z ^ (z >> 31) + } + + fn unit(&mut self) -> f64 { + (self.next_u64() >> 11) as f64 / (1u64 << 53) as f64 + } + + fn below(&mut self, n: usize) -> usize { + (self.next_u64() % n.max(1) as u64) as usize + } + + /// Next arrival of a Poisson process with `per_hour` events per hour. + fn after(&mut self, per_hour: f64) -> Option { + if per_hour <= 0.0 { + return None; + } + let secs = -(1.0 - self.unit()).ln() * 3600.0 / per_hour; + Some(Instant::now() + Duration::from_secs_f64(secs)) + } +} + +fn now_micros() -> u128 { + SystemTime::now() + .duration_since(UNIX_EPOCH) + .unwrap_or_default() + .as_micros() +} + +fn tag(parts: [&str; N]) -> Tag { + Tag::parse(parts).expect("static tag shape") +} + +fn kind(k: u32) -> Kind { + Kind::Custom(k as u16) +} + +fn stopped(shared: &Shared) -> bool { + *shared.stop.borrow() +} + +/// Resolves once the run is stopping. +async fn stopping(shared: &Shared) { + let mut stop = shared.stop.clone(); + let _ = stop.wait_for(|s| *s).await; +} + +/// Jittered exponential backoff between failed connects or failed sessions. +struct Backoff(f64); + +impl Backoff { + const MAX_SECS: f64 = 30.0; + + fn new() -> Self { + Self(1.0) + } + + /// Sleeps for the next interval, returning early when the run stops. + async fn wait(&mut self, shared: &Shared, rng: &mut Rng) { + let wait = Duration::from_secs_f64(self.0 * (0.5 + rng.unit())); + self.0 = (self.0 * 2.0).min(Self::MAX_SECS); + tokio::select! { + _ = tokio::time::sleep(wait) => {} + _ = stopping(shared) => {} + } + } +} + +/// A stream message a client can react to: its id and channel. +#[derive(Clone)] +struct Seen { + id: String, + channel: String, +} + +fn seen(message: &RelayMessage) -> Option { + let RelayMessage::Event { event, .. } = message else { + return None; + }; + if event.kind != kind(KIND_STREAM_MESSAGE) { + return None; + } + let channel = event + .tags + .iter() + .find(|t| t.kind().as_str() == "h") + .and_then(|t| t.content())?; + Some(Seen { + id: event.id.to_hex(), + channel: channel.to_string(), + }) +} + +/// Handles a message for a live subscription. `EOSE` completes a subscribe +/// and `CLOSED` is a rejection that ends the session. Events record delivery +/// latency and become reaction targets. Late messages for a history +/// subscription, including the relay's `CLOSED` reply to our `CLOSE`, are +/// ignored. +fn live( + stats: &Stats, + message: &RelayMessage, + subscribed: Instant, + last_message: &mut Option, +) -> Result<(), String> { + if subscription_of(message).is_some_and(|id| id.starts_with(HISTORY_PREFIX)) { + return Ok(()); + } + match message { + RelayMessage::Eose { .. } => stats.ok(Op::Subscribe, subscribed.elapsed()), + RelayMessage::Closed { + subscription_id, + message, + } => { + let reason = format!("CLOSED {subscription_id}: {message}"); + stats.err(Op::Subscribe, &reason); + return Err(reason); + } + _ => { + if let Some(target) = observe(stats, message) { + *last_message = Some(target); + } + } + } + Ok(()) +} + +/// Records end-to-end delivery latency for a live timed message and returns +/// it as a reaction target. Only live subscriptions call this: history +/// replays old messages, whose age is not delivery latency. +fn observe(stats: &Stats, message: &RelayMessage) -> Option { + if let RelayMessage::Event { event, .. } = message { + if let Some(sent) = event + .content + .strip_prefix(TIMED_PREFIX) + .and_then(|rest| rest.split_whitespace().next()) + .and_then(|micros| micros.parse::().ok()) + { + let micros = now_micros().saturating_sub(sent).min(u64::MAX as u128) as u64; + stats.ok(Op::Deliver, Duration::from_micros(micros)); + } + } + seen(message) +} + +/// Publishes over the socket and records send→OK latency. +async fn publish( + stats: &Stats, + conn: &mut NostrWsConnection, + op: Op, + event: Event, +) -> Result<(), WsClientError> { + let pending = stats.begin(op); + match conn.send_event(event).await { + Ok(ok) if ok.accepted => pending.ok(), + Ok(ok) => pending.err(&ok.message), + Err(WsClientError::Timeout) => pending.err("OK timeout"), + Err(e) => { + // The caller records the broken socket. + pending.forget(); + return Err(e); + } + } + Ok(()) +} + +/// Connects with retry until connected or stopped. Each attempt is bounded +/// by the connect deadline, which should be shorter than [`SHUTDOWN_GRACE`] +/// so an attempt under way at the stop settles before clients are cancelled. +async fn connect( + shared: &Shared, + community: &Community, + keys: &Keys, + auth: Option<&Tag>, + rng: &mut Rng, + backoff: &mut Backoff, +) -> Option { + while !stopped(shared) { + // An attempt already under way runs to its connect deadline even when + // the run stops, so it settles as a success or an error rather than + // vanishing; the loop then ends instead of starting another. + let pending = shared.stats.begin(Op::WsConnect); + match shared.net.connect(community, keys, auth).await { + Ok(conn) => { + pending.ok(); + // Shutdown drains work already in flight; it does not start + // a session for a connect that finished after the stop. + if stopped(shared) { + let _ = conn.disconnect().await; + return None; + } + return Some(conn); + } + Err(e) => pending.err(&e.to_string()), + } + backoff.wait(shared, rng).await; + } + None +} + +/// Thread roots of `channel`, cached once found. Concurrent first callers may +/// each query; a failed query is not cached, so a later reply retries it. +async fn roots( + shared: &Shared, + community: &Community, + reader: &Keys, + channel: &str, +) -> Arc> { + if let Some(found) = shared.roots.lock().await.get(channel) { + return found.clone(); + } + let owner = identity::keys(&shared.scenario.run_id, Role::Owner, community.index); + let pending = shared.stats.begin(Op::HttpQuery); + let found = match find_roots(&shared.net, community, reader, &owner, channel).await { + Ok(found) => { + pending.ok(); + found + } + Err(e) => { + pending.err(&e.to_string()); + return Arc::new(Vec::new()); + } + }; + let found = Arc::new(found); + shared + .roots + .lock() + .await + .insert(channel.to_string(), found.clone()); + found +} + +/// A human client: live subscriptions, history on connect, presence +/// heartbeats, typing, occasional messages, thread replies and reactions. +async fn human(shared: Arc, global: usize) { + let scenario = &shared.scenario; + let profile = &scenario.human; + let community = &shared.communities[global % shared.communities.len()]; + let keys = identity::keys(&scenario.run_id, Role::Human, global); + let me = keys.public_key().to_hex(); + let channels = scenario.member_channels(community.index, global); + let peers: Vec = (1..=10) + .map(|k| { + let peer = (global + k * shared.communities.len()) % scenario.population.humans.max(1); + identity::keys(&scenario.run_id, Role::Human, peer).public_key() + }) + .collect(); + let mut rng = Rng::new(global as u64); + let mut backoff = Backoff::new(); + let mut last_message: Option = None; + + while let Some(mut conn) = + connect(&shared, community, &keys, None, &mut rng, &mut backoff).await + { + let since = Timestamp::now(); + let subscribed = Instant::now(); + let subs = [ + ( + HUMAN_SUBS[0], + Filter::new() + .kinds([ + kind(KIND_STREAM_MESSAGE), + kind(KIND_REACTION), + Kind::Custom(5), + Kind::Custom(40003), + ]) + .custom_tags( + nostr::SingleLetterTag::lowercase(nostr::Alphabet::H), + channels.clone(), + ) + .since(since), + ), + ( + HUMAN_SUBS[1], + human_mentions_filter(keys.public_key(), since), + ), + ( + HUMAN_SUBS[2], + Filter::new().kind(kind(KIND_TYPING_INDICATOR)).custom_tags( + nostr::SingleLetterTag::lowercase(nostr::Alphabet::H), + channels.clone(), + ), + ), + ( + HUMAN_SUBS[3], + Filter::new() + .kind(kind(KIND_PRESENCE_UPDATE)) + .authors(peers.clone()), + ), + ]; + let mut ok = true; + for (id, filter) in subs { + if let Err(e) = send_req(&mut conn, id, filter).await { + shared + .stats + .err(Op::WsConnect, &format!("disconnected: {e}")); + ok = false; + break; + } + } + if ok && profile.history_limit > 0 { + for (n, channel) in channels.iter().enumerate() { + if stopped(&shared) { + break; + } + let id = format!("{HISTORY_PREFIX}{n}"); + let filter = Filter::new() + .kind(kind(KIND_STREAM_MESSAGE)) + .custom_tag( + nostr::SingleLetterTag::lowercase(nostr::Alphabet::H), + channel.clone(), + ) + .limit(profile.history_limit); + let fetched = history( + &shared.stats, + &mut conn, + &id, + filter, + subscribed, + &mut last_message, + ) + .await; + if let Err(e) = fetched { + if let HistoryError::Socket(e) = e { + shared + .stats + .err(Op::WsConnect, &format!("disconnected: {e}")); + } + ok = false; + break; + } + } + } + + let interval = profile.presence_interval_secs; + let mut next_presence = + Some(Instant::now() + Duration::from_secs_f64(interval * rng.unit())); + let mut next_typing = rng.after(profile.typing_per_hour); + let mut next_message = rng.after(profile.messages_per_hour); + let mut next_reaction = rng.after(profile.reactions_per_hour); + let session_end = (profile.mean_session_secs > 0.0) + .then(|| rng.after(3600.0 / profile.mean_session_secs)) + .flatten(); + + while ok && !stopped(&shared) { + let due = [ + next_presence, + next_typing, + next_message, + next_reaction, + session_end, + ] + .into_iter() + .flatten() + .min(); + let wait = due + .map(|d| d.saturating_duration_since(Instant::now())) + .unwrap_or(MAX_IDLE) + .min(MAX_IDLE); + // Fall through to the schedule after every message so a busy + // subscription cannot starve this client's own sends. + match conn.next_event(wait).await { + Ok(message) => { + if live(&shared.stats, &message, subscribed, &mut last_message).is_err() { + ok = false; + break; + } + } + Err(WsClientError::Timeout) => {} + Err(e) => { + shared + .stats + .err(Op::WsConnect, &format!("disconnected: {e}")); + ok = false; + break; + } + } + let now = Instant::now(); + let result = if next_presence.is_some_and(|t| t <= now) { + next_presence = Some(now + Duration::from_secs_f64(interval)); + let event = + EventBuilder::new(kind(KIND_PRESENCE_UPDATE), "online").sign_with_keys(&keys); + send(&shared.stats, &mut conn, Op::PublishEphemeral, event).await + } else if next_typing.is_some_and(|t| t <= now) { + next_typing = rng.after(profile.typing_per_hour); + let channel = &channels[rng.below(channels.len())]; + let event = EventBuilder::new(kind(KIND_TYPING_INDICATOR), "") + .tags([tag(["h", channel])]) + .sign_with_keys(&keys); + send(&shared.stats, &mut conn, Op::PublishEphemeral, event).await + } else if next_message.is_some_and(|t| t <= now) { + next_message = rng.after(profile.messages_per_hour); + let channel = &channels[rng.below(channels.len())]; + let mut tags = vec![tag(["h", channel])]; + if rng.unit() < profile.thread_reply_fraction { + let roots = roots(&shared, community, &keys, channel).await; + if !roots.is_empty() { + tags.push(tag(["e", &roots[rng.below(roots.len())], "", "reply"])); + } + } + let event = EventBuilder::new( + kind(KIND_STREAM_MESSAGE), + format!("{TIMED_PREFIX}{} {me}", now_micros()), + ) + .tags(tags) + .sign_with_keys(&keys); + send(&shared.stats, &mut conn, Op::PublishMessage, event).await + } else if next_reaction.is_some_and(|t| t <= now) { + next_reaction = rng.after(profile.reactions_per_hour); + // React to each seen message at most once; the relay rejects duplicates. + match last_message.take() { + Some(target) => { + let event = EventBuilder::new(kind(KIND_REACTION), "+") + .tags([tag(["h", &target.channel]), tag(["e", &target.id])]) + .sign_with_keys(&keys); + send(&shared.stats, &mut conn, Op::PublishMessage, event).await + } + None => Ok(()), + } + } else if session_end.is_some_and(|t| t <= now) { + break; + } else { + Ok(()) + }; + if let Err(e) = result { + shared + .stats + .err(Op::WsConnect, &format!("disconnected: {e}")); + ok = false; + break; + } + } + let _ = conn.disconnect().await; + if stopped(&shared) { + return; + } + // A failed session backs off like a failed connect, so a relay that + // rejects subscriptions cannot drive a tight reconnect loop. + if ok { + backoff = Backoff::new(); + } else { + backoff.wait(&shared, &mut rng).await; + } + } +} + +async fn send( + stats: &Stats, + conn: &mut NostrWsConnection, + op: Op, + event: Result, +) -> Result<(), WsClientError> { + match event { + Ok(event) => publish(stats, conn, op, event).await, + Err(e) => { + stats.err(op, &format!("sign: {e}")); + Ok(()) + } + } +} + +async fn send_req( + conn: &mut NostrWsConnection, + id: &str, + filter: Filter, +) -> Result<(), WsClientError> { + conn.send_raw(&json!(["REQ", id, filter])).await +} + +enum HistoryError { + /// The socket failed. + Socket(WsClientError), + /// The relay closed a live subscription (already recorded). + LiveClosed, +} + +impl From for HistoryError { + fn from(e: WsClientError) -> Self { + Self::Socket(e) + } +} + +/// Fetches history until EOSE, records the latency and closes the +/// subscription. Messages for the live subscriptions opened at `subscribed` +/// go through [`live`]. +async fn history( + stats: &Stats, + conn: &mut NostrWsConnection, + id: &str, + filter: Filter, + subscribed: Instant, + last_message: &mut Option, +) -> Result<(), HistoryError> { + let pending = stats.begin(Op::History); + if let Err(e) = send_req(conn, id, filter).await { + pending.forget(); + return Err(e.into()); + } + let deadline = Instant::now() + Duration::from_secs(30); + loop { + let remaining = deadline.saturating_duration_since(Instant::now()); + match conn.next_event(remaining).await { + Ok(RelayMessage::Eose { subscription_id }) if subscription_id == id => { + pending.ok(); + break; + } + Ok(RelayMessage::Closed { + subscription_id, + message, + }) if subscription_id == id => { + pending.err(&message); + return Ok(()); + } + Ok(message) if subscription_of(&message) == Some(id) => { + if let Some(target) = seen(&message) { + *last_message = Some(target); + } + } + Ok(message) => { + if live(stats, &message, subscribed, last_message).is_err() { + pending.forget(); + return Err(HistoryError::LiveClosed); + } + } + Err(WsClientError::Timeout) => { + pending.err("EOSE timeout"); + break; + } + Err(e) => { + pending.forget(); + return Err(e.into()); + } + } + } + Ok(conn.send_raw(&json!(["CLOSE", id])).await?) +} + +/// Global `#p` events addressed to `me`. Real clients name kinds here; a +/// kindless global filter would be matched against every global event. +fn mentions_filter(me: PublicKey, since: Timestamp) -> Filter { + Filter::new() + .kinds([ + kind(KIND_MEMBER_ADDED_NOTIFICATION), + kind(KIND_MEMBER_REMOVED_NOTIFICATION), + ]) + .pubkey(me) + .since(since) +} + +/// Humans own the simulated agents, so their mentions also carry the +/// owner-addressed observer frames, as desktop subscribes to them. +fn human_mentions_filter(me: PublicKey, since: Timestamp) -> Filter { + mentions_filter(me, since).kind(kind(KIND_AGENT_OBSERVER_FRAME)) +} + +fn subscription_of(message: &RelayMessage) -> Option<&str> { + match message { + RelayMessage::Event { + subscription_id, .. + } + | RelayMessage::Eose { subscription_id } + | RelayMessage::Closed { + subscription_id, .. + } => Some(subscription_id), + _ => None, + } +} + +/// An agent client: NIP-OA authenticated socket, HTTP `/query` polling, +/// HTTP `/events` posts and encrypted observer frames to its owner. +async fn agent(shared: Arc, global: usize) { + let scenario = &shared.scenario; + let profile = &scenario.agent; + let community = &shared.communities[global % shared.communities.len()]; + let keys = identity::keys(&scenario.run_id, Role::Agent, global); + let owner = identity::keys( + &scenario.run_id, + Role::Human, + global % scenario.population.humans.max(1), + ); + let auth = identity::nip_oa_tag(&owner, &keys); + let channels = scenario.member_channels(community.index, global); + let me = keys.public_key().to_hex(); + let mut rng = Rng::new(u64::MAX - global as u64); + let mut backoff = Backoff::new(); + let mut last_message = None; + + while let Some(mut conn) = connect( + &shared, + community, + &keys, + Some(&auth), + &mut rng, + &mut backoff, + ) + .await + { + let since = Timestamp::now(); + let subscribed = Instant::now(); + let mut ok = match send_req( + &mut conn, + "mentions", + mentions_filter(keys.public_key(), since), + ) + .await + { + Ok(()) => true, + Err(e) => { + shared + .stats + .err(Op::WsConnect, &format!("disconnected: {e}")); + false + } + }; + let query_per_hour = 3600.0 / profile.query_interval_secs; + let mut next_query = rng.after(query_per_hour); + let mut next_post = rng.after(profile.events_per_hour); + let mut next_frame = rng.after(profile.observer_frames_per_hour); + + while ok && !stopped(&shared) { + let due = [next_query, next_post, next_frame] + .into_iter() + .flatten() + .min(); + let wait = due + .map(|d| d.saturating_duration_since(Instant::now())) + .unwrap_or(MAX_IDLE) + .min(MAX_IDLE); + match conn.next_event(wait).await { + Ok(message) => { + if live(&shared.stats, &message, subscribed, &mut last_message).is_err() { + ok = false; + break; + } + } + Err(WsClientError::Timeout) => {} + Err(e) => { + shared + .stats + .err(Op::WsConnect, &format!("disconnected: {e}")); + ok = false; + break; + } + } + let now = Instant::now(); + let channel = &channels[rng.below(channels.len())]; + if next_query.is_some_and(|t| t <= now) { + next_query = rng.after(query_per_hour); + let window = Timestamp::now().as_secs().saturating_sub(3600); + let filters = json!([ + { "kinds": [KIND_STREAM_MESSAGE], "#h": [channel], "since": window, "limit": 20 }, + { "#p": [me], "since": window, "limit": 20 }, + ]); + http( + &shared, + Op::HttpQuery, + &keys, + community, + "/query", + &filters, + &auth, + ) + .await; + } else if next_post.is_some_and(|t| t <= now) { + next_post = rng.after(profile.events_per_hour); + let event = EventBuilder::new( + kind(KIND_STREAM_MESSAGE), + format!("{TIMED_PREFIX}{} {me}", now_micros()), + ) + .tags([tag(["h", channel])]) + .sign_with_keys(&keys); + match event { + Ok(event) => { + let pending = shared.stats.begin(Op::HttpEvents); + match shared + .net + .submit(&keys, community, &event, Some(&auth)) + .await + { + Ok(()) => pending.ok(), + Err(e) => pending.err(&e.to_string()), + } + } + Err(e) => shared.stats.err(Op::HttpEvents, &format!("sign: {e}")), + } + } else if next_frame.is_some_and(|t| t <= now) { + next_frame = rng.after(profile.observer_frames_per_hour); + let payload = json!({ "type": "loadgen", "sent_micros": now_micros().to_string() }); + let event = encrypt_observer_payload(&keys, &owner.public_key(), &payload) + .map_err(|e| e.to_string()) + .and_then(|content| { + EventBuilder::new(kind(KIND_AGENT_OBSERVER_FRAME), content) + .tags([ + tag(["p", &owner.public_key().to_hex()]), + tag([OBSERVER_AGENT_TAG, &me]), + tag([OBSERVER_FRAME_TAG, OBSERVER_FRAME_TELEMETRY]), + ]) + .sign_with_keys(&keys) + .map_err(|e| e.to_string()) + }); + let result = match event { + Ok(event) => { + publish(&shared.stats, &mut conn, Op::PublishEphemeral, event).await + } + Err(e) => { + shared + .stats + .err(Op::PublishEphemeral, &format!("observer: {e}")); + Ok(()) + } + }; + if let Err(e) = result { + shared + .stats + .err(Op::WsConnect, &format!("disconnected: {e}")); + ok = false; + } + } + } + let _ = conn.disconnect().await; + if stopped(&shared) { + return; + } + if ok { + backoff = Backoff::new(); + } else { + backoff.wait(&shared, &mut rng).await; + } + } +} + +async fn http( + shared: &Shared, + op: Op, + keys: &Keys, + community: &Community, + path: &str, + body: &Value, + auth: &Tag, +) { + let pending = shared.stats.begin(op); + let body = serde_json::to_vec(body).expect("serialize JSON body"); + match shared + .net + .post(keys, &community.http_url, path, body, Some(auth)) + .await + { + Ok((status, _)) if status.is_success() => pending.ok(), + Ok((status, text)) => pending.err(&http_reason(status, &text)), + Err(e) => pending.err(&e.to_string()), + } +} + +#[cfg(test)] +mod tests { + use super::*; + use std::sync::atomic::{AtomicUsize, Ordering}; + + #[test] + fn shards_partition_clients() { + let shards: Vec = (0..3) + .map(|i| Shard::parse(&format!("{i}/3")).unwrap()) + .collect(); + for g in 0..30 { + assert_eq!(shards.iter().filter(|s| s.owns(g)).count(), 1); + } + assert!(Shard::parse("3/3").is_err()); + assert!(Shard::parse("x").is_err()); + } + + #[test] + fn poisson_schedule_matches_rate() { + let mut rng = Rng::new(1); + let start = Instant::now(); + let n = 20_000; + let total: f64 = (0..n) + .map(|_| { + rng.after(3600.0) + .unwrap() + .duration_since(start) + .as_secs_f64() + }) + .sum(); + let mean = total / n as f64; + assert!((mean - 1.0).abs() < 0.05, "mean interval {mean}s for 1/s"); + assert!(rng.after(0.0).is_none()); + } + + #[test] + fn timed_messages_record_delivery() { + let stats = Stats::default(); + let keys = Keys::generate(); + let event = EventBuilder::new( + kind(KIND_STREAM_MESSAGE), + format!("{TIMED_PREFIX}{} x", now_micros()), + ) + .tags([tag(["h", "c"])]) + .sign_with_keys(&keys) + .unwrap(); + let target = observe( + &stats, + &RelayMessage::Event { + subscription_id: "live".into(), + event: Box::new(event.clone()), + }, + ) + .expect("stream message with an h tag is a reaction target"); + assert_eq!( + (target.id, target.channel), + (event.id.to_hex(), "c".to_string()) + ); + let summary = stats.snapshot().since(&Default::default()); + assert_eq!( + summary + .iter() + .find(|(op, _)| *op == Op::Deliver) + .unwrap() + .1 + .count, + 1 + ); + } + #[test] + fn human_mentions_receive_owner_observer_frames() { + let owner = Keys::generate(); + let agent = Keys::generate(); + let frame_to = |to: &Keys| { + EventBuilder::new(kind(KIND_AGENT_OBSERVER_FRAME), "x") + .tags([tag(["p", &to.public_key().to_hex()])]) + .sign_with_keys(&agent) + .unwrap() + }; + let opts = nostr::filter::MatchEventOptions::default(); + let since = Timestamp::from(0); + let human = human_mentions_filter(owner.public_key(), since); + assert!(human.match_event(&frame_to(&owner), opts)); + assert!(!human.match_event(&frame_to(&agent), opts)); + let agent_filter = mentions_filter(owner.public_key(), since); + assert!(!agent_filter.match_event(&frame_to(&owner), opts)); + } + + #[test] + fn closed_live_subscriptions_end_the_session_and_count() { + let stats = Stats::default(); + let mut last = None; + let subscribed = Instant::now(); + let eose = RelayMessage::Eose { + subscription_id: "live".into(), + }; + assert!(live(&stats, &eose, subscribed, &mut last).is_ok()); + // The relay acknowledges our own CLOSE of a history subscription. + let closed_history = RelayMessage::Closed { + subscription_id: format!("{HISTORY_PREFIX}0"), + message: String::new(), + }; + assert!(live(&stats, &closed_history, subscribed, &mut last).is_ok()); + let closed = RelayMessage::Closed { + subscription_id: "live".into(), + message: "error: overloaded".into(), + }; + assert!(live(&stats, &closed, subscribed, &mut last).is_err()); + let summary = stats.snapshot().since(&Default::default()); + let sub = summary + .iter() + .find(|(op, _)| *op == Op::Subscribe) + .unwrap() + .1; + assert_eq!((sub.count, sub.errors), (1, 1)); + let reason = abort_reason(&summary, 0.05, 5_000.0).expect("rejected subscriptions abort"); + assert!(reason.starts_with("subscribe_eose"), "{reason}"); + } + + #[tokio::test] + async fn stalled_connects_count_and_the_run_still_returns() { + // A peer that accepts TCP but never answers the WebSocket upgrade. + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let addr = listener.local_addr().unwrap(); + let held = tokio::spawn(async move { + let mut open = Vec::new(); + while let Ok((socket, _)) = listener.accept().await { + open.push(socket); + } + }); + let scenario = Scenario::parse(&format!( + r#" +run_id = "t" +[target] +ws_url = "ws://{addr}" +http_url = "http://{addr}" +connect_timeout_secs = 0.2 +[shape] +communities = 1 +channels_per_community = 1 +[population] +humans = 3 +agents = 1 +channels_per_member = 1 +connect_rate_per_sec = 100.0 +[[ramp]] +users = 1.0 +hold_secs = 1 +[[ramp]] +users = 2.0 +hold_secs = 1 +"# + )) + .unwrap(); + let net = Net::new(&scenario).await.unwrap(); + let started = Instant::now(); + let report = tokio::time::timeout( + Duration::from_secs(20), + run(scenario, net, Shard { index: 0, count: 1 }), + ) + .await + .expect("run returns while every connect stalls") + .expect_err("a step whose connects all fail aborts"); + assert!(started.elapsed() < Duration::from_secs(10)); + let steps = report["steps"].as_array().unwrap(); + assert_eq!(steps.len(), 1, "the ramp stops at the failing step"); + assert_eq!(steps[0]["aborted"], true); + let errors = steps[0]["ops"]["ws_connect_auth"]["errors"] + .as_u64() + .unwrap(); + assert!(errors >= 4, "{report}"); + assert_eq!(report["clients_cancelled"], 0); + held.abort(); + } + + #[tokio::test] + async fn stalled_connects_still_under_way_at_the_stop_settle_and_fail_the_run() { + // The 2s connect deadline outlasts the 1s hold, so every attempt is + // still in flight when the step is judged: the step passes, the stop + // lets the attempts run to their deadline, and they fail the run. + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let addr = listener.local_addr().unwrap(); + let held = tokio::spawn(async move { + let mut open = Vec::new(); + while let Ok((socket, _)) = listener.accept().await { + open.push(socket); + } + }); + let scenario = Scenario::parse(&format!( + r#" +run_id = "t" +[target] +ws_url = "ws://{addr}" +http_url = "http://{addr}" +connect_timeout_secs = 2.0 +[shape] +communities = 1 +channels_per_community = 1 +[population] +humans = 3 +agents = 1 +channels_per_member = 1 +connect_rate_per_sec = 100.0 +[[ramp]] +users = 1.0 +hold_secs = 1 +"# + )) + .unwrap(); + let net = Net::new(&scenario).await.unwrap(); + let started = Instant::now(); + let report = tokio::time::timeout( + Duration::from_secs(10), + run(scenario, net, Shard { index: 0, count: 1 }), + ) + .await + .expect("the stalled connects settle at their deadline") + .expect_err("clients that never connected fail the run"); + assert!(started.elapsed() >= Duration::from_secs(2), "{report}"); + let step = &report["steps"][0]; + assert_eq!(step["aborted"], false, "{report}"); + assert_eq!(step["ops"]["ws_connect_auth"]["pending"], 4, "{report}"); + assert_eq!(step["ops"]["ws_connect_auth"]["errors"], 0, "{report}"); + // The attempts finish as connect timeouts, not as abandoned ones. + let closing = &report["shutdown"]; + assert_eq!(closing["ops"]["ws_connect_auth"]["errors"], 4, "{report}"); + let reasons = closing["failures"].to_string(); + assert!(reasons.contains("connect timeout"), "{report}"); + assert!(!reasons.contains("unfinished at stop"), "{report}"); + assert!( + closing["abort_reason"] + .as_str() + .is_some_and(|r| r.starts_with("ws_connect_auth error rate 1.000")), + "{report}" + ); + assert_eq!(report["clients_cancelled"], 0, "{report}"); + held.abort(); + } + + #[test] + fn a_run_fails_unless_every_step_and_the_shutdown_were_clean() { + let reason = Some("publish_ephemeral_ok error rate 1.000 > 0.05"); + // (aborted, panicked, cancelled, shutdown) -> failed + let cases = [ + ((false, 0, 0, None), false), + ((true, 0, 0, None), true), + ((false, 1, 0, None), true), + ((false, 0, 1, None), true), + ((false, 0, 0, reason), true), + ]; + for ((aborted, panicked, cancelled, shutdown), expected) in cases { + assert_eq!( + failed(aborted, panicked, cancelled, shutdown), + expected, + "{aborted} {panicked} {cancelled} {shutdown:?}" + ); + } + } + + #[tokio::test] + async fn shutdown_cancels_stragglers_and_counts_what_they_left_in_flight() { + let stats = Arc::new(Stats::default()); + let before = stats.snapshot(); + let stuck = { + let stats = stats.clone(); + tokio::spawn(async move { + // A publish that never gets its OK. + let _pending = stats.begin(Op::PublishEphemeral); + std::future::pending::<()>().await; + }) + }; + let finished = tokio::spawn(async {}); + // Let the straggler register as in flight before shutting down. + while stats + .snapshot() + .since(&before) + .iter() + .all(|(_, s)| s.pending == 0) + { + tokio::task::yield_now().await; + } + let (panicked, cancelled) = + shutdown(vec![stuck, finished], Duration::from_millis(100)).await; + assert_eq!((panicked, cancelled), (0, 1)); + let closed = stats.snapshot().since(&before); + let publish = closed.iter().find(|(op, _)| *op == Op::PublishEphemeral); + let publish = publish.unwrap().1; + assert_eq!((publish.errors, publish.pending), (1, 0)); + let reason = abort_reason(&closed, 0.05, 5_000.0).expect("the straggler fails the run"); + assert!(reason.starts_with("publish_ephemeral_ok"), "{reason}"); + } + + /// How [`fake_ws_relay`] answers a client. + #[derive(Clone, Copy)] + struct FakeRelay { + /// Delay before the AUTH challenge. + challenge_delay: Duration, + /// Delay before each EVENT's `OK`; never answered when `None`. + ok_delay: Option, + /// Delay before the EOSE of each history REQ. + history_delay: Duration, + /// From this long after the handshake, each `OK` takes this long. + slow_ok: Option<(Duration, Duration)>, + } + + impl FakeRelay { + const PROMPT: Self = Self { + challenge_delay: Duration::ZERO, + ok_delay: Some(Duration::ZERO), + history_delay: Duration::ZERO, + slow_ok: None, + }; + } + + /// A loopback relay that speaks just enough NIP-42 and NIP-01 for a + /// client: it sends the AUTH challenge, accepts the AUTH event, answers + /// every REQ with EOSE and every EVENT with an `OK`, each as `relay` + /// says. Returns its address and what it has seen. + async fn fake_ws_relay(relay: FakeRelay) -> (String, Arc) { + use futures_util::{SinkExt, StreamExt}; + use tokio_tungstenite::tungstenite::Message; + + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let addr = listener.local_addr().unwrap().to_string(); + let seen = Arc::new(Seen::default()); + let counted = seen.clone(); + tokio::spawn(async move { + while let Ok((stream, _)) = listener.accept().await { + let seen = counted.clone(); + tokio::spawn(async move { + // Counted on every exit, so frames read before it are final. + let _closed = Closed(seen.clone()); + let Ok(mut ws) = tokio_tungstenite::accept_async(stream).await else { + return; + }; + let opened = Instant::now(); + tokio::time::sleep(relay.challenge_delay).await; + let challenge = json!(["AUTH", "challenge"]).to_string(); + if ws.send(Message::Text(challenge.into())).await.is_err() { + return; + } + while let Some(Ok(Message::Text(text))) = ws.next().await { + let frame: Value = serde_json::from_str(&text).unwrap(); + let reply = match frame[0].as_str() { + Some("AUTH") => json!(["OK", frame[1]["id"], true, ""]), + Some("REQ") => { + seen.reqs.fetch_add(1, Ordering::SeqCst); + let id = frame[1].as_str().unwrap_or_default(); + if id.starts_with(HISTORY_PREFIX) { + tokio::time::sleep(relay.history_delay).await; + } + json!(["EOSE", frame[1]]) + } + Some("EVENT") => { + let Some(mut delay) = relay.ok_delay else { + continue; + }; + if let Some((after, slow)) = relay.slow_ok { + if opened.elapsed() >= after { + delay = slow; + } + } + tokio::time::sleep(delay).await; + json!(["OK", frame[1]["id"], true, ""]) + } + _ => continue, + }; + if ws + .send(Message::Text(reply.to_string().into())) + .await + .is_err() + { + return; + } + } + }); + } + }); + (addr, seen) + } + + /// What [`fake_ws_relay`] has received. + #[derive(Default)] + struct Seen { + reqs: AtomicUsize, + closed: AtomicUsize, + } + + impl Seen { + /// REQs received over `sessions` connections, once all have closed. + /// A client sends its REQs before it closes, so the count is final. + async fn reqs_after(&self, sessions: usize) -> usize { + tokio::time::timeout(Duration::from_secs(5), async { + while self.closed.load(Ordering::SeqCst) < sessions { + tokio::time::sleep(Duration::from_millis(10)).await; + } + }) + .await + .expect("the client closes its connection"); + self.reqs.load(Ordering::SeqCst) + } + } + + struct Closed(Arc); + + impl Drop for Closed { + fn drop(&mut self) { + self.0.closed.fetch_add(1, Ordering::SeqCst); + } + } + + /// One human, no agents, one `hold_secs` step, and a presence heartbeat + /// every second; other traffic is off so the heartbeat is the only publish. + fn lone_human(addr: &str, hold_secs: u64, max_error_rate: f64) -> Scenario { + Scenario::parse(&format!( + r#" +run_id = "t" +[target] +ws_url = "ws://{addr}" +http_url = "http://{addr}" +[shape] +communities = 1 +channels_per_community = 1 +[population] +humans = 1 +agents = 0 +channels_per_member = 1 +[human] +presence_interval_secs = 1.0 +typing_per_hour = 0.0 +messages_per_hour = 0.0 +reactions_per_hour = 0.0 +history_limit = 1 +[abort] +max_error_rate = {max_error_rate} +[[ramp]] +users = 1.0 +hold_secs = {hold_secs} +"# + )) + .unwrap() + } + + async fn run_lone_human(scenario: Scenario, grace: Duration) -> Result { + let net = Net::new(&scenario).await.unwrap(); + tokio::time::timeout( + Duration::from_secs(20), + run_with(scenario, net, Shard { index: 0, count: 1 }, grace), + ) + .await + .expect("the run returns") + } + + #[tokio::test] + async fn a_publish_that_never_gets_its_ok_is_cancelled_and_fails_the_run() { + let (addr, _) = fake_ws_relay(FakeRelay { + ok_delay: None, + ..FakeRelay::PROMPT + }) + .await; + let report = run_lone_human(lone_human(&addr, 2, 0.05), Duration::from_millis(300)) + .await + .expect_err("a relay that never acknowledges a publish fails the run"); + // The heartbeat is still waiting for its OK, which is not judged yet, + // but the client cannot settle it within the grace period. + let step = &report["steps"][0]; + assert_eq!(step["aborted"], false, "{report}"); + assert_eq!( + step["ops"]["publish_ephemeral_ok"]["pending"], 1, + "{report}" + ); + assert_eq!(report["clients_cancelled"], 1, "{report}"); + let closing = &report["shutdown"]; + assert_eq!( + closing["ops"]["publish_ephemeral_ok"]["errors"], 1, + "{report}" + ); + assert!( + closing["failures"] + .to_string() + .contains("unfinished at stop"), + "{report}" + ); + } + + #[tokio::test] + async fn a_cancelled_client_fails_the_run_even_within_the_error_limit() { + // With the error limit lifted, only the cancellation can fail the run. + let (addr, _) = fake_ws_relay(FakeRelay { + ok_delay: None, + ..FakeRelay::PROMPT + }) + .await; + let report = run_lone_human(lone_human(&addr, 2, 1.0), Duration::from_millis(300)) + .await + .expect_err("cancelling a client fails the run"); + assert_eq!(report["clients_cancelled"], 1, "{report}"); + assert!(report["shutdown"]["abort_reason"].is_null(), "{report}"); + } + + #[tokio::test] + async fn a_publish_in_flight_at_the_snapshot_settles_and_the_run_passes() { + // The OK takes 1.5s, so a heartbeat is awaiting it when the 2s hold + // ends. The relay is healthy: the publish settles during shutdown. + let (addr, _) = fake_ws_relay(FakeRelay { + ok_delay: Some(Duration::from_millis(1500)), + ..FakeRelay::PROMPT + }) + .await; + let report = run_lone_human(lone_human(&addr, 2, 0.05), Duration::from_secs(5)) + .await + .expect("a healthy publish straddling the snapshot does not fail the run"); + assert_eq!( + report["steps"][0]["ops"]["publish_ephemeral_ok"]["pending"], 1, + "{report}" + ); + let closing = &report["shutdown"]["ops"]["publish_ephemeral_ok"]; + assert!(closing["count"].as_u64().unwrap() >= 1, "{report}"); + assert_eq!(closing["errors"], 0, "{report}"); + assert_eq!(report["clients_cancelled"], 0, "{report}"); + } + + #[tokio::test] + async fn a_connect_under_way_at_the_stop_settles_and_the_run_passes() { + // The relay takes 1.5s to challenge, so the first connect is still in + // flight when the 1s hold ends and the run stops. It is not abandoned, + // and it does not start a session: the client subscribes to nothing. + let (addr, seen) = fake_ws_relay(FakeRelay { + challenge_delay: Duration::from_millis(1500), + ..FakeRelay::PROMPT + }) + .await; + let report = run_lone_human(lone_human(&addr, 1, 0.05), Duration::from_secs(5)) + .await + .expect("a healthy connect straddling the stop does not fail the run"); + assert_eq!( + report["steps"][0]["ops"]["ws_connect_auth"]["pending"], 1, + "{report}" + ); + let closing = &report["shutdown"]["ops"]["ws_connect_auth"]; + assert_eq!( + (closing["count"].clone(), closing["errors"].clone()), + (json!(1), json!(0)), + "{report}" + ); + assert_eq!(seen.reqs_after(1).await, 0, "{report}"); + } + + #[tokio::test] + async fn a_stop_during_history_ends_the_session_after_the_fetch_under_way() { + // Ten channels whose history takes 400ms each: the 1s hold ends + // mid-history. Fetching the rest would outlast the 2s grace and get + // the client cancelled; the fetch under way settles and the rest are + // skipped. + let (addr, seen) = fake_ws_relay(FakeRelay { + history_delay: Duration::from_millis(400), + ..FakeRelay::PROMPT + }) + .await; + let scenario = Scenario::parse(&format!( + r#" +run_id = "t" +[target] +ws_url = "ws://{addr}" +http_url = "http://{addr}" +[shape] +communities = 1 +channels_per_community = 10 +[population] +humans = 1 +agents = 0 +channels_per_member = 10 +[human] +typing_per_hour = 0.0 +messages_per_hour = 0.0 +reactions_per_hour = 0.0 +history_limit = 1 +[[ramp]] +users = 1.0 +hold_secs = 1 +"# + )) + .unwrap(); + let report = run_lone_human(scenario, Duration::from_secs(2)) + .await + .expect("a stop mid-history does not fail a healthy run"); + assert_eq!(report["clients_cancelled"], 0, "{report}"); + // Four live subscriptions plus the history fetched before the stop. + let sent = seen.reqs_after(1).await; + assert!((5..4 + 10).contains(&sent), "{sent} REQs: {report}"); + } + + #[tokio::test] + async fn one_slow_publish_in_flight_at_the_stop_is_judged_with_the_last_step() { + // About 200 typing indicators a second are acknowledged at once until + // 1.7s in; from then each OK takes 1.5s, so exactly one is in flight + // when the 2s hold ends. On its own, the settlement's p99 would be + // that one slow publish; among the step's fast ones it is not. + let (addr, _) = fake_ws_relay(FakeRelay { + slow_ok: Some((Duration::from_millis(1700), Duration::from_millis(1500))), + ..FakeRelay::PROMPT + }) + .await; + let scenario = Scenario::parse(&format!( + r#" +run_id = "t" +[target] +ws_url = "ws://{addr}" +http_url = "http://{addr}" +[shape] +communities = 1 +channels_per_community = 1 +[population] +humans = 1 +agents = 0 +channels_per_member = 1 +[human] +presence_interval_secs = 3600.0 +typing_per_hour = 720000.0 +messages_per_hour = 0.0 +reactions_per_hour = 0.0 +history_limit = 1 +[abort] +max_p99_ms = 1000.0 +[[ramp]] +users = 1.0 +hold_secs = 2 +"# + )) + .unwrap(); + let report = run_lone_human(scenario, Duration::from_secs(5)) + .await + .expect("one slow publish among many fast ones does not fail the run"); + let step = &report["steps"][0]["ops"]["publish_ephemeral_ok"]; + assert!(step["count"].as_u64().unwrap() >= 100, "{report}"); + assert_eq!(step["pending"], 1, "{report}"); + let closing = &report["shutdown"]["ops"]["publish_ephemeral_ok"]; + assert_eq!( + (closing["count"].clone(), closing["errors"].clone()), + (json!(1), json!(0)), + "{report}" + ); + assert!(closing["p99_ms"].as_f64().unwrap() > 1000.0, "{report}"); + } +} diff --git a/crates/buzz-test-client/src/loadgen/scenario.rs b/crates/buzz-test-client/src/loadgen/scenario.rs new file mode 100644 index 0000000000..05680cb9d8 --- /dev/null +++ b/crates/buzz-test-client/src/loadgen/scenario.rs @@ -0,0 +1,499 @@ +//! Scenario file schema and derived fixture layout. + +use serde::Deserialize; + +use super::identity; + +/// A load scenario, parsed from TOML. +#[derive(Clone, Debug, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct Scenario { + /// Seeds every derived identity, host and channel id. Change it for a fresh fixture. + pub run_id: String, + pub target: Target, + pub shape: Shape, + pub population: Population, + #[serde(default)] + pub human: HumanProfile, + #[serde(default)] + pub agent: AgentProfile, + pub ramp: Vec, + #[serde(default)] + pub abort: Abort, +} + +#[derive(Clone, Debug, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct Target { + /// WebSocket URL of the relay (single community) or the scheme source for + /// templated community hosts, e.g. `ws://localhost:3000`. + pub ws_url: String, + /// HTTP base URL matching `ws_url`, e.g. `http://localhost:3000`. + pub http_url: String, + /// Community host template with `{run}` and `{i}` placeholders. Unset means + /// one community at `ws_url`'s host. + #[serde(default)] + pub community_host_template: Option, + /// Open TCP connections here (`host:port`) instead of resolving community + /// hosts, e.g. the relay's in-cluster service. Host headers still name the community. + #[serde(default)] + pub connect_addr: Option, + /// Deployment-root HTTP URL serving `POST /operator/communities`. + #[serde(default)] + pub operator_url: Option, + /// Environment variable holding the operator's hex secret key. + #[serde(default = "default_operator_key_env")] + pub operator_key_env: String, + /// Deadline for one WebSocket connect plus NIP-42 authentication. + #[serde(default = "default_connect_timeout")] + pub connect_timeout_secs: f64, +} + +fn default_operator_key_env() -> String { + "LOADGEN_OPERATOR_KEY".into() +} + +fn default_connect_timeout() -> f64 { + 10.0 +} + +/// Deepest reply nesting the relay accepts. +const MAX_THREAD_DEPTH: usize = 100; + +/// Data-shape vectors created by `seed`. +#[derive(Clone, Debug, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct Shape { + pub communities: usize, + pub channels_per_community: usize, + /// Top-level kind:9 messages seeded per channel. + #[serde(default)] + pub history_messages_per_channel: usize, + /// Thread roots seeded per channel, each a chain of `thread_depth` nested + /// replies (at most 100, the relay's limit). + #[serde(default)] + pub thread_roots_per_channel: usize, + #[serde(default)] + pub thread_depth: usize, + /// Concurrent seeding requests. + #[serde(default = "default_seed_concurrency")] + pub seed_concurrency: usize, +} + +fn default_seed_concurrency() -> usize { + 32 +} + +/// Base (1x) population; ramp steps multiply it. +#[derive(Clone, Debug, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct Population { + pub humans: usize, + pub agents: usize, + /// Channels each human and agent subscribes to and posts in. + pub channels_per_member: usize, + /// New connections per second per shard while ramping. + #[serde(default = "default_connect_rate")] + pub connect_rate_per_sec: f64, +} + +fn default_connect_rate() -> f64 { + 200.0 +} + +/// Per-connection rates for a human client. Defaults are round placeholders; +/// set rates per scenario to model a specific deployment. +#[derive(Clone, Debug, Deserialize)] +#[serde(default, deny_unknown_fields)] +pub struct HumanProfile { + pub presence_interval_secs: f64, + pub typing_per_hour: f64, + pub messages_per_hour: f64, + pub reactions_per_hour: f64, + /// Fraction of messages posted as replies into a seeded deep thread. + pub thread_reply_fraction: f64, + /// History events fetched per channel on connect (0 disables). + pub history_limit: usize, + /// Mean session length before a reconnect (0 keeps connections open). + pub mean_session_secs: f64, +} + +impl Default for HumanProfile { + fn default() -> Self { + Self { + presence_interval_secs: 60.0, + typing_per_hour: 10.0, + messages_per_hour: 1.0, + reactions_per_hour: 0.5, + thread_reply_fraction: 0.3, + history_limit: 50, + mean_session_secs: 0.0, + } + } +} + +/// Per-connection rates for an agent client. +#[derive(Clone, Debug, Deserialize)] +#[serde(default, deny_unknown_fields)] +pub struct AgentProfile { + /// Mean interval between HTTP `/query` polls. + pub query_interval_secs: f64, + /// Messages posted through HTTP `/events` per hour. + pub events_per_hour: f64, + /// Encrypted kind:24200 observer frames sent to the owner per hour. + pub observer_frames_per_hour: f64, +} + +impl Default for AgentProfile { + fn default() -> Self { + Self { + query_interval_secs: 5.0, + events_per_hour: 60.0, + observer_frames_per_hour: 60.0, + } + } +} + +#[derive(Clone, Debug, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct RampStep { + /// Multiplier on the base population. + pub users: f64, + pub hold_secs: u64, +} + +/// Stop the ramp at the end of a step that crosses these limits. +#[derive(Clone, Debug, Deserialize)] +#[serde(default, deny_unknown_fields)] +pub struct Abort { + pub max_error_rate: f64, + pub max_p99_ms: f64, +} + +impl Default for Abort { + fn default() -> Self { + Self { + max_error_rate: 0.05, + max_p99_ms: 5_000.0, + } + } +} + +/// One community's endpoints. +#[derive(Clone, Debug)] +pub struct Community { + pub index: usize, + pub host: Option, + pub ws_url: String, + pub http_url: String, +} + +impl Scenario { + pub fn parse(text: &str) -> Result { + let scenario: Self = toml::from_str(text).map_err(|e| e.to_string())?; + scenario.validate()?; + Ok(scenario) + } + + fn validate(&self) -> Result<(), String> { + if self.shape.communities == 0 || self.shape.channels_per_community == 0 { + return Err("shape needs at least one community and one channel".into()); + } + // `buffer_unordered(0)` never polls its input, so seeding would hang. + if self.shape.seed_concurrency == 0 { + return Err("shape.seed_concurrency must be at least 1".into()); + } + if self.shape.communities > 1 && self.target.community_host_template.is_none() { + return Err("more than one community needs target.community_host_template".into()); + } + // The relay rejects replies nested deeper than this. + if self.shape.thread_depth > MAX_THREAD_DEPTH { + return Err(format!( + "shape.thread_depth must be at most {MAX_THREAD_DEPTH}" + )); + } + // The HTTP override resolves a bare hostname, so a host with a port + // would reach `connect_addr` over WebSocket but not over HTTP. + if self.target.connect_addr.is_some() + && self + .target + .community_host_template + .as_deref() + .is_some_and(|template| template.contains(':')) + { + return Err( + "target.community_host_template must not include a port when target.connect_addr is set" + .into(), + ); + } + if self.population.channels_per_member == 0 + || self.population.channels_per_member > self.shape.channels_per_community + { + return Err("population.channels_per_member must be 1..=channels_per_community".into()); + } + if self.ramp.is_empty() { + return Err("ramp needs at least one step".into()); + } + // Clients are only ever added, so a step can never shrink the population. + let mut previous = 0.0; + for step in &self.ramp { + if !step.users.is_finite() || step.users <= previous { + return Err("ramp users must be finite and increase every step".into()); + } + previous = step.users; + } + let (h, a) = (&self.human, &self.agent); + let positive = [ + ( + "target.connect_timeout_secs", + self.target.connect_timeout_secs, + ), + ( + "population.connect_rate_per_sec", + self.population.connect_rate_per_sec, + ), + ("agent.query_interval_secs", a.query_interval_secs), + ("abort.max_p99_ms", self.abort.max_p99_ms), + ]; + for (name, value) in positive { + if !value.is_finite() || value <= 0.0 { + return Err(format!("{name} must be a finite number > 0")); + } + } + // Presence heartbeats are never sent more often than once a second. + if !h.presence_interval_secs.is_finite() || h.presence_interval_secs < 1.0 { + return Err("human.presence_interval_secs must be a finite number >= 1".into()); + } + let non_negative = [ + ("human.typing_per_hour", h.typing_per_hour), + ("human.messages_per_hour", h.messages_per_hour), + ("human.reactions_per_hour", h.reactions_per_hour), + ("human.mean_session_secs", h.mean_session_secs), + ("agent.events_per_hour", a.events_per_hour), + ("agent.observer_frames_per_hour", a.observer_frames_per_hour), + ]; + for (name, value) in non_negative { + if !value.is_finite() || value < 0.0 { + return Err(format!("{name} must be a finite number >= 0")); + } + } + let fractions = [ + ("human.thread_reply_fraction", h.thread_reply_fraction), + ("abort.max_error_rate", self.abort.max_error_rate), + ]; + for (name, value) in fractions { + if !(0.0..=1.0).contains(&value) { + return Err(format!("{name} must be between 0 and 1")); + } + } + Ok(()) + } + + pub fn communities(&self) -> Vec { + (0..self.shape.communities) + .map(|index| match &self.target.community_host_template { + None => Community { + index, + host: None, + ws_url: self.target.ws_url.clone(), + http_url: self.target.http_url.clone(), + }, + Some(template) => { + let host = template + .replace("{run}", &self.run_id) + .replace("{i}", &index.to_string()); + Community { + index, + ws_url: with_host(&self.target.ws_url, &host), + http_url: with_host(&self.target.http_url, &host), + host: Some(host), + } + } + }) + .collect() + } + + pub fn channel_ids(&self, community: usize) -> Vec { + (0..self.shape.channels_per_community) + .map(|i| identity::channel_id(&self.run_id, community, i)) + .collect() + } + + /// Channels member `member` (global human or agent index) belongs to. + pub fn member_channels(&self, community: usize, member: usize) -> Vec { + let n = self.shape.channels_per_community; + (0..self.population.channels_per_member) + .map(|k| identity::channel_id(&self.run_id, community, (member + k) % n)) + .collect() + } + + /// Largest multiplier any ramp step reaches. + pub fn peak_users(&self) -> f64 { + self.ramp.iter().map(|s| s.users).fold(0.0, f64::max) + } +} + +/// `url` with its host (and port) replaced by `host`. +fn with_host(url: &str, host: &str) -> String { + match url.split_once("://") { + Some((scheme, _)) => format!("{scheme}://{host}"), + None => format!("{url}://{host}"), + } +} + +#[cfg(test)] +mod tests { + use super::*; + + const MINIMAL: &str = r#" +run_id = "t" +[target] +ws_url = "ws://localhost:3000" +http_url = "http://localhost:3000" +[shape] +communities = 1 +channels_per_community = 4 +[population] +humans = 10 +agents = 2 +channels_per_member = 2 +[[ramp]] +users = 1.0 +hold_secs = 10 +"#; + + #[test] + fn minimal_scenario_parses_with_profile_defaults() { + let s = Scenario::parse(MINIMAL).unwrap(); + assert_eq!(s.communities().len(), 1); + assert_eq!(s.communities()[0].ws_url, "ws://localhost:3000"); + assert_eq!(s.human.presence_interval_secs, 60.0); + assert_eq!(s.member_channels(0, 3).len(), 2); + } + + #[test] + fn templated_hosts_keep_scheme_and_drop_port() { + let text = MINIMAL + .replace("communities = 1", "communities = 2") + .replace( + "http_url = \"http://localhost:3000\"", + "http_url = \"http://localhost:3000\"\ncommunity_host_template = \"lg-{run}-{i}.example\"", + ); + let s = Scenario::parse(&text).unwrap(); + let c = s.communities(); + assert_eq!(c[1].ws_url, "ws://lg-t-1.example"); + assert_eq!(c[1].http_url, "http://lg-t-1.example"); + assert_eq!(c[1].host.as_deref(), Some("lg-t-1.example")); + } + + #[test] + fn invalid_shapes_are_rejected() { + assert!(Scenario::parse(&MINIMAL.replace("communities = 1", "communities = 2")).is_err()); + assert!(Scenario::parse( + &MINIMAL.replace("channels_per_member = 2", "channels_per_member = 5") + ) + .is_err()); + assert!(Scenario::parse(&MINIMAL.replace("users = 1.0", "users = 0.0")).is_err()); + } + + #[test] + fn unschedulable_values_are_rejected() { + let cases = [ + "[human]\nmessages_per_hour = nan\n", + "[human]\ntyping_per_hour = inf\n", + "[human]\nreactions_per_hour = -1.0\n", + "[human]\npresence_interval_secs = 0.0\n", + "[human]\npresence_interval_secs = 0.5\n", + "[human]\nthread_reply_fraction = 1.5\n", + "[agent]\nquery_interval_secs = nan\n", + "[agent]\nobserver_frames_per_hour = inf\n", + "[abort]\nmax_error_rate = nan\n", + "[abort]\nmax_p99_ms = inf\n", + "[[ramp]]\nusers = inf\nhold_secs = 10\n", + // A down-ramp would report a smaller population than is running. + "[[ramp]]\nusers = 0.5\nhold_secs = 10\n", + "[[ramp]]\nusers = 1.0\nhold_secs = 10\n", + ]; + for case in cases { + let text = format!("{MINIMAL}{case}"); + assert!(Scenario::parse(&text).is_err(), "accepted: {case}"); + } + let text = MINIMAL.replace( + "http_url = \"http://localhost:3000\"", + "http_url = \"http://localhost:3000\"\nconnect_timeout_secs = nan", + ); + assert!(Scenario::parse(&text).is_err()); + let up = format!("{MINIMAL}[[ramp]]\nusers = 2.0\nhold_secs = 10\n"); + assert!(Scenario::parse(&up).is_ok()); + } + + #[test] + fn shipped_scenarios_validate() { + let dir = std::path::Path::new(env!("CARGO_MANIFEST_DIR")).join("../../perf/scenarios"); + let mut checked = 0; + for entry in std::fs::read_dir(dir).unwrap() { + let path = entry.unwrap().path(); + if path.extension().is_some_and(|ext| ext == "toml") { + let text = std::fs::read_to_string(&path).unwrap(); + Scenario::parse(&text).unwrap_or_else(|e| panic!("{}: {e}", path.display())); + checked += 1; + } + } + assert!(checked > 0); + } + + #[test] + fn presence_interval_of_one_second_is_the_minimum() { + let with = |secs: &str| format!("{MINIMAL}[human]\npresence_interval_secs = {secs}\n"); + assert!(Scenario::parse(&with("1.0")).is_ok()); + assert!(Scenario::parse(&with("0.99")).is_err()); + } + + #[test] + fn zero_seed_concurrency_is_rejected() { + let with = |n: usize| { + MINIMAL.replace( + "channels_per_community = 4", + &format!("channels_per_community = 4\nseed_concurrency = {n}"), + ) + }; + assert!(Scenario::parse(&with(1)).is_ok()); + let err = Scenario::parse(&with(0)).unwrap_err(); + assert!(err.contains("seed_concurrency"), "{err}"); + } + + #[test] + fn host_template_ports_conflict_with_connect_addr() { + let with = |template: &str, connect_addr: Option<&str>| { + let connect = connect_addr + .map(|a| format!("\nconnect_addr = \"{a}\"")) + .unwrap_or_default(); + MINIMAL.replace( + "http_url = \"http://localhost:3000\"", + &format!( + "http_url = \"http://localhost:3000\"\ncommunity_host_template = \"{template}\"{connect}" + ), + ) + }; + // A port is fine while hosts resolve normally, and so is an override + // for a portless host. + assert!(Scenario::parse(&with("lg-{run}-{i}.example:8443", None)).is_ok()); + assert!(Scenario::parse(&with("lg-{run}-{i}.example", Some("10.0.0.1:8080"))).is_ok()); + let err = + Scenario::parse(&with("lg-{run}-{i}.example:8443", Some("10.0.0.1:8080"))).unwrap_err(); + assert!(err.contains("port"), "{err}"); + } + + #[test] + fn thread_depth_is_limited_to_what_the_relay_accepts() { + let with_depth = |depth: usize| { + MINIMAL.replace( + "channels_per_community = 4", + &format!("channels_per_community = 4\nthread_depth = {depth}"), + ) + }; + assert!(Scenario::parse(&with_depth(100)).is_ok()); + let err = Scenario::parse(&with_depth(101)).unwrap_err(); + assert!(err.contains("thread_depth"), "{err}"); + } +} diff --git a/crates/buzz-test-client/src/loadgen/seed.rs b/crates/buzz-test-client/src/loadgen/seed.rs new file mode 100644 index 0000000000..4ff8af2158 --- /dev/null +++ b/crates/buzz-test-client/src/loadgen/seed.rs @@ -0,0 +1,690 @@ +//! Creates the scenario's fixture: communities, channels, history and deep threads. + +use std::collections::HashMap; +use std::sync::Arc; +use std::time::Instant; + +use futures_util::stream::{self, StreamExt}; +use nostr::{Event, EventBuilder, Keys, Kind, Tag}; +use reqwest::StatusCode; +use serde_json::json; + +use super::identity::{self, Role}; +use super::net::{http_reason, Net}; +use super::scenario::{Community, Scenario}; +use super::stats::{Op, Snapshot, Stats}; + +/// Content prefix of seeded thread roots; `run` finds roots by it. +pub const ROOT_PREFIX: &str = "lg-root"; + +/// Seeds every community. Idempotent for communities this run's fixture +/// owner owns, and for channels; history and threads are appended on every +/// call. Fails unless every planned post +/// was accepted. +pub async fn seed(scenario: &Scenario, net: &Net, stats: &Arc) -> anyhow::Result<()> { + for community in scenario.communities() { + let owner = identity::keys(&scenario.run_id, Role::Owner, community.index); + if let Some(host) = &community.host { + provision(scenario, net, host, &owner).await?; + } + create_channels(scenario, net, &community, &owner).await?; + eprintln!( + "seed: community {} ready ({} channels)", + community.index, scenario.shape.channels_per_community + ); + } + + let communities = scenario.communities(); + let started = Instant::now(); + let before = stats.snapshot(); + let (roots, rest) = plan_posts(scenario); + let planned: usize = roots.iter().chain(&rest).map(Vec::len).sum(); + let root_ids = post_all(scenario, net, stats, &communities, roots, &RootIds::new()).await; + post_all(scenario, net, stats, &communities, rest, &root_ids).await; + let accepted = seeded(&stats.snapshot(), &before); + eprintln!( + "seed: {accepted} of {planned} posts accepted in {:.1}s", + started.elapsed().as_secs_f64() + ); + if accepted < planned { + anyhow::bail!( + "seeded {accepted} of {planned} posts; {} missing", + planned - accepted + ); + } + Ok(()) +} + +fn seeded(now: &Snapshot, before: &Snapshot) -> usize { + now.since(before) + .iter() + .find(|(op, _)| *op == Op::Seed) + .map_or(0, |(_, s)| s.count as usize) +} + +/// Accepted post ids keyed by (community, channel, seq). +type RootIds = HashMap<(usize, usize, usize), String>; + +/// Posts submitted in order, each attaching to the previous accepted one. +type Chain = Vec; + +/// Submits the chains concurrently, each in order, and returns the ids of the +/// accepted posts. A chain stops making progress at its first failed post. +async fn post_all( + scenario: &Scenario, + net: &Net, + stats: &Stats, + communities: &[Community], + chains: Vec, + roots: &RootIds, +) -> RootIds { + stream::iter(chains) + .map(|chain| async move { + let mut accepted = Vec::new(); + let mut previous: Option = None; + for post in chain { + let community = &communities[post.community]; + let Some((keys, event)) = post.build(scenario, roots, previous.as_deref()) else { + stats.err(Op::Seed, "thread parent missing"); + previous = None; + continue; + }; + let t0 = Instant::now(); + match net.submit(&keys, community, &event, None).await { + Ok(()) => { + stats.ok(Op::Seed, t0.elapsed()); + let id = event.id.to_hex(); + accepted.push(((post.community, post.channel, post.seq), id.clone())); + previous = Some(id); + } + Err(e) => { + stats.err(Op::Seed, &e.to_string()); + previous = None; + } + } + } + accepted + }) + .buffer_unordered(scenario.shape.seed_concurrency) + .flat_map(stream::iter) + .collect() + .await +} + +async fn provision(scenario: &Scenario, net: &Net, host: &str, owner: &Keys) -> anyhow::Result<()> { + let base = scenario + .target + .operator_url + .as_deref() + .ok_or_else(|| anyhow::anyhow!("templated communities need target.operator_url"))?; + let secret = std::env::var(&scenario.target.operator_key_env).map_err(|_| { + anyhow::anyhow!( + "{} must hold the operator secret key", + scenario.target.operator_key_env + ) + })?; + let operator = Keys::parse(&secret)?; + provision_host(net, base, &operator, host, owner).await +} + +/// Creates `host` owned by `owner`. The request is `create_only`, so the relay +/// never converges an existing host or rotates its owner. A host that already +/// exists is reused only when `owner` owns it, e.g. when `seed` is re-run; +/// anything else is a collision with a community this run does not own. +async fn provision_host( + net: &Net, + base: &str, + operator: &Keys, + host: &str, + owner: &Keys, +) -> anyhow::Result<()> { + let owner_hex = owner.public_key().to_hex(); + let body = serde_json::to_vec(&json!({ + "host": host, + "initial_owner_pubkey": owner_hex, + "create_only": true, + }))?; + let (status, text) = net + .post(operator, base, "/operator/communities", body, None) + .await?; + if status.is_success() { + return Ok(()); + } + if status != StatusCode::CONFLICT || !text.contains("community already exists") { + anyhow::bail!("provision {host}: {}", http_reason(status, &text)); + } + // The relay reports no owner for a host, but lists the hosts an owner owns. + let url = format!("{base}/operator/communities?owner_pubkey={owner_hex}"); + let (status, text) = net.get(operator, &url).await?; + if !status.is_success() { + anyhow::bail!( + "provision {host}: it already exists and its owner could not be checked: {}", + http_reason(status, &text) + ); + } + let owned: serde_json::Value = serde_json::from_str(&text)?; + let owns = owned["communities"] + .as_array() + .is_some_and(|list| list.iter().any(|c| c["host"] == host)); + if !owns { + anyhow::bail!( + "provision {host}: it already exists and is not owned by this run's fixture owner \ + {owner_hex}; use a fresh run_id or a host template that is unique per run" + ); + } + Ok(()) +} + +async fn create_channels( + scenario: &Scenario, + net: &Net, + community: &Community, + owner: &Keys, +) -> anyhow::Result<()> { + let results: Vec> = stream::iter(scenario.channel_ids(community.index)) + .enumerate() + .map(|(i, channel)| async move { + let event = EventBuilder::new(Kind::Custom(9007), "") + .tags([ + tag(["h", &channel]), + tag(["name", &format!("lg-{}-{i}", scenario.run_id)]), + tag(["channel_type", "stream"]), + tag(["visibility", "open"]), + ]) + .sign_with_keys(owner)?; + match net.submit(owner, community, &event, None).await { + Err(e) if !e.to_string().contains("already exists") => Err(e), + _ => Ok(()), + } + }) + .buffer_unordered(scenario.shape.seed_concurrency) + .collect() + .await; + results.into_iter().collect() +} + +/// One seeded post: history message, thread root, or thread reply. +struct Post { + community: usize, + channel: usize, + kind: PostKind, + seq: usize, +} + +enum PostKind { + History, + Root, + Reply { root: usize }, +} + +/// Thread roots, then history and replies (posted once root ids are known). +/// A root's `seq` is its index within its channel. Each thread's replies form +/// one chain, so reply `k` sits at depth `k + 1`; threads and history posts +/// are independent of each other. +fn plan_posts(scenario: &Scenario) -> (Vec, Vec) { + let shape = &scenario.shape; + let mut roots = Vec::new(); + let mut rest = Vec::new(); + for community in 0..shape.communities { + for channel in 0..shape.channels_per_community { + for seq in 0..shape.history_messages_per_channel { + rest.push(vec![Post { + community, + channel, + kind: PostKind::History, + seq, + }]); + } + for root in 0..shape.thread_roots_per_channel { + roots.push(vec![Post { + community, + channel, + kind: PostKind::Root, + seq: root, + }]); + let replies: Chain = (0..shape.thread_depth) + .map(|seq| Post { + community, + channel, + kind: PostKind::Reply { root }, + seq, + }) + .collect(); + if !replies.is_empty() { + rest.push(replies); + } + } + } + } + (roots, rest) +} + +impl Post { + /// Builds the event. A reply attaches to `previous`, the accepted reply + /// before it in its thread; the first reply attaches to the thread root. + fn build( + &self, + scenario: &Scenario, + roots: &RootIds, + previous: Option<&str>, + ) -> Option<(Keys, Event)> { + let channel = identity::channel_id(&scenario.run_id, self.community, self.channel); + // Rotate authors across humans to stay under per-pubkey rate limits. + let humans = scenario.population.humans.max(1); + let author = identity::keys( + &scenario.run_id, + Role::Human, + (self.seq * scenario.shape.communities + self.community) % humans, + ); + let mut tags = vec![tag(["h", &channel])]; + let (keys, content) = match self.kind { + PostKind::History => (author, format!("lg-history {}", self.seq)), + PostKind::Root => ( + identity::keys(&scenario.run_id, Role::Owner, self.community), + format!("{ROOT_PREFIX} {}", self.seq), + ), + PostKind::Reply { root } => { + let root_id = roots.get(&(self.community, self.channel, root))?; + let parent = if self.seq == 0 { + root_id.as_str() + } else { + previous? + }; + tags.extend(reply_tags(root_id, parent)); + (author, format!("lg-reply {}", self.seq)) + } + }; + let event = EventBuilder::new(Kind::Custom(9), content) + .tags(tags) + .sign_with_keys(&keys) + .ok()?; + Some((keys, event)) + } +} + +/// NIP-10 markers for a reply. `root` always names the thread root, since the +/// relay rejects a reply whose root differs from its parent's, and `reply` +/// names the parent, from which the relay derives depth (parent depth + 1). +fn reply_tags(root: &str, parent: &str) -> [Tag; 2] { + [ + tag(["e", root, "", "root"]), + tag(["e", parent, "", "reply"]), + ] +} + +/// Up to the 100 newest seeded thread-root ids in `channel`, oldest first. +/// `run` replies into these; a sample of roots is enough load. +pub async fn find_roots( + net: &Net, + community: &Community, + reader: &Keys, + owner: &Keys, + channel: &str, +) -> anyhow::Result> { + let filter = json!([{ + "kinds": [9], + "authors": [owner.public_key().to_hex()], + "#h": [channel], + "limit": 100, + }]); + let (status, body) = net + .post( + reader, + &community.http_url, + "/query", + serde_json::to_vec(&filter)?, + None, + ) + .await?; + if !status.is_success() { + anyhow::bail!("{}", http_reason(status, &body)); + } + let events: Vec = serde_json::from_str(&body)?; + let mut roots: Vec<(u64, String)> = events + .into_iter() + .filter(|e| { + e.content.starts_with(ROOT_PREFIX) && !e.tags.iter().any(|t| t.kind().as_str() == "e") + }) + .map(|e| (e.created_at.as_secs(), e.id.to_hex())) + .collect(); + roots.sort(); + Ok(roots.into_iter().map(|(_, id)| id).collect()) +} + +fn tag(parts: [&str; N]) -> Tag { + Tag::parse(parts).expect("static tag shape") +} + +#[cfg(test)] +mod tests { + use super::*; + use tokio::io::{AsyncBufReadExt, AsyncReadExt, AsyncWriteExt, BufReader}; + + /// Serves HTTP/1.1 on loopback. `handler(method, target, body)` returns + /// the status and JSON body of each response. + async fn fake_http(handler: F) -> String + where + F: Fn(&str, &str, &[u8]) -> (u16, String) + Send + Sync + 'static, + { + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let addr = listener.local_addr().unwrap().to_string(); + let handler = Arc::new(handler); + tokio::spawn(async move { + while let Ok((socket, _)) = listener.accept().await { + let handler = handler.clone(); + tokio::spawn(async move { + let mut reader = BufReader::new(socket); + loop { + let mut request_line = String::new(); + if reader.read_line(&mut request_line).await.unwrap_or(0) == 0 { + return; + } + let mut length = 0; + let mut line = String::new(); + loop { + line.clear(); + if reader.read_line(&mut line).await.unwrap_or(0) == 0 { + return; + } + if line == "\r\n" { + break; + } + if let Some(v) = + line.to_ascii_lowercase().strip_prefix("content-length:") + { + length = v.trim().parse().unwrap(); + } + } + let mut body = vec![0; length]; + reader.read_exact(&mut body).await.unwrap(); + let mut parts = request_line.split_whitespace(); + let (method, target) = (parts.next().unwrap(), parts.next().unwrap()); + let (status, reply) = handler(method, target, &body); + let response = format!( + "HTTP/1.1 {status} X\r\ncontent-type: application/json\r\ncontent-length: {}\r\n\r\n{reply}", + reply.len() + ); + if reader + .get_mut() + .write_all(response.as_bytes()) + .await + .is_err() + { + return; + } + } + }); + } + }); + addr + } + + /// Accepted event id to (thread root id, depth); a root's depth is 0. + type Accepted = Arc>>; + + /// A minimal `/events` endpoint that rejects posts whose content is + /// `reject`. Like the relay it accepts a reply only when its parent was + /// accepted and its `root` marker names the parent's root, and it derives + /// depth as the parent's depth + 1. + async fn fake_relay(reject: Arc) -> (String, Accepted) { + let accepted: Accepted = Default::default(); + let known = accepted.clone(); + let addr = fake_http(move |_, _, body| { + let event: Event = serde_json::from_slice(body).unwrap(); + let mut known = known.lock().unwrap(); + let id = event.id.to_hex(); + let placed = match buzz_core::nip10::parse_thread_markers(&event.tags).resolve() { + None => Some((id.clone(), 0)), + Some((root, parent)) => known + .get(&parent) + .filter(|(parent_root, _)| *parent_root == root || parent == root) + .map(|(_, depth)| (root.clone(), depth + 1)), + }; + let placed = placed.filter(|_| event.content != *reject); + let ok = placed.is_some(); + if let Some(placed) = placed { + known.insert(id, placed); + } + let reply = json!({"accepted": ok, "message": if ok { "" } else { "blocked" }}); + (200, reply.to_string()) + }) + .await; + (addr, accepted) + } + + async fn seed_against(reject: &str) -> (anyhow::Result<()>, Arc, Accepted) { + let (addr, accepted) = fake_relay(reject.into()).await; + let scenario = Scenario::parse(&format!( + r#" +run_id = "t" +[target] +ws_url = "ws://{addr}" +http_url = "http://{addr}" +[shape] +communities = 1 +channels_per_community = 1 +history_messages_per_channel = 2 +thread_roots_per_channel = 2 +thread_depth = 3 +[population] +humans = 4 +agents = 0 +channels_per_member = 1 +[[ramp]] +users = 1.0 +hold_secs = 1 +"# + )) + .unwrap(); + let net = Net::new(&scenario).await.unwrap(); + let stats = Arc::new(Stats::default()); + (seed(&scenario, &net, &stats).await, stats, accepted) + } + + #[tokio::test] + async fn replies_use_accepted_roots_and_missing_posts_fail_the_seed() { + let (outcome, stats, _) = seed_against("none").await; + outcome.expect("every post accepted"); + assert_eq!(seeded(&stats.snapshot(), &Snapshot::default()), 10); + + let (outcome, stats, _) = seed_against("lg-history 1").await; + let err = outcome.expect_err("a rejected post fails the seed"); + assert!(err.to_string().contains("seeded 9 of 10"), "{err}"); + assert_eq!(seeded(&stats.snapshot(), &Snapshot::default()), 9); + + // A rejected root leaves its three replies unpostable. + let (outcome, _, _) = seed_against(&format!("{ROOT_PREFIX} 1")).await; + let err = outcome.expect_err("a rejected root fails the seed"); + assert!(err.to_string().contains("seeded 6 of 10"), "{err}"); + + // A rejected reply leaves the replies chained to it unpostable. + let (outcome, _, _) = seed_against("lg-reply 1").await; + let err = outcome.expect_err("a rejected reply fails the seed"); + assert!(err.to_string().contains("seeded 6 of 10"), "{err}"); + } + + #[tokio::test] + async fn thread_depth_builds_deep_threads() { + let (outcome, _, accepted) = seed_against("none").await; + outcome.expect("every post accepted"); + // Two threads of three chained replies, not six direct replies. + let mut depths: Vec = accepted + .lock() + .unwrap() + .values() + .map(|(_, depth)| *depth) + .filter(|depth| *depth > 0) + .collect(); + depths.sort(); + assert_eq!(depths, [1, 1, 2, 2, 3, 3]); + } + + #[test] + fn reply_k_points_reply_at_reply_k_minus_one_and_root_at_the_root() { + let scenario = Scenario::parse( + r#" +run_id = "t" +[target] +ws_url = "ws://localhost:3000" +http_url = "http://localhost:3000" +[shape] +communities = 1 +channels_per_community = 1 +thread_roots_per_channel = 1 +thread_depth = 4 +[population] +humans = 2 +agents = 0 +channels_per_member = 1 +[[ramp]] +users = 1.0 +hold_secs = 1 +"#, + ) + .unwrap(); + let (roots, rest) = plan_posts(&scenario); + assert_eq!(roots.len(), 1); + assert_eq!(rest.len(), 1, "one chain per thread"); + let root_id = "a".repeat(64); + let known: RootIds = [((0, 0, 0), root_id.clone())].into(); + + let mut previous: Option = None; + for (k, post) in rest[0].iter().enumerate() { + let (_, event) = post + .build(&scenario, &known, previous.as_deref()) + .expect("reply builds once its parent is known"); + let markers = buzz_core::nip10::parse_thread_markers(&event.tags); + assert_eq!(markers.root.as_deref(), Some(root_id.as_str()), "reply {k}"); + let parent = previous.as_deref().unwrap_or(&root_id); + assert_eq!(markers.reply.as_deref(), Some(parent), "reply {k}"); + previous = Some(event.id.to_hex()); + } + // A reply cannot be placed without its predecessor. + assert!(rest[0][2].build(&scenario, &known, None).is_none()); + } + + /// An operator endpoint that lists hosts by owner, as `GET + /// /operator/communities` does, and refuses a create that is not + /// `create_only` (which would converge ownership on an existing host). + async fn fake_operator(hosts: Vec<(String, String)>) -> String { + let hosts = std::sync::Mutex::new(hosts); + fake_http(move |method, target, body| { + let mut hosts = hosts.lock().unwrap(); + if method == "GET" { + let owner = target.rsplit('=').next().unwrap(); + let owned: Vec<_> = hosts + .iter() + .filter(|(_, o)| o == owner) + .map(|(host, _)| json!({ "host": host })) + .collect(); + return (200, json!({ "communities": owned }).to_string()); + } + let request: serde_json::Value = serde_json::from_slice(body).unwrap(); + if request["create_only"] != true { + return (400, json!({"error": "create_only required"}).to_string()); + } + let host = request["host"].as_str().unwrap().to_string(); + if hosts.iter().any(|(h, _)| *h == host) { + return ( + 409, + json!({"error": "community already exists"}).to_string(), + ); + } + let owner = request["initial_owner_pubkey"].as_str().unwrap(); + hosts.push((host, owner.to_string())); + (200, json!({"status": "created"}).to_string()) + }) + .await + } + + #[tokio::test] + async fn provisioning_never_takes_over_a_host_the_fixture_owner_does_not_own() { + let owner = identity::keys("t", Role::Owner, 0); + let someone_else = Keys::generate().public_key().to_hex(); + let operator = Keys::generate(); + let addr = fake_operator(vec![ + ("mine.example".to_string(), owner.public_key().to_hex()), + ("theirs.example".to_string(), someone_else), + ]) + .await; + let base = format!("http://{addr}"); + let scenario = Scenario::parse(&format!( + r#" +run_id = "t" +[target] +ws_url = "ws://{addr}" +http_url = "http://{addr}" +[shape] +communities = 1 +channels_per_community = 1 +[population] +humans = 1 +agents = 0 +channels_per_member = 1 +[[ramp]] +users = 1.0 +hold_secs = 1 +"# + )) + .unwrap(); + let net = Net::new(&scenario).await.unwrap(); + let provision = |host: &'static str| provision_host(&net, &base, &operator, host, &owner); + + provision("fresh.example") + .await + .expect("a new host is created"); + provision("fresh.example") + .await + .expect("re-running reuses a host the fixture owner owns"); + provision("mine.example") + .await + .expect("an existing host owned by the fixture owner is reused"); + let err = provision("theirs.example") + .await + .expect_err("another owner's host is not taken over"); + assert!(err.to_string().contains("not owned by"), "{err}"); + } + + #[tokio::test] + async fn only_already_exists_channel_errors_are_ignored() { + let outcome = |message: &'static str| async move { + let addr = fake_http(move |_, _, _| { + ( + 200, + json!({"accepted": false, "message": message}).to_string(), + ) + }) + .await; + let scenario = Scenario::parse(&format!( + r#" +run_id = "t" +[target] +ws_url = "ws://{addr}" +http_url = "http://{addr}" +[shape] +communities = 1 +channels_per_community = 2 +[population] +humans = 1 +agents = 0 +channels_per_member = 1 +[[ramp]] +users = 1.0 +hold_secs = 1 +"# + )) + .unwrap(); + let net = Net::new(&scenario).await.unwrap(); + let community = &scenario.communities()[0]; + let owner = identity::keys("t", Role::Owner, 0); + create_channels(&scenario, &net, community, &owner).await + }; + outcome("duplicate: channel already exists") + .await + .expect("an existing channel is fine"); + let err = outcome("invalid: community does not exist") + .await + .expect_err("a missing community is not an existing channel"); + assert!(err.to_string().contains("does not exist"), "{err}"); + } +} diff --git a/crates/buzz-test-client/src/loadgen/stats.rs b/crates/buzz-test-client/src/loadgen/stats.rs new file mode 100644 index 0000000000..08c2a47c9d --- /dev/null +++ b/crates/buzz-test-client/src/loadgen/stats.rs @@ -0,0 +1,593 @@ +//! Lock-free latency histograms and counters shared by every simulated client. + +use std::collections::HashMap; +use std::sync::atomic::{AtomicU64, Ordering}; +use std::sync::Mutex; +use std::time::{Duration, Instant}; + +use serde_json::{json, Map, Value}; + +/// A measured operation. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub enum Op { + WsConnect, + History, + PublishEphemeral, + PublishMessage, + Deliver, + HttpQuery, + HttpEvents, + Seed, + Subscribe, +} + +impl Op { + pub const ALL: [Op; 9] = [ + Op::WsConnect, + Op::History, + Op::PublishEphemeral, + Op::PublishMessage, + Op::Deliver, + Op::HttpQuery, + Op::HttpEvents, + Op::Seed, + Op::Subscribe, + ]; + + pub fn name(self) -> &'static str { + match self { + Op::WsConnect => "ws_connect_auth", + Op::History => "history_eose", + Op::PublishEphemeral => "publish_ephemeral_ok", + Op::PublishMessage => "publish_message_ok", + Op::Deliver => "deliver", + Op::HttpQuery => "http_query", + Op::HttpEvents => "http_events", + Op::Seed => "seed", + Op::Subscribe => "subscribe_eose", + } + } +} + +/// Why a step crossed the abort thresholds, if it did. Every operation but +/// seeding counts toward the latency limit; delivery has no error signal of +/// its own, so it is left out of the error-rate limit. Operations in flight +/// when the window ends are reported but not judged: they have not failed +/// yet. The run stops and each one settles, so a stalled operation shows up +/// as an error in the shutdown window. +pub fn abort_reason( + summaries: &[(Op, OpSummary)], + max_error_rate: f64, + max_p99_ms: f64, +) -> Option { + summaries + .iter() + .filter(|(op, s)| *op != Op::Seed && s.count + s.errors > 0) + .find_map(|(op, s)| { + if *op != Op::Deliver && s.error_rate() > max_error_rate { + Some(format!( + "{} error rate {:.3} > {max_error_rate}", + op.name(), + s.error_rate() + )) + } else if s.p99_ms > max_p99_ms { + Some(format!( + "{} p99 {:.1} ms > {max_p99_ms} ms", + op.name(), + s.p99_ms + )) + } else { + None + } + }) +} + +/// Bucket `i` holds latencies up to `GROWTH^i` microseconds (2% resolution). +const GROWTH: f64 = 1.02; +const BUCKETS: usize = 1024; + +fn bucket(d: Duration) -> usize { + let us = d.as_micros().max(1) as f64; + ((us.ln() / GROWTH.ln()).ceil() as usize).min(BUCKETS - 1) +} + +fn bucket_ms(i: usize) -> f64 { + GROWTH.powi(i as i32) / 1_000.0 +} + +struct OpStats { + buckets: Vec, + errors: AtomicU64, + throttled: AtomicU64, + /// Operations begun with [`Stats::begin`] and not yet finished. + in_flight: AtomicU64, +} + +impl OpStats { + fn new() -> Self { + Self { + buckets: (0..BUCKETS).map(|_| AtomicU64::new(0)).collect(), + errors: AtomicU64::new(0), + throttled: AtomicU64::new(0), + in_flight: AtomicU64::new(0), + } + } +} + +/// Cumulative counters for every [`Op`]. +pub struct Stats { + ops: Vec, + reasons: Mutex>, +} + +impl Default for Stats { + fn default() -> Self { + Self { + ops: Op::ALL.iter().map(|_| OpStats::new()).collect(), + reasons: Mutex::new(HashMap::new()), + } + } +} + +impl Stats { + fn op(&self, op: Op) -> &OpStats { + &self.ops[Op::ALL + .iter() + .position(|o| *o == op) + .expect("op listed in ALL")] + } + + pub fn ok(&self, op: Op, latency: Duration) { + self.op(op).buckets[bucket(latency)].fetch_add(1, Ordering::Relaxed); + } + + /// Starts timing an operation. The guard counts as in flight until it is + /// finished; one dropped unfinished, which only happens when its client is + /// cancelled, records an error. + pub fn begin(&self, op: Op) -> Pending<'_> { + self.op(op).in_flight.fetch_add(1, Ordering::Relaxed); + Pending { + stats: self, + op, + start: Instant::now(), + finished: false, + } + } + + /// Records a failure. Quota rejections count as throttled, not errors, so a + /// run can tell relay limits apart from relay failures. + pub fn err(&self, op: Op, reason: &str) { + let throttled = is_throttle(reason); + let stats = self.op(op); + if throttled { + stats.throttled.fetch_add(1, Ordering::Relaxed); + } else { + stats.errors.fetch_add(1, Ordering::Relaxed); + } + let key = format!("{}: {}", op.name(), normalize(reason)); + *self + .reasons + .lock() + .expect("reasons lock") + .entry(key) + .or_default() += 1; + } + + pub fn snapshot(&self) -> Snapshot { + Snapshot { + ops: self + .ops + .iter() + .map(|s| OpSnapshot { + buckets: s + .buckets + .iter() + .map(|b| b.load(Ordering::Relaxed)) + .collect(), + errors: s.errors.load(Ordering::Relaxed), + throttled: s.throttled.load(Ordering::Relaxed), + in_flight: s.in_flight.load(Ordering::Relaxed), + }) + .collect(), + reasons: self.reasons.lock().expect("reasons lock").clone(), + } + } +} + +/// An operation started with [`Stats::begin`]. +#[must_use = "an unfinished guard records an error when dropped"] +pub struct Pending<'a> { + stats: &'a Stats, + op: Op, + start: Instant, + finished: bool, +} + +impl Pending<'_> { + /// Records success with the latency since [`Stats::begin`]. + pub fn ok(mut self) { + self.stats.ok(self.op, self.start.elapsed()); + self.finish(); + } + + /// Records a failure. + pub fn err(mut self, reason: &str) { + self.stats.err(self.op, reason); + self.finish(); + } + + /// Ends the operation without recording it, for a failure the caller + /// accounts for elsewhere. + pub fn forget(mut self) { + self.finish(); + } + + /// Leaves the in-flight gauge after the outcome is recorded, so a + /// snapshot between the two sees the operation twice, never not at all. + fn finish(&mut self) { + self.finished = true; + self.stats + .op(self.op) + .in_flight + .fetch_sub(1, Ordering::Relaxed); + } +} + +impl Drop for Pending<'_> { + fn drop(&mut self) { + if !self.finished { + self.stats.err(self.op, "unfinished at stop"); + self.finish(); + } + } +} + +/// Start of a reason built by `net::http_reason` for an HTTP 429. +const HTTP_TOO_MANY_REQUESTS: &str = "HTTP 429 "; +/// The relay's quota rejection, over WebSocket (`OK`, `CLOSED`) and in HTTP bodies. +const QUOTA_EXCEEDED: &str = "rate-limited: quota exceeded"; + +/// Quota rejections only: an HTTP 429 status or the relay's quota text. Other +/// `rate-limited:` rejections, such as `shared admission unavailable` when the +/// relay's admission store is down, are relay failures and count as errors. +fn is_throttle(reason: &str) -> bool { + reason.starts_with(HTTP_TOO_MANY_REQUESTS) || reason.contains(QUOTA_EXCEEDED) +} + +/// Collapses ids and numbers so equal failures aggregate under one reason. +fn normalize(reason: &str) -> String { + let mut out: String = reason + .split_whitespace() + .map(|w| { + if w.chars().any(|c| c.is_ascii_digit()) && w.len() > 3 { + "" + } else { + w + } + }) + .collect::>() + .join(" "); + if let Some((cut, _)) = out.char_indices().nth(120) { + out.truncate(cut); + } + out +} + +#[derive(Clone, Default)] +struct OpSnapshot { + buckets: Vec, + errors: u64, + throttled: u64, + in_flight: u64, +} + +/// Point-in-time copy of [`Stats`]; subtract two to report one interval. +#[derive(Clone, Default)] +pub struct Snapshot { + ops: Vec, + reasons: HashMap, +} + +/// Summary of one operation over an interval. +#[derive(Clone, Copy, Debug, Default)] +pub struct OpSummary { + pub count: u64, + pub errors: u64, + pub throttled: u64, + /// Operations in flight when the interval ended. + pub pending: u64, + pub p50_ms: f64, + pub p90_ms: f64, + pub p99_ms: f64, + pub max_ms: f64, +} + +impl OpSummary { + pub fn error_rate(&self) -> f64 { + let total = self.count + self.errors; + if total == 0 { + 0.0 + } else { + self.errors as f64 / total as f64 + } + } +} + +impl Snapshot { + /// Per-op summaries for the interval `earlier..self`. + pub fn since(&self, earlier: &Snapshot) -> Vec<(Op, OpSummary)> { + Op::ALL + .iter() + .enumerate() + .map(|(i, op)| { + let now = &self.ops[i]; + let then = earlier.ops.get(i).cloned().unwrap_or_default(); + let buckets: Vec = now + .buckets + .iter() + .enumerate() + .map(|(b, n)| n - then.buckets.get(b).copied().unwrap_or(0)) + .collect(); + ( + *op, + summarize( + &buckets, + now.errors - then.errors, + now.throttled - then.throttled, + now.in_flight, + ), + ) + }) + .collect() + } + + /// Failure reasons counted during `earlier..self`, most frequent first. + pub fn reasons_since(&self, earlier: &Snapshot, limit: usize) -> Vec<(String, u64)> { + let mut out: Vec<(String, u64)> = self + .reasons + .iter() + .map(|(k, v)| (k.clone(), v - earlier.reasons.get(k).copied().unwrap_or(0))) + .filter(|(_, v)| *v > 0) + .collect(); + out.sort_by_key(|entry| std::cmp::Reverse(entry.1)); + out.truncate(limit); + out + } +} + +fn summarize(buckets: &[u64], errors: u64, throttled: u64, pending: u64) -> OpSummary { + let count: u64 = buckets.iter().sum(); + let pct = |q: f64| { + if count == 0 { + return 0.0; + } + let rank = ((count as f64) * q).ceil().max(1.0) as u64; + let mut seen = 0; + for (i, n) in buckets.iter().enumerate() { + seen += n; + if seen >= rank { + return bucket_ms(i); + } + } + bucket_ms(BUCKETS - 1) + }; + OpSummary { + count, + errors, + throttled, + pending, + p50_ms: pct(0.50), + p90_ms: pct(0.90), + p99_ms: pct(0.99), + max_ms: buckets + .iter() + .rposition(|n| *n > 0) + .map(bucket_ms) + .unwrap_or(0.0), + } +} + +/// JSON object of the non-empty op summaries in an interval of `secs` seconds. +pub fn ops_json(summaries: &[(Op, OpSummary)], secs: f64) -> Value { + let mut map = Map::new(); + for (op, s) in summaries { + if s.count + s.errors + s.throttled + s.pending == 0 { + continue; + } + map.insert( + op.name().into(), + json!({ + "count": s.count, + "errors": s.errors, + "throttled": s.throttled, + "pending": s.pending, + "per_sec": round(s.count as f64 / secs.max(1e-9)), + "p50_ms": round(s.p50_ms), + "p90_ms": round(s.p90_ms), + "p99_ms": round(s.p99_ms), + "max_ms": round(s.max_ms), + }), + ); + } + Value::Object(map) +} + +fn round(v: f64) -> f64 { + (v * 100.0).round() / 100.0 +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn percentiles_are_within_bucket_resolution() { + let stats = Stats::default(); + for ms in 1..=100 { + stats.ok(Op::HttpQuery, Duration::from_millis(ms)); + } + let base = Snapshot::default(); + let summary = stats.snapshot().since(&base)[5].1; + assert_eq!(summary.count, 100); + assert!((summary.p50_ms - 50.0).abs() <= 1.1, "{summary:?}"); + assert!((summary.p99_ms - 99.0).abs() <= 2.0, "{summary:?}"); + assert!((summary.max_ms - 100.0).abs() <= 2.0, "{summary:?}"); + } + + #[test] + fn intervals_subtract_and_throttles_are_not_errors() { + let stats = Stats::default(); + stats.ok(Op::PublishMessage, Duration::from_millis(5)); + let first = stats.snapshot(); + stats.ok(Op::PublishMessage, Duration::from_millis(500)); + stats.err( + Op::PublishMessage, + "rate-limited: quota exceeded; retry in 12345s", + ); + stats.err(Op::PublishMessage, "error: internal 98765"); + let interval = stats.snapshot().since(&first); + let s = interval + .iter() + .find(|(o, _)| *o == Op::PublishMessage) + .unwrap() + .1; + assert_eq!((s.count, s.errors, s.throttled), (1, 1, 1)); + assert!(s.p50_ms > 400.0); + let reasons = stats.snapshot().reasons_since(&first, 10); + assert!(reasons + .iter() + .any(|(r, _)| r == "publish_message_ok: error: internal ")); + } + + #[test] + fn long_reasons_truncate_on_char_boundaries() { + let stats = Stats::default(); + for reason in ["a".repeat(119) + "é tail", "é".repeat(200), "x".repeat(300)] { + stats.err(Op::HttpEvents, &reason); + } + let reasons = stats.snapshot().reasons_since(&Snapshot::default(), 10); + assert_eq!(reasons.len(), 3); + assert!(reasons + .iter() + .all(|(r, _)| r.chars().count() <= "http_events: ".len() + 120)); + } + + #[test] + fn abort_judges_error_rate_and_every_latency_including_delivery() { + let stats = Stats::default(); + stats.ok(Op::PublishMessage, Duration::from_millis(2)); + stats.ok(Op::Deliver, Duration::from_millis(900)); + let base = Snapshot::default(); + let slow_delivery = stats.snapshot().since(&base); + let reason = abort_reason(&slow_delivery, 0.05, 500.0).expect("slow delivery aborts"); + assert!(reason.starts_with("deliver p99"), "{reason}"); + assert_eq!(abort_reason(&slow_delivery, 0.05, 5_000.0), None); + + stats.err(Op::PublishMessage, "error: boom"); + let failing = stats.snapshot().since(&base); + let reason = abort_reason(&failing, 0.05, 5_000.0).expect("errors abort"); + assert!( + reason.starts_with("publish_message_ok error rate"), + "{reason}" + ); + + let seeding = Stats::default(); + seeding.err(Op::Seed, "error: boom"); + assert_eq!( + abort_reason(&seeding.snapshot().since(&base), 0.05, 1.0), + None + ); + } + + fn summary(stats: &Stats, op: Op) -> OpSummary { + let window = stats.snapshot().since(&Snapshot::default()); + window.iter().find(|(o, _)| *o == op).unwrap().1 + } + + #[test] + fn begun_operations_count_in_flight_until_finished_and_unfinished_drops_are_errors() { + let stats = Stats::default(); + let waiting = stats.begin(Op::WsConnect); + let succeeds = stats.begin(Op::WsConnect); + let fails = stats.begin(Op::WsConnect); + let handled_elsewhere = stats.begin(Op::WsConnect); + assert_eq!(summary(&stats, Op::WsConnect).pending, 4); + + succeeds.ok(); + fails.err("connect timeout"); + handled_elsewhere.forget(); + let s = summary(&stats, Op::WsConnect); + assert_eq!((s.count, s.errors, s.pending), (1, 1, 1)); + + // A guard dropped unfinished, as when its task is aborted, is an + // error, not a vanished sample. + drop(waiting); + let s = summary(&stats, Op::WsConnect); + assert_eq!((s.count, s.errors, s.pending), (1, 2, 0)); + let reasons = stats.snapshot().reasons_since(&Snapshot::default(), 10); + assert!(reasons + .iter() + .any(|(r, n)| r == "ws_connect_auth: unfinished at stop" && *n == 1)); + } + + #[test] + fn in_flight_operations_are_reported_but_not_judged() { + let stats = Stats::default(); + let straddling = stats.begin(Op::PublishEphemeral); + let window = stats.snapshot().since(&Snapshot::default()); + // One healthy operation still in flight is not a 100% error rate. + assert_eq!(abort_reason(&window, 0.0, 5_000.0), None); + assert_eq!(summary(&stats, Op::PublishEphemeral).error_rate(), 0.0); + assert_eq!(ops_json(&window, 1.0)["publish_ephemeral_ok"]["pending"], 1); + + // Once it settles as a failure, it counts. + straddling.err("OK timeout"); + let window = stats.snapshot().since(&Snapshot::default()); + let reason = abort_reason(&window, 0.05, 5_000.0).expect("a settled error aborts"); + assert!( + reason.starts_with("publish_ephemeral_ok error rate 1.000"), + "{reason}" + ); + } + + #[test] + fn only_quota_rejections_are_throttles() { + use crate::loadgen::net::http_reason; + use reqwest::StatusCode; + let throttles = [ + // WebSocket OK and CLOSED text from `rejection.rs`. + "rate-limited: quota exceeded; retry in 7s", + "CLOSED live: rate-limited: quota exceeded; retry in 7s", + // HTTP 429 from `api/bridge.rs`, with and without its body. + &http_reason( + StatusCode::TOO_MANY_REQUESTS, + r#"{"error":"rate-limited: quota exceeded; retry in 3s"}"#, + ), + &http_reason(StatusCode::TOO_MANY_REQUESTS, "slow down"), + ]; + let errors = [ + // The admission store is down: an outage, not a quota. + "rate-limited: shared admission unavailable", + &http_reason( + StatusCode::SERVICE_UNAVAILABLE, + r#"{"error":"rate-limited: shared admission unavailable"}"#, + ), + "rate-limited: too many concurrent requests", + // Numbers that merely contain 429, and prose that mentions limits. + "invalid: created_at 1793242912 is too far in the past", + "error: event 429abc is not valid", + "rate limit configuration is invalid", + &http_reason(StatusCode::BAD_REQUEST, "created_at 1793242912 rejected"), + &http_reason(StatusCode::INTERNAL_SERVER_ERROR, "HTTP 429 upstream"), + ]; + let stats = Stats::default(); + for reason in throttles.iter().chain(errors.iter()) { + stats.err(Op::HttpEvents, reason); + } + let s = summary(&stats, Op::HttpEvents); + assert_eq!( + (s.throttled, s.errors), + (throttles.len() as u64, errors.len() as u64) + ); + } +} diff --git a/crates/buzz-ws-client/src/connection.rs b/crates/buzz-ws-client/src/connection.rs index 5382292763..ade9ba6a9d 100644 --- a/crates/buzz-ws-client/src/connection.rs +++ b/crates/buzz-ws-client/src/connection.rs @@ -7,7 +7,9 @@ use serde_json::{json, Value}; use tokio::time::timeout; use tokio_tungstenite::tungstenite::client::IntoClientRequest; use tokio_tungstenite::tungstenite::http::HeaderMap; -use tokio_tungstenite::{connect_async, tungstenite::Message, MaybeTlsStream, WebSocketStream}; +use tokio_tungstenite::{ + client_async_tls, connect_async, tungstenite::Message, MaybeTlsStream, WebSocketStream, +}; use tracing::debug; use crate::error::WsClientError; @@ -85,12 +87,39 @@ impl NostrWsConnection { debug!("connected to relay at {url}"); - Ok(Self { + Ok(Self::from_stream(ws, url)) + } + + /// [`Self::connect`], but opens the TCP connection to `addr` (`host:port`) + /// instead of resolving the URL's host. + /// + /// The URL still supplies the `Host` header, TLS server name and NIP-42 + /// relay tag, so a client inside the relay's cluster can reach a community + /// host through the relay's service address. + pub async fn connect_to_addr(url: &str, addr: &str) -> Result { + let parsed = url + .parse::() + .map_err(|e| WsClientError::Url(e.to_string()))?; + let request = parsed.as_str().into_client_request()?; + let tcp = tokio::net::TcpStream::connect(addr) + .await + .map_err(|e| WsClientError::WebSocket(e.into()))?; + let (ws, _response) = client_async_tls(request, tcp) + .await + .map_err(WsClientError::WebSocket)?; + + debug!("connected to relay at {url} via {addr}"); + + Ok(Self::from_stream(ws, url)) + } + + fn from_stream(ws: WsStream, url: &str) -> Self { + Self { ws, buffer: VecDeque::new(), pending_challenge: None, relay_url: url.to_string(), - }) + } } /// Performs NIP-42 authentication using `keys` against the connected relay. diff --git a/perf/LOADGEN.md b/perf/LOADGEN.md new file mode 100644 index 0000000000..cc26fa5ebd --- /dev/null +++ b/perf/LOADGEN.md @@ -0,0 +1,125 @@ +# Relay load generator + +`buzz-loadgen` (in `crates/buzz-test-client`) seeds a data shape and then +drives a population of simulated humans and agents through a ramp of user +multipliers, reporting latency percentiles, errors and rate-limit rejections +per operation for each step. + +```bash +cargo run --release -p buzz-test-client --bin buzz-loadgen -- seed perf/scenarios/local-smoke.toml +cargo run --release -p buzz-test-client --bin buzz-loadgen -- run perf/scenarios/local-smoke.toml +``` + +## Scale vectors + +| Vector | Scenario knob | +|---|---| +| Communities | `shape.communities` (each provisioned through `POST /operator/communities`) | +| Channels | `shape.channels_per_community`, `population.channels_per_member` | +| History depth | `shape.history_messages_per_channel`, `human.history_limit` | +| Deep threads | `shape.thread_roots_per_channel`, `shape.thread_depth`, `human.thread_reply_fraction` | +| Humans / agents | `population.humans`, `population.agents`, multiplied by each `[[ramp]]` step's `users` | +| Throughput per client | `[human]` and `[agent]` rates | +| Reconnect churn | `human.mean_session_secs`, `population.connect_rate_per_sec` | + +`shape.thread_depth` seeds each thread as a chain: reply `k` replies to reply +`k-1`, so the thread's deepest reply sits at depth `thread_depth`. The relay +rejects depths above 100, so scenarios do too. Threads are seeded in parallel +with each other. + +A scenario that cannot be scheduled is rejected when it is parsed: for +example `seed_concurrency` of 0, `human.presence_interval_secs` below 1, or a +`community_host_template` with a port while `target.connect_addr` is set (the +override maps hostnames, so HTTP would ignore it). + +## What each client does + +- **Human**: authenticates over WebSocket and subscribes to its channels' + messages, its mentions, typing, and ten peers' presence. On connect it + fetches `history_limit` messages per channel. Then it sends presence + heartbeats, typing indicators, messages (some as replies into seeded deep + threads) and reactions on Poisson schedules. +- **Agent**: authenticates with a NIP-OA tag from its owner, polls HTTP + `/query` (channel window plus mentions), posts messages through HTTP + `/events`, and sends encrypted kind:24200 observer frames to its owner. + +Messages carry their send time, so every subscriber records publish→delivery +latency (`deliver`). Across machines this depends on clock sync. + +## Modeling a deployment + +The `[human]` and `[agent]` defaults are round placeholders. To model a known +deployment, divide its hourly event counts by its client counts and set the +results as the scenario's rates. Then a ramp step's `users` reads as a +multiple of that deployment's population. Keep such scenarios with that +deployment's own configuration. + +## Targeting and membership + +`target.connect_addr` (`host:port`) applies only together with +`target.community_host_template`: it then replaces host resolution for the +community hosts, over both WebSocket and HTTP, while TLS names, `Host` headers +and the NIP-42 relay tag still name the community. Without a host +template, the generator connects to `ws_url` and `http_url` as written and +`connect_addr` has no effect. + +Seeded identities do not join communities or channels. A relay that requires +membership rejects both `seed` and `run`. + +## Sharding + +`run --shard i/n` runs the clients whose global index `g` satisfies +`g % n == i`. Identities, community hosts and channel ids derive from +`run_id`, so shards need no coordination: run `seed` once, then start `n` +processes, such as a Kubernetes Indexed Job, with the same scenario. Each +shard reports its own clients; add counts and take the worst percentiles +across shards. + +## Abort + +Each step reports every operation from the start of its ramp to the end of +its hold, so connection failures during the ramp count. At the end of the +hold, the run stops if any operation's p99 (publish→delivery included) +exceeds `abort.max_p99_ms`, if any operation other than delivery has an +error rate above `abort.max_error_rate`, or if a client exited unexpectedly. +An operation still in flight when the hold ends is reported as `pending` +but not judged: it has not failed yet. A relay `CLOSED` on a live +subscription counts as a `subscribe_eose` error and ends that client's +session. Each connect plus authentication must finish within +`target.connect_timeout_secs` (default 10). Quota rejections, meaning an HTTP 429 or the relay's +`rate-limited: quota exceeded` text, are reported as `throttled`, not errors. +Every other rejection, including `rate-limited: shared admission unavailable` +(the relay's admission store is down), counts as an error. +Ramp steps must increase `users`, since clients are only ever added. + +When the run stops, clients get 15 seconds to settle what they have in flight +and disconnect. An operation already under way runs to its own deadline +instead of being abandoned: a connect attempt finishes within +`target.connect_timeout_secs`, and a publish waits at most 30 seconds for its +`OK`. A stalled connect therefore ends as a `connect timeout` error and a +healthy one as a success. What settles after the last step is reported in the +`shutdown` object (`ops`, `failures`, `abort_reason`) and judged by the same +limits as a step, so a connect that never completes fails the run there. A client +still busy when the grace period ends is cancelled; its in-flight operations +are recorded as `unfinished at stop` errors, and the cancellation fails the run +whatever the limits. Keep `target.connect_timeout_secs` below 15, or a stalled +connect is cancelled instead of timing out. Publishes, history fetches and +HTTP requests can outlast the grace period (30 s, 30 s and 60 s deadlines), in +which case the client is cancelled and the run fails. + +The process exits non-zero when a step aborts, a client panics, a client had to +be cancelled, or `shutdown.abort_reason` is set. + +`seed` exits non-zero unless the relay accepted every planned post. + +`seed` provisions each community with `create_only`, so it never rotates the +owner of an existing host. A host that already exists is reused only when the +fixture owner derived from `run_id` owns it (re-running `seed` for the same +scenario); otherwise `seed` fails. Use a fresh `run_id` or a host template +with `{run}` to avoid colliding with another community. + +## Safety + +The generator creates real communities, channels and events. Point it only at +environments built for load tests, never at a production relay. Use a fresh +`run_id` per fixture. diff --git a/perf/scenarios/local-smoke.toml b/perf/scenarios/local-smoke.toml new file mode 100644 index 0000000000..d52362aefe --- /dev/null +++ b/perf/scenarios/local-smoke.toml @@ -0,0 +1,41 @@ +# Two-minute smoke run against a local relay (`just relay`). +# cargo run -p buzz-test-client --bin buzz-loadgen -- seed perf/scenarios/local-smoke.toml +# cargo run -p buzz-test-client --bin buzz-loadgen -- run perf/scenarios/local-smoke.toml +run_id = "local-smoke" + +[target] +ws_url = "ws://localhost:3000" +http_url = "http://localhost:3000" + +[shape] +communities = 1 +channels_per_community = 10 +history_messages_per_channel = 20 +thread_roots_per_channel = 1 +thread_depth = 50 + +[population] +humans = 50 +agents = 5 +channels_per_member = 3 +connect_rate_per_sec = 50.0 + +# Rates far above the defaults so a short run exercises every operation. +[human] +presence_interval_secs = 10.0 +typing_per_hour = 360.0 +messages_per_hour = 120.0 +reactions_per_hour = 60.0 + +[agent] +query_interval_secs = 2.0 +events_per_hour = 360.0 +observer_frames_per_hour = 360.0 + +[[ramp]] +users = 1.0 +hold_secs = 30 + +[[ramp]] +users = 2.0 +hold_secs = 30 diff --git a/scripts/run-tests.sh b/scripts/run-tests.sh index 77319c0795..43a153a18b 100755 --- a/scripts/run-tests.sh +++ b/scripts/run-tests.sh @@ -422,6 +422,10 @@ run_unit_tests() { run_test_step "buzz-relay boot lifecycle tests" \ cargo test -p buzz-relay --test boot_lifecycle -- --nocapture + + # Load generator unit tests are infra-free. Keep in step with `just test-unit`. + run_test_step "buzz-test-client loadgen unit tests" \ + cargo test -p buzz-test-client --lib loadgen:: -- --nocapture } # ---- DB / integration tests (infra required) --------------------------------