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

feat(loadgen): add a hot-thread stressor and teardown (#8746)

Browse Source 
## Summary

`buzz-loadgen` spreads load evenly: every channel gets the same history
and the same short thread chains. That never puts heavy load on a single
thread. A busy conversation, such as agents talking to each other, can
give one root a huge number of replies, and thread-summary work grows
with it (see #8635). This PR adds an optional `[hot_thread]` scenario
section so a load test can measure that case.

- **Seed:** `seed` posts a hot root in community 0, channel 0, then
`seed_replies` replies directly under it. The root comes from a
dedicated identity that posts nothing else, so `run` finds it however
many other roots the channel has. The replies go out concurrently, like
the rest of the seed, so the run starts with the thread already large.
- **Run:** `authors` hot-thread clients are scaled by each ramp step's
`users`, the same way humans and agents are. Each one replies to the hot
root over HTTP `/events` and reads the thread's newest 100 replies over
HTTP `/query`, at `replies_per_hour` and `fetches_per_hour`.
- **Metrics:** two new report ops, `hot_reply` and `thread_fetch`. The
replies are timed, so humans subscribed to that channel also record
delivery latency for them.
- **No root, failed run:** an author that can't find the seeded root
counts a `thread_fetch` error on every attempt, so a run without the hot
thread fails instead of quietly measuring nothing.
- **Off by default:** the section is disabled while `seed_replies` and
`authors` are both 0, so existing scenarios are unchanged.

`perf/scenarios/local-smoke.toml` gets a small hot thread, and
`perf/LOADGEN.md` documents the knobs.

### Teardown

The PR also adds `buzz-loadgen teardown <scenario>`, so a staging
fixture can be removed after the load test.

- For each community from `target.community_host_template`, it first
calls `POST /operator/communities/archive` as the fixture owner, using
the operator key `seed` uses. It then calls `POST
/operator/communities/delete`.
- The deletion `request_id` is derived from `run_id` and the host, so a
re-run replays the same requests. That retries failures and reports
progress without a `deletion_request_conflict`.
- The relay's deletion drain does the actual deletion asynchronously.
- If the fixture owner does not own a host, the host is reported
`absent` and left untouched.
- A scenario without a host template is refused.
- Deleted hosts stay as tombstones, so `LOADGEN.md` says each fixture
needs a fresh `run_id`.

The signing has to happen as each derived owner, which is why this lives
in the generator rather than in deployment config.

## What got simpler

Nothing got simpler; this adds a scale vector. To keep it small, the PR
reuses what's already there:
- the seeder's post plan, plus `find_roots` with a content prefix
- the existing HTTP helper, whose auth tag is now optional
- humans' delivery measurement

A separate chain-depth knob and dedicated viewer clients were left out.
`shape.thread_depth` already seeds chains up to the relay's depth cap,
and humans in the hot channel already receive the replies.

## Testing

-
`loadgen::seed::tests::the_hot_thread_seeds_direct_replies_under_its_own_root`
checks that the hot root and its replies are seeded, that every reply is
direct (depth 1), and that a rejected root fails the seed.
-
`loadgen::run::tests::hot_thread_authors_reply_to_the_hot_root_and_fail_without_it`
checks several things:
  - authors post replies tagged to the hot root;
- root lookups query only the hot-root identity, and a newer
reply-shaped decoy is never taken as the root;
  - scheduled reads all target the hot root's channel and id;
  - the run fails when the root isn't seeded.
-
`loadgen::run::tests::hot_thread_authors_start_nothing_once_the_run_stops`
runs 8 hot authors directly. It holds every in-flight `/events` reply at
a gate until the stop is signalled, so each author's next reply is
already overdue when the gate opens. It then checks that exactly 8
replies were posted, none started after the stop. Removing the
post-`select!` stop check fails the test, and the gate is released
before any assertion, so a failure can't hang the job.
- The run now signals the stop before the last step's snapshot instead
of after building that step's report. Under a loaded parallel test run,
work started in that gap leaked into `shutdown`.
-
`loadgen::teardown::tests::teardown_deletes_only_the_fixture_owners_communities_and_replays_on_rerun`
runs teardown twice against a fake operator that checks the owner,
requires archive before delete, and conflicts on a different request id.
It checks that:
  - the owned host is archived and deletion-requested;
- another owner's host and an unseeded host are reported absent and left
untouched;
  - a 503 on archive is reported as a failure;
  - the second pass converges to the same result;
- a deletion already past membership purge, where archive returns 404,
is replayed and reported at its stage under `requested`;
- the same run under another host template gets its own request ids:
another owner's host is reported absent rather than conflicting with a
request id already used.

If the request id is replaced with a fresh one per run, the test fails.
-
`loadgen::teardown::tests::teardown_refuses_a_scenario_without_templated_communities`
- All of these run in the existing `loadgen::` unit lane.
- Teardown was run end to end at `28192004` against a local relay. The
relay had operator keys configured, local S3 behind it, and `buzz-admin
deletions drain`. The run covered:
  - seeding three templated communities;
- teardown, with every fixture archived and a deletion request at its
derived id;
  - a re-run that converged;
  - the drain, which tombstoned the fixtures;
- teardown again after the purge, which reported each fixture at
`retention_pending` and exited 0;
  - a re-seed of the torn-down run, which failed at provisioning.

A community outside the fixture stayed active, with no deletion request.
- At `28192004`, `cargo fmt --check`, `cargo clippy -p buzz-test-client
--all-targets -D warnings` and six parallel runs of `cargo test -p
buzz-test-client` pass.

🤖 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-30 13:52:30 -04:00
committed by GitHub
parent a56966b777
commit 4cf2a3bd66
10 changed files with 918 additions and 29 deletions
Download Patch File Download Diff File Expand all files Collapse all files

20
crates/buzz-test-client/src/bin/loadgen.rs
View File

@@ -3,11 +3,12 @@
//! Usage:
//! buzz-loadgen seed <scenario.toml>
//! buzz-loadgen run <scenario.toml> [--shard i/n] [--report <path>]
//! buzz-loadgen teardown <scenario.toml>
use std::path::PathBuf;
use std::sync::Arc;
use buzz_test_client::loadgen::{net::Net, run, scenario::Scenario, seed, stats};
use buzz_test_client::loadgen::{net::Net, run, scenario::Scenario, seed, stats, teardown};
use clap::{Parser, Subcommand};
#[derive(Parser)]
@@ -34,6 +35,8 @@ enum Command {
#[arg(long)]
report: Option<PathBuf>,
},
/// Archive the scenario's communities and request their deletion.
Teardown { scenario: PathBuf },
}
#[tokio::main]
@@ -48,7 +51,9 @@ async fn main() -> anyhow::Result<()> {
let cli = Cli::parse();
let path = match &cli.command {
Command::Seed { scenario } | Command::Run { scenario, .. } => scenario,
Command::Seed { scenario }
| Command::Run { scenario, .. }
| Command::Teardown { scenario } => scenario,
};
let scenario = Scenario::parse(&std::fs::read_to_string(path)?)
.map_err(|e| anyhow::anyhow!("{}: {e}", path.display()))?;
@@ -81,5 +86,16 @@ async fn main() -> anyhow::Result<()> {
}
outcome
}
Command::Teardown { .. } => {
let report = teardown::teardown(&scenario, &net).await?;
println!("{}", report.to_json());
if !report.failures.is_empty() {
anyhow::bail!(
"{} communities failed teardown; re-run to retry",
report.failures.len()
);
}
Ok(())
}
}
}

30
crates/buzz-test-client/src/loadgen/identity.rs
View File

@@ -12,6 +12,10 @@ pub enum Role {
Owner,
Human,
Agent,
HotAuthor,
/// Posts only hot-thread roots, so `run` finds them among its newest
/// posts however many other roots a channel has.
HotRoot,
}
impl Role {
@@ -20,6 +24,8 @@ impl Role {
Role::Owner => "owner",
Role::Human => "human",
Role::Agent => "agent",
Role::HotAuthor => "hot-author",
Role::HotRoot => "hot-root",
}
}
}
@@ -52,19 +58,29 @@ pub fn keys(run_id: &str, role: Role, index: usize) -> Keys {
}
}
fn uuid(parts: &[&str]) -> uuid::Uuid {
let seed = digest(parts);
let mut bytes = [0u8; 16];
bytes.copy_from_slice(&seed[..16]);
uuid::Builder::from_random_bytes(bytes).into_uuid()
}
/// Channel UUID for channel `index` of community `community`.
pub fn channel_id(run_id: &str, community: usize, index: usize) -> String {
let seed = digest(&[
uuid(&[
"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()
])
.to_string()
}
/// Deletion request id for `host` in run `run_id`, so a re-run of `teardown`
/// replays the same request instead of conflicting with it. Keyed by host,
/// so a changed host template never reuses another host's request id.
pub fn teardown_request_id(run_id: &str, host: &str) -> uuid::Uuid {
uuid(&["buzz-loadgen-teardown", run_id, host])
}
/// NIP-OA `auth` tag in which `owner` authorizes `agent`.

4
crates/buzz-test-client/src/loadgen/mod.rs
View File

@@ -4,7 +4,8 @@
//! 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.
//! ramp of user multipliers and reports latency percentiles per operation;
//! `teardown` archives the seeded communities and requests their deletion.
//!
//! Every identity, community host and channel id is derived from the
//! scenario's `run_id`, so independent shards (`--shard i/n`) agree on the
@@ -16,3 +17,4 @@ pub mod run;
pub mod scenario;
pub mod seed;
pub mod stats;
pub mod teardown;

364
crates/buzz-test-client/src/loadgen/run.rs
View File

@@ -19,7 +19,7 @@ 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::seed::{find_roots, hot_root_keys, reply_tags, HOT_ROOT_PREFIX, ROOT_PREFIX};
use super::stats::{abort_reason, ops_json, Op, Stats};
/// Content prefix of timed messages: `lg:<send unix micros>`.
@@ -95,7 +95,7 @@ async fn run_with(
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 humans, mut agents, mut authors) = (0usize, 0usize, 0usize);
let mut steps = Vec::new();
let mut aborted = false;
// End of the last reported window and start of the last judged one.
@@ -105,6 +105,7 @@ async fn run_with(
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;
let want_authors = (scenario.hot_thread.authors 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();
@@ -132,16 +133,28 @@ async fn run_with(
humans += 1;
}
}
// Hot-thread authors hold no socket, so they start without pacing.
while authors < want_authors {
if shard.owns(authors) {
let (shared, index) = (shared.clone(), authors);
tasks.push(tokio::spawn(hot_author(shared, index)));
}
authors += 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();
// Stop before the last step's snapshot, so the shutdown settles only
// work already in flight when that step's window closed.
if n + 1 == scenario.ramp.len() {
let _ = stop_tx.send(true);
}
let after = stats.snapshot();
let summaries = after.since(&before);
let reason = if lost > 0 {
Some(format!("{lost} clients exited unexpectedly"))
} else {
@@ -156,6 +169,7 @@ async fn run_with(
"users": step.users,
"humans": want_humans,
"agents": want_agents,
"hot_authors": want_authors,
"shard": format!("{}/{}", shard.index, shard.count),
"ramp_secs": (ramp_secs * 10.0).round() / 10.0,
"hold_secs": (hold_secs * 10.0).round() / 10.0,
@@ -467,7 +481,8 @@ async fn roots(
}
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 {
let found = match find_roots(&shared.net, community, reader, &owner, channel, ROOT_PREFIX).await
{
Ok(found) => {
pending.ok();
found
@@ -914,7 +929,7 @@ async fn agent(shared: Arc<Shared>, global: usize) {
community,
"/query",
&filters,
&auth,
Some(&auth),
)
.await;
} else if next_post.is_some_and(|t| t <= now) {
@@ -985,6 +1000,103 @@ async fn agent(shared: Arc<Shared>, global: usize) {
}
}
/// A hot-thread author: replies to the hot root through HTTP `/events` and
/// reads the thread's newest replies through HTTP `/query`. Replies carry
/// their send time, so humans in the hot channel record their delivery.
async fn hot_author(shared: Arc<Shared>, index: usize) {
let scenario = &shared.scenario;
let profile = &scenario.hot_thread;
let community = &shared.communities[0];
let keys = identity::keys(&scenario.run_id, Role::HotAuthor, index);
let root_author = hot_root_keys(&scenario.run_id);
let channel = identity::channel_id(&scenario.run_id, 0, 0);
let me = keys.public_key().to_hex();
let mut rng = Rng::new(u64::MAX / 2 - index as u64);
let mut backoff = Backoff::new();
// A missing root is an error every attempt, so a run without the seeded
// hot thread fails its steps instead of measuring nothing.
let root = loop {
if stopped(&shared) {
return;
}
let pending = shared.stats.begin(Op::ThreadFetch);
let newest = find_roots(
&shared.net,
community,
&keys,
&root_author,
&channel,
HOT_ROOT_PREFIX,
)
.await
.map(|roots| roots.last().cloned());
match newest {
Ok(Some(root)) => {
pending.ok();
break root;
}
Ok(None) => pending.err("hot thread root not seeded"),
Err(e) => pending.err(&e.to_string()),
}
backoff.wait(&shared, &mut rng).await;
};
let read = json!([{
"kinds": [KIND_STREAM_MESSAGE],
"#h": [channel],
"#e": [root],
"limit": 100,
}]);
let mut next_reply = rng.after(profile.replies_per_hour);
let mut next_fetch = rng.after(profile.fetches_per_hour);
loop {
let Some(due) = [next_reply, next_fetch].into_iter().flatten().min() else {
return stopping(&shared).await;
};
tokio::select! {
_ = tokio::time::sleep_until(due.into()) => {}
_ = stopping(&shared) => return,
}
// An overdue arrival and the stop can be ready together, and `select!`
// picks either; start nothing new once the run is stopping.
if stopped(&shared) {
return;
}
if next_reply == Some(due) {
next_reply = rng.after(profile.replies_per_hour);
let event = EventBuilder::new(
kind(KIND_STREAM_MESSAGE),
format!("{TIMED_PREFIX}{} {me}", now_micros()),
)
.tags(
[tag(["h", &channel])]
.into_iter()
.chain(reply_tags(&root, &root)),
)
.sign_with_keys(&keys);
let pending = shared.stats.begin(Op::HotReply);
match event {
Ok(event) => match shared.net.submit(&keys, community, &event, None).await {
Ok(()) => pending.ok(),
Err(e) => pending.err(&e.to_string()),
},
Err(e) => pending.err(&format!("sign: {e}")),
}
} else {
next_fetch = rng.after(profile.fetches_per_hour);
http(
&shared,
Op::ThreadFetch,
&keys,
community,
"/query",
&read,
None,
)
.await;
}
}
}
async fn http(
shared: &Shared,
op: Op,
@@ -992,13 +1104,13 @@ async fn http(
community: &Community,
path: &str,
body: &Value,
auth: &Tag,
auth: Option<&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))
.post(keys, &community.http_url, path, body, auth)
.await
{
Ok((status, _)) if status.is_success() => pending.ok(),
@@ -1651,4 +1763,238 @@ hold_secs = 2
);
assert!(closing["p99_ms"].as_f64().unwrap() > 1000.0, "{report}");
}
/// What a hot-thread run sent: its outcome, the events and `/query`
/// filters it posted, and the hot root's id and channel.
struct HotRun {
outcome: Result<Value, Value>,
posted: Vec<Event>,
queries: Vec<Value>,
root_id: String,
channel: String,
}
type Log<T> = Arc<std::sync::Mutex<Vec<T>>>;
/// A fake relay for the hot thread: the events it received, the `/query`
/// filters it answered, and the hot root's id and channel.
struct HotRelay {
addr: String,
posted: Log<Event>,
queries: Log<Value>,
root_id: String,
channel: String,
}
/// Holds `/events` responses until opened, counting the requests held.
#[derive(Default)]
struct Gate {
arrived: AtomicUsize,
open: std::sync::Mutex<bool>,
opened: std::sync::Condvar,
}
impl Gate {
fn release(&self) {
*self.open.lock().unwrap() = true;
self.opened.notify_all();
}
}
/// Serves a relay whose root lookup returns a newer reply-shaped decoy by
/// the root's author, plus the hot root itself when `seeded`. With a
/// `gate`, every event waits for it to open before it is accepted.
async fn hot_relay(seeded: bool, gate: Option<Arc<Gate>>) -> HotRelay {
let author = hot_root_keys("t");
let channel = identity::channel_id("t", 0, 0);
let root = EventBuilder::new(kind(KIND_STREAM_MESSAGE), HOT_ROOT_PREFIX)
.tags([tag(["h", &channel])])
.sign_with_keys(&author)
.unwrap();
let root_id = root.id.to_hex();
let decoy = EventBuilder::new(kind(KIND_STREAM_MESSAGE), "lg-hot-reply 0")
.tags(
[tag(["h", &channel])]
.into_iter()
.chain(reply_tags(&"0".repeat(64), &"0".repeat(64))),
)
.custom_created_at(Timestamp::from(root.created_at.as_secs() + 1))
.sign_with_keys(&author)
.unwrap();
let posted: Log<Event> = Default::default();
let queries: Log<Value> = Default::default();
let (post_log, query_log) = (posted.clone(), queries.clone());
let addr = super::super::seed::tests::fake_http(move |_, target, body| {
if target == "/events" {
let event: Event = serde_json::from_slice(body).unwrap();
post_log.lock().unwrap().push(event);
if let Some(gate) = &gate {
gate.arrived.fetch_add(1, Ordering::SeqCst);
tokio::task::block_in_place(|| {
let open = gate.open.lock().unwrap();
drop(gate.opened.wait_while(open, |open| !*open).unwrap());
});
}
return (200, json!({"accepted": true, "message": ""}).to_string());
}
let filter: Value = serde_json::from_slice(body).unwrap();
let lookup = filter[0].get("authors").is_some();
query_log.lock().unwrap().push(filter[0].clone());
let found = match (lookup, seeded) {
(true, true) => json!([root, decoy]),
(true, false) => json!([decoy]),
(false, _) => json!([]),
};
(200, found.to_string())
})
.await;
HotRelay {
addr,
posted,
queries,
root_id,
channel,
}
}
/// A one-second scenario with `[hot_thread]` section `hot` against `addr`.
fn hot_scenario(addr: &str, hot: &str) -> 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 = 0
agents = 0
channels_per_member = 1
[hot_thread]
{hot}
[[ramp]]
users = 1.0
hold_secs = 1
"#
))
.unwrap()
}
/// Runs `[hot_thread]` section `hot` for a second against `hot_relay`.
async fn hot_thread_run(seeded: bool, hot: &str) -> HotRun {
let relay = hot_relay(seeded, None).await;
let scenario = hot_scenario(&relay.addr, hot);
let net = Net::new(&scenario).await.unwrap();
let shard = Shard::parse("0/1").unwrap();
let outcome = run_with(scenario, net, shard, Duration::from_secs(5)).await;
let posted = relay.posted.lock().unwrap().clone();
let queries = relay.queries.lock().unwrap().clone();
HotRun {
outcome,
posted,
queries,
root_id: relay.root_id,
channel: relay.channel,
}
}
const TWO_AUTHORS: &str = "authors = 2\nreplies_per_hour = 36000.0\nfetches_per_hour = 36000.0";
#[tokio::test]
async fn hot_thread_authors_reply_to_the_hot_root_and_fail_without_it() {
let run = hot_thread_run(true, TWO_AUTHORS).await;
let report = run.outcome.expect("hot-thread run passes");
assert_eq!(report["steps"][0]["hot_authors"], 2, "{report}");
assert!(!run.posted.is_empty(), "{report}");
for event in &run.posted {
let markers = buzz_core::nip10::parse_thread_markers(&event.tags);
assert_eq!(markers.root.as_deref(), Some(run.root_id.as_str()));
assert_eq!(markers.reply.as_deref(), Some(run.root_id.as_str()));
let h = event
.tags
.iter()
.find(|t| t.kind().as_str() == "h")
.unwrap();
assert_eq!(h.content(), Some(run.channel.as_str()));
}
// The root is looked up among its own author's posts, so other roots
// in the channel can't push it out of the window.
let author = hot_root_keys("t").public_key().to_hex();
let (lookups, reads): (Vec<&Value>, Vec<&Value>) =
run.queries.iter().partition(|q| q.get("authors").is_some());
assert!(!lookups.is_empty());
for lookup in lookups {
assert_eq!(lookup["authors"], json!([author]), "{lookup}");
}
// Scheduled reads fetch the hot thread itself.
let read = json!({
"kinds": [KIND_STREAM_MESSAGE],
"#h": [run.channel],
"#e": [run.root_id],
"limit": 100,
});
assert!(!reads.is_empty(), "no scheduled thread reads: {report}");
assert!(reads.iter().all(|q| **q == read), "{reads:?}");
let ops = &report["steps"][0]["ops"];
assert!(ops["hot_reply"]["count"].as_u64() > Some(0), "{report}");
// Without a seeded hot root the authors post nothing and the step aborts.
let run = hot_thread_run(false, TWO_AUTHORS).await;
let report = run.outcome.expect_err("a missing hot root fails the run");
assert!(run.posted.is_empty());
let reason = report["steps"][0]["abort_reason"]
.as_str()
.unwrap_or_default();
assert!(reason.starts_with("thread_fetch"), "{report}");
}
/// Every author is mid-reply at the stop, held there by the relay, with
/// its next reply already due. Each must finish the reply it started and
/// start no other: exactly one reply per author settles after the stop,
/// and the relay receives no more.
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn hot_thread_authors_start_nothing_once_the_run_stops() {
let gate = Arc::new(Gate::default());
let relay = hot_relay(true, Some(gate.clone())).await;
let scenario = hot_scenario(&relay.addr, "authors = 8\nreplies_per_hour = 36000000.0");
let net = Net::new(&scenario).await.unwrap();
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 authors: Vec<_> = (0..8)
.map(|i| tokio::spawn(hot_author(shared.clone(), i)))
.collect();
let deadline = Duration::from_secs(10);
let arrived = tokio::time::timeout(deadline, async {
while gate.arrived.load(Ordering::SeqCst) < 8 {
tokio::time::sleep(Duration::from_millis(5)).await;
}
})
.await;
stop_tx.send(true).unwrap();
let at_stop = stats.snapshot();
// Release before any assertion, so a failure never leaves handlers
// blocked in the gate and the runtime unable to shut down.
gate.release();
arrived.expect("every author starts a reply");
for author in authors {
tokio::time::timeout(deadline, author)
.await
.expect("authors return once stopped")
.unwrap();
}
let settled = ops_json(&stats.snapshot().since(&at_stop), 1.0);
assert_eq!(settled["hot_reply"]["count"], 8, "{settled}");
assert_eq!(relay.posted.lock().unwrap().len(), 8, "{settled}");
}
}

33
crates/buzz-test-client/src/loadgen/scenario.rs
View File

@@ -17,6 +17,8 @@ pub struct Scenario {
pub human: HumanProfile,
#[serde(default)]
pub agent: AgentProfile,
#[serde(default)]
pub hot_thread: HotThread,
pub ramp: Vec<RampStep>,
#[serde(default)]
pub abort: Abort,
@@ -154,6 +156,29 @@ impl Default for AgentProfile {
}
}
/// One thread in community 0, channel 0 that keeps growing: seeded with
/// many direct replies, then replied to and read throughout the run.
/// Disabled while `seed_replies` and `authors` are both 0.
#[derive(Clone, Debug, Default, Deserialize)]
#[serde(default, deny_unknown_fields)]
pub struct HotThread {
/// Direct replies seeded under the hot root.
pub seed_replies: usize,
/// Clients replying to the hot root at 1x; ramp steps multiply them.
pub authors: usize,
/// Replies each author posts per hour through HTTP `/events`.
pub replies_per_hour: f64,
/// Thread reads (the root's newest 100 replies) each author makes per
/// hour through HTTP `/query`.
pub fetches_per_hour: f64,
}
impl HotThread {
pub fn enabled(&self) -> bool {
self.seed_replies > 0 || self.authors > 0
}
}
#[derive(Clone, Debug, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct RampStep {
@@ -271,6 +296,14 @@ impl Scenario {
("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),
(
"hot_thread.replies_per_hour",
self.hot_thread.replies_per_hour,
),
(
"hot_thread.fetches_per_hour",
self.hot_thread.fetches_per_hour,
),
];
for (name, value) in non_negative {
if !value.is_finite() || value < 0.0 {

103
crates/buzz-test-client/src/loadgen/seed.rs
View File

@@ -16,6 +16,13 @@ use super::stats::{Op, Snapshot, Stats};
/// Content prefix of seeded thread roots; `run` finds roots by it.
pub const ROOT_PREFIX: &str = "lg-root";
/// Content prefix of the seeded hot-thread root; `run` finds it by this.
pub const HOT_ROOT_PREFIX: &str = "lg-hot";
/// Author of every hot-thread root in run `run_id`.
pub fn hot_root_keys(run_id: &str) -> Keys {
identity::keys(run_id, Role::HotRoot, 0)
}
/// Seeds every community. Idempotent for communities this run's fixture
/// owner owns, and for channels; history and threads are appended on every
@@ -112,6 +119,12 @@ async fn post_all(
}
async fn provision(scenario: &Scenario, net: &Net, host: &str, owner: &Keys) -> anyhow::Result<()> {
let (base, operator) = operator(scenario)?;
provision_host(net, base, &operator, host, owner).await
}
/// The operator API base URL and the operator's keys, for templated communities.
pub(super) fn operator(scenario: &Scenario) -> anyhow::Result<(&str, Keys)> {
let base = scenario
.target
.operator_url
@@ -123,8 +136,7 @@ async fn provision(scenario: &Scenario, net: &Net, host: &str, owner: &Keys) ->
scenario.target.operator_key_env
)
})?;
let operator = Keys::parse(&secret)?;
provision_host(net, base, &operator, host, owner).await
Ok((base, Keys::parse(&secret)?))
}
/// Creates `host` owned by `owner`. The request is `create_only`, so the relay
@@ -215,12 +227,20 @@ enum PostKind {
History,
Root,
Reply { root: usize },
HotRoot,
HotReply,
}
/// Key of the hot root in [`RootIds`]: the root after channel 0's seeded ones.
fn hot_key(scenario: &Scenario) -> (usize, usize, usize) {
(0, 0, scenario.shape.thread_roots_per_channel)
}
/// 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.
/// are independent of each other. The hot thread's replies all answer its
/// root directly, so they post concurrently.
fn plan_posts(scenario: &Scenario) -> (Vec<Chain>, Vec<Chain>) {
let shape = &scenario.shape;
let mut roots = Vec::new();
@@ -256,6 +276,23 @@ fn plan_posts(scenario: &Scenario) -> (Vec<Chain>, Vec<Chain>) {
}
}
}
if scenario.hot_thread.enabled() {
let (community, channel, seq) = hot_key(scenario);
roots.push(vec![Post {
community,
channel,
kind: PostKind::HotRoot,
seq,
}]);
rest.extend((0..scenario.hot_thread.seed_replies).map(|seq| {
vec![Post {
community,
channel,
kind: PostKind::HotReply,
seq,
}]
}));
}
(roots, rest)
}
@@ -293,6 +330,12 @@ impl Post {
tags.extend(reply_tags(root_id, parent));
(author, format!("lg-reply {}", self.seq))
}
PostKind::HotRoot => (hot_root_keys(&scenario.run_id), HOT_ROOT_PREFIX.to_string()),
PostKind::HotReply => {
let root_id = roots.get(&hot_key(scenario))?;
tags.extend(reply_tags(root_id, root_id));
(author, format!("lg-hot-reply {}", self.seq))
}
};
let event = EventBuilder::new(Kind::Custom(9), content)
.tags(tags)
@@ -305,21 +348,23 @@ impl Post {
/// 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] {
pub 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.
/// Ids of `owner`'s seeded roots in `channel` whose content starts with
/// `prefix`, oldest first, among its 100 newest posts there. `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,
prefix: &str,
) -> anyhow::Result<Vec<String>> {
let filter = json!([{
"kinds": [9],
@@ -343,7 +388,7 @@ pub async fn find_roots(
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")
e.content.starts_with(prefix) && !e.tags.iter().any(|t| t.kind().as_str() == "e")
})
.map(|e| (e.created_at.as_secs(), e.id.to_hex()))
.collect();
@@ -356,13 +401,13 @@ fn tag<const N: usize>(parts: [&str; N]) -> Tag {
}
#[cfg(test)]
mod tests {
pub(super) 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
pub(in crate::loadgen) async fn fake_http<F>(handler: F) -> String
where
F: Fn(&str, &str, &[u8]) -> (u16, String) + Send + Sync + 'static,
{
@@ -453,6 +498,12 @@ mod tests {
}
async fn seed_against(reject: &str) -> (anyhow::Result<()>, Arc<Stats>, Accepted) {
seed_with(reject, "").await
}
/// Seeds two history posts and two chains of three replies, plus `extra`
/// scenario TOML, against a relay that rejects content `reject`.
async fn seed_with(reject: &str, extra: &str) -> (anyhow::Result<()>, Arc<Stats>, Accepted) {
let (addr, accepted) = fake_relay(reject.into()).await;
let scenario = Scenario::parse(&format!(
r#"
@@ -473,6 +524,7 @@ channels_per_member = 1
[[ramp]]
users = 1.0
hold_secs = 1
{extra}
"#
))
.unwrap();
@@ -519,6 +571,39 @@ hold_secs = 1
assert_eq!(depths, [1, 1, 2, 2, 3, 3]);
}
#[tokio::test]
async fn the_hot_thread_seeds_direct_replies_under_its_own_root() {
let hot = "[hot_thread]\nseed_replies = 4";
let (outcome, stats, accepted) = seed_with("none", hot).await;
outcome.expect("every post accepted");
// The base 10 posts plus the hot root and its 4 replies.
assert_eq!(seeded(&stats.snapshot(), &Snapshot::default()), 15);
let hot_depths: Vec<usize> = {
let accepted = accepted.lock().unwrap();
let roots: Vec<&String> = accepted
.iter()
.filter(|(_, (_, depth))| *depth == 0)
.map(|(id, _)| id)
.collect();
// Every reply under the most-replied root sits directly on it.
let hot_root = roots
.iter()
.max_by_key(|root| accepted.values().filter(|(r, _)| r == **root).count())
.unwrap();
accepted
.iter()
.filter(|(id, (root, _))| root == *hot_root && id != hot_root)
.map(|(_, (_, depth))| *depth)
.collect()
};
assert_eq!(hot_depths, [1, 1, 1, 1]);
// A rejected hot root leaves its replies unpostable.
let (outcome, _, _) = seed_with(HOT_ROOT_PREFIX, hot).await;
let err = outcome.expect_err("a rejected hot root fails the seed");
assert!(err.to_string().contains("seeded 10 of 15"), "{err}");
}
#[test]
fn reply_k_points_reply_at_reply_k_minus_one_and_root_at_the_root() {
let scenario = Scenario::parse(

8
crates/buzz-test-client/src/loadgen/stats.rs
View File

@@ -19,10 +19,12 @@ pub enum Op {
HttpEvents,
Seed,
Subscribe,
HotReply,
ThreadFetch,
}
impl Op {
pub const ALL: [Op; 9] = [
pub const ALL: [Op; 11] = [
Op::WsConnect,
Op::History,
Op::PublishEphemeral,
@@ -32,6 +34,8 @@ impl Op {
Op::HttpEvents,
Op::Seed,
Op::Subscribe,
Op::HotReply,
Op::ThreadFetch,
];
pub fn name(self) -> &'static str {
@@ -45,6 +49,8 @@ impl Op {
Op::HttpEvents => "http_events",
Op::Seed => "seed",
Op::Subscribe => "subscribe_eose",
Op::HotReply => "hot_reply",
Op::ThreadFetch => "thread_fetch",
}
}
}

345
crates/buzz-test-client/src/loadgen/teardown.rs Normal file
View File

@@ -0,0 +1,345 @@
//! Archives and requests deletion of the communities `seed` provisioned.
use std::collections::BTreeMap;
use futures_util::stream::{self, StreamExt};
use nostr::Keys;
use reqwest::StatusCode;
use serde_json::json;
use super::identity::{self, Role};
use super::net::{http_reason, Net};
use super::scenario::Scenario;
use super::seed;
/// The acknowledgement version the relay accepts for owner deletions.
const ACKNOWLEDGEMENT_VERSION: i32 = 1;
/// What `teardown` did across the scenario's communities.
#[derive(Debug, Default)]
pub struct Report {
/// Deletion requests by the stage the relay reported.
pub requested: BTreeMap<String, usize>,
/// Communities with neither this run's fixture owner nor this run's
/// deletion request: never seeded, or another owner's host.
pub absent: usize,
/// One reason per community whose archive or deletion request failed.
pub failures: Vec<String>,
}
impl Report {
pub fn to_json(&self) -> serde_json::Value {
json!({
"requested": self.requested,
"absent": self.absent,
"failed": self.failures.len(),
"failures": self.failures.iter().take(10).collect::<Vec<_>>(),
})
}
}
enum Outcome {
Requested(String),
Absent,
}
/// Archives every templated community of the scenario as its fixture owner,
/// then requests its deletion under a request id derived from `run_id` and
/// the host, so a re-run replays the same requests. Deletion itself runs
/// asynchronously on the relay's deletion drain.
pub async fn teardown(scenario: &Scenario, net: &Net) -> anyhow::Result<Report> {
if scenario.target.community_host_template.is_none() {
anyhow::bail!(
"teardown deletes only communities from target.community_host_template; \
this scenario targets {} directly",
scenario.target.http_url
);
}
let (base, operator) = seed::operator(scenario)?;
Ok(teardown_with(scenario, net, base, &operator).await)
}
async fn teardown_with(scenario: &Scenario, net: &Net, base: &str, operator: &Keys) -> Report {
let outcomes: Vec<Result<Outcome, String>> = stream::iter(scenario.communities())
.filter_map(|community| async move { Some((community.index, community.host?)) })
.map(|(index, host)| async move {
teardown_community(net, base, operator, &scenario.run_id, index, &host).await
})
.buffer_unordered(scenario.shape.seed_concurrency)
.collect()
.await;
let mut report = Report::default();
for outcome in outcomes {
match outcome {
Ok(Outcome::Requested(stage)) => *report.requested.entry(stage).or_default() += 1,
Ok(Outcome::Absent) => report.absent += 1,
Err(reason) => report.failures.push(reason),
}
}
report
}
async fn teardown_community(
net: &Net,
base: &str,
operator: &Keys,
run_id: &str,
index: usize,
host: &str,
) -> Result<Outcome, String> {
let owner = identity::keys(run_id, Role::Owner, index)
.public_key()
.to_hex();
let post = |path: &'static str, body: serde_json::Value| async move {
let body = serde_json::to_vec(&body).map_err(|e| e.to_string())?;
net.post(operator, base, path, body, None)
.await
.map_err(|e| format!("{path} {host}: {e}"))
};
let (status, text) = post(
"/operator/communities/archive",
json!({ "host": host, "owner_pubkey": owner }),
)
.await?;
// Not found means this owner holds no live community here. A deletion
// already past membership purge looks the same, so replay it below.
if !status.is_success() && status != StatusCode::NOT_FOUND {
return Err(format!("archive {host}: {}", http_reason(status, &text)));
}
let (status, text) = post(
"/operator/communities/delete",
json!({
"host": host,
"owner_pubkey": owner,
"request_id": identity::teardown_request_id(run_id, host).to_string(),
"acknowledgement_version": ACKNOWLEDGEMENT_VERSION,
}),
)
.await?;
if status == StatusCode::NOT_FOUND {
return Ok(Outcome::Absent);
}
if !status.is_success() {
return Err(format!("delete {host}: {}", http_reason(status, &text)));
}
let accepted: serde_json::Value = serde_json::from_str(&text)
.map_err(|e| format!("delete {host}: unreadable response: {e}"))?;
let stage = accepted["status"].as_str().unwrap_or("unknown").to_string();
Ok(Outcome::Requested(stage))
}
#[cfg(test)]
mod tests {
use super::*;
use crate::loadgen::seed::tests::fake_http;
use std::collections::HashMap;
use std::sync::{Arc, Mutex};
#[derive(Debug, Default)]
struct Community {
/// Empty once deletion has purged membership.
owner: String,
archived: bool,
/// Request id and stage of the community's deletion request.
deletion: Option<(String, &'static str)>,
}
/// The operator's archive and delete endpoints: owner-checked, archive
/// before delete, and a request id replayed for its community at any
/// stage, even after membership purge. A request id already used for
/// another host, or a second id for one community, conflicts.
/// `unavailable` answers every archive with 503.
async fn fake_operator(
communities: Arc<Mutex<HashMap<String, Community>>>,
unavailable: &'static str,
) -> String {
fake_http(move |_, target, body| {
let request: serde_json::Value = serde_json::from_slice(body).unwrap();
let host = request["host"].as_str().unwrap();
let owner = request["owner_pubkey"].as_str().unwrap();
let mut communities = communities.lock().unwrap();
let not_found = (404, json!({"error": "community not found"}).to_string());
match target {
"/operator/communities/archive" => {
let Some(community) = communities.get_mut(host).filter(|c| c.owner == owner)
else {
return not_found;
};
if host == unavailable {
return (503, json!({"error": "propagation pending"}).to_string());
}
community.archived = true;
(200, json!({"status": "archived"}).to_string())
}
"/operator/communities/delete" => {
if request["acknowledgement_version"] != 1 {
return (400, json!({"error": "unsupported"}).to_string());
}
let id = request["request_id"].as_str().unwrap();
let known = communities.iter().find_map(|(h, c)| {
c.deletion
.as_ref()
.filter(|(known, _)| known == id)
.map(|(_, stage)| (h.as_str(), *stage))
});
match known {
Some((h, stage)) if h == host => {
return (202, json!({ "status": stage }).to_string());
}
Some(_) => {
return (
409,
json!({"code": "deletion_request_conflict"}).to_string(),
);
}
None => {}
}
let Some(community) = communities.get_mut(host).filter(|c| c.owner == owner)
else {
return not_found;
};
if !community.archived {
return (409, json!({"code": "community_not_archived"}).to_string());
}
if community.deletion.is_some() {
return (
409,
json!({"code": "deletion_request_conflict"}).to_string(),
);
}
community.deletion = Some((id.to_string(), "submitted"));
(202, json!({"status": "submitted"}).to_string())
}
other => panic!("unexpected {other}"),
}
})
.await
}
fn scenario(addr: &str, template: Option<&str>) -> Scenario {
let template = template
.map(|t| format!("community_host_template = \"{t}\"\noperator_url = \"http://{addr}\""))
.unwrap_or_default();
Scenario::parse(&format!(
r#"
run_id = "t"
[target]
ws_url = "ws://{addr}"
http_url = "http://{addr}"
{template}
[shape]
communities = {}
channels_per_community = 1
[population]
humans = 1
agents = 0
channels_per_member = 1
[[ramp]]
users = 1.0
hold_secs = 1
"#,
if template.is_empty() { 1 } else { 5 }
))
.unwrap()
}
#[tokio::test]
async fn teardown_deletes_only_the_fixture_owners_communities_and_replays_on_rerun() {
let owner = |i| identity::keys("t", Role::Owner, i).public_key().to_hex();
let communities = Arc::new(Mutex::new(HashMap::from([
(
"lg-t-0.example".to_string(),
Community {
owner: owner(0),
..Default::default()
},
),
(
// A host this run does not own is left alone.
"lg-t-1.example".to_string(),
Community {
owner: Keys::generate().public_key().to_hex(),
..Default::default()
},
),
// lg-t-2 was never seeded.
(
"lg-t-3.example".to_string(),
Community {
owner: owner(3),
..Default::default()
},
),
(
// Deletion already purged membership, so archive finds no
// owner; the request is still replayed for its stage.
"lg-t-4.example".to_string(),
Community {
archived: true,
deletion: Some((
identity::teardown_request_id("t", "lg-t-4.example").to_string(),
"postgres_purged",
)),
..Default::default()
},
),
])));
let addr = fake_operator(communities.clone(), "lg-t-3.example").await;
let fixture = scenario(&addr, Some("lg-{run}-{i}.example"));
let net = Net::new(&fixture).await.unwrap();
let base = format!("http://{addr}");
let operator = Keys::generate();
for pass in ["first", "re-run"] {
let report = teardown_with(&fixture, &net, &base, &operator).await;
assert_eq!(
report.requested,
BTreeMap::from([("postgres_purged".into(), 1), ("submitted".into(), 1)]),
"{pass}"
);
assert_eq!(report.absent, 2, "{pass}");
assert_eq!(report.failures.len(), 1, "{pass}: {:?}", report.failures);
assert!(
report.failures[0].starts_with("archive lg-t-3.example: HTTP 503"),
"{pass}: {:?}",
report.failures
);
}
// The same run under another host template names another owner's
// host; it gets its own request id and is left alone.
communities.lock().unwrap().insert(
"other-0.example".to_string(),
Community {
owner: Keys::generate().public_key().to_hex(),
..Default::default()
},
);
let other = scenario(&addr, Some("other-{i}.example"));
let report = teardown_with(&other, &net, &base, &operator).await;
assert!(report.failures.is_empty(), "{:?}", report.failures);
assert_eq!(report.absent, 5);
let communities = communities.lock().unwrap();
assert!(!communities["other-0.example"].archived);
assert_eq!(
communities["lg-t-0.example"].deletion,
Some((
identity::teardown_request_id("t", "lg-t-0.example").to_string(),
"submitted"
))
);
assert!(!communities["lg-t-1.example"].archived);
assert!(communities["lg-t-1.example"].deletion.is_none());
assert!(communities["lg-t-3.example"].deletion.is_none());
}
#[tokio::test]
async fn teardown_refuses_a_scenario_without_templated_communities() {
let scenario = scenario("127.0.0.1:9", None);
let net = Net::new(&scenario).await.unwrap();
let err = teardown(&scenario, &net).await.unwrap_err();
assert!(err.to_string().contains("community_host_template"), "{err}");
}
}

33
perf/LOADGEN.md
View File

@@ -8,6 +8,7 @@ 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
cargo run --release -p buzz-test-client --bin buzz-loadgen -- teardown perf/scenarios/<templated>.toml
```
## Scale vectors
@@ -18,6 +19,7 @@ cargo run --release -p buzz-test-client --bin buzz-loadgen -- run perf/scenarios
| 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` |
| One hot thread | `[hot_thread]`: `seed_replies`, `authors`, `replies_per_hour`, `fetches_per_hour` |
| 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` |
@@ -32,6 +34,14 @@ 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).
`[hot_thread]` models one thread that keeps growing, such as a conversation
between busy agents. `seed` posts its root in community 0, channel 0 and
then `seed_replies` replies directly under it, concurrently. During `run`,
`authors` clients (multiplied by each ramp step's `users`, like humans and
agents) reply to that root and read its newest 100 replies over HTTP. Their
replies are timed, so humans subscribed to that channel record delivery.
Leave `seed_replies` and `authors` at 0 to disable it.
## What each client does
- **Human**: authenticates over WebSocket and subscribes to its channels'
@@ -42,6 +52,10 @@ override maps hostnames, so HTTP would ignore it).
- **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.
- **Hot-thread author**: finds the seeded hot root, then posts replies to it
through HTTP `/events` (`hot_reply`) and reads the thread through HTTP
`/query` (`thread_fetch`). Until the root is found every attempt counts as
a `thread_fetch` error, so a run without the seeded thread fails.
Messages carry their send time, so every subscriber records publish→delivery
latency (`deliver`). Across machines this depends on clock sync.
@@ -126,6 +140,25 @@ 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.
## Teardown
`teardown <scenario>` removes the communities `seed` provisioned from
`target.community_host_template`, using the operator key `seed` used. For each
host it archives the community as its fixture owner, then requests owner
deletion with a request id derived from `run_id` and the host. The
relay's deletion drain does the deletion asynchronously; `teardown` reports
the requests by the stage the relay returned (`requested`), hosts this run's
fixture owner does not own (`absent`), and failures. It never touches a host
owned by anyone else, and refuses a scenario without a host template.
Re-running `teardown` replays the same request ids, so it retries failures
and reports progress without conflicting. It exits non-zero while any
community failed.
Deleted hosts stay reserved as tombstones, so a torn-down `run_id` cannot be
seeded again: `seed` fails on the first host. Give each fixture a new
`run_id`.
## Safety
The generator creates real communities, channels and events. Point it only at

7
perf/scenarios/local-smoke.toml
View File

@@ -32,6 +32,13 @@ query_interval_secs = 2.0
events_per_hour = 360.0
observer_frames_per_hour = 360.0
# One thread that keeps growing while the population runs.
[hot_thread]
seed_replies = 200
authors = 3
replies_per_hour = 1800.0
fetches_per_hour = 360.0
[[ramp]]
users = 1.0
hold_secs = 30