From 4813b8cd15411355f05599e263ab15dd04e7afcc Mon Sep 17 00:00:00 2001 From: Luke Tornquist Date: Wed, 4 Nov 2026 13:15:51 -0500 Subject: [PATCH] relay: time every WebSocket read and every stored EVENT outcome (#9052) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ## Summary Two gaps in relay telemetry: WebSocket reads (`REQ`, NIP-AR sync, `COUNT`) have no latency or outcome metric, and `buzz_event_processing_seconds` only measures stored EVENTs that succeed. **What this does** - **`buzz_ws_read_seconds{type, outcome, reason}`** (new histogram). It's recorded from `access::ReadLog::drop`. Every WebSocket read already finishes through that one seam, whether it ends `served`, `denied`, `error`, or `aborted` (future dropped mid-read), so this adds no new exit paths and changes no handler. Duration runs from the moment the handler takes the frame until `EOSE` is queued (`REQ`), the initial set ends (sync), or the total is sent (`COUNT`). `reason` is the existing fixed denial/error code (`query_timed_out`, `database_error`, `not_channel_member`, …), or `none`. The histogram's `_count` doubles as the outcome counter, so one metric gives both read latency and read failure rate. - It's recorded *before* the refusal budget check, because the budget only exists to suppress log lines. When `refused_read` is over budget, it records the sample without building the log line, so its cost under a refusal flood stays the same. - **`buzz_event_processing_seconds{outcome}`**. Before this change it was recorded only when `ingest_event` returned `Ok`. A store that was slow *and failing* (statement timeouts, pool exhaustion) produced no samples, so latency looked better while the store was degrading. It's now recorded on every persistent-ingest outcome: `accepted`, `not_accepted` (for example a duplicate), `invalid`, `auth`, `error`. Queries that sum across all outcomes still work. **Cardinality.** Every label value is a `&'static str` from a code-defined set: 3 types × 4 outcomes, with about 25 reasons that only appear alongside `denied`/`error`. On purpose, there's no `kind` label: `bounded_kind_label` still admits around 200 values, which is too many to cross with histogram buckets. **What got simpler:** `ReadOutcome::parts()` replaces the inline outcome→label match in `Drop`, so the log line and the metric share one mapping. There's no new instrumentation seam: the read metric reuses the access log's lifecycle. ### Related issue None found. This came out of a monitoring-gap review: without these signals, alerts can't catch a slow or failing WebSocket read path, or a failing EVENT store. ### Testing - `access::tests::every_read_is_measured_even_when_its_line_is_suppressed` (unit). It drives the real `ReadLog` constructors and `Drop`: 20 admission refusals against a 16-line budget, a handler denial past the budget, a `COUNT` error, a served sync and an aborted `REQ`. It asserts the exact `(type, outcome, reason, samples)` rows. Two mutants each fail it: removing the over-budget branch in `refused_read`, and moving the measurement after the budget check in `Drop`. - `handlers::event::tests::postgres_tests::event_processing_is_timed_for_accepted_and_refused_ingest` (Postgres lane). It sends one accepted text note, then a refused one from a cached-ban sender, through `handle_event`, and asserts one sample each under `accepted` and `auth` (a ban maps to the `auth` class). Removing the `Err`-arm measurement fails it. - Blox gate at `b314df43f2` (tree `4ca9ebc1`): `cargo fmt --check` clean. The new Postgres test passes, and removing the `Err`-arm measurement fails it. buzz-relay unit lane (`cargo nextest run -p buzz-relay --lib --bins`): 1708/1708 passed. - At `539796bd51` (the same code; only the second commit's test assertion differs): workspace clippy (`--all-targets -D warnings`) passed. The new unit test passed, and each of its two mutants failed it. buzz-relay Postgres lane: 604/605 passed. The one failure was this test's old `invalid` expectation, which `b314df43f2` corrects. - CI at `b314df43f2`: both new tests appear in the CI logs as passing (Rust / Unit Tests, PostgreSQL Domain / PostgreSQL Tests). 🤖 Generated with [Claude Code](https://claude.com/claude-code) --------- Signed-off-by: Luke Tornquist Co-authored-by: Elrond <28d6302a099e5225b02c4155ac4236e4912603df2ab08dbfc2f4fef08ce598c8@buzz.block.builderlab.xyz> Co-authored-by: Claude Opus 5.5 --- crates/buzz-relay/src/access.rs | 94 +++++++++++++++++++++++-- crates/buzz-relay/src/handlers/event.rs | 68 +++++++++++++++++- crates/buzz-relay/src/metrics.rs | 23 ++++++ 3 files changed, 177 insertions(+), 8 deletions(-) diff --git a/crates/buzz-relay/src/access.rs b/crates/buzz-relay/src/access.rs index e71c9f21e9..b14ada9e67 100644 --- a/crates/buzz-relay/src/access.rs +++ b/crates/buzz-relay/src/access.rs @@ -238,6 +238,18 @@ enum ReadOutcome { Error(&'static str), } +impl ReadOutcome { + /// The `outcome` and `reason` this read logs and is measured under. + fn parts(self) -> (&'static str, Option<&'static str>) { + match self { + Self::Aborted => ("aborted", None), + Self::Served => ("served", None), + Self::Denied(r) => ("denied", Some(r)), + Self::Error(r) => ("error", Some(r)), + } + } +} + /// One access line per WebSocket REQ, COUNT or sync, written when the handler /// finishes (a sync's initial set completes), including when its future is /// dropped mid-read. @@ -254,6 +266,8 @@ pub(crate) struct ReadLog { results: u64, /// Charged if the read ends denied; `None` once already charged. budget: Option>, + /// When the handler took the frame; the read's duration ends at drop. + started: std::time::Instant, } impl ReadLog { @@ -318,6 +332,9 @@ impl ReadLog { let mut log = Self::new(event, conn, sub_id, filters); log.budget = None; log.deny(reason); + } else { + // Over budget the line is suppressed, but the read still counts. + crate::metrics::record_ws_read_finished(event, "denied", Some(reason), 0.0); } } @@ -362,6 +379,7 @@ impl ReadLog { outcome: ReadOutcome::Aborted, results: 0, budget: Some(Arc::clone(&conn.refusal_budget)), + started: std::time::Instant::now(), } } @@ -404,17 +422,19 @@ impl ReadLog { impl Drop for ReadLog { fn drop(&mut self) { + let (outcome, reason) = self.outcome.parts(); + // Measured before the refusal budget, which only suppresses the line. + crate::metrics::record_ws_read_finished( + self.event, + outcome, + reason, + self.started.elapsed().as_secs_f64(), + ); if let (ReadOutcome::Denied(_), Some(budget)) = (self.outcome, &self.budget) { if !budget.read() { return; } } - let (outcome, reason) = match self.outcome { - ReadOutcome::Aborted => ("aborted", None), - ReadOutcome::Served => ("served", None), - ReadOutcome::Denied(r) => ("denied", Some(r)), - ReadOutcome::Error(r) => ("error", Some(r)), - }; let s = &self.summary; let (events_queued, count) = match (self.event, self.outcome) { ("count", ReadOutcome::Served) => (None, Some(self.results)), @@ -1569,6 +1589,7 @@ pub(crate) mod tests { outcome, results: 0, budget: None, + started: std::time::Instant::now(), }; let (_, out) = capture(|| { drop(read(ReadOutcome::Error("database_error"))); @@ -1710,6 +1731,7 @@ pub(crate) mod tests { outcome: ReadOutcome::Aborted, results: 0, budget: None, + started: std::time::Instant::now(), }; for _ in 0..3 { log.queued_one(); @@ -1744,6 +1766,7 @@ pub(crate) mod tests { outcome: ReadOutcome::Aborted, results: 0, budget: None, + started: std::time::Instant::now(), }; }); let line = &out.lines()[0]; @@ -1758,6 +1781,65 @@ pub(crate) mod tests { assert!(line.get("pubkey").is_none()); } + /// Every read is measured under its type, outcome and reason, including + /// the refusals whose lines the budget suppresses. + /// Mutation: record after the budget check in `Drop`, or drop the + /// over-budget branch of `refused_read` → RED. + #[test] + fn every_read_is_measured_even_when_its_line_is_suppressed() { + let (conn, _rx) = + crate::connection::tests::test_conn_with_auth(crate::connection::AuthState::Failed); + let recorder = metrics_util::debugging::DebuggingRecorder::new(); + let snapshotter = recorder.snapshotter(); + metrics::with_local_recorder(&recorder, || { + // More refusals than the budget logs: the rest are only counted. + for _ in 0..REFUSAL_LINES + 4 { + ReadLog::refused_read("req", &conn, "r", &[], "not_admitted"); + } + // A handler denial past the budget: `Drop` suppresses its line. + ReadLog::req(&conn, "d", &[]).deny("auth_required"); + ReadLog::count(&conn, "c", &[]).error("database_error"); + ReadLog::sync(&conn, "s", &[]).served(); + drop(ReadLog::req(&conn, "a", &[])); + }); + let mut measured: Vec<(String, String, String, usize)> = snapshotter + .snapshot() + .into_vec() + .into_iter() + .filter(|(key, ..)| key.key().name() == "buzz_ws_read_seconds") + .map(|(key, _, _, value)| { + let metrics_util::debugging::DebugValue::Histogram(samples) = value else { + panic!("buzz_ws_read_seconds must be a histogram"); + }; + let label = |name: &str| { + key.key() + .labels() + .find(|l| l.key() == name) + .map(|l| l.value().to_owned()) + .unwrap_or_default() + }; + ( + label("type"), + label("outcome"), + label("reason"), + samples.len(), + ) + }) + .collect(); + measured.sort(); + let row = |t: &str, o: &str, r: &str, n| (t.to_owned(), o.to_owned(), r.to_owned(), n); + assert_eq!( + measured, + [ + row("count", "error", "database_error", 1), + row("req", "aborted", "none", 1), + row("req", "denied", "auth_required", 1), + row("req", "denied", "not_admitted", REFUSAL_LINES as usize + 4), + row("sync", "served", "none", 1), + ] + ); + } + #[test] fn auth_failure_is_warn_and_marks_the_claimed_key_unverified() { let key = nostr::Keys::generate().public_key(); diff --git a/crates/buzz-relay/src/handlers/event.rs b/crates/buzz-relay/src/handlers/event.rs index 256a7d09c7..a73a4c2383 100644 --- a/crates/buzz-relay/src/handlers/event.rs +++ b/crates/buzz-relay/src/handlers/event.rs @@ -52,6 +52,15 @@ pub(crate) fn bounded_kind_label(kind: u32) -> String { } } +/// Time one persistent WebSocket EVENT from frame to OK under `outcome`: +/// `accepted`, `not_accepted` (for example a duplicate), or the ingest error +/// class (`invalid`, `auth`, `error`). Failures are timed too, so a slow +/// failing store shows in the latency instead of hiding from it. +fn record_event_processing(start: std::time::Instant, outcome: &'static str) { + metrics::histogram!("buzz_event_processing_seconds", "outcome" => outcome) + .record(start.elapsed().as_secs_f64()); +} + fn event_frame_for_sub(sub_id: &str, event_json: &str) -> String { format!(r#"["EVENT","{}",{}]"#, sub_id, event_json) } @@ -921,8 +930,14 @@ pub async fn handle_event(event: Event, conn: Arc, state: Arc, state: Arc = snapshotter + .snapshot() + .into_vec() + .into_iter() + .filter(|(key, ..)| key.key().name() == "buzz_event_processing_seconds") + .map(|(key, _, _, value)| { + let metrics_util::debugging::DebugValue::Histogram(samples) = value else { + panic!("buzz_event_processing_seconds must be a histogram"); + }; + let outcome = key + .key() + .labels() + .find(|l| l.key() == "outcome") + .map(|l| l.value().to_owned()) + .unwrap_or_default(); + (outcome, samples.len()) + }) + .collect(); + timed.sort(); + assert_eq!(timed, [("accepted".to_owned(), 1), ("auth".to_owned(), 1)]); + } + /// Stored writes skip the cached serving state: a cached "fenced" /// for a community Postgres says is active does not refuse them. /// Mutation: make ingest's fence call `is_serving_active_cached` → RED. diff --git a/crates/buzz-relay/src/metrics.rs b/crates/buzz-relay/src/metrics.rs index b4bc7df953..a6ebbcd513 100644 --- a/crates/buzz-relay/src/metrics.rs +++ b/crates/buzz-relay/src/metrics.rs @@ -552,6 +552,11 @@ fn describe_ws_read_metrics() { "buzz_ws_reads_total", "WebSocket REQ (including NIP-AR sync) and COUNT frames received, by type, whatever the log level" ); + metrics::describe_histogram!( + "buzz_ws_read_seconds", + metrics::Unit::Seconds, + "WebSocket REQ, sync and COUNT reads from handler start to their end (REQ: EOSE queued; sync: initial set; COUNT: total sent), by type, outcome and reason, whatever the log level" + ); } /// Export every WebSocket read series at zero. @@ -566,6 +571,24 @@ pub(crate) fn record_ws_read(kind: &'static str) { metrics::counter!("buzz_ws_reads_total", "type" => kind).increment(1); } +/// Record how one WebSocket read ended and how long it took. `outcome` is +/// `served`, `denied`, `error` or `aborted`; `reason` is the fixed denial or +/// error code, or `none`. +pub(crate) fn record_ws_read_finished( + kind: &'static str, + outcome: &'static str, + reason: Option<&'static str>, + seconds: f64, +) { + metrics::histogram!( + "buzz_ws_read_seconds", + "type" => kind, + "outcome" => outcome, + "reason" => reason.unwrap_or("none"), + ) + .record(seconds); +} + /// Export the partition audit counters at zero before the first audit, so /// `increase()`-based alerts count the first failure and every new outcome. pub(crate) fn initialize_partition_audit_metric_series() {