jerry/buzz
1
0
Fork 0
mirror of https://github.com/block/buzz.git synced 2026-11-04 16:31:38 +00:00
Code Issues Packages Projects Releases Wiki Activity

Add a scenario-driven relay load generator (#8636)

Browse Source 
## 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 <scenario.toml>` creates communities (through `POST
/operator/communities`), channels, channel history, and deep threads.
- `buzz-loadgen run <scenario.toml>` 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 <tornquist@squareup.com>
Co-authored-by: Elrond <28d6302a099e5225b02c4155ac4236e4912603df2ab08dbfc2f4fef08ce598c8@buzz.block.builderlab.xyz>
Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
Luke Tornquist
2026-10-29 13:49:37 -04:00
committed by GitHub
parent 8caf8351c1
commit aed4be8592
17 changed files with 4074 additions and 121 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
Cargo.lock generated
View File

@@ -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",

3
Justfile
View File

@@ -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

10
crates/buzz-test-client/Cargo.toml
View File

@@ -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"

85
crates/buzz-test-client/src/bin/loadgen.rs Normal file
View File

@@ -0,0 +1,85 @@
//! Scenario-driven relay load generator. See `perf/LOADGEN.md`.
//!
//! Usage:
//! buzz-loadgen seed <scenario.toml>
//! buzz-loadgen run <scenario.toml> [--shard i/n] [--report <path>]
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<PathBuf>,
},
}
#[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
}
}
}

118
crates/buzz-test-client/src/bin/wamp_bench.rs
View File

@@ -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 <channel_uuid> <qps> <duration_secs> <conns> <latency_out>
//! 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<String> = std::env::args().collect();
if args.len() != 6 {
eprintln!("Usage: wamp-bench <channel_uuid> <qps> <duration_secs> <conns> <latency_out>");
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<f64> = 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<f64> = 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(())
}

5
crates/buzz-test-client/src/lib.rs
View File

@@ -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 {

113
crates/buzz-test-client/src/loadgen/identity.rs Normal file
View File

@@ -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());
}
}

18
crates/buzz-test-client/src/loadgen/mod.rs Normal file
View File

@@ -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;

200
crates/buzz-test-client/src/loadgen/net.rs Normal file
View File

@@ -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<String>,
connect_timeout: Duration,
}
impl Net {
pub async fn new(scenario: &Scenario) -> anyhow::Result<Self> {
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<NostrWsConnection> {
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<u8>,
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 <status>`,
/// 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();
}
}

1654
crates/buzz-test-client/src/loadgen/run.rs Normal file
View File

File diff suppressed because it is too large Load Diff

499
crates/buzz-test-client/src/loadgen/scenario.rs Normal file
View File

@@ -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<RampStep>,
#[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<String>,
/// 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<String>,
/// Deployment-root HTTP URL serving `POST /operator/communities`.
#[serde(default)]
pub operator_url: Option<String>,
/// 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<String>,
pub ws_url: String,
pub http_url: String,
}
impl Scenario {
pub fn parse(text: &str) -> Result<Self, String> {
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<Community> {
(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<String> {
(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<String> {
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}");
}
}

690
crates/buzz-test-client/src/loadgen/seed.rs Normal file
View File

@@ -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<Stats>) -> 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<Post>;
/// 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<Chain>,
roots: &RootIds,
) -> RootIds {
stream::iter(chains)
.map(|chain| async move {
let mut accepted = Vec::new();
let mut previous: Option<String> = 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<anyhow::Result<()>> = 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<Chain>, Vec<Chain>) {
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<Vec<String>> {
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<Event> = 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<const N: usize>(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<F>(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<std::sync::Mutex<HashMap<String, (String, usize)>>>;
/// 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<str>) -> (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<Stats>, 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<usize> = 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<String> = 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}");
}
}

593
crates/buzz-test-client/src/loadgen/stats.rs Normal file
View File

@@ -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<String> {
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<AtomicU64>,
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<OpStats>,
reasons: Mutex<HashMap<String, u64>>,
}
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 {
"<n>"
} else {
w
}
})
.collect::<Vec<_>>()
.join(" ");
if let Some((cut, _)) = out.char_indices().nth(120) {
out.truncate(cut);
}
out
}
#[derive(Clone, Default)]
struct OpSnapshot {
buckets: Vec<u64>,
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<OpSnapshot>,
reasons: HashMap<String, u64>,
}
/// 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<u64> = 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 <n>"));
}
#[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)
);
}
}

35
crates/buzz-ws-client/src/connection.rs
View File

@@ -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<Self, WsClientError> {
let parsed = url
.parse::<url::Url>()
.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.

125
perf/LOADGEN.md Normal file
View File

@@ -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.

41
perf/scenarios/local-smoke.toml Normal file
View File

@@ -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

4
scripts/run-tests.sh
View File

@@ -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) --------------------------------