From 955b24bbb403bbefb7b8ed4df72cc53c01a6ed0e Mon Sep 17 00:00:00 2001 From: npub1mn7jgtj4w2pd0g0zeuhxsa6jy6p0rewxz4kujt98my82ahfmp72sxjexk7 Date: Mon, 13 Jul 2026 11:10:35 -0400 Subject: [PATCH 1/3] feat(relay): gate usage metrics behind stable leader Run DB-derived usage rollups on one advisory-lock leader while preserving per-pod in-memory gauges and safely expiring idle series. Co-authored-by: Will Pfleger Signed-off-by: Will Pfleger --- Cargo.lock | 39 +++ Cargo.toml | 1 + crates/buzz-db/src/lib.rs | 83 +++++- crates/buzz-relay/Cargo.toml | 1 + crates/buzz-relay/src/main.rs | 352 +++++++++++++++++--------- crates/buzz-relay/src/metrics.rs | 10 +- crates/buzz-relay/src/subscription.rs | 3 +- 7 files changed, 370 insertions(+), 119 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 96e1bf1ed7..9876dfe33e 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1108,6 +1108,7 @@ dependencies = [ "mesh-llm-sdk", "metrics", "metrics-exporter-prometheus", + "metrics-util", "moka", "nostr", "opentelemetry 0.32.0", @@ -2454,6 +2455,12 @@ version = "1.1.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "66b7e2430c6dff6a955451e2cfc438f09cea1965a9d6f87f7e3b90decc014099" +[[package]] +name = "endian-type" +version = "0.1.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c34f04666d835ff5d62e058c3995147c06f42fe86ff053337632bca83e42702d" + [[package]] name = "enum-assoc" version = "1.3.0" @@ -4696,11 +4703,15 @@ version = "0.20.4" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "96f8722f8562635f92f8ed992f26df0532266eb03d5202607c20c0d7e9745e13" dependencies = [ + "aho-corasick", "crossbeam-epoch", "crossbeam-utils", "hashbrown 0.16.1", + "indexmap", "metrics", + "ordered-float", "quanta", + "radix_trie", "rand 0.9.4", "rand_xoshiro", "rapidhash", @@ -5084,6 +5095,15 @@ dependencies = [ "wmi", ] +[[package]] +name = "nibble_vec" +version = "0.1.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "77a5d83df9f36fe23f0c3648c6bbb8b0298bb5f1939c8f2704431371f4b84d43" +dependencies = [ + "smallvec", +] + [[package]] name = "nix" version = "0.29.0" @@ -5728,6 +5748,15 @@ version = "0.2.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "04744f49eae99ab78e0d5c0b603ab218f515ea8cfe5a456d7629ad883a3b6e7d" +[[package]] +name = "ordered-float" +version = "5.3.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b7d950ca161dc355eaf28f82b11345ed76c6e1f6eb1f4f4479e0323b9e2fbd0e" +dependencies = [ + "num-traits", +] + [[package]] name = "ordered-multimap" version = "0.7.3" @@ -6471,6 +6500,16 @@ version = "6.0.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "f8dcc9c7d52a811697d2151c701e0d08956f92b0e24136cf4cf27b57a6a0d9bf" +[[package]] +name = "radix_trie" +version = "0.2.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c069c179fcdc6a2fe24d8d18305cf085fdbd4f922c041943e203685d6a1c58fd" +dependencies = [ + "endian-type", + "nibble_vec", +] + [[package]] name = "rand" version = "0.8.6" diff --git a/Cargo.toml b/Cargo.toml index 62de389807..3499285f91 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -79,6 +79,7 @@ opentelemetry_sdk = { version = "0.32", features = ["trace", "rt-tokio opentelemetry-otlp = { version = "0.32", default-features = false, features = ["trace", "grpc-tonic", "tls-ring"] } metrics = "0.24" metrics-exporter-prometheus = "0.18" +metrics-util = "0.20" # Error handling thiserror = "2" diff --git a/crates/buzz-db/src/lib.rs b/crates/buzz-db/src/lib.rs index 424d29df4e..d942eac0c5 100644 --- a/crates/buzz-db/src/lib.rs +++ b/crates/buzz-db/src/lib.rs @@ -52,8 +52,8 @@ pub use error::{DbError, Result}; pub use event::{EventQuery, ReactionEventInsertOutcome}; use chrono::{DateTime, Utc}; -use sqlx::postgres::PgPoolOptions; -use sqlx::{PgPool, QueryBuilder, Row}; +use sqlx::postgres::{PgConnection, PgPoolOptions}; +use sqlx::{Connection, PgPool, QueryBuilder, Row}; use std::time::Duration; use uuid::Uuid; @@ -181,6 +181,22 @@ pub struct DbPoolStats { pub max: u32, } +/// Owns the detached Postgres session holding the relay usage-metrics advisory lock. +/// +/// The connection deliberately does not return to the main pool: session advisory +/// locks must remain bound to this exact physical connection, and the poller +/// pings it before each leader-only collection tick. +pub struct UsageMetricsLeader { + connection: PgConnection, +} + +impl UsageMetricsLeader { + /// Returns whether the lock-owning session is still reachable. + pub async fn is_live(&mut self) -> bool { + self.connection.ping().await.is_ok() + } +} + /// Configuration for the Postgres connection pool. #[derive(Debug, Clone)] pub struct DbConfig { @@ -343,6 +359,30 @@ impl Db { } } + /// Try to acquire the detached session advisory lock for relay usage metrics. + /// + /// The returned guard owns the exact connection that acquired the lock. It is + /// detached from the shared pool so a stable leader neither returns a locked + /// session to other callers nor permanently consumes a pool slot. Dropping the + /// guard closes the connection and releases the session-scoped lock. + pub async fn try_lock_usage_metrics( + &self, + lock_key: i64, + ) -> Result> { + let mut connection = self.pool.acquire().await?; + let acquired = sqlx::query_scalar::<_, bool>("SELECT pg_try_advisory_lock($1)") + .bind(lock_key) + .fetch_one(&mut *connection) + .await?; + if acquired { + Ok(Some(UsageMetricsLeader { + connection: connection.detach(), + })) + } else { + Ok(None) + } + } + /// Return total number of communities on this relay. pub async fn usage_community_count(&self) -> Result { usage::community_count(&self.pool).await @@ -3510,6 +3550,7 @@ mod tests { //! channels, that fail-closed chain would go blind. use super::*; use buzz_core::CommunityId; + use sqlx::postgres::PgPoolOptions; use sqlx::{Acquire, PgPool}; use uuid::Uuid; @@ -4163,6 +4204,44 @@ mod tests { assert_eq!(watermark.1, c.id.as_bytes().as_slice()); } + #[tokio::test] + #[ignore = "requires Postgres"] + async fn test_usage_metrics_lock_has_single_owner_and_releases_on_drop() { + let pool = PgPoolOptions::new() + .max_connections(2) + .connect(TEST_DB_URL) + .await + .expect("connect to test DB"); + let first = Db::from_pool(pool.clone()); + let second = Db::from_pool(pool); + let key = 0x4255_5A5A_4D45_5452; + + let mut leader = first + .try_lock_usage_metrics(key) + .await + .expect("first lock attempt") + .expect("first database handle becomes leader"); + assert!(leader.is_live().await, "lock owner remains reachable"); + assert!( + second + .try_lock_usage_metrics(key) + .await + .expect("second lock attempt") + .is_none(), + "another session cannot become leader while the guard exists" + ); + + drop(leader); + assert!( + second + .try_lock_usage_metrics(key) + .await + .expect("lock attempt after leader drop") + .is_some(), + "dropping the detached session releases its advisory lock" + ); + } + #[tokio::test] #[ignore = "requires Postgres"] async fn lookup_community_by_host_matches_case_insensitive_host_index() { diff --git a/crates/buzz-relay/Cargo.toml b/crates/buzz-relay/Cargo.toml index e5f4b47617..d85c514db4 100644 --- a/crates/buzz-relay/Cargo.toml +++ b/crates/buzz-relay/Cargo.toml @@ -76,6 +76,7 @@ url = { workspace = true } moka = { workspace = true } metrics = { workspace = true } metrics-exporter-prometheus = { workspace = true } +metrics-util = { workspace = true } [features] dev = ["buzz-auth/dev"] diff --git a/crates/buzz-relay/src/main.rs b/crates/buzz-relay/src/main.rs index de7f0c172b..def51c0b48 100644 --- a/crates/buzz-relay/src/main.rs +++ b/crates/buzz-relay/src/main.rs @@ -1,4 +1,4 @@ -use std::collections::HashMap; +use std::collections::{HashMap, HashSet}; use std::sync::atomic::Ordering; use std::sync::Arc; @@ -47,31 +47,37 @@ fn buzz_auto_migrate_enabled(value: Option<&str>) -> bool { /// is planned as a fast-follow once the series-lifecycle (gauge idle-timeout /// and stable tie-breaking across pods) is fully designed. #[derive(Debug, Clone)] -enum PerCommunityMode { +enum EmissionScope { All, Off, } -impl PerCommunityMode { +impl EmissionScope { fn from_env() -> Self { let raw = std::env::var("BUZZ_USAGE_METRICS_PER_COMMUNITY") .unwrap_or_default() .trim() .to_ascii_lowercase(); match raw.as_str() { - "" | "all" => PerCommunityMode::All, - "off" => PerCommunityMode::Off, + "" | "all" => EmissionScope::All, + "off" => EmissionScope::Off, other => { warn!( value = other, "BUZZ_USAGE_METRICS_PER_COMMUNITY: unknown value — defaulting to all" ); - PerCommunityMode::All + EmissionScope::All } } } + + fn allows(&self, _community_id: &Uuid) -> bool { + matches!(self, Self::All) + } } +const USAGE_METRICS_LOCK_KEY: i64 = 0x4255_5A5A_4D45_5452; + #[tokio::main] async fn main() -> anyhow::Result<()> { // Install the ring CryptoProvider for rustls. Required before any rustls @@ -125,9 +131,12 @@ async fn main() -> anyhow::Result<()> { "Config loaded" ); - relay_metrics::install(config.metrics_port); + let usage_interval_secs = usage_metrics_interval_secs(); + let usage_idle_timeout_secs = usage_metrics_idle_timeout_secs(usage_interval_secs); + relay_metrics::install(config.metrics_port, usage_idle_timeout_secs); info!( port = config.metrics_port, + idle_timeout_secs = usage_idle_timeout_secs, "Prometheus metrics exporter started" ); @@ -926,15 +935,10 @@ async fn main() -> anyhow::Result<()> { // In-memory: each pod exports its partition → dashboard uses sum() { let usage_state = Arc::clone(&state); - let per_community_mode = PerCommunityMode::from_env(); - let interval_secs = std::env::var("BUZZ_USAGE_METRICS_INTERVAL_SECS") - .ok() - .and_then(|v| v.parse::().ok()) - .unwrap_or(300) // 300s default: adoption stocks don't need minute freshness; - // if event-table rollups (message counts, DAU/WAU/MAU) become - // slow at scale, move them to a maintained rollup table and drop - // the interval back to 60s. - .max(5); // 5s minimum: cheaper than pool poller's 1s minimum + let emission_scope = EmissionScope::from_env(); + let interval_secs = usage_interval_secs; + let mut leader = None; + let mut emitted_in_memory = HashSet::new(); tokio::spawn(async move { // Jitter the first tick by a random fraction of the interval so // that a rolling deploy with N pods doesn't hammer the DB @@ -951,7 +955,14 @@ async fn main() -> anyhow::Result<()> { interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip); loop { interval.tick().await; - if let Err(e) = run_usage_metrics_tick(&usage_state, &per_community_mode).await { + if let Err(e) = run_usage_metrics_tick( + &usage_state, + &emission_scope, + &mut leader, + &mut emitted_in_memory, + ) + .await + { error!(error = %e, "Usage metrics tick failed — skipping"); } } @@ -1160,22 +1171,173 @@ fn reminder_to_event(reminder: &buzz_db::event::DueReminder) -> nostr::Event { serde_json::from_value(event_json).expect("valid event JSON from DB row") } -/// Run one tick of the usage metrics poller. -/// -/// Queries the DB for per-community stock counts and snapshots in-memory -/// state for connection/subscription gauges. On any DB error, returns `Err` -/// so the caller can log and skip the tick without crashing. -/// -/// All DB-derived gauges use absolute SET (not increment), so they self-heal -/// after a missed tick or a pod restart. +/// Return the usage poll interval, with a floor that prevents a busy loop. +fn usage_metrics_interval_secs() -> u64 { + std::env::var("BUZZ_USAGE_METRICS_INTERVAL_SECS") + .ok() + .and_then(|value| value.parse().ok()) + .unwrap_or(300) + .max(5) +} + +/// Return a gauge lifetime that always outlives several usage-poller ticks. +fn usage_metrics_idle_timeout_secs(interval_secs: u64) -> u64 { + let configured = std::env::var("BUZZ_USAGE_METRICS_IDLE_TIMEOUT_SECS") + .ok() + .and_then(|value| value.parse().ok()); + idle_timeout_secs(configured, interval_secs) +} + +fn idle_timeout_secs(configured: Option, interval_secs: u64) -> u64 { + configured + .unwrap_or(900) + .max(interval_secs.saturating_mul(3)) +} + +#[derive(Debug, Clone, PartialEq, Eq, Hash)] +enum InMemoryMetricKey { + WsConnections(String), + UsersOnline(String), + Subscriptions(String), +} + +impl InMemoryMetricKey { + fn set(&self, value: f64) { + match self { + Self::WsConnections(community) => { + metrics::gauge!("buzz_community_ws_connections", "community" => community.clone()) + .set(value); + } + Self::UsersOnline(community) => { + metrics::gauge!("buzz_community_users_online_pod", "community" => community.clone()) + .set(value); + } + Self::Subscriptions(community) => { + metrics::gauge!("buzz_community_subscriptions", "community" => community.clone()) + .set(value); + } + } + } +} + +/// Emit pod-local gauges and zero only label keys that disappeared since the +/// preceding tick. The key stores the resolved host label so a removed or +/// renamed community can still receive its final zero. +fn emit_in_memory_usage_metrics( + state: &AppState, + emission_scope: &EmissionScope, + host_map: Option<&HashMap>, + previously_emitted: &mut HashSet, +) { + let connections = state.conn_manager.per_community_ws_connections(); + let users_online = state.conn_manager.per_community_users_online(); + let subscriptions = state.sub_registry.per_community_subscriptions(); + let total_connections = connections.values().sum::(); + let total_subscriptions = subscriptions.values().sum::(); + + metrics::gauge!("buzz_total_ws_connections").set(total_connections as f64); + metrics::gauge!("buzz_total_users_online_pod").set(users_online.values().sum::() as f64); + metrics::gauge!("buzz_total_subscriptions").set(total_subscriptions as f64); + // These event-driven gauges share the same live registries. Refresh them + // here so the gauge-wide idle timeout cannot remove a steady nonzero value. + metrics::gauge!("buzz_ws_connections_active").set(total_connections as f64); + metrics::gauge!("buzz_subscriptions_active").set(total_subscriptions as f64); + + let Some(host_map) = host_map else { + return; + }; + + let mut current = HashSet::new(); + for (id, host) in host_map { + if !emission_scope.allows(id) { + continue; + } + let community_id = CommunityId::from_uuid(*id); + let keys_and_values = [ + ( + InMemoryMetricKey::WsConnections(host.clone()), + connections.get(&community_id).copied(), + ), + ( + InMemoryMetricKey::UsersOnline(host.clone()), + users_online.get(&community_id).copied(), + ), + ( + InMemoryMetricKey::Subscriptions(host.clone()), + subscriptions.get(&community_id).copied(), + ), + ]; + for (key, value) in keys_and_values { + if let Some(value) = value { + key.set(value as f64); + current.insert(key); + } + } + } + + for key in dropped_in_memory_keys(previously_emitted, ¤t) { + key.set(0.0); + } + *previously_emitted = current; +} + +fn dropped_in_memory_keys( + previously_emitted: &HashSet, + current: &HashSet, +) -> Vec { + previously_emitted.difference(current).cloned().collect() +} + +/// Run one usage-metrics tick. Every pod emits its own in-memory gauges, while +/// one leader owns the heavier database-derived snapshot. async fn run_usage_metrics_tick( state: &AppState, - per_community_mode: &PerCommunityMode, + emission_scope: &EmissionScope, + leader: &mut Option, + emitted_in_memory: &mut HashSet, ) -> anyhow::Result<()> { - // --- community id → host label map (one query, cached for this tick) --- - let hosts = state.db.usage_community_hosts().await?; - let host_map: HashMap = hosts.into_iter().map(|c| (c.id, c.host)).collect(); + let host_map: HashMap = match state.db.usage_community_hosts().await { + Ok(hosts) => hosts + .into_iter() + .map(|community| (community.id, community.host)) + .collect(), + Err(error) => { + emit_in_memory_usage_metrics(state, emission_scope, None, emitted_in_memory); + return Err(error.into()); + } + }; + emit_in_memory_usage_metrics(state, emission_scope, Some(&host_map), emitted_in_memory); + + let mut demoted = false; + if let Some(leader_guard) = leader.as_mut() { + if !leader_guard.is_live().await { + warn!("Usage metrics leader lock connection failed liveness check; demoting"); + *leader = None; + demoted = true; + } + } + if leader.is_none() && !demoted { + *leader = state + .db + .try_lock_usage_metrics(USAGE_METRICS_LOCK_KEY) + .await?; + if leader.is_some() { + info!("Acquired usage metrics leader lock"); + } + } + if leader.is_some() { + emit_db_usage_metrics(state, emission_scope, &host_map).await?; + } + Ok(()) +} + +/// Emit the database-derived usage snapshot from the stable leader only. +async fn emit_db_usage_metrics( + state: &AppState, + emission_scope: &EmissionScope, + host_map: &HashMap, +) -> anyhow::Result<()> { // --- Collect all DB results before emitting any metrics (C4) --- // // All `.await?` calls happen here. If any query fails the function returns @@ -1196,19 +1358,15 @@ async fn run_usage_metrics_tick( let active_channels_1d = state.db.usage_active_channel_counts("1 day").await?; let active_channels_7d = state.db.usage_active_channel_counts("7 days").await?; - // In-memory snapshots are infallible — snapshot once before publish phase. - let conns_snapshot = state.conn_manager.per_community_ws_connections(); - let online_snapshot = state.conn_manager.per_community_users_online(); - let subs_snapshot = state.sub_registry.per_community_subscriptions(); - // --- Determine which community IDs receive per-community series (K1) --- // // `active_set` is the subset of host_map IDs that get per-community gauges // this tick. Fleet-wide totals (buzz_total_*) always emit regardless. - let active_set: std::collections::HashSet = match per_community_mode { - PerCommunityMode::All => host_map.keys().copied().collect(), - PerCommunityMode::Off => std::collections::HashSet::new(), - }; + let active_set: HashSet = host_map + .keys() + .filter(|id| emission_scope.allows(id)) + .copied() + .collect(); // --- Publish phase: emit all metrics now that every query succeeded --- @@ -1229,7 +1387,7 @@ async fn run_usage_metrics_tick( metrics::gauge!("buzz_total_users", "type" => "human").set(total_human as f64); metrics::gauge!("buzz_total_users", "type" => "agent").set(total_agent as f64); // Per-community series (gated by active_set). - for (&id, community) in &host_map { + for (&id, community) in host_map { if !active_set.contains(&id) { continue; } @@ -1271,7 +1429,7 @@ async fn run_usage_metrics_tick( metrics::gauge!("buzz_total_channels", "type" => ct).set(total as f64); } // Per-community series (gated by active_set). - for (&id, community) in &host_map { + for (&id, community) in host_map { if !active_set.contains(&id) { continue; } @@ -1298,7 +1456,7 @@ async fn run_usage_metrics_tick( let total: i64 = rows.values().sum(); metrics::gauge!("buzz_total_messages").set(total as f64); // Per-community series (gated by active_set). - for (&id, community) in &host_map { + for (&id, community) in host_map { if !active_set.contains(&id) { continue; } @@ -1338,7 +1496,7 @@ async fn run_usage_metrics_tick( metrics::gauge!("buzz_total_relay_members", "role" => role).set(total as f64); } // Per-community series (gated by active_set). - for (&id, community) in &host_map { + for (&id, community) in host_map { if !active_set.contains(&id) { continue; } @@ -1384,7 +1542,7 @@ async fn run_usage_metrics_tick( metrics::gauge!("buzz_total_workflows", "status" => status).set(total as f64); } // Per-community series (gated by active_set). - for (&id, community) in &host_map { + for (&id, community) in host_map { if !active_set.contains(&id) { continue; } @@ -1411,7 +1569,7 @@ async fn run_usage_metrics_tick( let total: i64 = rows.values().sum(); metrics::gauge!("buzz_total_git_repos").set(total as f64); // Per-community series (gated by active_set). - for (&id, community) in &host_map { + for (&id, community) in host_map { if !active_set.contains(&id) { continue; } @@ -1444,7 +1602,7 @@ async fn run_usage_metrics_tick( metrics::gauge!("buzz_total_active_users", "window" => label, "type" => "unknown") .set(total_unknown as f64); // Per-community series (gated by active_set). - for (&id, community) in &host_map { + for (&id, community) in host_map { if !active_set.contains(&id) { continue; } @@ -1485,7 +1643,7 @@ async fn run_usage_metrics_tick( let total: i64 = rows.values().sum(); metrics::gauge!("buzz_total_active_channels", "window" => label).set(total as f64); // Per-community series (gated by active_set). - for (&id, community) in &host_map { + for (&id, community) in host_map { if !active_set.contains(&id) { continue; } @@ -1499,84 +1657,22 @@ async fn run_usage_metrics_tick( } } - // --- D. Realtime — in-memory snapshots --- - // All three gauges are derived from the in-memory conn/sub registries. - // We emit from host_map so communities that just dropped to zero - // (last connection closed, all subs removed) receive an explicit 0 - // rather than keeping the stale last value. - - // buzz_community_ws_connections{community} - { - // Fleet total (always emitted). - let total: u64 = conns_snapshot.values().sum(); - metrics::gauge!("buzz_total_ws_connections").set(total as f64); - // Per-community series (gated by active_set). - for (&id, community) in &host_map { - if !active_set.contains(&id) { - continue; - } - let count = conns_snapshot - .get(&CommunityId::from_uuid(id)) - .copied() - .unwrap_or(0); - metrics::gauge!("buzz_community_ws_connections", "community" => community.clone()) - .set(count as f64); - } - } - - // buzz_community_users_online_pod{community} - // This is a pod-local distinct count — a pubkey connected to N pods is - // counted once per pod. Dashboard queries should sum across pods to get - // the fleet-wide total (with the caveat that multi-pod connections are - // counted N times). The metric name "_pod" suffix makes this explicit. - { - // Fleet total (always emitted; same pod-local caveat applies). - let total: u64 = online_snapshot.values().sum(); - metrics::gauge!("buzz_total_users_online_pod").set(total as f64); - // Per-community series (gated by active_set). - for (&id, community) in &host_map { - if !active_set.contains(&id) { - continue; - } - let count = online_snapshot - .get(&CommunityId::from_uuid(id)) - .copied() - .unwrap_or(0); - metrics::gauge!("buzz_community_users_online_pod", "community" => community.clone()) - .set(count as f64); - } - } - - // buzz_community_subscriptions{community} - { - // Fleet total (always emitted). - let total: u64 = subs_snapshot.values().sum(); - metrics::gauge!("buzz_total_subscriptions").set(total as f64); - // Per-community series (gated by active_set). - for (&id, community) in &host_map { - if !active_set.contains(&id) { - continue; - } - let count = subs_snapshot - .get(&CommunityId::from_uuid(id)) - .copied() - .unwrap_or(0); - metrics::gauge!("buzz_community_subscriptions", "community" => community.clone()) - .set(count as f64); - } - } - Ok(()) } #[cfg(test)] mod tests { + use std::collections::HashSet; use std::sync::Arc; use std::time::Duration; use tokio_util::sync::CancellationToken; + use uuid::Uuid; - use super::{buzz_auto_migrate_enabled, run_periodic_until_cancelled}; + use super::{ + buzz_auto_migrate_enabled, dropped_in_memory_keys, idle_timeout_secs, + run_periodic_until_cancelled, EmissionScope, InMemoryMetricKey, + }; #[tokio::test(start_paused = true)] async fn periodic_loop_exits_immediately_on_cancellation() { @@ -1617,4 +1713,32 @@ mod tests { assert!(buzz_auto_migrate_enabled(Some("yes"))); assert!(buzz_auto_migrate_enabled(Some("on"))); } + + #[test] + fn test_emission_scope_off_disallows_every_community() { + assert!(EmissionScope::All.allows(&Uuid::new_v4())); + assert!(!EmissionScope::Off.allows(&Uuid::new_v4())); + } + + #[test] + fn test_dropped_in_memory_keys_preserves_resolved_host_label() { + let previous = HashSet::from([ + InMemoryMetricKey::WsConnections("removed.example".to_owned()), + InMemoryMetricKey::UsersOnline("live.example".to_owned()), + ]); + let current = HashSet::from([InMemoryMetricKey::UsersOnline("live.example".to_owned())]); + + assert_eq!( + dropped_in_memory_keys(&previous, ¤t), + vec![InMemoryMetricKey::WsConnections( + "removed.example".to_owned() + )] + ); + } + + #[test] + fn test_idle_timeout_is_at_least_three_usage_intervals() { + assert_eq!(idle_timeout_secs(None, 300), 900); + assert_eq!(idle_timeout_secs(Some(10), 1_000), 3_000); + } } diff --git a/crates/buzz-relay/src/metrics.rs b/crates/buzz-relay/src/metrics.rs index 954e78434d..5c0b40b49d 100644 --- a/crates/buzz-relay/src/metrics.rs +++ b/crates/buzz-relay/src/metrics.rs @@ -14,7 +14,7 @@ //! recorded by [`track_metrics`] middleware on the app router. Buzz-specific //! metrics are recorded inline at their call sites. -use std::time::Instant; +use std::time::{Duration, Instant}; use axum::{ extract::{MatchedPath, Request}, @@ -22,6 +22,7 @@ use axum::{ response::Response, }; use metrics_exporter_prometheus::{Matcher, PrometheusBuilder}; +use metrics_util::MetricKindMask; /// HTTP latency buckets (milliseconds) — only for `http_request_latency_ms`. const LATENCY_BUCKETS_MS: [f64; 11] = [ @@ -41,9 +42,14 @@ const FANOUT_BUCKETS: [f64; 9] = [0.0, 1.0, 5.0, 10.0, 25.0, 50.0, 100.0, 500.0, /// /// Must be called from within a Tokio runtime. /// Panics if a recorder is already installed or the port is in use. -pub fn install(port: u16) { +pub fn install(port: u16, gauge_idle_timeout_secs: u64) { let (recorder, exporter) = PrometheusBuilder::new() .with_http_listener(([0, 0, 0, 0], port)) + // Remove gauge series that the relay intentionally stops emitting. + .idle_timeout( + MetricKindMask::GAUGE, + Some(Duration::from_secs(gauge_idle_timeout_secs)), + ) // Per-metric buckets: ms for HTTP latency, seconds for internal processing. .set_buckets_for_metric( Matcher::Full("http_request_latency_ms".to_owned()), diff --git a/crates/buzz-relay/src/subscription.rs b/crates/buzz-relay/src/subscription.rs index 9831e8eee4..68f0fea3c4 100644 --- a/crates/buzz-relay/src/subscription.rs +++ b/crates/buzz-relay/src/subscription.rs @@ -1532,7 +1532,8 @@ mod tests { fn per_community_subscriptions_drops_to_zero_when_all_subs_removed() { // Regression: when the last subscription for a community is removed, // per_community_subscriptions() must return no entry (not stale nonzero). - // The poller zero-fills from host_map, so absence == 0 is correct. + // The usage poller emits one explicit zero for the previously observed + // label before allowing its idle timeout to remove the series. let registry = SubscriptionRegistry::new(); let community = CommunityId::from_uuid(Uuid::from_u128(0xcccc)); let conn = Uuid::new_v4(); From 54ba7cf3381ec76c252b410f26ecad311ddd4f9c Mon Sep 17 00:00:00 2001 From: npub1mn7jgtj4w2pd0g0zeuhxsa6jy6p0rewxz4kujt98my82ahfmp72sxjexk7 Date: Mon, 13 Jul 2026 11:55:40 -0400 Subject: [PATCH 2/3] fix(relay): refresh legacy gauge recency without overwrites Touch event-driven gauges with a value-neutral increment so steady series survive the exporter idle timeout without racing lifecycle deltas. The regression coverage verifies both generation advancement and lifecycle value preservation. Co-authored-by: Will Pfleger Signed-off-by: Will Pfleger --- crates/buzz-relay/src/main.rs | 70 ++++++++++++++++++++++++++++++++--- 1 file changed, 65 insertions(+), 5 deletions(-) diff --git a/crates/buzz-relay/src/main.rs b/crates/buzz-relay/src/main.rs index def51c0b48..000e405414 100644 --- a/crates/buzz-relay/src/main.rs +++ b/crates/buzz-relay/src/main.rs @@ -1220,6 +1220,18 @@ impl InMemoryMetricKey { } } +/// Refresh the exporter recency for legacy event-driven gauges without changing +/// their values. +/// +/// `metrics-util` 0.20.4 increments the Prometheus recorder's generation on +/// every gauge operation, including `increment(0.0)`. The recency policy uses +/// that generation, so this retains a steady gauge without a snapshot `set()` +/// racing the lifecycle-relative increments and decrements. +fn refresh_legacy_active_gauge_recency() { + metrics::gauge!("buzz_ws_connections_active").increment(0.0); + metrics::gauge!("buzz_subscriptions_active").increment(0.0); +} + /// Emit pod-local gauges and zero only label keys that disappeared since the /// preceding tick. The key stores the resolved host label so a removed or /// renamed community can still receive its final zero. @@ -1238,10 +1250,7 @@ fn emit_in_memory_usage_metrics( metrics::gauge!("buzz_total_ws_connections").set(total_connections as f64); metrics::gauge!("buzz_total_users_online_pod").set(users_online.values().sum::() as f64); metrics::gauge!("buzz_total_subscriptions").set(total_subscriptions as f64); - // These event-driven gauges share the same live registries. Refresh them - // here so the gauge-wide idle timeout cannot remove a steady nonzero value. - metrics::gauge!("buzz_ws_connections_active").set(total_connections as f64); - metrics::gauge!("buzz_subscriptions_active").set(total_subscriptions as f64); + refresh_legacy_active_gauge_recency(); let Some(host_map) = host_map else { return; @@ -1671,7 +1680,13 @@ mod tests { use super::{ buzz_auto_migrate_enabled, dropped_in_memory_keys, idle_timeout_secs, - run_periodic_until_cancelled, EmissionScope, InMemoryMetricKey, + refresh_legacy_active_gauge_recency, run_periodic_until_cancelled, EmissionScope, + InMemoryMetricKey, + }; + use metrics::GaugeFn; + use metrics_util::{ + debugging::DebugValue, + registry::{GenerationalAtomicStorage, Registry}, }; #[tokio::test(start_paused = true)] @@ -1736,6 +1751,51 @@ mod tests { ); } + #[test] + fn test_legacy_gauge_recency_refresh_preserves_lifecycle_deltas() { + let recorder = metrics_util::debugging::DebuggingRecorder::new(); + let snapshotter = recorder.snapshotter(); + + metrics::with_local_recorder(&recorder, || { + let connections = metrics::gauge!("buzz_ws_connections_active"); + let subscriptions = metrics::gauge!("buzz_subscriptions_active"); + connections.increment(1.0); + subscriptions.increment(1.0); + + refresh_legacy_active_gauge_recency(); + + connections.decrement(1.0); + subscriptions.increment(1.0); + }); + + let values = snapshotter + .snapshot() + .into_vec() + .into_iter() + .map(|(key, _, _, value)| { + let DebugValue::Gauge(value) = value else { + panic!("{} must be a gauge", key.key().name()); + }; + (key.key().name().to_owned(), value.into_inner()) + }) + .collect::>(); + + assert_eq!(values.get("buzz_ws_connections_active"), Some(&0.0)); + assert_eq!(values.get("buzz_subscriptions_active"), Some(&2.0)); + } + + #[test] + fn test_legacy_gauge_recency_refresh_advances_generation() { + let registry = Registry::new(GenerationalAtomicStorage::atomic()); + let key = metrics::Key::from_name("legacy"); + let gauge = registry.get_or_create_gauge(&key, Clone::clone); + gauge.increment(1.0); + let generation_before = gauge.get_generation(); + gauge.increment(0.0); + + assert!(gauge.get_generation() > generation_before); + } + #[test] fn test_idle_timeout_is_at_least_three_usage_intervals() { assert_eq!(idle_timeout_secs(None, 300), 900); From 862bcd2fa84c74e48fee918eac91e3561f663418 Mon Sep 17 00:00:00 2001 From: Will Pfleger Date: Tue, 14 Jul 2026 11:33:23 -0400 Subject: [PATCH 3/3] fix(relay): demote usage-metrics leader on collection failure A leader whose DB collection path fails (pool exhaustion, host-map query error) now releases the advisory lock so another pod can take over, instead of holding the lock indefinitely while emitting nothing. Also bounds the liveness ping to 5 seconds (prevents a blackholed TCP connection from stalling the entire tick loop) and adds a per-pod `buzz_usage_poller_is_leader` gauge for dashboard/alerting visibility. --- crates/buzz-db/src/lib.rs | 7 ++++++- crates/buzz-relay/src/main.rs | 15 ++++++++++++++- 2 files changed, 20 insertions(+), 2 deletions(-) diff --git a/crates/buzz-db/src/lib.rs b/crates/buzz-db/src/lib.rs index d942eac0c5..c0b6bf78c6 100644 --- a/crates/buzz-db/src/lib.rs +++ b/crates/buzz-db/src/lib.rs @@ -192,8 +192,13 @@ pub struct UsageMetricsLeader { impl UsageMetricsLeader { /// Returns whether the lock-owning session is still reachable. + /// + /// Bounded to 5 seconds — a blackholed connection (no RST) would otherwise + /// stall the entire poller tick until the OS TCP timeout. pub async fn is_live(&mut self) -> bool { - self.connection.ping().await.is_ok() + tokio::time::timeout(std::time::Duration::from_secs(5), self.connection.ping()) + .await + .is_ok_and(|r| r.is_ok()) } } diff --git a/crates/buzz-relay/src/main.rs b/crates/buzz-relay/src/main.rs index 000e405414..1579cc2dfa 100644 --- a/crates/buzz-relay/src/main.rs +++ b/crates/buzz-relay/src/main.rs @@ -965,6 +965,11 @@ async fn main() -> anyhow::Result<()> { { error!(error = %e, "Usage metrics tick failed — skipping"); } + metrics::gauge!("buzz_usage_poller_is_leader").set(if leader.is_some() { + 1.0 + } else { + 0.0 + }); } }); } @@ -1311,6 +1316,10 @@ async fn run_usage_metrics_tick( .map(|community| (community.id, community.host)) .collect(), Err(error) => { + if leader.is_some() { + warn!("Usage metrics leader demoting: host map collection failed"); + *leader = None; + } emit_in_memory_usage_metrics(state, emission_scope, None, emitted_in_memory); return Err(error.into()); } @@ -1335,7 +1344,11 @@ async fn run_usage_metrics_tick( } } if leader.is_some() { - emit_db_usage_metrics(state, emission_scope, &host_map).await?; + if let Err(error) = emit_db_usage_metrics(state, emission_scope, &host_map).await { + warn!("Usage metrics leader demoting: DB collection failed"); + *leader = None; + return Err(error); + } } Ok(())