feat(relay): gate usage metrics behind stable leader (#1814)

Signed-off-by: Will Pfleger <pfleger.will@gmail.com>
Co-authored-by: npub1mn7jgtj4w2pd0g0zeuhxsa6jy6p0rewxz4kujt98my82ahfmp72sxjexk7 <dcfd242e557282d7a1e2cf2e6877522682f1e5c6156dc92ca7d90eaedd3b0f95@sprout-oss.stage.blox.sqprod.co>

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"

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"

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,27 @@ 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.
+    ///
+    /// 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 {
+        tokio::time::timeout(std::time::Duration::from_secs(5), self.connection.ping())
+            .await
+            .is_ok_and(|r| r.is_ok())
+    }
+}
+
 /// Configuration for the Postgres connection pool.
 #[derive(Debug, Clone)]
 pub struct DbConfig {
@@ -343,6 +364,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<Option<UsageMetricsLeader>> {
+        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<i64> {
         usage::community_count(&self.pool).await
@@ -3510,6 +3555,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 +4209,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() {

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"]

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::<u64>().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,9 +955,21 @@ 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");
                 }
+                metrics::gauge!("buzz_usage_poller_is_leader").set(if leader.is_some() {
+                    1.0
+                } else {
+                    0.0
+                });
             }
         });
     }
@@ -1160,22 +1176,190 @@ 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.
+/// 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<u64>, 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.
 ///
-/// 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.
+/// `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<Uuid, String>>,
+    previously_emitted: &mut HashSet<InMemoryMetricKey>,
+) {
+    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::<u64>();
+    let total_subscriptions = subscriptions.values().sum::<u64>();
+
+    metrics::gauge!("buzz_total_ws_connections").set(total_connections as f64);
+    metrics::gauge!("buzz_total_users_online_pod").set(users_online.values().sum::<u64>() 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, &current) {
+        key.set(0.0);
+    }
+    *previously_emitted = current;
+}
+
+fn dropped_in_memory_keys(
+    previously_emitted: &HashSet<InMemoryMetricKey>,
+    current: &HashSet<InMemoryMetricKey>,
+) -> Vec<InMemoryMetricKey> {
+    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<buzz_db::UsageMetricsLeader>,
+    emitted_in_memory: &mut HashSet<InMemoryMetricKey>,
 ) -> 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<Uuid, String> = hosts.into_iter().map(|c| (c.id, c.host)).collect();
+    let host_map: HashMap<Uuid, String> = match state.db.usage_community_hosts().await {
+        Ok(hosts) => hosts
+            .into_iter()
+            .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());
+        }
+    };
+    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() {
+        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(())
+}
+
+/// 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<Uuid, String>,
+) -> 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 +1380,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<Uuid> = match per_community_mode {
-        PerCommunityMode::All => host_map.keys().copied().collect(),
-        PerCommunityMode::Off => std::collections::HashSet::new(),
-    };
+    let active_set: HashSet<Uuid> = host_map
+        .keys()
+        .filter(|id| emission_scope.allows(id))
+        .copied()
+        .collect();
 
     // --- Publish phase: emit all metrics now that every query succeeded ---
 
@@ -1229,7 +1409,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 +1451,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 +1478,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 +1518,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 +1564,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 +1591,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 +1624,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 +1665,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 +1679,28 @@ 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,
+        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)]
     async fn periodic_loop_exits_immediately_on_cancellation() {
@@ -1617,4 +1741,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, &current),
+            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::<std::collections::HashMap<_, _>>();
+
+        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);
+    }
 }

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()),

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();