mirror of
https://github.com/block/buzz.git
synced 2026-11-06 01:11:33 +00:00
relay: time every WebSocket read and every stored EVENT outcome (#9052)
## 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 <tornquist@squareup.com>
Co-authored-by: Elrond <28d6302a099e5225b02c4155ac4236e4912603df2ab08dbfc2f4fef08ce598c8@buzz.block.builderlab.xyz>
Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
Luke Tornquist
@@ -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<Arc<RefusalBudget>>,
|
||||
/// 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();
|
||||
|
||||
@@ -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<ConnectionState>, state: Arc<A
|
||||
"Event ingested"
|
||||
);
|
||||
}
|
||||
metrics::histogram!("buzz_event_processing_seconds")
|
||||
.record(start.elapsed().as_secs_f64());
|
||||
record_event_processing(
|
||||
start,
|
||||
if result.accepted {
|
||||
"accepted"
|
||||
} else {
|
||||
"not_accepted"
|
||||
},
|
||||
);
|
||||
conn.send(RelayMessage::ok(
|
||||
&result.event_id,
|
||||
result.accepted,
|
||||
@@ -941,6 +956,7 @@ pub async fn handle_event(event: Event, conn: Arc<ConnectionState>, state: Arc<A
|
||||
("error: internal server error".to_string(), "error")
|
||||
}
|
||||
};
|
||||
record_event_processing(start, reason);
|
||||
reject(reason);
|
||||
conn.send(RelayMessage::ok(&event_id_hex, false, &msg));
|
||||
}
|
||||
@@ -3977,6 +3993,54 @@ mod tests {
|
||||
);
|
||||
}
|
||||
|
||||
/// Every persistent ingest is timed under its outcome, failures
|
||||
/// included, so a failing store cannot hide from the latency.
|
||||
/// Mutation: drop `record_event_processing` from the `Err` arm → RED.
|
||||
#[tokio::test(flavor = "current_thread")]
|
||||
#[ignore = "requires Postgres — runs in postgres-ci nextest lane"]
|
||||
async fn event_processing_is_timed_for_accepted_and_refused_ingest() {
|
||||
let (state, tenant, user) = unrestricted_fixture().await;
|
||||
let scopes = buzz_auth::Scope::all_non_admin();
|
||||
let recorder = metrics_util::debugging::DebuggingRecorder::new();
|
||||
let snapshotter = recorder.snapshotter();
|
||||
{
|
||||
let _guard = metrics::set_default_local_recorder(&recorder);
|
||||
let frame =
|
||||
ok_frame_scoped(&state, &tenant, &user, text_note(&user), scopes.clone()).await;
|
||||
assert!(frame.contains(",true,"), "{frame}");
|
||||
state.seed_restriction(
|
||||
(tenant.community(), user.public_key().to_bytes().to_vec()),
|
||||
buzz_db::moderation::RestrictionState {
|
||||
banned: true,
|
||||
own_banned: true,
|
||||
muted_until: None,
|
||||
},
|
||||
);
|
||||
let frame = ok_frame_scoped(&state, &tenant, &user, text_note(&user), scopes).await;
|
||||
assert!(frame.contains("blocked: you are banned"), "{frame}");
|
||||
}
|
||||
let mut timed: Vec<(String, usize)> = 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.
|
||||
|
||||
@@ -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() {
|
||||
|
||||
Reference in New Issue
Block a user