diff --git a/crates/buzz-test-client/src/bin/ingest_load.rs b/crates/buzz-test-client/src/bin/ingest_load.rs new file mode 100644 index 00000000000..416dbed68f4 --- /dev/null +++ b/crates/buzz-test-client/src/bin/ingest_load.rs @@ -0,0 +1,455 @@ +//! Multi-community paced ingest load generator. +//! +//! Drives several communities at independently settable rates against one relay +//! process, which is what lets the harness tell a per-pod ceiling from a +//! per-community one. `perf/RELAY_INGEST_CEILING.md` owns that reasoning. +//! +//! Reports two latencies per target, because they diverge exactly when the relay +//! saturates and the gap between them is the measurement: +//! +//! - `service_ms` — signed event on the wire until the relay's OK. +//! - `scheduled_ms` — the send's *intended* slot until the OK. Delay the +//! generator itself added by falling behind stays visible here instead of +//! silently redefining the offered rate downward. +//! +//! Signing happens before the service clock starts, so BIP340 cost lands in +//! `scheduled_ms` and never inflates the relay's number. +//! +//! Raw per-send samples are deliberately not written: the harness verdict is a +//! ratio of rates, and p50/p95/p99/max cover what a human reads. +//! +//! Usage: +//! ingest_load [ ...] +//! target: url=,channel=,rate=[,conns=] +//! +//! Env: +//! BENCH_PRIVATE_KEY hex secret key; must be a channel member on every target +//! BENCH_METRICS_URL when set, the relay's Prometheus endpoint is sampled at +//! this run's timed-window edges and reported as +//! `counters_before`/`counters_after` + +use std::time::Duration; + +use anyhow::{anyhow, bail, Context}; +use buzz_core::kind::KIND_STREAM_MESSAGE; +use buzz_test_client::BuzzTestClient; +use nostr::{EventBuilder, Keys, Kind, Tag}; +use serde_json::{json, Value}; +use tokio::time::Instant; + +/// Connections per target when the spec omits `conns`. +const DEFAULT_CONNS: usize = 8; + +/// Padding that puts each event in the size range of a real chat message. +const PADDING: &str = "the quick brown fox jumps over the lazy dog 0123456789"; + +/// The run's observation window: what was asked for, and what it took. +struct Window { + requested_secs: f64, + elapsed_secs: f64, +} + +/// One community's offered load: where to send, and how fast. +#[derive(Debug, Clone)] +struct Target { + url: String, + channel: String, + rate: f64, + conns: usize, +} + +/// What one connection, or one whole target, observed. +/// +/// `attempted` counts *settled* sends only. A send that fails in flight leaves +/// no latency sample and is not counted here either — it shows up as +/// `first_transport_error`, which the runner treats as invalidating the cell. +#[derive(Debug, Default)] +struct Outcome { + attempted: u64, + accepted: u64, + rejected: u64, + service_ms: Vec, + scheduled_ms: Vec, + first_rejection: Option, + first_transport_error: Option, + /// Time from when this connection was free to send until the send happened: + /// `sent_at - max(slot, previous settled_at)`. Signing and scheduler delay + /// land here, and relay backpressure does not, which is what makes it a + /// generator-vs-relay discriminator. The closed-loop rate metrics cannot be: + /// at saturation a connection's throughput *is* the relay's, so an apparent + /// generator shortfall is the treatment effect. + generator_lag_ms: Vec, +} + +impl Outcome { + fn absorb(&mut self, other: Self) { + self.attempted += other.attempted; + self.accepted += other.accepted; + self.rejected += other.rejected; + self.service_ms.extend(other.service_ms); + self.scheduled_ms.extend(other.scheduled_ms); + self.generator_lag_ms.extend(other.generator_lag_ms); + self.first_rejection = self.first_rejection.take().or(other.first_rejection); + self.first_transport_error = self + .first_transport_error + .take() + .or(other.first_transport_error); + } +} + +fn parse_target(spec: &str) -> anyhow::Result { + let mut url = None; + let mut channel = None; + let mut rate = None; + let mut conns = DEFAULT_CONNS; + + for field in spec.split(',') { + let (key, value) = field + .split_once('=') + .ok_or_else(|| anyhow!("target field {field:?} is not key=value"))?; + match key { + "url" => url = Some(value.to_string()), + "channel" => channel = Some(value.to_string()), + "rate" => rate = Some(value.parse::().context("rate")?), + "conns" => conns = value.parse::().context("conns")?, + other => bail!("unknown target field {other:?}"), + } + } + + let target = Target { + url: url.ok_or_else(|| anyhow!("target {spec:?} is missing url="))?, + channel: channel.ok_or_else(|| anyhow!("target {spec:?} is missing channel="))?, + rate: rate.ok_or_else(|| anyhow!("target {spec:?} is missing rate="))?, + conns, + }; + if !target.rate.is_finite() || target.rate <= 0.0 { + bail!("target rate must be a positive number, got {}", target.rate); + } + if target.conns == 0 { + bail!("target conns must be at least 1"); + } + Ok(target) +} + +/// Sends on one connection against a fixed schedule. +/// +/// The schedule advances by a constant interval and is never rebased on the +/// response, so a slow relay produces a rising `scheduled_ms` and a visible +/// shortfall in `attempted` rather than a quietly reduced offer. +async fn drive_connection( + mut client: BuzzTestClient, + keys: Keys, + channel: String, + label: String, + first_slot: Instant, + interval: Duration, + deadline: Instant, +) -> anyhow::Result { + let h_tag = Tag::parse(["h", channel.as_str()]).map_err(|e| anyhow!("h tag: {e}"))?; + let kind = Kind::Custom(u16::try_from(KIND_STREAM_MESSAGE).context("stream message kind")?); + + let mut out = Outcome::default(); + let mut slot = first_slot; + let mut seq: u64 = 0; + let mut settled_at = first_slot; + + while slot < deadline && Instant::now() < deadline { + tokio::time::sleep_until(slot).await; + // Ready to send once both the slot has arrived and the previous send has + // settled; anything after this instant is the generator's own delay. + let ready_at = slot.max(settled_at); + seq += 1; + let event = EventBuilder::new(kind, format!("{label} seq={seq} {PADDING}")) + .tags([h_tag.clone()]) + .sign_with_keys(&keys)?; + + let sent_at = Instant::now(); + out.generator_lag_ms + .push((sent_at - ready_at).as_secs_f64() * 1e3); + let response = client.send_event(event).await; + settled_at = Instant::now(); + + let ok = match response { + Ok(ok) => ok, + Err(e) => { + // A dead connection ends this sender but keeps its samples: + // losing them would hide the saturation that killed it. + out.first_transport_error = Some(e.to_string()); + break; + } + }; + + out.attempted += 1; + out.service_ms + .push((settled_at - sent_at).as_secs_f64() * 1e3); + out.scheduled_ms + .push((settled_at - slot).as_secs_f64() * 1e3); + if ok.accepted { + out.accepted += 1; + } else { + out.rejected += 1; + if out.first_rejection.is_none() { + out.first_rejection = Some(ok.message); + } + } + slot += interval; + } + + if let Err(e) = client.disconnect().await { + out.first_transport_error = out.first_transport_error.or(Some(e.to_string())); + } + Ok(out) +} + +/// Prometheus counters this harness reads, sampled at the timed window's edges. +/// +/// Sampled here rather than by the caller because the caller can only bracket +/// the whole process: its "before" lands before the connection phase and its +/// "after" after teardown, while the rates are divided by the window that starts +/// once every connection is authenticated. Backlog draining during setup then +/// lands in the delta but not the divisor, which can push a busy fraction above +/// 1.0 and overstate completions. +async fn scrape_counters(metrics_url: &str) -> anyhow::Result { + const WANTED: [(&str, &str); 6] = [ + ("buzz_audit_log_seconds_count", "audit_count"), + ("buzz_audit_log_seconds_sum", "audit_sum"), + ("buzz_audit_log_errors_total", "audit_log_errors"), + ("buzz_audit_send_errors_total", "audit_send_errors"), + ( + "buzz_admission_rejections_total{transport=\"websocket\",reason=\"quota\"}", + "quota", + ), + ( + "buzz_admission_rejections_total{transport=\"websocket\",reason=\"unavailable\"}", + "unavailable", + ), + ]; + + // Keep in step with `scrape()` in perf/relay_ingest_ceiling.py, which reads + // the same series names for the un-aligned fallback path. + let parsed = metrics_url + .parse::() + .map_err(|e| anyhow!("metrics url {metrics_url:?}: {e}"))?; + let host = parsed + .host_str() + .ok_or_else(|| anyhow!("metrics url {metrics_url:?} has no host"))?; + let port = parsed.port().unwrap_or(80); + let authority = format!("{host}:{port}"); + + // HTTP/1.0 so the server closes the body and a read-to-end terminates. The + // endpoint is a local Prometheus exporter; a full HTTP client would be a new + // dependency on this crate for one GET. + let mut stream = tokio::net::TcpStream::connect(&authority).await?; + let request = format!( + "GET {} HTTP/1.0\r\nHost: {}\r\nConnection: close\r\n\r\n", + parsed.path(), + authority + ); + tokio::io::AsyncWriteExt::write_all(&mut stream, request.as_bytes()).await?; + let mut body = Vec::new(); + tokio::io::AsyncReadExt::read_to_end(&mut stream, &mut body).await?; + let text = String::from_utf8_lossy(&body); + // A wrong path or a 404 would parse every counter to 0.0, and in the + // audit-off arm all-zeros reads as "the audit series stayed flat" — faking + // the positive control in exactly the misconfigured case it exists to catch. + let status_ok = text + .lines() + .next() + .is_some_and(|line| line.contains(" 200 ") || line.ends_with(" 200")); + if !status_ok { + let status = text.lines().next().unwrap_or(""); + bail!("metrics endpoint {metrics_url} did not return 200: {status}"); + } + + let mut out = serde_json::Map::new(); + for (needle, name) in WANTED { + let value = text + .lines() + .find_map(|line| { + let rest = line.strip_prefix(needle)?; + // Require a delimiter so `..._count` cannot match `..._count_x`. + if !rest.starts_with(' ') { + return None; + } + rest.trim().parse::().ok() + }) + // Absent means never incremented, which is zero. Safe only because + // the quota series has been positive-controlled on this rig. + .unwrap_or(0.0); + out.insert((*name).to_string(), json!(value)); + } + Ok(Value::Object(out)) +} + +fn percentiles(samples: &mut [f64]) -> Value { + samples.sort_by(|a, b| a.total_cmp(b)); + let at = |p: f64| -> Option { + let last = samples.len().checked_sub(1)?; + let idx = (last as f64 * p).round() as usize; + samples.get(idx).copied() + }; + json!({ "p50": at(0.50), "p95": at(0.95), "p99": at(0.99), "max": at(1.0) }) +} + +fn summarize(target: &Target, window: &Window, out: &mut Outcome) -> Value { + let service = percentiles(&mut out.service_ms); + let achieved = out.accepted as f64 / window.elapsed_secs; + + // The ceiling this generator imposes on itself: each connection is + // closed-loop, so it cannot exceed one send per service time. Derived from + // the *mean*, not the median — closed-loop throughput depends on mean + // service demand, and a median understates it on a skewed distribution. + // A reader comparing `achieved_per_s` against this can tell a relay ceiling + // from a generator ceiling; raise `conns` when they are close. + let service_mean_ms = (!out.service_ms.is_empty()) + .then(|| out.service_ms.iter().sum::() / out.service_ms.len() as f64); + let conn_capacity = service_mean_ms + .filter(|mean| *mean > 0.0) + .map(|mean| target.conns as f64 / (mean / 1e3)); + + json!({ + "url": target.url, + "channel": target.channel, + "conns": target.conns, + "offered_per_s": target.rate, + "attempted": out.attempted, + "accepted": out.accepted, + "rejected": out.rejected, + "achieved_per_s": achieved, + // Fraction of the whole offer that was accepted. The denominator is the + // offer the run asked for, not the rate re-derived from elapsed time, so + // a run that finishes a hair early cannot report better than 1.0. + "achieved_over_offered": out.accepted as f64 / (target.rate * window.requested_secs), + "conn_capacity_per_s": conn_capacity, + "service_mean_ms": service_mean_ms, + "generator_lag_ms": percentiles(&mut out.generator_lag_ms), + "service_ms": service, + "scheduled_ms": percentiles(&mut out.scheduled_ms), + "first_rejection": out.first_rejection, + "first_transport_error": out.first_transport_error, + }) +} + +#[tokio::main] +async fn main() -> anyhow::Result<()> { + // Errors only when another initializer already installed a provider, which + // is the same outcome we want. + let _ = rustls::crypto::CryptoProvider::install_default( + rustls::crypto::aws_lc_rs::default_provider(), + ); + + let args: Vec = std::env::args().skip(1).collect(); + let (duration_arg, target_args) = args + .split_first() + .filter(|(_, targets)| !targets.is_empty()) + .ok_or_else(|| { + anyhow!( + "usage: ingest_load [ ...]\n \ + target: url=,channel=,rate=[,conns=]" + ) + })?; + + let duration_secs: u64 = duration_arg.parse().context("duration_secs")?; + let targets: Vec = target_args + .iter() + .map(|spec| parse_target(spec)) + .collect::>()?; + + let keys = Keys::parse( + &std::env::var("BENCH_PRIVATE_KEY") + .context("BENCH_PRIVATE_KEY is required (a channel member's secret key)")?, + )?; + + // Connect everything before the clock starts. Otherwise one target is + // already publishing while another is still in its NIP-42 handshake, and + // the per-community rates were never concurrent. + let mut connected = Vec::new(); + for (t_idx, target) in targets.iter().enumerate() { + for conn_idx in 0..target.conns { + let client = BuzzTestClient::connect(&target.url, &keys) + .await + .with_context(|| format!("connecting to {}", target.url))?; + connected.push((t_idx, conn_idx, client)); + } + } + + let metrics_url = std::env::var("BENCH_METRICS_URL").ok(); + let counters_before = match metrics_url.as_deref() { + Some(url) => Some(scrape_counters(url).await.context("metrics before")?), + None => None, + }; + + let start = Instant::now(); + let deadline = start + Duration::from_secs(duration_secs); + let mut tasks = Vec::new(); + for (t_idx, conn_idx, client) in connected { + let target = targets + .get(t_idx) + .ok_or_else(|| anyhow!("target index {t_idx} vanished"))? + .clone(); + // Stagger connections within a target so the aggregate offer is evenly + // spaced at `rate` rather than arriving in bursts of `conns`. + let first_slot = start + Duration::from_secs_f64(conn_idx as f64 / target.rate); + let interval = Duration::from_secs_f64(target.conns as f64 / target.rate); + let label = format!("ingest-load t{t_idx} c{conn_idx}"); + let keys = keys.clone(); + tasks.push(tokio::spawn(async move { + let channel = target.channel.clone(); + let out = + drive_connection(client, keys, channel, label, first_slot, interval, deadline) + .await?; + Ok::<_, anyhow::Error>((t_idx, out)) + })); + } + + let mut per_target: Vec = targets.iter().map(|_| Outcome::default()).collect(); + for task in tasks { + let (t_idx, out) = task.await??; + per_target + .get_mut(t_idx) + .ok_or_else(|| anyhow!("target index {t_idx} vanished"))? + .absorb(out); + } + + let window = Window { + requested_secs: duration_secs as f64, + elapsed_secs: start.elapsed().as_secs_f64(), + }; + let counters_after = match metrics_url.as_deref() { + Some(url) => Some(scrape_counters(url).await.context("metrics after")?), + None => None, + }; + let mut aggregate = Outcome::default(); + let mut offered_total = 0.0; + let mut summaries = Vec::new(); + for (target, mut out) in targets.iter().zip(per_target) { + offered_total += target.rate; + summaries.push(summarize(target, &window, &mut out)); + aggregate.absorb(out); + } + + let achieved_total = aggregate.accepted as f64 / window.elapsed_secs; + println!( + "{}", + json!({ + "duration_secs": duration_secs, + "elapsed_secs": window.elapsed_secs, + "counters_before": counters_before, + "counters_after": counters_after, + "targets": summaries, + "aggregate": { + "offered_per_s": offered_total, + "attempted": aggregate.attempted, + "accepted": aggregate.accepted, + "rejected": aggregate.rejected, + "achieved_per_s": achieved_total, + "achieved_over_offered": + aggregate.accepted as f64 / (offered_total * window.requested_secs), + "service_ms": percentiles(&mut aggregate.service_ms), + "scheduled_ms": percentiles(&mut aggregate.scheduled_ms), + "generator_lag_ms": percentiles(&mut aggregate.generator_lag_ms), + }, + }) + ); + Ok(()) +} diff --git a/perf/RELAY_INGEST_CEILING.md b/perf/RELAY_INGEST_CEILING.md new file mode 100644 index 00000000000..82124136d2c --- /dev/null +++ b/perf/RELAY_INGEST_CEILING.md @@ -0,0 +1,333 @@ +# Buzz relay ingest ceiling harness + +Measures where accepted-event throughput stops tracking the offered rate, and +whether the audit write path is what stops it. + +It exists because the audit write path costs a fixed number of sequential +statements per accepted event, which predicts a hard ingest ceiling, and nothing +in this repo could measure it: there are no criterion benches, and `perf/` +otherwise covers only the Redis fan-out boundary. The structural claim is +verifiable from the source cited below; this harness is the part that can be wrong +out loud. (See `RESEARCH/BUZZ_BACKEND_PERF_FINDINGS.md` for the original survey +and its local checkout line references.) + +## What is under test + +`buzz-audit`'s `log` is six sequential client/server exchanges per entry — +advisory lock, BEGIN, head read, INSERT, COMMIT, unlock +(`crates/buzz-audit/src/service.rs`) — plus a synchronous durability wait inside +the COMMIT. It sits on the OK path: `dispatch_persistent_event` awaits +`audit_tx.send()` on a bounded channel before the rest of the dispatch is spawned. +So sustained ingest cannot exceed one audit entry per that fixed cost. + +The exchanges and the durability wait are separate costs, and conflating them is +easy: disabling `synchronous_commit` removes the flush *wait* but not the COMMIT +exchange, so a model of "five network round trips plus a commit" overcounts what +that setting removes. Both are amortized by batching, which is why the direction +of the proposed fix does not depend on the split. + +There are **two** ceilings and they coincide numerically: + +* **Per-pod worker ceiling.** One `tokio::spawn` per `AppState` drains the audit + channel serially for every community on the pod + (`crates/buzz-relay/src/state.rs`). Aggregate across communities. More pods + raise it. +* **Per-community lock ceiling.** The advisory lock is DB-global, so the six + round trips serialize per community cluster-wide. More pods do not raise it. + +**The worker ceiling is the lower of the two, so it masks the lock ceiling.** A +sweep that does not surface the lock is *structurally blind to it* — it is not +evidence the lock is fine. Exposing the lock needs a second measurement round +after the worker is fixed. Do not quote a passing run as clearing the lock. + +## Run it + +A full experiment is two half-runs, one per arm, judged together. Both arms bind +the same port, so they cannot run at once. + +```bash +./scripts/start-perf-ingest-rig.sh --reset > /tmp/rig-on.json +./perf/relay_ingest_ceiling.py --rig /tmp/rig-on.json --json /tmp/on.json + +./scripts/start-perf-ingest-rig.sh --reset --audit off > /tmp/rig-off.json +./perf/relay_ingest_ceiling.py --rig /tmp/rig-off.json --json /tmp/off.json + +./perf/relay_ingest_ceiling.py --combine /tmp/on.json /tmp/off.json +``` + +**`--reset` on both arms is not optional.** Running audit-on first and audit-off +second against a database that grew in between confounds the arm with time, cache +state and index size — a difference that has nothing to do with the audit path. +Restoring the same schema at both boundaries makes the arms comparable, and +`database_reset` is part of the experiment identity so `--combine` rejects a pair +where only one arm was reset. Randomised or interleaved ordering is the stronger +follow-up; the shared snapshot is what makes the immediate comparison valid. + +Each half-run exits non-zero on its own — a single arm is a partial experiment by +construction — and writes its `--json` first, so the workflow above still works. +`--skip-relay` attaches to a relay someone else supervises, which is required in +environments that reap detached processes. + +The rig is one relay process serving two communities, resolved by `Host`: +`a.localhost:3030` and `b.localhost:3030` both resolve to 127.0.0.1, so the URL +host *is* the `Host` header and no proxy is involved. That is what lets the +harness drive two communities at independent rates through one worker. Backing +services run under the `buzz-harness` Compose project, so the shared `:3000` dev +stack is untouched. + +Verdict logic without any services: + +```bash +./perf/relay_ingest_ceiling.py --mode model +python3 -m unittest discover -s perf -p 'test_*.py' +``` + +## What it asserts + +The contract is an **arm separation**, not a threshold. Each rate is run `--repeats` +times in both arms, and the verdict asks whether the difference in accepted/offered +between audit-off and audit-on excludes zero at any rate. No noise floor is needed +for that. + +An earlier version derived its pass threshold from the spread of repeated runs +(`1 - 3s`). That is retired, and the reason is worth keeping: in the unsaturated +region accepted/offered is pinned at 1.0, so the spread is ~0 and the threshold +absorbs nothing; in the saturated region the spread is the system's own throughput +variability, which is the signal, not the noise. No placement of that control +rescues the formula. + +Overlapping intervals are also not used as evidence of anything. "The arms' +intervals overlap" would not establish that they are equal — absence of a +significant difference is not evidence of absence — so the predicate is on the +interval of the *difference*. + +Non-zero exit on any of these, with every failure reported rather than the first: + +Cells that cannot be evidence are **excluded and reported**, not treated as run +failures — see the next section for why that distinction is load-bearing. Only +three things fail a run: a dead audit worker, a missing control, and too little +surviving evidence to compare the arms. + +1. **A cell was contaminated by admission control.** Either `reason="quota"` or + `reason="unavailable"` moved. The second matters as much as the first: a + rejected event takes the same NOTICE-without-OK path either way, and admission + itself costs Redis round trips against the same rig the sweep is loading — so + unavailability is load-correlated and can forge a knee that *persists* across + repeats at exactly the rate a reader would trust. +2. **A cell saw relay rejections, generator transport errors, or audit-write or + audit-enqueue failures.** An enqueue failure specifically means the worker is + gone: a bounded `mpsc::Sender::send` awaits when full and errors only when the + receiver is dropped. +3. **The generator had less than 1.5x headroom over the offered rate**, so the cell + was partly measuring the generator rather than the relay. +4. **The audit-off control did not run.** A single-arm dataset is a partial + experiment and reports `control.ran: false`; `--combine` judges the pair. The + verdict carries an explicit ran/skipped marker, because "no knee on the + audit-off arm" otherwise reads identically for "the control ran and the knee + was gone" and "the control never ran". +6. **The two halves handed to `--combine` are not the same experiment.** Rates, + duration, repeats, community hosts, both limiter settings, the generator path, + and the source revision must all match. +7. **The grid never saturated the audit path.** If no evidence cell reached a busy + worker, the run is *inconclusive*, not negative — the audit path was never + reached, so nothing here can exonerate it, and the grid needs extending. This + is a separate verdict from "compared and found nothing" because the two carry + opposite consequences: the first says re-run, the second would say stop. +8. **Only secondary rates separated.** The pass depends on one predeclared primary + contrast, at the highest rate that produced a comparison. Passing on any-of-N + rates runs an unadjusted test per rate: with five rates at a two-sided 95% + interval that is a **~10% false-pass rate**, measured on this module against + identical populations, not the nominal 5%. +9. **Audit-off did significantly worse than audit-on** at some rate. That + contradicts the hypothesis rather than failing to support it, and a separation + elsewhere cannot be read past it. +10. **The audit-off arm did not hold its offer** at the primary contrast. Both arms + collapsing is not "removing the audit path restores ingest", however large the + gap between them — the control has to be positive, not merely better. +11. **No rate kept two evidence cells in both arms**, so the arms cannot be + compared at all and too much of the dataset was excluded — or, when they can be + compared, **no rate separated them**, so the audit path is not shown to limit + ingest. + +Of those, items 1 and 2 exclude the offending cell; an audit *enqueue* failure is +the exception and fails the run outright, because a bounded `mpsc::Sender::send` +errors only when the receiver is dropped, so it means the worker is gone and every +later cell is suspect. + +### Why a non-steady cell is excluded rather than fatal + +The audit channel starts empty, so **the first repeat of the first saturating rate +banks acceptance credit by construction** — the preceding unsaturated rates never +filled it. A rate just above the drain rate accumulates over several repeats until +the channel caps, and that is the transition region where the bracket is decided. +A contract that failed the run on any non-steady cell would therefore fail exactly +the datasets this harness exists to judge. + +So a non-steady cell is dropped from the arm intervals and from the worker-rate +estimate and listed in `excluded_cells` with its reason. Dropping it also removes +the bias: the banked credit lands in `accepted`, inflating that cell's +accepted/offered, so it does not belong in the interval either way. A rate needs at +least two surviving repeats per arm to produce a difference interval, and the run +fails if no rate clears that — exclusion must not become a way to pass by +discarding almost everything. + +`model()` reproduces this fill sequence rather than setting every cell steady, so +the green path is tested against the physics. The two-community cells are +report-only by scope but carry `problems` and `steady` annotations, because an +unannotated contaminated cell in a table nobody judges is how a bad number gets +quoted later. + +`perf/test_relay_ingest_ceiling.py` pairs every passing case with a mutant that +must fail — a lone dip that must not count as a knee, a control that did not run, +an arm separation in the wrong direction, a cell that banked the whole channel, a +`--combine` across mismatched durations. A contract that cannot go red is +decoration. Nothing wires this suite into CI, a Justfile target, or a hook (the +same is true of the pre-existing bus-scaling tests), so run it by hand: + +```bash +python3 -m unittest discover -s perf -p 'test_*.py' +``` + +## Two throughput series, and why neither replaces the other + +- `accepted_per_s` — user-visible ingest. Arm separation is computed on this, + because it is the series both arms have. +- `audit_completed_per_s` — audit-worker completions, from + `buzz_audit_log_seconds`. **N/A in the audit-off arm**: with + `BUZZ_AUDIT_ENABLED=false` there is no worker and no series, so substituting it + for accepted throughput would make the positive control read as total collapse + and invert the predicate. + +Accepted throughput needs the steady-state gate because the audit channel is a +bounded `mpsc::channel(1000)`: a cell that starts with it empty accepts up to +1000 events before backpressure. Measured on this rig, `accepted - completed` came +to **exactly +1000** from a known-empty start and **exactly 0** from a full one. +That credit is a *bias*, identical across repeats, so an interval over n runs +converges tightly on a wrong number — precision and bias are different axes and n +only buys the first. It is exactly 1000 only in deep saturation; a transition cell +banks a partial amount depending on both offer and duration, and the knee lives in +the transition region, so the bias is least tractable exactly where the bracket is +decided. + +`audit_completed_per_s` is free of that credit, and it licenses a capacity claim +only where `audit_busy_fraction` is near 1 and no error counter moved — below +saturation a completion rate just tracks the offer. Two further bounds: the +histogram is a per-pod aggregate, so it cannot be split per community in the +two-community cells; and it measures the audit worker, which is the subject only +while the audit path is the binding constraint. Once the worker is fixed, it and +ingest throughput part ways. + +Counters are sampled **by the generator, at its own timed-window edges**, and a +cell that lacks aligned samples is not evidence. The runner can only bracket the +whole subprocess: its own "before" lands ahead of the connection phase and its +"after" after teardown, while every rate divides by the window that starts once +the connections are authenticated. Backlog draining during setup then lands in the +delta but not the divisor, which can push a busy fraction above 1.0, overstate +completions, and — because the exclusion decision reads `outstanding_delta` — flip +the verdict rather than merely shifting an estimate. `setup_overhead_fraction` +bounds what is left. + +`audit_busy_fraction` is reported explicitly rather than left implicit. Completion +rate and `1/mean(service)` are `C/T` and `C/S` over the same count, so their ratio +is exactly `S/T` — they are one measurement reported two ways, and their agreeing +tells you the worker was busy, not that two instruments corroborate each other. + +## The two-community arm, and what it can show + +Each rate is also run split evenly across two communities. The expected relation +follows from which defect binds: + +- If the **per-pod worker** is the ceiling, the two-community aggregate knee sits + at roughly the same place as the one-community knee — the worker drains all + communities serially, so splitting the offer buys nothing. +- If the **per-community lock** were the ceiling, two communities would reach + roughly double the combined rate. + +Round 1 observed the first shape. The cells are recorded but **not** judged: a +CI-backed equivalence test on the difference between the arms, with a predeclared +margin, is the designed follow-up. Overlapping marginal intervals would not +establish that the two ceilings are equal, so nothing here asserts that they are. + + +## Two latencies, and why both + +`ingest_load` reports each send twice: + +* `service_ms` — signed event on the wire until the relay's OK. Relay time. +* `scheduled_ms` — the send's *intended* slot until the OK. + +The schedule advances by a fixed interval and is never rebased on the response. +A generator that paces with a self-correcting timer and measures from the actual +send silently redefines its own offer downward when the relay slows: throughput +caps at `connections / latency` and the queueing delay never enters the +percentiles. That is coordinated omission, and it hides the damage exactly when +the damage is the point. Signing happens before the service clock starts, so +BIP340 cost lands in `scheduled_ms` and never inflates the relay's number. + +`conn_capacity_per_s` is the ceiling the generator imposes on itself — each +connection is closed-loop, so it cannot exceed one send per service latency. +Compare `achieved_per_s` against it before believing any knee; raise `conns` when +they are close. + +## The trap this harness exists to avoid + +At default settings one identity is capped at **50 events per 5 seconds**: +`human_ws_events_per_sec` defaults to 10 (`crates/buzz-auth/src/rate_limit.rs`) +and `ws_admission_budget` turns that into a fixed 5s window with a limit of 50 +(`crates/buzz-relay/src/admission.rs`), keyed on `(tenant, pubkey)`. + +A rejected EVENT gets a **NOTICE**, and a NOTICE carries no event id +(`request_rejection_message` with no `sub_id` — `crates/buzz-relay/src/connection.rs`). +A NIP-01 client waiting for an `OK` therefore never sees the rejection and blocks +for its whole publish timeout (30s in `buzz-ws-client`). Measured on this rig: 50 +events land in 2.4s, the next send stalls 30s, and the run self-truncates. A +reader seeing the resulting ~1.5/s would have a limiter artifact that looks like +a textbook saturation knee. + +So the rig raises both admission limits — both matter, because `WsEvents` is a 5s +window and `Messages` a 60s one, and a short run only ever exercises the first. +The configured values are printed in the run metadata, and the harness invalidates +any run where the quota-rejection counter moves. Scoped to `reason="quota"`: +`reason="unavailable"` means the limiter itself was unreachable, which is a +different diagnosis. + +## What this harness can and cannot support + +**It characterizes the mechanism, not deployability at scale.** A raised-limit +sweep is valid evidence that the audit path caps a community's ingest. It is +*not* evidence that a real community reaches that rate. At production defaults +one identity sustains far less, so reaching a few hundred events/s inside one +community needs hundreds of concurrent identities — each carrying its own +WebSocket and its own per-event admission round trip that this sweep never pays. + +The **N-identity variant** is the queued fidelity check for exactly that gap. +Two constraints on whoever builds it: + +* **Self-hosted isolated stack only.** `SECURITY.md` asks reporters not to + disrupt production systems. Never point it at a hosted or shared relay. +* Nothing on the relay side will stop it opening hundreds of connections from one + host: `LimitType::IpConnections` is defined in the enum but wired up nowhere, + and the only connection cap is a global semaphore. Convenient here, and not to + be mistaken for a control that exists. + +`load_per_cpu` is recorded per cell, but read it with its resolution in mind: it +is a 1-minute average sampled inside a much shorter cell, so consecutive cells are +autocorrelated and every cell in a sweep reads about the same. It can catch a +sweep-long compile storm on a shared machine; it cannot exonerate one cell. Run +sweeps on an otherwise idle machine. + +## Not yet measured + +* **Sensitivity to round-trip latency.** Injected delay between relay and + Postgres should move the ceiling, and a sweep across at least three injected + values can measure how much — a single value agreeing with one predicted number + is a coincidence indistinguishable from a correct prediction. But predeclare + the expected shape as *six* added exchange delays plus a durability intercept, + not five: the COMMIT exchange is still a round trip. Attributing the resulting + slope needs per-statement timing, since the measured interval also contains pool + acquisition, SQL execution, hash chaining and WAL generation. No clean + round-trip decomposition is available from anything run so far. Local Postgres is a loopback socket, so absolute rates from this + rig are not comparable to a same-VPC deployment — the harness prints measured + latency beside every rate for that reason, and no absolute events/s figure from + it should be quoted as a production number. +* **The lock ceiling**, per the blindness note above. diff --git a/perf/relay_ingest_ceiling.py b/perf/relay_ingest_ceiling.py new file mode 100755 index 00000000000..49df9ae27e1 --- /dev/null +++ b/perf/relay_ingest_ceiling.py @@ -0,0 +1,1187 @@ +#!/usr/bin/env python3 +"""Ingest-ceiling harness for the Buzz relay's audit write path. + +Answers one causal question — does the audit log limit ingest? — and reports a +worker-rate estimate alongside it. Stdlib only; `ingest_load` (Rust) does the +measuring and this script owns the experiment and the verdict. + +The contract is an **arm separation**, not a threshold. With the audit log +enabled, accepted/offered falls away from 1.0 as the offer rises; with +`BUZZ_AUDIT_ENABLED=false` it does not. Repeats give each arm an interval, and +the verdict asks whether the difference between the arms excludes zero at any +rate. No noise floor is needed for that, which matters: an earlier design derived +its pass threshold from the spread of repeated runs, and in the saturated region +that spread is the system's own throughput variability — the signal, not noise. + +Two throughput series are reported and neither replaces the other: + + * `accepted_per_s` — user-visible ingest, and the quantity arm separation is + computed on, because it is the series both arms have. + * `audit_completed_per_s` — audit-worker completions, from + `buzz_audit_log_seconds`. Free of the acceptance credit below, and **N/A in + the audit-off arm**, where there is no worker and no series. It licenses a + capacity claim only where the worker was demonstrably busy + (`audit_busy_fraction` near 1) and no error counter moved. + +Why accepted throughput needs a validity gate: the audit channel is a bounded +`mpsc::channel(1000)`, so a cell starting with it empty accepts up to 1000 events +before backpressure — measured on this rig as exactly +1000 from an empty start +and 0 from a full one. That credit is a *bias*, identical across repeats, so an +interval over n runs converges tightly on a wrong number. And it is exactly 1000 +only in deep saturation: a transition cell banks a partial amount depending on +both offer and duration, and the knee lives in the transition region, so the bias +is least tractable exactly where the bracket is decided. `outstanding_delta` +reports it per cell. + +Two bounds on `audit_completed_per_s`, since it invites over-reading: the +histogram is a per-pod aggregate, so it cannot be split per community in the +two-community cells; and it measures the audit worker, which is the subject only +while the audit path is the binding constraint. Once the worker is fixed, it and +ingest throughput part ways. + +Two ceilings are under test and they coincide numerically. The per-pod audit +worker drains all communities serially; the per-community advisory lock +serializes cluster-wide. The minimum always wins and it is always the worker, so +a first sweep is *structurally blind* to the lock. A run that does not surface the +lock is not evidence the lock is fine. See perf/RELAY_INGEST_CEILING.md. + +Usage: + ./scripts/start-perf-ingest-rig.sh --reset > /tmp/rig-on.json + ./perf/relay_ingest_ceiling.py --rig /tmp/rig-on.json --json /tmp/on.json + # restart the rig with --audit off, sweep again into /tmp/off.json, then: + ./perf/relay_ingest_ceiling.py --combine /tmp/on.json /tmp/off.json + + ./perf/relay_ingest_ceiling.py --mode model # verdict logic, no services + +Exits non-zero when any cell is invalid or the contract is not met. +""" + +from __future__ import annotations + +import argparse +import json +import math +import os +import subprocess +import sys +import time +import urllib.request +from typing import Callable + +# Offered events per second per connection. Raising this does not raise what a +# closed-loop generator can push through a saturated relay — measured on this rig, +# going from 32 to 80 connections at 800/s left throughput flat and stretched +# service time from 57ms to 187ms, because the extra sends just queued. Enough +# connections that an *unsaturated* cell can meet its offer is all this needs to +# buy; above the ceiling the relay sets the pace. +RATE_PER_CONN = 25.0 +MIN_CONNS = 4 + +# Reported alongside every cell, never gated on — see `cell_problems`. +GENERATOR_HEADROOM_MARGIN = 1.5 + +# Depth of the relay's audit channel, and the share of it that `outstanding_delta` +# may move before a cell counts as not in steady state. +AUDIT_CHANNEL_DEPTH = 1000 +OUTSTANDING_TOLERANCE_FRACTION = 0.05 + +# Below this, the worker idled inside the window, so its completion rate tracks +# the offer rather than its own limit. +BUSY_FRACTION_FOR_CAPACITY = 0.95 + +# The audit-off arm must hold its offer, not merely beat the audit-on arm. Both +# arms collapsing is not "audit removal restores ingest". +CONTROL_EQUIVALENCE_MARGIN = 0.05 + +# Counter deltas are read around the whole subprocess, but the generator's own +# window starts only after every connection is authenticated. A cell is rejected +# when that setup overhead is a material share of the window, because rates +# divided by mismatched windows can exceed 1.0 and overstate completions. +MAX_SETUP_OVERHEAD_FRACTION = 0.05 + +# Reported, never gated on — see `cell_problems`. +MIN_ATTEMPTED_FRACTION = 0.98 + +# Bound on the one generator-vs-relay discriminator that survives saturation; +# `Outcome::generator_lag_ms` in ingest_load.rs defines it. Only signing and +# scheduler delay land in it, and signing is tens of microseconds, so a p99 in +# milliseconds means the generator itself was starved. +MAX_GENERATOR_LAG_P99_MS = 10.0 + +# The grid has to straddle the drain rate, or the separation contract has nothing +# to fire on and the run reports "the audit path is not shown to limit ingest" — +# a false negative phrased as a conclusion. Every drain figure measured on this +# rig falls in 390-495/s, so 100-200 anchor the unsaturated arm, 400 brackets the +# ceiling from below, and 800/1600 are in clear saturation. Re-derive this if the +# rig's drain rate moves: a grid whose top rate sits at the ceiling makes the +# only informative cell a coin flip. +DEFAULT_RATES = [100.0, 200.0, 400.0, 800.0, 1600.0] +DEFAULT_REPEATS = 5 + +# Drain rate observed on this rig, used only to keep `model()` honest about where +# its ceiling sits relative to the grid. Not a capacity claim. +MEASURED_DRAIN_BAND_PER_S = 450.0 + +# Two-tailed 95% critical values by degrees of freedom; the fallback is the +# large-sample limit. Enough for the repeat counts this harness runs. +_T95 = { + 1: 12.706, 2: 4.303, 3: 3.182, 4: 2.776, 5: 2.571, 6: 2.447, 7: 2.365, + 8: 2.306, 9: 2.262, 10: 2.228, 12: 2.179, 15: 2.131, 20: 2.086, 25: 2.060, + 30: 2.042, +} + + +# -- statistics (pure) ------------------------------------------------------ + + +def mean(values: list[float]) -> float: + if not values: + raise ValueError("mean of no observations") + return sum(values) / len(values) + + +def sample_stddev(values: list[float]) -> float: + """Sample standard deviation, n-1. + + Not `(max - min) / mean`: the range's expectation grows with n (1.128 sigma + at n=2, 2.059 sigma at n=4), so a range-based spread rises with sample size + on its own and cannot be compared across cells with different repeat counts. + """ + if len(values) < 2: + raise ValueError("standard deviation needs at least two observations") + avg = mean(values) + return math.sqrt(sum((v - avg) ** 2 for v in values) / (len(values) - 1)) + + +def t95(df: float) -> float: + """Two-tailed 95% critical value, rounded to the conservative side. + + The table is sparse, so an exact df is often missing. Taking the next *higher* + stored df would pick a *smaller* critical value and under-cover: df 11 would + get 2.179 against a true 2.201, and df 13 would get 2.131 against 2.160. Take + the largest stored df at or below the real one instead, which errs wide. + """ + if df < 1: + raise ValueError("t95 needs at least one degree of freedom") + key = int(math.floor(df)) + usable = [cutoff for cutoff in _T95 if cutoff <= key] + return _T95[max(usable)] if usable else _T95[min(_T95)] + + +def confidence_interval(values: list[float]) -> dict: + """Mean with a 95% t-interval. Bounds are None when n < 2.""" + avg = mean(values) + if len(values) < 2: + return {"n": len(values), "mean": avg, "lo": None, "hi": None} + half = t95(len(values) - 1) * sample_stddev(values) / math.sqrt(len(values)) + return {"n": len(values), "mean": avg, "lo": avg - half, "hi": avg + half} + + +def difference_interval(a: list[float], b: list[float]) -> dict: + """95% Welch interval for mean(a) - mean(b), unequal variances. + + Arm separation asks whether this interval excludes zero. "The two arms' + intervals do not overlap" would be a weaker test and "they do overlap" would + prove nothing at all: absence of a significant difference is not evidence of + equality. + """ + if len(a) < 2 or len(b) < 2: + raise ValueError("difference interval needs two observations per arm") + va, vb = sample_stddev(a) ** 2, sample_stddev(b) ** 2 + na, nb = len(a), len(b) + se = math.sqrt(va / na + vb / nb) + diff = mean(a) - mean(b) + if se == 0.0: + return {"diff": diff, "lo": diff, "hi": diff, "excludes_zero": diff != 0.0} + df = (va / na + vb / nb) ** 2 / ( + (va / na) ** 2 / (na - 1) + (vb / nb) ** 2 / (nb - 1) + ) + half = t95(df) * se + lo, hi = diff - half, diff + half + return {"diff": diff, "lo": lo, "hi": hi, "excludes_zero": lo > 0.0 or hi < 0.0} + + +# -- knee, reporting only --------------------------------------------------- + + +def find_knee(points: list[tuple[float, float]], threshold: float) -> float | None: + """Lowest offered rate whose shortfall persists at the next rate too. + + Reporting only - nothing gates on it. A knee is a grid point, not a + measurement, so `threshold` here is a presentation choice and not a contract. + Saturation is monotone, so a lone dip is noise; the highest rate may stand + alone because it has no successor. + """ + ordered = sorted(points) + for idx, (rate, fraction) in enumerate(ordered): + if fraction >= threshold: + continue + if idx == len(ordered) - 1 or ordered[idx + 1][1] < threshold: + return rate + return None + + +def knee_bracket( + points: list[tuple[float, float]], threshold: float +) -> tuple[float | None, float | None]: + """The interval the ceiling lies in: (highest passing rate, knee).""" + knee = find_knee(points, threshold) + if knee is None: + return (None, None) + passing = [r for r, f in sorted(points) if r < knee and f >= threshold] + return (passing[-1] if passing else None, knee) + + +# -- validity and verdict (pure) -------------------------------------------- + + +def cell_problems(cell: dict) -> list[str]: + """Everything that disqualifies one cell from being evidence.""" + problems = [] + rate = cell["offered_per_s"] + + for key, why in ( + ("quota_rejections_delta", + "admission quota rejections: the limiter was measured, not the relay"), + ("unavailable_rejections_delta", + "admission reported unavailable: those events take the same " + "NOTICE-without-OK path as quota, and admission itself costs Redis round " + "trips against the rig the sweep is loading, so it is load-correlated " + "and can forge a knee that persists"), + ("audit_log_errors_delta", "audit writes failed"), + ("audit_send_errors_delta", "audit enqueue failed, which means the worker is gone"), + ("rejected", "the relay rejected events"), + ("transport_errors", "generator transport errors"), + ): + if cell.get(key): + problems.append("{:g}/s: {} ({}={})".format(rate, why, key, cell[key])) + + if cell.get("counters_window_aligned") is False: + problems.append( + "{:g}/s: audit counters were sampled around the whole subprocess " + "rather than the timed window, so completion and outstanding-work " + "readings are not comparable with the rates".format(rate) + ) + + overhead = cell.get("setup_overhead_fraction") + if overhead is not None and overhead > MAX_SETUP_OVERHEAD_FRACTION: + problems.append( + "{:g}/s: setup and teardown were {:.1%} of the window, over the {:.0%} " + "bound, so window-edge readings are unreliable".format( + rate, overhead, MAX_SETUP_OVERHEAD_FRACTION + ) + ) + + if cell.get("audit_activity_in_control_arm"): + problems.append( + "{:g}/s: the audit series moved by {} in the audit-off arm, so that " + "relay was still auditing and the control is not a control".format( + rate, cell["audit_activity_in_control_arm"] + ) + ) + + lag = cell.get("generator_lag_p99_ms") + if lag is not None and lag > MAX_GENERATOR_LAG_P99_MS: + problems.append( + "{:g}/s: the generator took {:.1f}ms at p99 between being free to " + "send and sending, over the {:.0f}ms bound, so it was starved and " + "this cell measures the generator".format( + rate, lag, MAX_GENERATOR_LAG_P99_MS + ) + ) + + # `attempted_over_offered` and `generator_headroom` are reported, never + # gated. By Little's law, a closed-loop sender against a relay whose + # completion rate is fixed has service time L/rate, so `conns / service` + # equals the relay's own throughput at any connection count — see the + # measurement at `RATE_PER_CONN`. Gating on either would reject every + # saturated cell, which is every cell that matters. Issuability comes from + # the control arm instead: audit-off holding its offer at rate R shows the + # generator can issue R. Overdriving a saturated relay needs an open-loop + # generator, a different instrument. + return problems + + +def steady_state(cell: dict) -> bool | None: + """Whether outstanding audit work held level across the window. + + The criterion is stability, not emptiness. A saturating cell settles with the + channel full and backpressure engaged; an unsaturated one settles near zero. + Both are steady. Requiring "empty at start" would make every saturated cell - + every cell that matters for a ceiling - permanently unmeasurable. + """ + delta = cell.get("outstanding_delta") + if delta is None: + return None + return abs(delta) <= OUTSTANDING_TOLERANCE_FRACTION * AUDIT_CHANNEL_DEPTH + + +# A dead audit worker invalidates every cell after it, not just its own, so this +# one problem fails the run instead of dropping a cell. +FATAL_PROBLEM_KEYS = ("audit_send_errors_delta",) + + +def fatal_problems(cells: list[dict]) -> list[str]: + out = [] + for cell in cells: + for key in FATAL_PROBLEM_KEYS: + if cell.get(key): + out.append( + "{:g}/s: audit enqueue failed, which means the worker is gone " + "and every later cell is suspect ({}={})".format( + cell["offered_per_s"], key, cell[key] + ) + ) + return out + + +def cell_exclusions(cells: list[dict]) -> list[dict]: + """Cells that cannot be evidence, with why. + + Excluded rather than run-fatal. A saturating cell that starts with the audit + channel empty banks its whole depth in accepted events — measured as exactly + +1000 — so the *first* repeat of the first saturating rate in any sweep is + non-steady by construction, and a transition rate accumulates over several + repeats. Failing the run on that would fail precisely the datasets this + harness exists to judge; the bias also means the cell does not belong in the + interval, since the credit inflates accepted/offered. + """ + excluded = [] + for cell in cells: + # Fatal problems are reported as run failures; repeating them here would + # print the same sentence twice under two different headings. + reasons = [ + r for r in cell_problems(cell) + if not any(key in r for key in FATAL_PROBLEM_KEYS) + ] + if steady_state(cell) is False: + reasons.append( + "{:g}/s: outstanding audit work moved by {}, so the cell carries " + "acceptance credit rather than a steady rate".format( + cell["offered_per_s"], cell["outstanding_delta"] + ) + ) + if reasons: + excluded.append( + { + "offered_per_s": cell["offered_per_s"], + "audit_enabled": cell["audit_enabled"], + "reasons": reasons, + } + ) + return excluded + + +def cell_is_evidence(cell: dict) -> bool: + return not cell_problems(cell) and steady_state(cell) is not False + + +def arm_separation(on_cells: list[dict], off_cells: list[dict]) -> dict: + """Per-rate difference in accepted/offered between the arms. + + Separation holds when at least one rate's difference interval lies wholly + above zero with audit-off higher. + """ + by_rate: dict = {} + for cells, arm in ((on_cells, "on"), (off_cells, "off")): + for cell in cells: + entry = by_rate.setdefault( + cell["offered_per_s"], {"on": [], "off": [], "dropped": 0} + ) + if cell_is_evidence(cell): + entry[arm].append(cell["accepted_over_offered"]) + else: + entry["dropped"] += 1 + + rates = [] + separated = False + contradicted = [] + comparable = 0 + for rate in sorted(by_rate): + on_vals, off_vals = by_rate[rate]["on"], by_rate[rate]["off"] + entry = { + "offered_per_s": rate, + "evidence_cells": {"audit_on": len(on_vals), "audit_off": len(off_vals)}, + "dropped_cells": by_rate[rate]["dropped"], + "audit_on": confidence_interval(on_vals) if on_vals else None, + "audit_off": confidence_interval(off_vals) if off_vals else None, + } + if len(on_vals) >= 2 and len(off_vals) >= 2: + comparable += 1 + diff = difference_interval(off_vals, on_vals) + entry["off_minus_on"] = diff + if diff["excludes_zero"] and diff["diff"] > 0.0: + separated = True + entry["separated_here"] = True + elif diff["excludes_zero"] and diff["diff"] < 0.0: + # Audit-off did significantly *worse*. That contradicts the + # hypothesis rather than failing to support it. + contradicted.append(rate) + entry["contradicted_here"] = True + else: + entry["off_minus_on"] = None + entry["note"] = "fewer than two evidence cells in one arm" + rates.append(entry) + + # One predeclared primary contrast, at the highest rate that produced a + # comparison. Passing on "any of N rates" runs an unadjusted test per rate: + # with five rates at a two-sided 95% interval the false-pass rate is ~10%, + # measured on this code with identical populations, not the nominal 5%. + comparisons = [e for e in rates if e.get("off_minus_on")] + primary = comparisons[-1] if comparisons else None + primary_separated = bool(primary and primary.get("separated_here")) + + control_holds = None + if primary and primary["audit_off"]: + lo = primary["audit_off"]["lo"] + control_holds = lo is not None and lo >= 1.0 - CONTROL_EQUIVALENCE_MARGIN + + return { + "separated": separated, + "comparable_rates": comparable, + "primary_rate": primary["offered_per_s"] if primary else None, + "primary_separated": primary_separated, + "primary_control_holds_offer": control_holds, + "contradicted_rates": contradicted, + "secondary_separated_rates": [ + e["offered_per_s"] for e in rates + if e.get("separated_here") and e is not primary + ], + "by_rate": rates, + } + + +def worker_rate(on_cells: list[dict]) -> dict: + """Audit-worker completion rate, from cells where it means capacity. + + Only cells with a busy worker, steady outstanding work and no errors qualify: + below saturation the completion rate tracks the offer, not the worker's limit. + """ + usable = [ + c for c in on_cells + if cell_is_evidence(c) + and steady_state(c) + and (c.get("audit_busy_fraction") or 0.0) >= BUSY_FRACTION_FOR_CAPACITY + ] + if not usable: + return { + "cells": 0, + "estimate": None, + "note": "no cell had a demonstrably busy worker in steady state", + } + by_rate: dict = {} + for c in usable: + by_rate.setdefault(c["offered_per_s"], []).append(c) + return { + "cells": len(usable), + # Per rate, not pooled: different offered rates are different load and + # database regimes, not repeats of one estimand. + "per_rate": [ + { + "offered_per_s": rate, + "completed_per_s": confidence_interval( + [c["audit_completed_per_s"] for c in cells] + ), + "service_ms": confidence_interval( + [c["audit_service_mean_ms"] for c in cells] + ), + } + for rate, cells in sorted(by_rate.items()) + ], + "estimate": confidence_interval([c["audit_completed_per_s"] for c in usable]), + "service_ms": confidence_interval( + [c["audit_service_mean_ms"] for c in usable] + ), + "note": ( + "audit-worker completion rate; not ingest capacity, and not the " + "subject at all once the worker is fixed" + ), + } + + +def verdict( + on_cells: list[dict], off_cells: list[dict] | None, control_ran: bool +) -> dict: + """Whether the dataset supports the audit-attribution claim. + + Cells that cannot be evidence are excluded and reported, not treated as run + failures. Only three things fail a run: a dead audit worker, a missing + control, and too little surviving evidence to compare the arms. + """ + all_cells = list(on_cells) + list(off_cells or []) + failures = fatal_problems(all_cells) + excluded = cell_exclusions(all_cells) + + busy = [c.get("audit_busy_fraction") or 0.0 for c in on_cells if cell_is_evidence(c)] + max_busy = max(busy) if busy else 0.0 + any_saturated = max_busy >= BUSY_FRACTION_FOR_CAPACITY + + if not control_ran: + failures.append( + "the audit-off control did not run: this dataset is a partial " + "experiment and cannot attribute a ceiling to the audit path" + ) + + separation = ( + arm_separation(on_cells, off_cells) + if control_ran and off_cells + else {"separated": False, "comparable_rates": 0, "by_rate": []} + ) + # These states carry different program consequences, so the run must say + # which one it is in. "Compared and found nothing" exonerates the audit path + # and would stop the work; "never saturated" or "lost the informative cells" + # mean re-run. A single message covering all three makes the loudest line the + # one a reader acts on hardest, and it is only correct for the first. + uncomparable = [ + entry["offered_per_s"] + for entry in separation.get("by_rate", []) + if entry.get("off_minus_on") is None and entry["dropped_cells"] + ] + if control_ran: + if not separation["comparable_rates"]: + failures.append( + "no rate kept two evidence cells in both arms, so the arms were " + "never compared: too much of this dataset was excluded to conclude " + "anything" + ) + elif not any_saturated: + failures.append( + "inconclusive, not negative: no audit-on cell reached a busy " + "worker (max busy fraction {:.2f} against the {:.2f} gate), so " + "this grid never saturated the audit path and cannot exonerate " + "it. Extend the rates above the drain rate and re-run".format( + max_busy, BUSY_FRACTION_FOR_CAPACITY + ) + ) + elif separation["contradicted_rates"]: + failures.append( + "audit-off was significantly *worse* than audit-on at {}: that " + "contradicts the hypothesis rather than failing to support it, " + "and no separation elsewhere can be read past it".format( + ", ".join( + "{:g}/s".format(r) for r in separation["contradicted_rates"] + ) + ) + ) + elif uncomparable and not separation["separated"]: + failures.append( + "inconclusive rather than negative: {} rate(s) ({}) lost their " + "evidence to exclusions, so a missing separation cannot be told " + "apart from missing data. A bracket-refinement run near the " + "ceiling drops its most informative cells exactly this way".format( + len(uncomparable), + ", ".join("{:g}".format(r) for r in uncomparable), + ) + ) + elif not separation["separated"]: + failures.append( + "every rate was comparable and none separated: the audit path is " + "not shown to limit ingest at these rates" + ) + elif not separation["primary_separated"]: + failures.append( + "only secondary rates separated; the predeclared primary contrast " + "at {:g}/s did not. Passing on any-of-N rates runs an unadjusted " + "test per rate: with five rates at a two-sided 95% interval that " + "is a ~10% false-pass rate, measured on this code with identical " + "populations".format(separation["primary_rate"]) + ) + elif separation["primary_control_holds_offer"] is False: + failures.append( + "the audit-off arm did not hold its offer at the primary contrast " + "({:g}/s): a control that also collapsed does not show that " + "removing the audit path restores ingest, however large the gap " + "between the arms".format(separation["primary_rate"]) + ) + elif uncomparable: + failures.append( + "separation was found, but {} rate(s) ({}) lost their evidence to " + "exclusions and contributed nothing".format( + len(uncomparable), + ", ".join("{:g}".format(r) for r in uncomparable), + ) + ) + + return { + "ok": not failures, + "control": {"ran": control_ran, "arm": "audit_off"}, + "failures": failures, + "excluded_cells": excluded, + "saturation": { + "max_busy_fraction": max_busy, + "gate": BUSY_FRACTION_FOR_CAPACITY, + "any_rate_saturated": any_saturated, + }, + "arm_separation": separation, + "worker_rate": worker_rate(on_cells), + "lock_ceiling": ( + "structurally blind - the per-pod worker ceiling is lower and masks " + "the per-community lock, so this dataset says nothing about the lock" + ), + } + + +# -- measurement ------------------------------------------------------------ + + +def load_per_cpu() -> float: + """1-minute load average per CPU. + + A lagging aggregate: sampled inside a short cell it is autocorrelated with + the previous cell, so it can catch a sweep-long compile storm and cannot + attribute load to any one cell. + """ + return os.getloadavg()[0] / (os.cpu_count() or 1) + + +def conns_for(rate: float) -> int: + return max(MIN_CONNS, int(math.ceil(rate / RATE_PER_CONN))) + + +def scrape(metrics_url: str) -> dict: + """Audit and admission counters. + + Every series is created on first increment, so absent reads as zero. That is + only safe because the quota series has been positive-controlled on this rig: + it does appear and increment when the limiter binds. + """ + with urllib.request.urlopen(metrics_url, timeout=10) as response: + body = response.read().decode("utf-8", "replace") + wanted = [ + ("buzz_audit_log_seconds_count", "audit_count"), + ("buzz_audit_log_seconds_sum", "audit_sum"), + ("buzz_audit_log_errors_total", "audit_log_errors"), + ("buzz_audit_send_errors_total", "audit_send_errors"), + ('buzz_admission_rejections_total{transport="websocket",reason="quota"}', "quota"), + ('buzz_admission_rejections_total{transport="websocket",reason="unavailable"}', + "unavailable"), + ] + out = {name: 0.0 for _, name in wanted} + for line in body.splitlines(): + for needle, name in wanted: + if line.startswith(needle + " "): + out[name] = float(line.split()[-1]) + return out + + +def generator_env(rig: dict) -> dict: + """Environment for the generator subprocess. + + `BENCH_METRICS_URL` is not optional in practice: without it the generator + reports no window-edge counters, every cell fails the alignment gate, and the + whole run is excluded. Extracted so that wiring is testable without a rig — + the first version of it was missing and no unit test could see it, because the + tests construct cells directly and never build this environment. + """ + env = dict( + os.environ, + BENCH_PRIVATE_KEY=rig["bench_private_key"], + BENCH_METRICS_URL=rig["metrics_url"], + ) + for stale in ("BUZZ_AUTH_TAG", "BUZZ_RELAY_URL", "BUZZ_PRIVATE_KEY"): + env.pop(stale, None) + return env + + +def run_generator(rig: dict, duration: int, offers: list) -> dict: + specs = [] + for index, rate in offers: + target = rig["targets"][index] + specs.append( + "url={},channel={},rate={},conns={}".format( + target["url"], target["channel"], rate, conns_for(rate) + ) + ) + env = generator_env(rig) + try: + # stderr is deliberately left on the inherited handle: capturing it would + # bury the generator's error context inside the exception, and a bare + # CalledProcessError is not diagnosable in the field. + completed = subprocess.run( + [rig["generator"], str(duration)] + specs, + env=env, + cwd=rig.get("repo_root") or ".", + stdout=subprocess.PIPE, + check=True, + ) + except subprocess.CalledProcessError as e: + raise SystemExit( + "generator failed (exit {}); its output is above".format(e.returncode) + ) + return json.loads(completed.stdout) + + +def run_cell(rig: dict, duration: int, offers: list, audit_on: bool) -> dict: + load_before = load_per_cpu() + outer_before = scrape(rig["metrics_url"]) + outer_start = time.monotonic() + result = run_generator(rig, duration, offers) + outer_elapsed = time.monotonic() - outer_start + outer_after = scrape(rig["metrics_url"]) + + # Prefer the counters the generator sampled at its own timed-window edges. + # The runner's own pair brackets the whole subprocess — connection setup and + # teardown included — while every rate is divided by the post-connect window, + # so backlog draining during setup lands in the delta but not the divisor. + # That can push a busy fraction above 1.0 and overstate completions, and it + # feeds the exclusion decision, so it can change the verdict rather than only + # the estimate. + aligned = bool(result.get("counters_before") and result.get("counters_after")) + before = result["counters_before"] if aligned else outer_before + after = result["counters_after"] if aligned else outer_after + + agg = result["aggregate"] + window = result["elapsed_secs"] + completed = after["audit_count"] - before["audit_count"] + service_sum = after["audit_sum"] - before["audit_sum"] + accepted = agg["accepted"] + + service_means = [ + t["service_mean_ms"] for t in result["targets"] if t.get("service_mean_ms") + ] + conns = sum(t["conns"] for t in result["targets"]) + headroom = None + if service_means and agg["offered_per_s"]: + # Closed-loop bound on what the generator could have offered, so the + # sweep does not end up measuring the generator. Mean service demand, + # not a median: closed-loop throughput depends on the mean, and a median + # understates it on a skewed distribution. + headroom = conns * (1000.0 / mean(service_means)) / agg["offered_per_s"] + + setup_overhead = max(0.0, outer_elapsed - window) / window if window else None + + cell = { + "offered_per_s": agg["offered_per_s"], + "audit_enabled": audit_on, + "duration_secs": duration, + "elapsed_secs": window, + "outer_elapsed_secs": outer_elapsed, + "setup_overhead_fraction": setup_overhead, + "counters_window_aligned": aligned, + "accepted": accepted, + "accepted_per_s": accepted / window if window else None, + "accepted_over_offered": agg["achieved_over_offered"], + "rejected": agg["rejected"], + "transport_errors": sum( + 1 for t in result["targets"] if t["first_transport_error"] + ), + "quota_rejections_delta": int(after["quota"] - before["quota"]), + "unavailable_rejections_delta": int( + after["unavailable"] - before["unavailable"] + ), + "audit_log_errors_delta": int( + after["audit_log_errors"] - before["audit_log_errors"] + ), + "audit_send_errors_delta": int( + after["audit_send_errors"] - before["audit_send_errors"] + ), + "generator_headroom": headroom, + "generator_lag_p99_ms": agg.get("generator_lag_ms", {}).get("p99"), + "attempted_over_offered": ( + agg["attempted"] / (agg["offered_per_s"] * duration) + if agg["offered_per_s"] else None + ), + "load_per_cpu_before": load_before, + "load_per_cpu_after": load_per_cpu(), + "service_ms": agg["service_ms"], + "scheduled_ms": agg["scheduled_ms"], + } + + if audit_on: + cell["audit_completed"] = int(completed) + cell["audit_completed_per_s"] = completed / window if window else None + cell["audit_service_mean_ms"] = ( + service_sum / completed * 1000.0 if completed else None + ) + # Share of the window the worker spent inside a timed `audit.log`. This is + # what "the completion rate agrees with 1/mean" actually measures: those + # are C/T and C/S over the same count, so their ratio is exactly S/T. + cell["audit_busy_fraction"] = service_sum / window if window else None + # accepted - completed = queued + in flight, valid only with both audit + # error deltas at zero and this generator as the sole producer. + cell["outstanding_delta"] = int(accepted - completed) + else: + cell["audit_completed_per_s"] = None + cell["audit_service_mean_ms"] = None + cell["audit_busy_fraction"] = None + cell["outstanding_delta"] = None + # Recorded rather than assumed: the rig JSON saying audit is off is a + # claim about how the relay was started, and with --skip-relay nobody + # verified it. A moving audit series here means the control arm was + # auditing after all, which would make the whole comparison meaningless. + cell["audit_activity_in_control_arm"] = int( + (after["audit_count"] - before["audit_count"]) + + (after["audit_log_errors"] - before["audit_log_errors"]) + ) + cell["audit_note"] = "audit disabled: no worker, so no completion series" + return cell + + +def sweep( + rig: dict, + rates: list[float], + duration: int, + repeats: int, + two_community: bool, + audit_on: bool, + log: Callable[[str], None], +) -> list: + cells = [] + for rate in rates: + offers = [(0, rate / 2.0), (1, rate / 2.0)] if two_community else [(0, rate)] + for repeat in range(repeats): + cell = run_cell(rig, duration, offers, audit_on) + cells.append(cell) + log( + " {:>6.0f}/s r{} accepted {:.4f} completed {} outstanding {}" + " busy {} svc_p50 {}".format( + rate, + repeat + 1, + cell["accepted_over_offered"], + _fmt(cell["audit_completed_per_s"], "{:.1f}/s"), + _fmt(cell["outstanding_delta"], "{}"), + _fmt(cell["audit_busy_fraction"], "{:.3f}"), + _fmt(cell["service_ms"]["p50"], "{:.2f}ms"), + ) + ) + return cells + + +def _fmt(value, spec: str) -> str: + """Format a value that is legitimately absent. + + Percentiles are null when every connection died before its first settled + send, and the audit series is null on the audit-off arm. Formatting those + directly aborts the sweep mid-run with no report written. + """ + return "n/a" if value is None else spec.format(value) + + +def experiment_identity(rig: dict, args: argparse.Namespace) -> dict: + """What two half-runs must agree on before they may be combined.""" + return { + "rates": args.rates, + "duration_secs": args.duration, + "repeats": args.repeats, + "targets": [t["community_host"] for t in rig["targets"]], + "ws_events_per_sec_limit": rig["ws_events_per_sec_limit"], + "messages_per_min_limit": rig["messages_per_min_limit"], + "generator": rig["generator"], + "source_revision": rig.get("source_revision"), + # Two dirty trees at the same commit are two different builds. + "source_diff_digest": rig.get("source_diff_digest"), + # What actually ran, as opposed to what the tree said: the diff digest is + # taken after the build and misses untracked inputs. + "binary_digest": rig.get("binary_digest"), + # A fixed audit-on-then-audit-off order against a database that grew in + # between confounds arm with time, cache and index size. Restoring the + # same snapshot at both arm boundaries makes the arms comparable; the + # identity records it so a pair where only one arm was reset is rejected. + "database_reset": rig.get("database_reset"), + } + + +def measure(args: argparse.Namespace, log: Callable[[str], None]) -> dict: + with open(args.rig) as handle: + rig = json.load(handle) + audit_on = bool(rig["audit_enabled"]) + + log("Sweep, audit {}, one community".format("enabled" if audit_on else "disabled")) + cells = sweep(rig, args.rates, args.duration, args.repeats, False, audit_on, log) + + two_cells = [] + if audit_on and not args.skip_two_community: + log("Sweep, audit enabled, two communities at half rate each") + two_cells = sweep(rig, args.rates, args.duration, args.repeats, True, True, log) + # Report-only by agreed scope, but annotated: an unannotated contaminated + # cell in a table nobody judges is how a bad number gets quoted later. + for cell in two_cells: + cell["problems"] = cell_problems(cell) + cell["steady"] = steady_state(cell) + + # A single-arm dataset is never a verdict; --combine judges the pair. + if audit_on: + result = verdict(cells, None, control_ran=False) + else: + result = { + "ok": False, + "control": {"ran": False, "arm": "this dataset is itself the audit-off arm"}, + "failures": [ + "audit-off half-run: combine it with the audit-on half to get a verdict" + ], + } + return { + "identity": experiment_identity(rig, args), + "audit_enabled": audit_on, + "partial": True, + "cells": cells, + "two_community_cells": two_cells, + "verdict": result, + } + + +def combine(first: dict, second: dict) -> dict: + """Judge one audit-on and one audit-off half-run together.""" + if first["audit_enabled"] == second["audit_enabled"]: + raise ValueError( + "combine needs one audit-on and one audit-off report; both say " + "audit_enabled={}".format(first["audit_enabled"]) + ) + on_report, off_report = ( + (first, second) if first["audit_enabled"] else (second, first) + ) + + mismatched = sorted( + key for key in on_report["identity"] + if on_report["identity"][key] != off_report["identity"].get(key) + ) + if mismatched: + raise ValueError( + "the two halves are not the same experiment; differing: " + + ", ".join(mismatched) + ) + + # An absent validity field skips its gate, so a cell missing them reads as + # valid. Require the schema at the boundary rather than trusting the producer. + required = ( + "offered_per_s", + "audit_enabled", + "duration_secs", + "accepted_over_offered", + "rejected", + "transport_errors", + "quota_rejections_delta", + "unavailable_rejections_delta", + "audit_log_errors_delta", + "audit_send_errors_delta", + "counters_window_aligned", + "setup_overhead_fraction", + "generator_lag_p99_ms", + "attempted_over_offered", + ) + for report, arm in ((on_report, True), (off_report, False)): + for cell in report["cells"]: + missing = [key for key in required if key not in cell] + if missing: + raise ValueError( + "a {}/s cell is missing {}; an absent validity field skips " + "its gate, so an incomplete cell would read as valid".format( + cell.get("offered_per_s", "?"), ", ".join(missing) + ) + ) + identity = report["identity"] + expected = sorted(float(r) for r in identity["rates"]) + seen: dict = {} + for cell in report["cells"]: + if cell["audit_enabled"] != arm: + raise ValueError( + "a cell labelled audit_enabled={} appears in the audit_enabled={} " + "report".format(cell["audit_enabled"], arm) + ) + if cell["duration_secs"] != identity["duration_secs"]: + raise ValueError( + "a {:g}/s cell ran for {}s but the identity declares {}s".format( + cell["offered_per_s"], + cell["duration_secs"], + identity["duration_secs"], + ) + ) + seen[cell["offered_per_s"]] = seen.get(cell["offered_per_s"], 0) + 1 + if sorted(seen) != expected: + raise ValueError( + "the cells do not cover the declared rate grid: declared {}, " + "present {}".format(expected, sorted(seen)) + ) + wrong = {r: n for r, n in seen.items() if n != identity["repeats"]} + if wrong: + raise ValueError( + "the declared {} repeats are not present at every rate: {}".format( + identity["repeats"], wrong + ) + ) + + # Equality is not truth: two halves that both skipped the reset agree, and + # that is exactly the fixed-order-against-a-growing-database confound the + # field exists to prevent. Same shape as a blank secret passing a + # decrypt-only check. + for report, name in ((on_report, "audit-on"), (off_report, "audit-off")): + if not report["identity"].get("database_reset"): + raise ValueError( + "the {} half did not restore the database snapshot; a fixed arm " + "order against a database that grew in between confounds the arm " + "with time, cache state and index size".format(name) + ) + + threshold = 0.99 + return { + "mode": "combine", + "identity": on_report["identity"], + "verdict": verdict(on_report["cells"], off_report["cells"], control_ran=True), + "knee_reporting_only": { + "threshold": threshold, + "audit_on": knee_bracket( + [(c["offered_per_s"], c["accepted_over_offered"]) + for c in on_report["cells"]], + threshold, + ), + "audit_off": knee_bracket( + [(c["offered_per_s"], c["accepted_over_offered"]) + for c in off_report["cells"]], + threshold, + ), + "note": "presentation only; nothing gates on the knee", + }, + } + + +def model( + ceiling_per_s: float = MEASURED_DRAIN_BAND_PER_S, + rates: list[float] | None = None, +) -> dict: + """Deterministic queueing arithmetic, no services. + + Documents the contract's shape, and deliberately reproduces the physics the + exclusion rule exists for: the audit channel starts empty, so a saturating + rate banks acceptance credit on its first repeats and only reaches steady + state once the channel is full. An earlier version of this model set + `outstanding_delta` to zero at every rate, including one above its own + ceiling — which made the green path a test of steady cells rather than of the + behaviour the harness meets in the field. + + The default ceiling is the drain rate measured on this rig, *not* a round + number below the grid. An earlier version fixed it at 333/s while the grid + topped out at 400/s, so the model ran at 120% utilization and separated + cleanly while the rig would have sat at 81-102% and separated by coin flip. + A fixture that cannot produce the failing input turns green into a statement + about the fixture. `ceiling_per_s` is a parameter so a test can push it above + the top grid rate and exercise the no-separation path. + + For review and for the unit tests, never as evidence. + """ + ceiling = ceiling_per_s + duration = 20.0 + repeats = 5 + rates = list(DEFAULT_RATES if rates is None else rates) + + def jitter(repeat: int) -> float: + return (repeat - (repeats - 1) / 2.0) * 0.002 + + def audit_on_cells() -> list: + cells = [] + fill = 0.0 + for rate in rates: + offered = rate * duration + drain_capacity = ceiling * duration + for repeat in range(repeats): + headroom = AUDIT_CHANNEL_DEPTH - fill + accepted = min(offered, drain_capacity + headroom) + drained = min(drain_capacity, fill + accepted) + delta = accepted - drained + fill += delta + cells.append( + { + "offered_per_s": rate, + "audit_enabled": True, + "accepted_over_offered": min( + 1.0, accepted / offered * (1.0 + jitter(repeat)) + ), + "accepted_per_s": accepted / duration, + "rejected": 0, + "transport_errors": 0, + "quota_rejections_delta": 0, + "unavailable_rejections_delta": 0, + "audit_log_errors_delta": 0, + "audit_send_errors_delta": 0, + # Derived, not fixtured: a closed-loop generator's + # apparent capacity collapses onto the relay's throughput + # at saturation, and hard-coding these is how two gates + # built on them stayed invisible to the suite. + "attempted_over_offered": accepted / offered, + "generator_headroom": (drained / duration) / rate, + "generator_lag_p99_ms": 0.2, + "audit_completed_per_s": drained / duration, + "audit_service_mean_ms": 1000.0 / ceiling, + "audit_busy_fraction": min(1.0, drained / drain_capacity), + "outstanding_delta": int(round(delta)), + } + ) + return cells + + def audit_off_cells() -> list: + return [ + { + "offered_per_s": rate, + "audit_enabled": False, + "accepted_over_offered": min(1.0, 1.0 * (1.0 + jitter(repeat))), + "accepted_per_s": rate, + "rejected": 0, + "transport_errors": 0, + "quota_rejections_delta": 0, + "unavailable_rejections_delta": 0, + "audit_log_errors_delta": 0, + "audit_send_errors_delta": 0, + "attempted_over_offered": 1.0, + "generator_headroom": 1.0, + "generator_lag_p99_ms": 0.2, + "audit_completed_per_s": None, + "audit_service_mean_ms": None, + "audit_busy_fraction": None, + "outstanding_delta": None, + } + for rate in rates + for repeat in range(repeats) + ] + + return { + "mode": "model", + "verdict": verdict(audit_on_cells(), audit_off_cells(), control_ran=True), + } + + +def main(argv: list[str] | None = None) -> int: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--mode", choices=("measure", "model"), default="measure") + parser.add_argument( + "--combine", + nargs=2, + metavar="REPORT", + help="judge one audit-on and one audit-off report written by --json", + ) + parser.add_argument("--rig", help="rig JSON from scripts/start-perf-ingest-rig.sh") + parser.add_argument("--duration", type=int, default=20) + parser.add_argument("--repeats", type=int, default=DEFAULT_REPEATS) + parser.add_argument("--rates", default=",".join(str(r) for r in DEFAULT_RATES)) + parser.add_argument("--skip-two-community", action="store_true") + parser.add_argument("--json", help="write the full report here") + args = parser.parse_args(argv) + args.rates = [float(r) for r in args.rates.split(",")] + + def log(message: str) -> None: + print(message, file=sys.stderr) + + if args.combine: + with open(args.combine[0]) as a, open(args.combine[1]) as b: + report = combine(json.load(a), json.load(b)) + elif args.mode == "model": + report = model() + else: + if not args.rig: + parser.error("--mode measure needs --rig") + if args.repeats < 2: + parser.error("--repeats must be at least 2; an interval needs a spread") + report = measure(args, log) + + if args.json: + with open(args.json, "w") as handle: + json.dump(report, handle, indent=2) + print(json.dumps(report["verdict"], indent=2)) + + if not report["verdict"]["ok"]: + for failure in report["verdict"]["failures"]: + print("NOT ESTABLISHED: " + failure, file=sys.stderr) + return 1 + return 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/perf/test_relay_ingest_ceiling.py b/perf/test_relay_ingest_ceiling.py new file mode 100644 index 00000000000..c9b2a75b79a --- /dev/null +++ b/perf/test_relay_ingest_ceiling.py @@ -0,0 +1,779 @@ +#!/usr/bin/env python3 +"""Unit tests for the ingest-ceiling harness verdict logic. + +Every test that asserts a *pass* is paired with a mutant that must fail. A +harness whose contract cannot go red is decoration. +""" + +from __future__ import annotations + +import unittest + +import relay_ingest_ceiling as harness + + +def cell(**overrides) -> dict: + """A valid, steady, busy audit-on cell. Mutants override one field.""" + base = { + "offered_per_s": 800.0, + "audit_enabled": True, + "accepted_over_offered": 0.60, + "accepted_per_s": 480.0, + "rejected": 0, + "transport_errors": 0, + "quota_rejections_delta": 0, + "unavailable_rejections_delta": 0, + "audit_log_errors_delta": 0, + "audit_send_errors_delta": 0, + "duration_secs": 20, + "counters_window_aligned": True, + "setup_overhead_fraction": 0.01, + "generator_headroom": 0.56, + "generator_lag_p99_ms": 0.3, + "attempted_over_offered": 0.56, + "audit_completed_per_s": 450.0, + "audit_service_mean_ms": 2.2, + "audit_busy_fraction": 0.99, + "outstanding_delta": 0, + } + base.update(overrides) + return base + + +def arm(fractions: list[float], audit: bool = True, rate: float = 800.0) -> list: + return [ + cell( + offered_per_s=rate, + accepted_over_offered=f, + audit_enabled=audit, + audit_completed_per_s=450.0 if audit else None, + audit_service_mean_ms=2.2 if audit else None, + audit_busy_fraction=0.99 if audit else None, + outstanding_delta=0 if audit else None, + ) + for f in fractions + ] + + +class StatisticsTests(unittest.TestCase): + def test_sample_stddev_is_n_minus_one(self) -> None: + self.assertAlmostEqual(harness.sample_stddev([2.0, 4.0, 4.0, 4.0, 5.0]), 1.0954451, 6) + + def test_stddev_needs_two_observations(self) -> None: + with self.assertRaises(ValueError): + harness.sample_stddev([1.0]) + + def test_stddev_does_not_grow_with_n_the_way_range_does(self) -> None: + # MUTANT guard for the estimator that was replaced: (max-min)/mean rises + # with sample size on its own, so the same population read at n=2 and + # n=6 would report different spreads. Stddev does not. + small = [10.0, 12.0] + large = [10.0, 10.4, 10.8, 11.2, 11.6, 12.0] + range_small = (max(small) - min(small)) / harness.mean(small) + range_large = (max(large) - min(large)) / harness.mean(large) + self.assertAlmostEqual(range_small, range_large, 6) # range is blind here + self.assertLess(harness.sample_stddev(large), harness.sample_stddev(small)) + + def test_interval_is_none_bounded_for_a_single_observation(self) -> None: + self.assertEqual(harness.confidence_interval([5.0])["lo"], None) + + def test_difference_interval_excludes_zero_when_arms_differ(self) -> None: + d = harness.difference_interval([1.0, 1.0, 0.999, 1.0], [0.6, 0.61, 0.59, 0.6]) + self.assertTrue(d["excludes_zero"]) + self.assertGreater(d["diff"], 0.0) + + def test_difference_interval_spans_zero_when_arms_agree(self) -> None: + # MUTANT: same population twice. A predicate built on "the intervals do + # not overlap" would be weaker; one built on "they overlap, so the arms + # are equal" would be invalid. Neither is what this returns. + d = harness.difference_interval([0.60, 0.62, 0.58, 0.61], [0.61, 0.59, 0.60, 0.62]) + self.assertFalse(d["excludes_zero"]) + + +class CellValidityTests(unittest.TestCase): + def test_a_clean_cell_has_no_problems(self) -> None: + self.assertEqual(harness.cell_problems(cell()), []) + + def test_quota_rejections_invalidate(self) -> None: + self.assertTrue( + any("limiter" in p for p in harness.cell_problems(cell(quota_rejections_delta=2))) + ) + + def test_unavailable_rejections_invalidate(self) -> None: + # MUTANT for the hole the first version had: `unavailable` takes the same + # NOTICE-without-OK path as quota and is load-correlated, so it can forge + # a knee that survives repeats. + problems = harness.cell_problems(cell(unavailable_rejections_delta=1)) + self.assertTrue(any("load-correlated" in p for p in problems)) + + def test_audit_send_failure_invalidates(self) -> None: + problems = harness.cell_problems(cell(audit_send_errors_delta=1)) + self.assertTrue(any("worker is gone" in p for p in problems)) + + def test_relay_rejections_and_transport_errors_invalidate(self) -> None: + self.assertTrue(harness.cell_problems(cell(rejected=3))) + self.assertTrue(harness.cell_problems(cell(transport_errors=1))) + + def test_a_starved_generator_disqualifies_its_cell(self) -> None: + # MUTANT for the replacement gate. + problems = harness.cell_problems(cell(generator_lag_p99_ms=45.0)) + self.assertTrue(any("free to send" in p for p in problems), problems) + + def test_a_healthy_generator_at_a_saturated_rate_passes(self) -> None: + # A saturated cell has a poor apparent capacity and a poor attempted + # fraction, and is still good evidence. + self.assertEqual( + harness.cell_problems( + cell( + generator_lag_p99_ms=0.4, + generator_headroom=0.56, + attempted_over_offered=0.56, + ) + ), + [], + ) + + def test_headroom_is_reported_not_gated(self) -> None: + # Issuability is established by the control arm holding its offer, not by + # a per-cell margin the saturated arm can never clear. + self.assertEqual(harness.cell_problems(cell(generator_headroom=1.1)), []) + self.assertEqual(harness.cell_problems(cell(generator_headroom=2.0)), []) + + +class SteadyStateTests(unittest.TestCase): + def test_level_outstanding_work_is_steady(self) -> None: + self.assertTrue(harness.steady_state(cell(outstanding_delta=0))) + + def test_a_full_queue_start_and_end_is_steady(self) -> None: + # The saturated regime settles with the channel full. A gate written as + # "the queue must be empty at the start" would delete every cell that + # matters for a ceiling; this one accepts any stable level. + self.assertTrue(harness.steady_state(cell(outstanding_delta=10))) + + def test_banking_the_whole_channel_is_not_steady(self) -> None: + # MUTANT: +1000 is exactly the measured empty-start acceptance credit. + self.assertFalse(harness.steady_state(cell(outstanding_delta=1000))) + + def test_audit_off_cells_have_no_steady_state_reading(self) -> None: + self.assertIsNone(harness.steady_state(cell(outstanding_delta=None))) + + def test_an_audit_off_cell_is_still_evidence(self) -> None: + # No queue means no steady-state reading, which must not be mistaken for + # failing the check - otherwise the whole control arm drops out. + self.assertTrue(harness.cell_is_evidence(cell(outstanding_delta=None))) + + def test_a_banking_cell_is_not_evidence(self) -> None: + self.assertFalse(harness.cell_is_evidence(cell(outstanding_delta=1000))) + + +class ArmSeparationTests(unittest.TestCase): + def test_separation_is_found_where_the_arms_diverge(self) -> None: + result = harness.arm_separation( + arm([0.60, 0.62, 0.58, 0.61]), arm([1.0, 1.0, 0.999, 1.0], audit=False) + ) + self.assertTrue(result["separated"]) + + def test_identical_arms_are_not_separated(self) -> None: + # MUTANT: the control changed nothing, so the audit path is not shown to + # limit ingest and the run must not pass. + result = harness.arm_separation( + arm([0.60, 0.62, 0.58, 0.61]), arm([0.61, 0.59, 0.60, 0.62], audit=False) + ) + self.assertFalse(result["separated"]) + + def test_separation_in_the_wrong_direction_does_not_count(self) -> None: + # MUTANT: audit-off *worse* than audit-on refutes the hypothesis; it must + # not satisfy a predicate that only looks at "the interval excludes zero". + result = harness.arm_separation( + arm([1.0, 1.0, 0.999, 1.0]), arm([0.60, 0.62, 0.58, 0.61], audit=False) + ) + self.assertFalse(result["separated"]) + + +class WorkerRateTests(unittest.TestCase): + def test_busy_steady_cells_give_an_estimate(self) -> None: + result = harness.worker_rate([cell(), cell(audit_completed_per_s=460.0)]) + self.assertEqual(result["cells"], 2) + self.assertAlmostEqual(result["estimate"]["mean"], 455.0) + + def test_an_idle_worker_cannot_report_capacity(self) -> None: + # MUTANT: below saturation the completion rate tracks the offer, so it is + # not the worker's limit. + result = harness.worker_rate([cell(audit_busy_fraction=0.40)]) + self.assertEqual(result["cells"], 0) + self.assertIsNone(result["estimate"]) + + def test_a_cell_banking_the_channel_cannot_report_capacity(self) -> None: + result = harness.worker_rate([cell(outstanding_delta=1000)]) + self.assertEqual(result["cells"], 0) + + def test_an_invalid_cell_cannot_report_capacity(self) -> None: + result = harness.worker_rate([cell(unavailable_rejections_delta=1)]) + self.assertEqual(result["cells"], 0) + + +class VerdictTests(unittest.TestCase): + def passing(self) -> dict: + return harness.verdict( + arm([0.60, 0.62, 0.58, 0.61]), + arm([1.0, 1.0, 0.999, 1.0], audit=False), + control_ran=True, + ) + + def test_separated_arms_with_clean_cells_pass(self) -> None: + result = self.passing() + self.assertTrue(result["ok"], result["failures"]) + self.assertTrue(result["control"]["ran"]) + + def test_a_missing_control_cannot_pass(self) -> None: + # MUTANT: this is the --skip-audit-off shape. Attribution was never + # tested, so no verdict is available at any cell quality. + result = harness.verdict(arm([0.60, 0.62, 0.58, 0.61]), None, control_ran=False) + self.assertFalse(result["ok"]) + self.assertFalse(result["control"]["ran"]) + self.assertTrue(any("partial experiment" in f for f in result["failures"])) + + def test_the_report_says_whether_the_control_ran(self) -> None: + # "no knee on the audit-off arm" used to read identically for "the + # control ran and the knee was gone" and "the control never ran". + self.assertTrue(self.passing()["control"]["ran"]) + self.assertFalse( + harness.verdict(arm([0.6, 0.61]), None, control_ran=False)["control"]["ran"] + ) + + def test_an_unsteady_cell_is_excluded_not_fatal(self) -> None: + # The first repeat of the first saturating rate starts with the audit + # channel empty and banks its whole depth, so failing the run on a + # non-steady cell would fail every genuine saturating dataset. It is + # dropped from the evidence and reported instead. + on = arm([0.60, 0.62, 0.58, 0.61]) + on[0]["outstanding_delta"] = 1000 + on[0]["accepted_over_offered"] = 0.73 # the credit inflates this + result = harness.verdict(on, arm([1.0, 1.0, 0.999, 1.0], audit=False), True) + self.assertTrue(result["ok"], result["failures"]) + self.assertEqual(len(result["excluded_cells"]), 1) + self.assertTrue( + any("acceptance credit" in r for r in result["excluded_cells"][0]["reasons"]) + ) + + def test_an_excluded_cell_is_kept_out_of_the_interval(self) -> None: + # Not only must it not fail the run, its inflated accepted/offered must + # not widen or shift the arm's interval. + on = arm([0.60, 0.62, 0.58, 0.61]) + on[0]["outstanding_delta"] = 1000 + on[0]["accepted_over_offered"] = 0.73 + result = harness.verdict(on, arm([1.0, 1.0, 0.999, 1.0], audit=False), True) + rate = result["arm_separation"]["by_rate"][0] + self.assertEqual(rate["evidence_cells"]["audit_on"], 3) + self.assertEqual(rate["dropped_cells"], 1) + self.assertLess(rate["audit_on"]["mean"], 0.63) + + def test_a_dead_audit_worker_is_fatal_not_merely_excluded(self) -> None: + # An enqueue failure means the receiver is gone, so every later cell is + # suspect - not just the one that noticed. + on = arm([0.60, 0.62, 0.58, 0.61]) + on[1]["audit_send_errors_delta"] = 1 + result = harness.verdict(on, arm([1.0, 1.0, 0.999, 1.0], audit=False), True) + self.assertFalse(result["ok"]) + self.assertTrue(any("worker is gone" in f for f in result["failures"])) + + def test_too_few_surviving_cells_fails_the_run(self) -> None: + # MUTANT: exclusion must not become a way to pass by discarding almost + # everything. One evidence cell per arm cannot support an interval. + on = arm([0.60, 0.62]) + on[0]["outstanding_delta"] = 1000 + off = arm([1.0, 1.0], audit=False) + result = harness.verdict(on, off, control_ran=True) + self.assertFalse(result["ok"]) + self.assertTrue(any("never compared" in f for f in result["failures"])) + + def test_lost_evidence_reports_inconclusive_not_negative(self) -> None: + # Two rates: one comparable and unseparated, one that lost its evidence. + # Reporting "not shown to limit ingest" here would state a result about + # the relay when the truth is that the informative cells were dropped. + on = arm([0.60, 0.62, 0.61], rate=100.0) + arm([0.60, 0.62], rate=800.0) + on[-1]["outstanding_delta"] = 1000 + off = arm([0.61, 0.60, 0.62], audit=False, rate=100.0) + arm( + [1.0, 1.0], audit=False, rate=800.0 + ) + result = harness.verdict(on, off, control_ran=True) + self.assertFalse(result["ok"]) + self.assertTrue( + any("inconclusive rather than negative" in f for f in result["failures"]), + result["failures"], + ) + + def test_a_fatal_problem_is_not_also_listed_as_an_exclusion(self) -> None: + on = arm([0.60, 0.62, 0.58, 0.61]) + on[0]["audit_send_errors_delta"] = 1 + result = harness.verdict(on, arm([1.0, 1.0, 0.999, 1.0], audit=False), True) + self.assertTrue(any("worker is gone" in f for f in result["failures"])) + self.assertFalse( + any( + "worker is gone" in r + for c in result["excluded_cells"] + for r in c["reasons"] + ) + ) + + def test_no_separation_fails_the_run(self) -> None: + result = harness.verdict( + arm([0.60, 0.62, 0.58, 0.61]), + arm([0.61, 0.59, 0.60, 0.62], audit=False), + control_ran=True, + ) + self.assertFalse(result["ok"]) + self.assertTrue(any("not shown to limit ingest" in f for f in result["failures"])) + + def test_every_failure_is_reported_not_just_the_first(self) -> None: + on = arm([0.60, 0.62, 0.58, 0.61]) + on[0]["audit_send_errors_delta"] = 1 + result = harness.verdict(on, None, control_ran=False) + self.assertGreaterEqual(len(result["failures"]), 2) + self.assertTrue(any("worker is gone" in f for f in result["failures"])) + self.assertTrue(any("partial experiment" in f for f in result["failures"])) + + def test_exclusions_and_failures_are_reported_separately(self) -> None: + # A contaminated cell and a banking cell are both excluded with reasons; + # neither is silently dropped, and neither is confused with a failure. + on = arm([0.60, 0.62, 0.58, 0.61]) + on[0]["quota_rejections_delta"] = 1 + on[1]["outstanding_delta"] = 1000 + result = harness.verdict(on, arm([1.0, 1.0, 0.999, 1.0], audit=False), True) + self.assertEqual(len(result["excluded_cells"]), 2) + self.assertEqual(result["arm_separation"]["by_rate"][0]["dropped_cells"], 2) + + def test_the_lock_ceiling_is_never_reported_as_absent(self) -> None: + # The worker ceiling is lower and masks the lock, so a passing run says + # nothing about the lock. This wording guards against a later reader + # quoting the run as evidence the lock is fine. + self.assertIn("structurally blind", self.passing()["lock_ceiling"]) + + +class CausalValidityTests(unittest.TestCase): + """The gates that stop a precise answer to the wrong experiment.""" + + def test_both_arms_collapsing_is_not_a_positive_control(self) -> None: + # MUTANT, and the shipped bug: audit-on ~0.40 against audit-off ~0.50 + # separates cleanly and means nothing. Removing the audit path has to + # restore the offer, not merely do better than keeping it. + on = arm([0.40, 0.41, 0.39, 0.40]) + off = arm([0.50, 0.51, 0.49, 0.50], audit=False) + result = harness.verdict(on, off, control_ran=True) + self.assertFalse(result["ok"]) + self.assertTrue( + any("did not hold its offer" in f for f in result["failures"]), + result["failures"], + ) + + def test_a_control_that_holds_its_offer_passes(self) -> None: + result = harness.verdict( + arm([0.60, 0.62, 0.58, 0.61]), + arm([1.0, 0.999, 1.0, 0.998], audit=False), + control_ran=True, + ) + self.assertTrue(result["ok"], result["failures"]) + self.assertTrue( + result["arm_separation"]["primary_control_holds_offer"] + ) + + def test_only_secondary_rates_separating_does_not_pass(self) -> None: + # MUTANT for the familywise hole: passing on any-of-N rates gives a ~10% + # false-pass rate at five rates, measured on this module. One predeclared + # primary contrast is the fix, so a separation at a lower rate while the + # primary does not separate must fail. + on = arm([0.60, 0.62, 0.58, 0.61], rate=800.0) + arm( + [1.0, 0.999, 1.0, 0.998], rate=1600.0 + ) + off = arm([1.0, 0.999, 1.0, 0.998], audit=False, rate=800.0) + arm( + [1.0, 0.999, 1.0, 0.998], audit=False, rate=1600.0 + ) + result = harness.verdict(on, off, control_ran=True) + self.assertFalse(result["ok"]) + self.assertTrue( + any("only secondary rates separated" in f for f in result["failures"]), + result["failures"], + ) + + def test_a_reverse_separation_contradicts_rather_than_fails_to_support(self) -> None: + on = arm([1.0, 0.999, 1.0, 0.998], rate=800.0) + off = arm([0.60, 0.62, 0.58, 0.61], audit=False, rate=800.0) + result = harness.verdict(on, off, control_ran=True) + self.assertFalse(result["ok"]) + self.assertTrue( + any("contradicts the hypothesis" in f for f in result["failures"]), + result["failures"], + ) + + def test_an_auditing_control_arm_is_not_a_control(self) -> None: + # MUTANT: the rig JSON claiming audit is off is a claim about how the + # relay was started, and under --skip-relay nobody verified it. + off = arm([1.0, 0.999, 1.0, 0.998], audit=False) + off[0]["audit_activity_in_control_arm"] = 4105 + result = harness.verdict(arm([0.60, 0.62, 0.58, 0.61]), off, True) + self.assertTrue( + any( + "still auditing" in r + for c in result["excluded_cells"] + for r in c["reasons"] + ), + result["excluded_cells"], + ) + + def test_unaligned_counter_windows_disqualify_a_cell(self) -> None: + # MUTANT: counters bracketing the whole subprocess while rates divide by + # the post-connect window can hide banking and push busy above 1.0. + problems = harness.cell_problems(cell(counters_window_aligned=False)) + self.assertTrue(any("timed window" in p for p in problems)) + + def test_aligned_counter_windows_pass(self) -> None: + self.assertEqual(harness.cell_problems(cell(counters_window_aligned=True)), []) + + def test_heavy_setup_overhead_disqualifies_a_cell(self) -> None: + problems = harness.cell_problems(cell(setup_overhead_fraction=0.20)) + self.assertTrue(any("setup and teardown" in p for p in problems)) + + def test_closed_loop_shortfall_does_not_disqualify_a_saturated_cell(self) -> None: + # MUTANT for a gate that was added and then had to be removed: at a + # saturated rate a closed-loop sender can reach neither its scheduled + # slots nor a capacity margin, so gating on either rejected every cell + # that matters. + problems = harness.cell_problems( + cell(attempted_over_offered=0.57, generator_headroom=0.59) + ) + self.assertEqual(problems, []) + + +class GeneratorEnvironmentTests(unittest.TestCase): + RIG = { + "bench_private_key": "ab" * 32, + "metrics_url": "http://localhost:9202/metrics", + } + + def test_the_metrics_url_reaches_the_generator(self) -> None: + # Its absence excludes the whole run, and nothing else in the suite can + # see that: these tests build cells directly and the model bypasses + # run_cell. See `generator_env`. + env = harness.generator_env(self.RIG) + self.assertEqual(env["BENCH_METRICS_URL"], self.RIG["metrics_url"]) + self.assertEqual(env["BENCH_PRIVATE_KEY"], self.RIG["bench_private_key"]) + + def test_inherited_cli_credentials_are_scrubbed(self) -> None: + # A stale BUZZ_AUTH_TAG fails the dev relay's first write outright. + import os as _os + + for name in ("BUZZ_AUTH_TAG", "BUZZ_RELAY_URL", "BUZZ_PRIVATE_KEY"): + _os.environ[name] = "stale" + try: + env = harness.generator_env(self.RIG) + for name in ("BUZZ_AUTH_TAG", "BUZZ_RELAY_URL", "BUZZ_PRIVATE_KEY"): + self.assertNotIn(name, env) + finally: + for name in ("BUZZ_AUTH_TAG", "BUZZ_RELAY_URL", "BUZZ_PRIVATE_KEY"): + _os.environ.pop(name, None) + + +class WorkerRateRegimeTests(unittest.TestCase): + def test_estimates_are_reported_per_rate_not_pooled(self) -> None: + # Different offered rates are different load and database regimes, not + # repeats of one estimand. + cells = [ + cell(offered_per_s=800.0, audit_completed_per_s=450.0), + cell(offered_per_s=800.0, audit_completed_per_s=460.0), + cell(offered_per_s=1600.0, audit_completed_per_s=400.0), + cell(offered_per_s=1600.0, audit_completed_per_s=410.0), + ] + per_rate = harness.worker_rate(cells)["per_rate"] + self.assertEqual([r["offered_per_s"] for r in per_rate], [800.0, 1600.0]) + self.assertAlmostEqual(per_rate[0]["completed_per_s"]["mean"], 455.0) + self.assertAlmostEqual(per_rate[1]["completed_per_s"]["mean"], 405.0) + + +class TTableTests(unittest.TestCase): + def test_rounding_errs_wide_not_narrow(self) -> None: + # The sparse table must never return a smaller critical value than the + # true one: df 11 is 2.201 and df 13 is 2.160, and picking the next higher + # stored df would under-cover at exactly the n>=16 cells the plan specs. + self.assertGreaterEqual(harness.t95(11), 2.201) + self.assertGreaterEqual(harness.t95(13), 2.160) + self.assertGreaterEqual(harness.t95(14), 2.145) + + def test_exact_entries_are_returned_unchanged(self) -> None: + self.assertEqual(harness.t95(4), 2.776) + self.assertEqual(harness.t95(10), 2.228) + + +class CombineTests(unittest.TestCase): + def half(self, audit: bool, **identity) -> dict: + base = { + "rates": [800.0], + "duration_secs": 20, + "repeats": 4, + "targets": ["a.localhost:3030", "b.localhost:3030"], + "ws_events_per_sec_limit": 100000, + "messages_per_min_limit": 6000000, + "generator": "./target/ci/ingest_load", + "source_revision": "deadbeef", + "source_diff_digest": "clean", + "binary_digest": "cafe", + "database_reset": True, + } + base.update(identity) + fractions = [0.60, 0.62, 0.58, 0.61] if audit else [1.0, 0.999, 1.0, 0.998] + cells = [] + for rate in base["rates"]: + for f in fractions[: base["repeats"]]: + cells.extend(arm([f], audit=audit, rate=rate)) + return {"audit_enabled": audit, "identity": base, "cells": cells} + + def test_matched_halves_combine_and_pass(self) -> None: + report = harness.combine(self.half(True), self.half(False)) + self.assertTrue(report["verdict"]["ok"], report["verdict"]["failures"]) + + def test_order_of_the_halves_does_not_matter(self) -> None: + self.assertTrue(harness.combine(self.half(False), self.half(True))["verdict"]["ok"]) + + def test_two_halves_from_the_same_arm_are_rejected(self) -> None: + with self.assertRaises(ValueError): + harness.combine(self.half(True), self.half(True)) + + def test_a_duration_mismatch_is_rejected(self) -> None: + # MUTANT: the shipped bug. A 10s three-rate on-report combined happily + # with a 60s two-rate off-report and returned ok. + with self.assertRaises(ValueError) as ctx: + harness.combine(self.half(True), self.half(False, duration_secs=60)) + self.assertIn("duration_secs", str(ctx.exception)) + + def test_a_rate_grid_mismatch_is_rejected(self) -> None: + with self.assertRaises(ValueError) as ctx: + harness.combine(self.half(True), self.half(False, rates=[100.0, 200.0])) + self.assertIn("rates", str(ctx.exception)) + + def test_a_limiter_or_revision_mismatch_is_rejected(self) -> None: + for field, value in ( + ("ws_events_per_sec_limit", 10), + ("source_revision", "cafe1234"), + ("targets", ["a.localhost:3030"]), + ): + with self.assertRaises(ValueError): + harness.combine(self.half(True), self.half(False, **{field: value})) + + def test_cells_must_cover_the_declared_rate_grid(self) -> None: + # MUTANT, probe-confirmed by the reviewers: an identity declaring + # [800, 1600] with only 800-cells supplied used to return ok. + broken = self.half(True, rates=[800.0, 1600.0]) + broken["cells"] = [c for c in broken["cells"] if c["offered_per_s"] == 800.0] + with self.assertRaises(ValueError) as ctx: + harness.combine(broken, self.half(False, rates=[800.0, 1600.0])) + self.assertIn("rate grid", str(ctx.exception)) + + def test_cells_must_carry_the_declared_repeat_count(self) -> None: + broken = self.half(True) + broken["cells"] = broken["cells"][:-1] + with self.assertRaises(ValueError) as ctx: + harness.combine(broken, self.half(False)) + self.assertIn("repeats", str(ctx.exception)) + + def test_a_cell_labelled_for_the_other_arm_is_rejected(self) -> None: + # MUTANT: audit-off-labelled cells inside the audit-on report used to pass. + broken = self.half(True) + broken["cells"] = arm([0.6, 0.62, 0.58, 0.61], audit=False) + with self.assertRaises(ValueError) as ctx: + harness.combine(broken, self.half(False)) + self.assertIn("audit_enabled", str(ctx.exception)) + + def test_a_dirty_tree_digest_mismatch_is_rejected(self) -> None: + with self.assertRaises(ValueError): + harness.combine( + self.half(True), self.half(False, source_diff_digest="beef") + ) + + def test_a_half_that_skipped_the_reset_is_rejected(self) -> None: + # MUTANT, probe-confirmed: two halves that both skipped the reset agree, + # and equality passed them. Equality is not truth. + with self.assertRaises(ValueError) as ctx: + harness.combine( + self.half(True, database_reset=False), + self.half(False, database_reset=False), + ) + self.assertIn("database snapshot", str(ctx.exception)) + + def test_an_incomplete_cell_is_rejected(self) -> None: + # MUTANT: an absent validity field skips its gate, so a cell missing + # `counters_window_aligned` read as valid evidence. + broken = self.half(True) + del broken["cells"][0]["counters_window_aligned"] + with self.assertRaises(ValueError) as ctx: + harness.combine(broken, self.half(False)) + self.assertIn("counters_window_aligned", str(ctx.exception)) + + def test_a_cell_without_a_duration_is_rejected(self) -> None: + broken = self.half(True) + del broken["cells"][0]["duration_secs"] + with self.assertRaises(ValueError): + harness.combine(broken, self.half(False)) + + def test_a_binary_digest_mismatch_is_rejected(self) -> None: + # Two builds at one commit are two experiments; the source digest is + # taken after the build and cannot prove what actually ran. + with self.assertRaises(ValueError): + harness.combine(self.half(True), self.half(False, binary_digest="beef")) + + def test_arms_reset_differently_are_rejected(self) -> None: + # A fixed arm order against a database that grew in between confounds arm + # with time and index size; restoring the same snapshot at both boundaries + # is what makes them comparable, so a pair where only one arm was reset + # is not one experiment. + with self.assertRaises(ValueError) as ctx: + harness.combine(self.half(True), self.half(False, database_reset=False)) + self.assertIn("database_reset", str(ctx.exception)) + + def test_combine_signals_misuse_the_way_its_neighbours_do(self) -> None: + # ValueError, not SystemExit: the guard on the path that produced the + # shipped verdict has to be assertable from a test. + with self.assertRaises(ValueError): + harness.combine(self.half(True), self.half(True)) + + +class ModelTests(unittest.TestCase): + def test_the_documented_model_satisfies_its_own_contract(self) -> None: + report = harness.model() + self.assertTrue(report["verdict"]["ok"], report["verdict"]["failures"]) + + def test_the_model_separates_only_where_it_saturates(self) -> None: + rates = [ + r["offered_per_s"] + for r in harness.model()["verdict"]["arm_separation"]["by_rate"] + if r.get("separated_here") + ] + self.assertEqual(rates, [800.0, 1600.0]) + + def test_the_default_grid_straddles_the_measured_drain_band(self) -> None: + # The regression guard for a grid that cannot fire the contract. A top + # rate at the drain rate makes the only informative cell a coin flip, and + # the run then reports "not shown to limit ingest" — a false negative + # phrased as a conclusion. + band = harness.MEASURED_DRAIN_BAND_PER_S + self.assertLess(min(harness.DEFAULT_RATES), band) + self.assertGreaterEqual(max(harness.DEFAULT_RATES), 2.0 * band) + + def test_the_model_ceiling_is_not_below_its_own_grid_by_construction(self) -> None: + # The earlier model fixed its ceiling at 333/s under a 400/s top rate, so + # it separated at 120% utilization while the rig sat at 81-102%. Green + # then described the fixture rather than the contract. + self.assertGreater( + harness.MEASURED_DRAIN_BAND_PER_S, max(harness.DEFAULT_RATES) / 8.0 + ) + self.assertLess(harness.MEASURED_DRAIN_BAND_PER_S, max(harness.DEFAULT_RATES)) + + def test_a_ceiling_above_the_grid_reports_a_negative_not_a_pass(self) -> None: + # MUTANT: nothing saturates, so there is nothing to separate. The run must + # fail, and must say it compared every rate and found nothing rather than + # implying the evidence was missing. + verdict = harness.model(ceiling_per_s=2.0 * max(harness.DEFAULT_RATES))["verdict"] + self.assertFalse(verdict["ok"]) + # And the diagnosis is the saturation one, not the exoneration one: a + # ceiling above the grid means the grid never reached the audit path, so + # "none separated" would be the wrong thing for a reader to act on. + self.assertTrue( + any("never saturated" in f for f in verdict["failures"]), verdict["failures"] + ) + self.assertFalse(verdict["saturation"]["any_rate_saturated"]) + + def test_a_transition_rate_fills_the_channel_over_several_repeats(self) -> None: + # Partial banking, as distinct from the deep +1000-from-empty shape: an + # offer just above the drain rate accumulates across repeats until the + # channel caps. + verdict = harness.model(ceiling_per_s=450.0, rates=[100.0, 470.0])["verdict"] + deltas = [c["offered_per_s"] for c in verdict["excluded_cells"]] + self.assertTrue(deltas) + self.assertTrue(all(r == 470.0 for r in deltas)) + self.assertGreater(len(deltas), 1) + + def test_the_model_derives_its_generator_metrics(self) -> None: + # Fixturing these is how two gates built on them stayed invisible: an + # absent key skips a gate, and a hard-coded healthy value passes it. The + # model must produce the saturated regime's real shape. + on = harness.model()["verdict"] + saturated = [ + r for r in on["arm_separation"]["by_rate"] if r.get("separated_here") + ] + self.assertTrue(saturated) + # And the derived headroom at a saturated rate is below 1, which the old + # gate would have rejected. + cells = harness.model(ceiling_per_s=450.0, rates=[1600.0]) + self.assertTrue(cells["verdict"]["saturation"]["any_rate_saturated"]) + + def test_the_model_reproduces_sequential_channel_fill(self) -> None: + # The green path has to be tested against the physics, not against a + # dataset where every cell is conveniently steady. The first saturating + # rate banks acceptance credit until the channel is full, so the model + # must produce excluded cells and still pass. + verdict = harness.model()["verdict"] + self.assertTrue(verdict["ok"], verdict["failures"]) + self.assertGreater(len(verdict["excluded_cells"]), 0) + self.assertTrue( + all( + any("acceptance credit" in r for r in c["reasons"]) + for c in verdict["excluded_cells"] + ) + ) + + def test_the_model_keeps_every_rate_comparable(self) -> None: + separation = harness.model()["verdict"]["arm_separation"] + self.assertEqual(separation["comparable_rates"], len(harness.DEFAULT_RATES)) + + def test_the_model_reports_the_serialized_worker_rate(self) -> None: + # The worker rate the model recovers is the ceiling it was given, which is + # now the drain rate measured on this rig rather than a round number + # chosen to sit below the grid. + estimate = harness.model()["verdict"]["worker_rate"]["estimate"] + self.assertAlmostEqual(estimate["mean"], harness.MEASURED_DRAIN_BAND_PER_S, 6) + + +class KneeReportingTests(unittest.TestCase): + def test_knee_is_the_first_persistent_shortfall(self) -> None: + points = [(20.0, 1.0), (50.0, 1.0), (100.0, 0.80), (200.0, 0.41)] + self.assertEqual(harness.find_knee(points, 0.99), 100.0) + + def test_a_lone_dip_is_not_a_knee(self) -> None: + # MUTANT: one rate dips and the next recovers. Saturation is monotone, so + # this is noise, and a harness that called it a knee would report a + # ceiling that moves run to run. + points = [(20.0, 1.0), (50.0, 0.90), (100.0, 1.0), (200.0, 1.0)] + self.assertIsNone(harness.find_knee(points, 0.99)) + + def test_bracket_spans_the_last_pass_and_the_knee(self) -> None: + points = [(20.0, 1.0), (50.0, 1.0), (100.0, 0.80), (200.0, 0.41)] + self.assertEqual(harness.knee_bracket(points, 0.99), (50.0, 100.0)) + + def test_points_need_not_arrive_sorted(self) -> None: + points = [(200.0, 0.41), (20.0, 1.0), (100.0, 0.80), (50.0, 1.0)] + self.assertEqual(harness.find_knee(points, 0.99), 100.0) + + +class FormattingTests(unittest.TestCase): + def test_absent_values_format_instead_of_aborting_the_sweep(self) -> None: + # A null percentile (every connection died before its first settled send) + # or a null audit series (audit-off arm) used to abort mid-sweep with a + # TypeError, losing the report and the --json file. + self.assertEqual(harness._fmt(None, "{:.2f}ms"), "n/a") + self.assertEqual(harness._fmt(2.5, "{:.2f}ms"), "2.50ms") + + +class ConnectionSizingTests(unittest.TestCase): + def test_connection_count_scales_with_offered_rate(self) -> None: + self.assertEqual(harness.conns_for(20.0), harness.MIN_CONNS) + self.assertEqual(harness.conns_for(800.0), 32) + + def test_sizing_lets_an_unsaturated_cell_meet_its_offer(self) -> None: + # That is all the sizing has to buy. Above the ceiling the relay sets the + # pace and more connections only lengthen the queue, so the target is the + # unsaturated service time, not the saturated one. + unsaturated_service_s = 0.008 + for rate in harness.DEFAULT_RATES: + self.assertGreater(harness.conns_for(rate) / unsaturated_service_s, rate) + + +if __name__ == "__main__": + unittest.main() diff --git a/scripts/start-perf-ingest-rig.sh b/scripts/start-perf-ingest-rig.sh new file mode 100755 index 00000000000..39c2eff2a66 --- /dev/null +++ b/scripts/start-perf-ingest-rig.sh @@ -0,0 +1,268 @@ +#!/usr/bin/env bash +# ============================================================================= +# start-perf-ingest-rig.sh — isolated two-community relay for the ingest-ceiling +# harness (perf/RELAY_INGEST_CEILING.md). +# ============================================================================= +# Two communities on ONE relay process, resolved by Host: a.localhost and +# b.localhost both reach 127.0.0.1, so the URL host is the Host header and no +# proxy is needed. That is what lets the harness drive two communities at +# independent rates and tell the per-pod audit worker apart from the +# per-community audit lock. +# +# Reuses the `buzz-harness` Compose project and ports from +# docker-compose.harness.yml, so the shared :3000 dev stack is never touched. +# +# The relay's admission limits are raised deliberately. At defaults one identity +# is capped at 50 events per 5s, and a rejected EVENT gets a NOTICE with no OK, +# which stalls a NIP-01 client for its whole publish timeout. A sweep run at +# defaults measures the limiter, not the relay. perf/relay_ingest_ceiling.py +# invalidates any run where the quota-rejection metric moves. +# +# Emits the rig's coordinates as JSON on stdout; progress goes to stderr. +# +# ./scripts/start-perf-ingest-rig.sh --reset # first run +# ./scripts/start-perf-ingest-rig.sh --audit off # attribution control +# +# Teardown (the script verifies the pid is still this relay before signalling it; +# do the same by hand, since pids are recycled): +# pid=$(cat /tmp/buzz-perf-ingest-rig.pid) +# ps -p "$pid" -o command= | grep -qF "$PWD/target/ci/buzz-relay" && kill "$pid" +# docker compose -p buzz-harness -f docker-compose.harness.yml down -v +# ============================================================================= +set -euo pipefail + +SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)" +REPO_ROOT="$(cd "${SCRIPT_DIR}/.." && pwd)" +cd "${REPO_ROOT}" + +PROJECT="buzz-harness" +COMPOSE_FILE="docker-compose.harness.yml" +PG_PORT=5471 +REDIS_PORT=6471 +MINIO_PORT=9471 +RELAY_MAIN=3030 +RELAY_HEALTH=8088 +RELAY_METRICS=9202 +HOST_A="a.localhost:${RELAY_MAIN}" +HOST_B="b.localhost:${RELAY_MAIN}" +PIDFILE=/tmp/buzz-perf-ingest-rig.pid +RELAY_LOG="${RELAY_LOG:-/tmp/buzz-perf-ingest-rig.log}" + +# Far above any rate the harness offers, so the limiter cannot become the +# binding constraint. Both gates must be lifted: WsEvents is a 5s window, +# Messages a 60s one, so a short run only ever exercises the first. +WS_EVENTS_PER_SEC=100000 +MESSAGES_PER_MIN=6000000 + +AUDIT=on +RESET=no +SKIP_RELAY=no +CARGO_PROFILE="${CARGO_PROFILE:-ci}" +while [[ $# -gt 0 ]]; do + case "$1" in + --audit) AUDIT="$2"; shift 2 ;; + --reset) RESET=yes; shift ;; + --skip-relay) SKIP_RELAY=yes; shift ;; + --profile) CARGO_PROFILE="$2"; shift 2 ;; + *) echo "Unknown option: $1" >&2; exit 1 ;; + esac +done +case "${AUDIT}" in on|off) ;; *) echo "--audit must be on or off" >&2; exit 1 ;; esac + +case "${CARGO_PROFILE}" in + dev|debug) CARGO_BUILD_PROFILE=dev; CARGO_TARGET_PROFILE=debug ;; + *) CARGO_BUILD_PROFILE="${CARGO_PROFILE}"; CARGO_TARGET_PROFILE="${CARGO_PROFILE}" ;; +esac + +log() { echo "[perf-rig] $*" >&2; } + +psql_h() { + docker compose -p "${PROJECT}" -f "${COMPOSE_FILE}" exec -T postgres \ + psql -U buzz -d buzz -v ON_ERROR_STOP=1 "$@" +} + +log "Bringing up backing services (project=${PROJECT})..." +docker compose -p "${PROJECT}" -f "${COMPOSE_FILE}" up -d >&2 + +for _ in $(seq 1 60); do + if psql_h -c 'SELECT 1' >/dev/null 2>&1; then break; fi + sleep 2 +done +psql_h -c 'SELECT 1' >/dev/null + +if [[ "${RESET}" == yes ]]; then + log "Resetting isolated database and applying schema..." + psql_h -c 'DROP SCHEMA public CASCADE; CREATE SCHEMA public;' >/dev/null + PGSCHEMA_PLAN_HOST=localhost PGSCHEMA_PLAN_PORT="${PG_PORT}" \ + PGSCHEMA_PLAN_DB=buzz PGSCHEMA_PLAN_USER=buzz PGSCHEMA_PLAN_PASSWORD=buzz_dev \ + PGHOST=localhost PGPORT="${PG_PORT}" PGUSER=buzz PGDATABASE=buzz PGPASSWORD=buzz_dev \ + ./bin/pgschema apply --file schema/schema.sql --auto-approve >&2 + psql_h < scripts/attach-schema-partitions.sql >/dev/null +fi + +# The rustup shim honours rust-toolchain.toml; a stray Homebrew cargo does not. +if [[ -x "${HOME}/.cargo/bin/cargo" ]]; then + export PATH="${HOME}/.cargo/bin:${PATH}" +fi +log "Building relay, CLI, and generator (profile=${CARGO_BUILD_PROFILE})..." +cargo build --profile "${CARGO_BUILD_PROFILE}" \ + -p buzz-relay -p buzz-cli -p buzz-test-client >&2 + +AUDIT_ENABLED=true +[[ "${AUDIT}" == off ]] && AUDIT_ENABLED=false + +start_relay() { + # Only signal a pid that is still our relay: pids are recycled, and a stale + # file from a relay that already exited would otherwise kill a stranger. + if [[ -f "${PIDFILE}" ]]; then + stale_pid="$(cat "${PIDFILE}")" + # The exact binary this rig launches, not any command containing + # "buzz-relay": another checkout's relay must not be killed. + if [[ "${stale_pid}" =~ ^[0-9]+$ ]] \ + && ps -p "${stale_pid}" -o command= 2>/dev/null \ + | grep -qF "${REPO_ROOT}/target/${CARGO_TARGET_PROFILE}/buzz-relay"; then + kill "${stale_pid}" 2>/dev/null || true + else + log "pidfile ${PIDFILE} is stale (pid ${stale_pid} is not our relay); removing it" + fi + rm -f "${PIDFILE}" + fi + # The relay panics rather than reporting a conflict if the metrics port is taken, + # so refuse to start instead of reporting somebody else's relay as this rig. + for port in "${RELAY_MAIN}" "${RELAY_HEALTH}" "${RELAY_METRICS}"; do + for _ in $(seq 1 15); do + lsof -nP -iTCP:"${port}" -sTCP:LISTEN >/dev/null 2>&1 || break + sleep 1 + done + if lsof -nP -iTCP:"${port}" -sTCP:LISTEN >/dev/null 2>&1; then + echo "[perf-rig] port ${port} is still in use; refusing to start" >&2 + exit 1 + fi + done + + log "Starting relay on :${RELAY_MAIN} (audit_enabled=${AUDIT_ENABLED})..." + # `env -u` scrubs any inherited CLI credentials: a stale BUZZ_AUTH_TAG fails the + # local dev relay's first write outright. + # + # The relay is exec'd through `os.setsid()` because this script is normally + # invoked from an ephemeral shell whose process group is reaped on return, which + # SIGTERMs a plain background child seconds after the rig reports ready. A new + # session detaches it. (The repo's other harness uses tmux for the same reason; + # setsid needs nothing installed.) + nohup env -u BUZZ_PRIVATE_KEY -u BUZZ_AUTH_TAG -u BUZZ_RELAY_URL \ + DATABASE_URL="postgres://buzz:buzz_dev@localhost:${PG_PORT}/buzz" \ + REDIS_URL="redis://localhost:${REDIS_PORT}" \ + RELAY_URL="ws://${HOST_A}" \ + BUZZ_BIND_ADDR="0.0.0.0:${RELAY_MAIN}" \ + BUZZ_HEALTH_PORT="${RELAY_HEALTH}" \ + BUZZ_METRICS_PORT="${RELAY_METRICS}" \ + BUZZ_S3_ENDPOINT="http://localhost:${MINIO_PORT}" \ + BUZZ_S3_ACCESS_KEY=buzz_dev \ + BUZZ_S3_SECRET_KEY=buzz_dev_secret \ + BUZZ_S3_BUCKET=buzz-media \ + BUZZ_REQUIRE_AUTH_TOKEN=false \ + BUZZ_AUDIT_ENABLED="${AUDIT_ENABLED}" \ + BUZZ_RATE_LIMIT_HUMAN_WS_EVENTS_PER_SEC="${WS_EVENTS_PER_SEC}" \ + BUZZ_RATE_LIMIT_HUMAN_MESSAGES_PER_MIN="${MESSAGES_PER_MIN}" \ + RUST_LOG=info \ + python3 -c 'import os, sys; os.setsid(); os.execv(sys.argv[1], sys.argv[1:])' \ + "${REPO_ROOT}/target/${CARGO_TARGET_PROFILE}/buzz-relay" > "${RELAY_LOG}" 2>&1 & + echo $! > "${PIDFILE}" + + for _ in $(seq 1 60); do + if curl -fs -o /dev/null "http://localhost:${RELAY_MAIN}/health"; then break; fi + sleep 1 + done + if ! curl -fs -o /dev/null "http://localhost:${RELAY_MAIN}/health"; then + echo "[perf-rig] relay did not come up on :${RELAY_MAIN} — see ${RELAY_LOG}" >&2 + exit 1 + fi +} + +if [[ "${SKIP_RELAY}" == yes ]]; then + # Attach to a relay someone else is supervising — a debugger, a CI service + # container, or an agent harness that reaps detached processes. The caller owns + # matching --audit to how that relay was actually started; nothing here can + # check it, so the audit-row control in perf/relay_ingest_ceiling.py is what + # catches a mismatch. + log "Attaching to the relay already listening on :${RELAY_MAIN}..." + if ! curl -fs -o /dev/null "http://localhost:${RELAY_MAIN}/health"; then + echo "[perf-rig] --skip-relay given but nothing is serving :${RELAY_MAIN}" >&2 + exit 1 + fi +else + start_relay +fi + +# Host A's community is seeded by the relay itself from RELAY_URL. Host B has no +# such hook, and the operator provisioning endpoint needs a NIP-98 signer the +# harness does not have, so insert the same row the startup path would. +psql_h -c "INSERT INTO communities (host) VALUES ('${HOST_B}') ON CONFLICT DO NOTHING;" >/dev/null + +BENCH_KEY="$(openssl rand -hex 32)" +create_channel() { + local host="$1" name="$2" + env -u BUZZ_AUTH_TAG BUZZ_RELAY_URL="http://${host}" BUZZ_PRIVATE_KEY="${BENCH_KEY}" \ + "./target/${CARGO_TARGET_PROFILE}/buzz" \ + channels create --name "${name}" --type stream --visibility open \ + | python3 -c 'import json,sys; print(json.load(sys.stdin)["channel_id"])' +} +log "Creating a channel in each community..." +CHANNEL_A="$(create_channel "${HOST_A}" "perf-ingest-a")" +CHANNEL_B="$(create_channel "${HOST_B}" "perf-ingest-b")" + +# Under --skip-relay this process manages no relay, so reporting a pid would +# hand the header's teardown instruction a number that is not ours to kill. +if [[ "${SKIP_RELAY}" == yes ]]; then + RELAY_PID=null +else + RELAY_PID="$(cat "${PIDFILE}")" +fi +SOURCE_REVISION="$(git -C "${REPO_ROOT}" rev-parse HEAD)" +# Two dirty trees at the same commit are two different builds, so the identity +# carries a digest of the working-tree diff rather than the commit alone. +SOURCE_DIFF_DIGEST="$(git -C "${REPO_ROOT}" diff HEAD | shasum -a 256 | cut -d' ' -f1)" +# The diff digest misses untracked inputs and is taken after the build, so it +# cannot prove the running binaries came from it. Hash the binaries themselves: +# that is the thing whose behaviour the dataset records. +BINARY_DIGEST="$(shasum -a 256 \ + "${REPO_ROOT}/target/${CARGO_TARGET_PROFILE}/buzz-relay" \ + "${REPO_ROOT}/target/${CARGO_TARGET_PROFILE}/ingest_load" \ + | shasum -a 256 | cut -d' ' -f1)" +DATABASE_RESET=false +[[ "${RESET}" == yes ]] && DATABASE_RESET=true +python3 -c ' +import json, sys +(pid, log, gen, metrics, db, project, key, audit, ws_limit, msg_limit, + host_a, chan_a, host_b, chan_b, revision, repo_root, diff_digest, + database_reset, binary_digest) = sys.argv[1:] +print(json.dumps({ + "relay_pid": None if pid == "null" else int(pid), + "source_revision": revision, + "source_diff_digest": diff_digest, + "binary_digest": binary_digest, + "database_reset": database_reset == "true", + "repo_root": repo_root, + "relay_log": log, + "generator": gen, + "metrics_url": metrics, + "database_url": db, + "compose_project": project, + "bench_private_key": key, + "audit_enabled": audit == "true", + "ws_events_per_sec_limit": int(ws_limit), + "messages_per_min_limit": int(msg_limit), + "targets": [ + {"community_host": host_a, "url": "ws://" + host_a, "channel": chan_a}, + {"community_host": host_b, "url": "ws://" + host_b, "channel": chan_b}, + ], +}, indent=2)) +' "${RELAY_PID}" "${RELAY_LOG}" "./target/${CARGO_TARGET_PROFILE}/ingest_load" \ + "http://localhost:${RELAY_METRICS}/metrics" \ + "postgres://buzz:buzz_dev@localhost:${PG_PORT}/buzz" \ + "${PROJECT}" "${BENCH_KEY}" "${AUDIT_ENABLED}" \ + "${WS_EVENTS_PER_SEC}" "${MESSAGES_PER_MIN}" \ + "${HOST_A}" "${CHANNEL_A}" "${HOST_B}" "${CHANNEL_B}" \ + "${SOURCE_REVISION}" "${REPO_ROOT}" "${SOURCE_DIFF_DIGEST}" "${DATABASE_RESET}" \ + "${BINARY_DIGEST}" +log "Rig ready. Relay pid ${RELAY_PID}, log ${RELAY_LOG}, revision ${SOURCE_REVISION}"