diff --git a/Cargo.lock b/Cargo.lock index bb66f39b0..3cbb1f557 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1056,6 +1056,7 @@ dependencies = [ "mesh-llm-sdk", "metrics", "metrics-exporter-prometheus", + "metrics-util", "moka", "nostr", "opentelemetry 0.32.0", @@ -2317,6 +2318,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" @@ -4520,11 +4527,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", @@ -4902,6 +4913,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" @@ -5546,6 +5566,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" @@ -6249,6 +6278,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 17267207a..06a53495c 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -73,6 +73,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 0e747fbcf..3925403ba 100644 --- a/crates/buzz-db/src/lib.rs +++ b/crates/buzz-db/src/lib.rs @@ -48,8 +48,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; @@ -150,6 +150,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 { @@ -288,6 +304,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 @@ -3009,6 +3049,7 @@ mod tests { //! channels, that fail-closed chain would go blind. use super::*; use buzz_core::CommunityId; + use sqlx::postgres::PgPoolOptions; use sqlx::PgPool; use uuid::Uuid; @@ -3033,6 +3074,44 @@ mod tests { id } + #[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 7cf665b1b..db86464e8 100644 --- a/crates/buzz-relay/Cargo.toml +++ b/crates/buzz-relay/Cargo.toml @@ -72,6 +72,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 df68ec270..06510b44b 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; @@ -46,31 +46,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 @@ -124,9 +130,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" ); @@ -863,15 +872,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 @@ -888,7 +892,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"); } } @@ -1060,22 +1071,182 @@ 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. +/// 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); + } + } + } +} + +/// Refresh the exporter recency for legacy event-driven gauges without changing +/// their values. /// -/// All DB-derived gauges use absolute SET (not increment), so they self-heal -/// after a missed tick or a pod restart. +/// `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. +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); + refresh_legacy_active_gauge_recency(); + + 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 @@ -1096,19 +1267,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 --- @@ -1129,7 +1296,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; } @@ -1171,7 +1338,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; } @@ -1198,7 +1365,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; } @@ -1238,7 +1405,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; } @@ -1284,7 +1451,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; } @@ -1311,7 +1478,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; } @@ -1344,7 +1511,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; } @@ -1385,7 +1552,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; } @@ -1399,79 +1566,23 @@ 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 super::buzz_auto_migrate_enabled; + use std::collections::HashSet; + + use super::{ + buzz_auto_migrate_enabled, dropped_in_memory_keys, idle_timeout_secs, + refresh_legacy_active_gauge_recency, EmissionScope, InMemoryMetricKey, + }; + use metrics::GaugeFn; + use metrics_util::{ + debugging::DebugValue, + registry::{GenerationalAtomicStorage, Registry}, + }; + use uuid::Uuid; #[test] fn buzz_auto_migrate_is_opt_in() { @@ -1487,4 +1598,77 @@ 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_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); + 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 954e78434..5c0b40b49 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 9831e8eee..68f0fea3c 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();