From b029457384c59d58052044ed3cecd9ca1344d2ec Mon Sep 17 00:00:00 2001 From: npub1mn7jgtj4w2pd0g0zeuhxsa6jy6p0rewxz4kujt98my82ahfmp72sxjexk7 Date: Fri, 10 Jul 2026 14:36:13 -0400 Subject: [PATCH 1/6] feat(relay): per-community usage metrics Emit per-community usage gauges and counter tags to Datadog: - New usage poller task (BUZZ_USAGE_METRICS_INTERVAL_SECS, default 60, min 5): DB-polled gauges for users, channels, messages, relay members, workflows, git repos, DAU/WAU/MAU, and active channels, all grouped by community; plus in-memory snapshot gauges for ws_connections, users_online, and subscriptions per community - Write-path community tag: events_received/stored, ws_connections, media_uploads, workflow_runs now carry a 'community' label - New counters: buzz_users_created_total and buzz_channels_created_total tagged by community, incremented only on actual first insert/creation - Fix kind 44200 missing from bounded_kind_label (was collapsing to 'other') ensure_user() now returns Result (true = new insert, false = conflict DO NOTHING) so callers can count new users without a separate query. Co-authored-by: Will Pfleger Signed-off-by: Will Pfleger --- crates/buzz-db/src/lib.rs | 66 ++- crates/buzz-db/src/usage.rs | 538 ++++++++++++++++++ crates/buzz-db/src/user.rs | 10 +- crates/buzz-relay/src/api/media.rs | 7 +- crates/buzz-relay/src/connection.rs | 6 +- crates/buzz-relay/src/handlers/auth.rs | 28 +- .../src/handlers/command_executor.rs | 14 +- crates/buzz-relay/src/handlers/event.rs | 24 +- crates/buzz-relay/src/handlers/ingest.rs | 6 + .../buzz-relay/src/handlers/side_effects.rs | 44 +- crates/buzz-relay/src/main.rs | 165 ++++++ crates/buzz-relay/src/state.rs | 36 ++ crates/buzz-relay/src/subscription.rs | 60 ++ 13 files changed, 980 insertions(+), 24 deletions(-) create mode 100644 crates/buzz-db/src/usage.rs diff --git a/crates/buzz-db/src/lib.rs b/crates/buzz-db/src/lib.rs index dd00ab1ff4..0e747fbcf9 100644 --- a/crates/buzz-db/src/lib.rs +++ b/crates/buzz-db/src/lib.rs @@ -37,6 +37,8 @@ pub mod reaction; pub mod relay_members; /// Thread metadata persistence. pub mod thread; +/// Per-community usage rollup queries for Prometheus gauges. +pub mod usage; /// User profile persistence. pub mod user; /// Workflow, run, and approval persistence. @@ -286,6 +288,64 @@ impl Db { } } + /// Return total number of communities on this relay. + pub async fn usage_community_count(&self) -> Result { + usage::community_count(&self.pool).await + } + + /// Return per-community user counts split by human/agent. + pub async fn usage_user_counts(&self) -> Result> { + usage::user_counts(&self.pool).await + } + + /// Return per-community channel counts by type. + pub async fn usage_channel_counts(&self) -> Result> { + usage::channel_counts(&self.pool).await + } + + /// Return per-community kind=9 message counts. + pub async fn usage_message_counts(&self) -> Result> { + usage::message_counts(&self.pool).await + } + + /// Return per-community relay-member counts by role. + pub async fn usage_relay_member_counts(&self) -> Result> { + usage::relay_member_counts(&self.pool).await + } + + /// Return per-community workflow counts by status. + pub async fn usage_workflow_counts(&self) -> Result> { + usage::workflow_counts(&self.pool).await + } + + /// Return per-community git-repo counts. + pub async fn usage_git_repo_counts(&self) -> Result> { + usage::git_repo_counts(&self.pool).await + } + + /// Return per-community distinct active-user counts for a given SQL interval. + /// + /// `interval_sql` must be a trusted literal such as `"1 day"` or `"7 days"`. + pub async fn usage_active_user_counts( + &self, + interval_sql: &'static str, + ) -> Result> { + usage::active_user_counts(&self.pool, interval_sql).await + } + + /// Return per-community active-channel counts for a given SQL interval. + pub async fn usage_active_channel_counts( + &self, + interval_sql: &'static str, + ) -> Result> { + usage::active_channel_counts(&self.pool, interval_sql).await + } + + /// Return all community id → host mappings. + pub async fn usage_community_hosts(&self) -> Result> { + usage::community_hosts(&self.pool).await + } + /// Begin a database transaction for atomic multi-statement operations. /// /// Returns a `'static` transaction because `PgPool` is `Arc`-backed internally. @@ -1154,7 +1214,11 @@ impl Db { } /// Ensure a user record exists (upsert). - pub async fn ensure_user(&self, community_id: CommunityId, pubkey: &[u8]) -> Result<()> { + /// + /// Returns `true` if a new row was inserted (first time), `false` if it + /// already existed. Callers use the `true` return to increment + /// `buzz_users_created_total`. + pub async fn ensure_user(&self, community_id: CommunityId, pubkey: &[u8]) -> Result { user::ensure_user(&self.pool, community_id, pubkey).await } diff --git a/crates/buzz-db/src/usage.rs b/crates/buzz-db/src/usage.rs new file mode 100644 index 0000000000..0bed64c03b --- /dev/null +++ b/crates/buzz-db/src/usage.rs @@ -0,0 +1,538 @@ +//! Per-community usage rollup queries for Prometheus gauges. +//! +//! All queries use `GROUP BY community_id` against indexed columns — +//! no per-community loops, no full-table scans. +//! +//! Returned structs are plain data; the caller (relay poller) maps them +//! to Prometheus labels and calls `metrics::gauge!(...).set(...)`. + +use crate::error::Result; +use sqlx::PgPool; +use uuid::Uuid; + +/// Total number of communities registered on this relay. +pub async fn community_count(pool: &PgPool) -> Result { + let row = sqlx::query_scalar::<_, i64>("SELECT COUNT(*) FROM communities") + .fetch_one(pool) + .await?; + Ok(row) +} + +/// Per-community user counts split by human/agent. +#[derive(Debug)] +pub struct CommunityUserCounts { + /// The UUID of the community. + pub community_id: Uuid, + /// Number of active human users (no `agent_owner_pubkey`). + pub human: i64, + /// Number of active agent users (`agent_owner_pubkey IS NOT NULL`). + pub agent: i64, +} + +/// Return active (non-deactivated) user counts per community, split by type. +/// +/// Agent discriminator: `agent_owner_pubkey IS NOT NULL`. +pub async fn user_counts(pool: &PgPool) -> Result> { + // Single GROUP BY query; two conditional SUMs avoid two round-trips. + let rows = sqlx::query_as::<_, (Uuid, i64, i64)>( + r#" + SELECT + community_id, + COUNT(*) FILTER (WHERE agent_owner_pubkey IS NULL) AS human, + COUNT(*) FILTER (WHERE agent_owner_pubkey IS NOT NULL) AS agent + FROM users + WHERE deactivated_at IS NULL + GROUP BY community_id + "#, + ) + .fetch_all(pool) + .await?; + + Ok(rows + .into_iter() + .map(|(community_id, human, agent)| CommunityUserCounts { + community_id, + human, + agent, + }) + .collect()) +} + +/// Per-community channel counts by type. +#[derive(Debug)] +pub struct CommunityChannelCount { + /// The UUID of the community. + pub community_id: Uuid, + /// Channel type string (e.g. `"stream"`, `"dm"`, `"forum"`, `"workflow"`). + pub channel_type: String, + /// Number of non-deleted channels of this type. + pub count: i64, +} + +/// Return non-deleted channel counts per community per type. +pub async fn channel_counts(pool: &PgPool) -> Result> { + let rows = sqlx::query_as::<_, (Uuid, String, i64)>( + r#" + SELECT community_id, channel_type::text, COUNT(*) AS count + FROM channels + WHERE deleted_at IS NULL + GROUP BY community_id, channel_type + "#, + ) + .fetch_all(pool) + .await?; + + Ok(rows + .into_iter() + .map( + |(community_id, channel_type, count)| CommunityChannelCount { + community_id, + channel_type, + count, + }, + ) + .collect()) +} + +/// Per-community message (kind=9) count. +#[derive(Debug)] +pub struct CommunityMessageCount { + /// The UUID of the community. + pub community_id: Uuid, + /// Number of stored non-deleted kind=9 events. + pub count: i64, +} + +/// Return non-deleted kind=9 event counts per community. +pub async fn message_counts(pool: &PgPool) -> Result> { + let rows = sqlx::query_as::<_, (Uuid, i64)>( + r#" + SELECT community_id, COUNT(*) AS count + FROM events + WHERE kind = 9 AND deleted_at IS NULL + GROUP BY community_id + "#, + ) + .fetch_all(pool) + .await?; + + Ok(rows + .into_iter() + .map(|(community_id, count)| CommunityMessageCount { + community_id, + count, + }) + .collect()) +} + +/// Per-community relay-member counts by role. +#[derive(Debug)] +pub struct CommunityMemberCount { + /// The UUID of the community. + pub community_id: Uuid, + /// Role string (e.g. `"owner"`, `"admin"`, `"member"`). + pub role: String, + /// Number of members with this role. + pub count: i64, +} + +/// Return relay-member counts per community per role. +pub async fn relay_member_counts(pool: &PgPool) -> Result> { + let rows = sqlx::query_as::<_, (Uuid, String, i64)>( + r#" + SELECT community_id, role::text, COUNT(*) AS count + FROM relay_members + GROUP BY community_id, role + "#, + ) + .fetch_all(pool) + .await?; + + Ok(rows + .into_iter() + .map(|(community_id, role, count)| CommunityMemberCount { + community_id, + role, + count, + }) + .collect()) +} + +/// Per-community workflow counts by status. +#[derive(Debug)] +pub struct CommunityWorkflowCount { + /// The UUID of the community. + pub community_id: Uuid, + /// Workflow status string (e.g. `"active"`, `"inactive"`). + pub status: String, + /// Number of workflows in this status. + pub count: i64, +} + +/// Return workflow counts per community per status. +pub async fn workflow_counts(pool: &PgPool) -> Result> { + let rows = sqlx::query_as::<_, (Uuid, String, i64)>( + r#" + SELECT community_id, status::text, COUNT(*) AS count + FROM workflows + GROUP BY community_id, status + "#, + ) + .fetch_all(pool) + .await?; + + Ok(rows + .into_iter() + .map(|(community_id, status, count)| CommunityWorkflowCount { + community_id, + status, + count, + }) + .collect()) +} + +/// Per-community git-repo count. +#[derive(Debug)] +pub struct CommunityGitRepoCount { + /// The UUID of the community. + pub community_id: Uuid, + /// Number of git repos registered for this community. + pub count: i64, +} + +/// Return git repo counts per community. +pub async fn git_repo_counts(pool: &PgPool) -> Result> { + let rows = sqlx::query_as::<_, (Uuid, i64)>( + r#" + SELECT community_id, COUNT(*) AS count + FROM git_repos + GROUP BY community_id + "#, + ) + .fetch_all(pool) + .await?; + + Ok(rows + .into_iter() + .map(|(community_id, count)| CommunityGitRepoCount { + community_id, + count, + }) + .collect()) +} + +/// Per-community active-user counts for a given window (e.g. 1d, 7d, 30d), +/// split by human/agent. +#[derive(Debug)] +pub struct CommunityActiveUsers { + /// The UUID of the community. + pub community_id: Uuid, + /// Distinct human pubkeys that published at least one event in the window. + pub human: i64, + /// Distinct agent pubkeys that published at least one event in the window. + pub agent: i64, +} + +/// Return distinct-publisher counts for events in `[now - interval, now]` +/// per community, split by human/agent. +/// +/// `interval_sql` must be a trusted literal (e.g. `"1 day"`, `"7 days"`) — +/// it is not user-controlled; callers are in the relay process. +pub async fn active_user_counts( + pool: &PgPool, + interval_sql: &'static str, +) -> Result> { + // JOIN users so we can discriminate human vs. agent. Events are authored + // by pubkeys; the users.agent_owner_pubkey column is the discriminator. + // We use encode(e.pubkey, 'hex') and encode(u.pubkey, 'hex') to join on + // bytea — the columns are the same type so direct equality works. + let sql = format!( + r#" + SELECT + e.community_id, + COUNT(DISTINCT e.pubkey) FILTER (WHERE u.agent_owner_pubkey IS NULL) AS human, + COUNT(DISTINCT e.pubkey) FILTER (WHERE u.agent_owner_pubkey IS NOT NULL) AS agent + FROM events e + LEFT JOIN users u + ON u.community_id = e.community_id AND u.pubkey = e.pubkey + WHERE e.created_at >= NOW() - INTERVAL '{interval_sql}' + AND e.deleted_at IS NULL + GROUP BY e.community_id + "# + ); + let rows = sqlx::query_as::<_, (Uuid, i64, i64)>(sqlx::AssertSqlSafe(sql)) + .fetch_all(pool) + .await?; + + Ok(rows + .into_iter() + .map(|(community_id, human, agent)| CommunityActiveUsers { + community_id, + human, + agent, + }) + .collect()) +} + +/// Per-community active-channel counts for a given window. +#[derive(Debug)] +pub struct CommunityActiveChannels { + /// The UUID of the community. + pub community_id: Uuid, + /// Distinct channel IDs with ≥1 kind=9 message in the window. + pub count: i64, +} + +/// Return distinct channel IDs with ≥1 kind=9 message in `[now - interval, now]`. +pub async fn active_channel_counts( + pool: &PgPool, + interval_sql: &'static str, +) -> Result> { + let sql = format!( + r#" + SELECT community_id, COUNT(DISTINCT channel_id) AS count + FROM events + WHERE kind = 9 + AND channel_id IS NOT NULL + AND created_at >= NOW() - INTERVAL '{interval_sql}' + AND deleted_at IS NULL + GROUP BY community_id + "# + ); + let rows = sqlx::query_as::<_, (Uuid, i64)>(sqlx::AssertSqlSafe(sql)) + .fetch_all(pool) + .await?; + + Ok(rows + .into_iter() + .map(|(community_id, count)| CommunityActiveChannels { + community_id, + count, + }) + .collect()) +} + +/// Mapping from community UUID to host string, used by the poller to resolve +/// Prometheus label values. +#[derive(Debug)] +pub struct CommunityHost { + /// The UUID of the community. + pub id: Uuid, + /// The canonical host string for this community (used as the Prometheus label value). + pub host: String, +} + +/// Fetch all community id → host mappings in one query. +pub async fn community_hosts(pool: &PgPool) -> Result> { + let rows = sqlx::query_as::<_, (Uuid, String)>("SELECT id, host FROM communities") + .fetch_all(pool) + .await?; + Ok(rows + .into_iter() + .map(|(id, host)| CommunityHost { id, host }) + .collect()) +} + +#[cfg(test)] +mod tests { + use super::*; + use buzz_core::CommunityId; + use nostr::Keys; + use sqlx::PgPool; + + const TEST_DB_URL: &str = "postgres://buzz:buzz_dev@localhost:5432/buzz"; + + async fn get_pool() -> PgPool { + PgPool::connect(TEST_DB_URL) + .await + .expect("connect to test DB") + } + + fn random_pubkey() -> Vec { + Keys::generate().public_key().to_bytes().to_vec() + } + + async fn make_community(pool: &PgPool) -> (Uuid, CommunityId, String) { + let id = uuid::Uuid::new_v4(); + let host = format!("usage-test-{}.example", id.simple()); + sqlx::query("INSERT INTO communities (id, host) VALUES ($1, $2)") + .bind(id) + .bind(&host) + .execute(pool) + .await + .expect("insert test community"); + (id, CommunityId::from_uuid(id), host) + } + + async fn insert_user(pool: &PgPool, community_id: Uuid, pubkey: &[u8], is_agent: bool) { + if is_agent { + let owner = random_pubkey(); + // Insert owner first (FK constraint). + sqlx::query( + "INSERT INTO users (community_id, pubkey) VALUES ($1, $2) ON CONFLICT DO NOTHING", + ) + .bind(community_id) + .bind(&owner) + .execute(pool) + .await + .expect("insert owner"); + sqlx::query( + "INSERT INTO users (community_id, pubkey, agent_owner_pubkey) VALUES ($1, $2, $3) ON CONFLICT DO NOTHING", + ) + .bind(community_id) + .bind(pubkey) + .bind(&owner) + .execute(pool) + .await + .expect("insert agent user"); + } else { + sqlx::query( + "INSERT INTO users (community_id, pubkey) VALUES ($1, $2) ON CONFLICT DO NOTHING", + ) + .bind(community_id) + .bind(pubkey) + .execute(pool) + .await + .expect("insert human user"); + } + } + + /// user_counts returns correct human/agent split and is scoped per community. + #[tokio::test] + #[ignore = "requires Postgres"] + async fn test_user_counts_scoped_per_community() { + let pool = get_pool().await; + let (comm_a_uuid, _, _) = make_community(&pool).await; + let (comm_b_uuid, _, _) = make_community(&pool).await; + + // Community A: 2 human, 1 agent + insert_user(&pool, comm_a_uuid, &random_pubkey(), false).await; + insert_user(&pool, comm_a_uuid, &random_pubkey(), false).await; + insert_user(&pool, comm_a_uuid, &random_pubkey(), true).await; + + // Community B: 0 human, 1 agent + insert_user(&pool, comm_b_uuid, &random_pubkey(), true).await; + + let counts = user_counts(&pool).await.expect("user_counts"); + + let a = counts.iter().find(|r| r.community_id == comm_a_uuid); + let b = counts.iter().find(|r| r.community_id == comm_b_uuid); + + let a = a.expect("community A row"); + assert_eq!(a.human, 2, "community A: 2 humans"); + assert_eq!(a.agent, 1, "community A: 1 agent"); + + let b = b.expect("community B row"); + assert_eq!(b.human, 0, "community B: 0 humans"); + assert_eq!(b.agent, 1, "community B: 1 agent"); + } + + /// Deactivated users are excluded from user_counts. + #[tokio::test] + #[ignore = "requires Postgres"] + async fn test_user_counts_excludes_deactivated() { + let pool = get_pool().await; + let (comm_uuid, _, _) = make_community(&pool).await; + + let active_pk = random_pubkey(); + let deactivated_pk = random_pubkey(); + + insert_user(&pool, comm_uuid, &active_pk, false).await; + insert_user(&pool, comm_uuid, &deactivated_pk, false).await; + // Deactivate the second user. + sqlx::query( + "UPDATE users SET deactivated_at = NOW() WHERE community_id = $1 AND pubkey = $2", + ) + .bind(comm_uuid) + .bind(&deactivated_pk) + .execute(&pool) + .await + .expect("deactivate user"); + + let counts = user_counts(&pool).await.expect("user_counts"); + let row = counts + .iter() + .find(|r| r.community_id == comm_uuid) + .expect("row"); + assert_eq!(row.human, 1, "only active user counted"); + } + + /// channel_counts is scoped per community and excludes deleted channels. + #[tokio::test] + #[ignore = "requires Postgres"] + async fn test_channel_counts_scoped_and_excludes_deleted() { + let pool = get_pool().await; + let (comm_uuid, comm_id, _) = make_community(&pool).await; + let owner = random_pubkey(); + insert_user(&pool, comm_uuid, &owner, false).await; + + // Insert a stream and a DM channel. + sqlx::query( + "INSERT INTO channels (id, community_id, name, channel_type, visibility, owner_pubkey) + VALUES ($1, $2, 'test-stream', 'stream', 'open', $3)", + ) + .bind(uuid::Uuid::new_v4()) + .bind(comm_uuid) + .bind(&owner) + .execute(&pool) + .await + .expect("insert stream channel"); + + let dm_id = uuid::Uuid::new_v4(); + sqlx::query( + "INSERT INTO channels (id, community_id, name, channel_type, visibility, owner_pubkey) + VALUES ($1, $2, 'test-dm', 'dm', 'private', $3)", + ) + .bind(dm_id) + .bind(comm_uuid) + .bind(&owner) + .execute(&pool) + .await + .expect("insert dm channel"); + + // Soft-delete the DM. + sqlx::query("UPDATE channels SET deleted_at = NOW() WHERE id = $1") + .bind(dm_id) + .execute(&pool) + .await + .expect("delete channel"); + + // Use comm_id to satisfy unused import warning. + let _ = comm_id; + + let counts = channel_counts(&pool).await.expect("channel_counts"); + let comm_counts: Vec<_> = counts + .iter() + .filter(|r| r.community_id == comm_uuid) + .collect(); + + // Only the stream channel should be counted. + assert_eq!(comm_counts.len(), 1); + assert_eq!(comm_counts[0].channel_type, "stream"); + assert_eq!(comm_counts[0].count, 1); + } + + /// community_hosts returns id → host mapping. + #[tokio::test] + #[ignore = "requires Postgres"] + async fn test_community_hosts_returns_mapping() { + let pool = get_pool().await; + let (id, _, host) = make_community(&pool).await; + + let hosts = community_hosts(&pool).await.expect("community_hosts"); + let found = hosts.iter().find(|h| h.id == id); + assert!(found.is_some(), "inserted community not found"); + assert_eq!(found.unwrap().host, host); + } + + /// community_count reflects newly inserted communities. + #[tokio::test] + #[ignore = "requires Postgres"] + async fn test_community_count_increases() { + let pool = get_pool().await; + let before = community_count(&pool).await.expect("count before"); + make_community(&pool).await; + let after = community_count(&pool).await.expect("count after"); + assert!(after > before, "count should increase after insert"); + } +} diff --git a/crates/buzz-db/src/user.rs b/crates/buzz-db/src/user.rs index a4a5a9d7e8..066fb5f5c0 100644 --- a/crates/buzz-db/src/user.rs +++ b/crates/buzz-db/src/user.rs @@ -35,8 +35,12 @@ pub struct UserSearchProfile { /// Ensure a user record exists for the given pubkey (upsert). /// Creates with minimal fields if not present; no-op if already exists. -pub async fn ensure_user(pool: &PgPool, community_id: CommunityId, pubkey: &[u8]) -> Result<()> { - sqlx::query( +/// +/// Returns `true` if a new row was inserted, `false` if the user already existed. +/// The `true` case is the reliable signal for "user was just registered" — used +/// by callers to increment `buzz_users_created_total`. +pub async fn ensure_user(pool: &PgPool, community_id: CommunityId, pubkey: &[u8]) -> Result { + let result = sqlx::query( r#" INSERT INTO users (community_id, pubkey) VALUES ($1, $2) @@ -47,7 +51,7 @@ pub async fn ensure_user(pool: &PgPool, community_id: CommunityId, pubkey: &[u8] .bind(pubkey) .execute(pool) .await?; - Ok(()) + Ok(result.rows_affected() == 1) } /// Get a single user record by pubkey. diff --git a/crates/buzz-relay/src/api/media.rs b/crates/buzz-relay/src/api/media.rs index c5498cc5ec..b7b6db369e 100644 --- a/crates/buzz-relay/src/api/media.rs +++ b/crates/buzz-relay/src/api/media.rs @@ -358,7 +358,12 @@ pub async fn upload_blob( } _ => "other", }; - metrics::counter!("buzz_media_uploads_total", "mime" => mime_label.to_owned()).increment(1); + metrics::counter!( + "buzz_media_uploads_total", + "mime" => mime_label.to_owned(), + "community" => auth.tenant.host().to_owned() + ) + .increment(1); // Audit via bounded channel — same pattern as event audit. let desc = descriptor.clone(); diff --git a/crates/buzz-relay/src/connection.rs b/crates/buzz-relay/src/connection.rs index 0a72f76dc0..294c3b1b4e 100644 --- a/crates/buzz-relay/src/connection.rs +++ b/crates/buzz-relay/src/connection.rs @@ -157,7 +157,11 @@ pub async fn handle_connection( }); info!(conn_id = %conn_id, addr = %addr, "WebSocket connection established"); - metrics::counter!("buzz_ws_connections_total").increment(1); + metrics::counter!( + "buzz_ws_connections_total", + "community" => conn.tenant.host().to_owned() + ) + .increment(1); let challenge_msg = RelayMessage::auth_challenge(&challenge); if tx diff --git a/crates/buzz-relay/src/handlers/auth.rs b/crates/buzz-relay/src/handlers/auth.rs index a5d6be50c7..dcb49df6a2 100644 --- a/crates/buzz-relay/src/handlers/auth.rs +++ b/crates/buzz-relay/src/handlers/auth.rs @@ -257,19 +257,39 @@ pub async fn handle_auth(event: nostr::Event, conn: Arc, state: if let Some(owner) = nip_oa_owner { // Ensure both agent and owner have users rows (BYO agents may not, // and agent_owner_pubkey has a FK constraint to users.pubkey). - if let Err(e) = state + match state .db .ensure_user(conn.tenant.community(), pubkey.as_bytes()) .await { - warn!(conn_id = %conn_id, error = %e, "ensure_user(agent) failed during NIP-OA backfill"); + Ok(true) => { + metrics::counter!( + "buzz_users_created_total", + "community" => conn.tenant.host().to_owned() + ) + .increment(1); + } + Ok(false) => {} + Err(e) => { + warn!(conn_id = %conn_id, error = %e, "ensure_user(agent) failed during NIP-OA backfill"); + } } - if let Err(e) = state + match state .db .ensure_user(conn.tenant.community(), owner.as_bytes()) .await { - warn!(conn_id = %conn_id, error = %e, "ensure_user(owner) failed during NIP-OA backfill"); + Ok(true) => { + metrics::counter!( + "buzz_users_created_total", + "community" => conn.tenant.host().to_owned() + ) + .increment(1); + } + Ok(false) => {} + Err(e) => { + warn!(conn_id = %conn_id, error = %e, "ensure_user(owner) failed during NIP-OA backfill"); + } } // Idempotent backfill: record agent→owner in DB so cross-connection diff --git a/crates/buzz-relay/src/handlers/command_executor.rs b/crates/buzz-relay/src/handlers/command_executor.rs index e3b19770b6..4cb88a570a 100644 --- a/crates/buzz-relay/src/handlers/command_executor.rs +++ b/crates/buzz-relay/src/handlers/command_executor.rs @@ -42,12 +42,22 @@ pub async fn handle_command( // Ensure the authenticated user exists in the users table (foreign key requirement). // The old REST handlers did this via extract_auth_context; command executor must do it explicitly. let pubkey_bytes = auth.pubkey().to_bytes().to_vec(); - if let Err(e) = state + match state .db .ensure_user(tenant.community(), &pubkey_bytes) .await { - tracing::warn!("command_executor: ensure_user failed: {e}"); + Ok(true) => { + metrics::counter!( + "buzz_users_created_total", + "community" => tenant.host().to_owned() + ) + .increment(1); + } + Ok(false) => {} + Err(e) => { + tracing::warn!("command_executor: ensure_user failed: {e}"); + } } let kind = event.kind.as_u16() as u32; diff --git a/crates/buzz-relay/src/handlers/event.rs b/crates/buzz-relay/src/handlers/event.rs index 31911e4c73..ad70edcf4a 100644 --- a/crates/buzz-relay/src/handlers/event.rs +++ b/crates/buzz-relay/src/handlers/event.rs @@ -43,6 +43,7 @@ fn bounded_kind_label(kind: u32) -> String { 41001 | 41010..=41012 => kind.to_string(), 43001..=43006 => kind.to_string(), 44100..=44101 => kind.to_string(), + 44200 => kind.to_string(), 45001..=45003 => kind.to_string(), 46001..=46012 | 46020 | 46030..=46031 => kind.to_string(), 48001 | 48100..=48103 | 48106 => kind.to_string(), @@ -509,6 +510,7 @@ async fn dispatch_persistent_event_inner( let workflow_engine = Arc::clone(&state.workflow_engine); let workflow_event = stored_event.clone(); let trigger_kind = kind_u32.to_string(); + let workflow_community_host = tenant.host().to_owned(); // The event was stored under `tenant.community()`; `StoredEvent` does // not carry the community, so pass it explicitly. The same channel UUID // can exist in another community — scoping the workflow lookup to this @@ -522,8 +524,12 @@ async fn dispatch_persistent_event_inner( { tracing::error!(event_id = ?workflow_event.event.id, "Workflow trigger failed: {e}"); } else { - metrics::counter!("buzz_workflow_runs_total", "trigger" => trigger_kind) - .increment(1); + metrics::counter!( + "buzz_workflow_runs_total", + "trigger" => trigger_kind, + "community" => workflow_community_host + ) + .increment(1); } }); } @@ -585,7 +591,12 @@ pub async fn handle_event(event: Event, conn: Arc, state: Arc kind_str.clone()).increment(1); + metrics::counter!( + "buzz_events_received_total", + "kind" => kind_str.clone(), + "community" => conn.tenant.host().to_owned() + ) + .increment(1); let (conn_id, pubkey_bytes, auth_pubkey, scopes, channel_ids) = { let auth = conn.auth_state.read().await; @@ -684,7 +695,12 @@ pub async fn handle_event(event: Event, conn: Arc, state: Arc { if result.accepted { - metrics::counter!("buzz_events_stored_total", "kind" => kind_str).increment(1); + metrics::counter!( + "buzz_events_stored_total", + "kind" => kind_str, + "community" => conn.tenant.host().to_owned() + ) + .increment(1); info!( event_id = %result.event_id, kind = kind_u32, diff --git a/crates/buzz-relay/src/handlers/ingest.rs b/crates/buzz-relay/src/handlers/ingest.rs index adfdaf2b20..69457e2dff 100644 --- a/crates/buzz-relay/src/handlers/ingest.rs +++ b/crates/buzz-relay/src/handlers/ingest.rs @@ -1997,6 +1997,12 @@ async fn ingest_event_inner( }); } pre_created_channel = Some(client_uuid); + metrics::counter!( + "buzz_channels_created_total", + "community" => tenant.host().to_owned(), + "type" => channel_type.to_string() + ) + .increment(1); } } diff --git a/crates/buzz-relay/src/handlers/side_effects.rs b/crates/buzz-relay/src/handlers/side_effects.rs index c64ae28530..a6a06ebd59 100644 --- a/crates/buzz-relay/src/handlers/side_effects.rs +++ b/crates/buzz-relay/src/handlers/side_effects.rs @@ -1073,10 +1073,17 @@ async fn handle_agent_profile( .ok_or_else(|| anyhow::anyhow!("kind:10100 missing channel_add_policy field"))?; let pubkey_bytes = event.pubkey.to_bytes().to_vec(); - state + if state .db .ensure_user(tenant.community(), &pubkey_bytes) - .await?; + .await? + { + metrics::counter!( + "buzz_users_created_total", + "community" => tenant.host().to_owned() + ) + .increment(1); + } state .db .set_channel_add_policy(tenant.community(), &pubkey_bytes, policy) @@ -1124,10 +1131,17 @@ async fn handle_kind0_profile( let pubkey_bytes = event.pubkey.to_bytes().to_vec(); - state + if state .db .ensure_user(tenant.community(), &pubkey_bytes) - .await?; + .await? + { + metrics::counter!( + "buzz_users_created_total", + "community" => tenant.host().to_owned() + ) + .increment(1); + } // Pass all fields as Some — empty string clears the field in the DB. // This ensures kind:0 is treated as absolute state, not a partial update. @@ -1659,7 +1673,7 @@ async fn handle_create_group( Err(_) => { // Channel not found — shouldn't happen (ingest_event pre-created it), // but fall back to creation to stay resilient. - state + let ch = state .db .create_channel( tenant.community(), @@ -1670,11 +1684,18 @@ async fn handle_create_group( &actor_bytes, ttl_seconds, ) - .await? + .await?; + metrics::counter!( + "buzz_channels_created_total", + "community" => tenant.host().to_owned(), + "type" => channel_type.to_string() + ) + .increment(1); + ch } } } else { - state + let ch = state .db .create_channel( tenant.community(), @@ -1685,7 +1706,14 @@ async fn handle_create_group( &actor_bytes, ttl_seconds, ) - .await? + .await?; + metrics::counter!( + "buzz_channels_created_total", + "community" => tenant.host().to_owned(), + "type" => channel_type.to_string() + ) + .increment(1); + ch }; // Creator becomes owner — evict any stale negative membership lookup. diff --git a/crates/buzz-relay/src/main.rs b/crates/buzz-relay/src/main.rs index e56e9dbffd..cba53bf9bb 100644 --- a/crates/buzz-relay/src/main.rs +++ b/crates/buzz-relay/src/main.rs @@ -3,6 +3,7 @@ use std::sync::Arc; use tracing::{error, info, warn}; use tracing_subscriber::{fmt, prelude::*, EnvFilter}; +use uuid::Uuid; use buzz_audit::AuditService; use buzz_auth::AuthService; @@ -806,6 +807,34 @@ async fn main() -> anyhow::Result<()> { }); } + // Usage metrics: periodic background task polling per-community stats. + // + // DB-derived gauges (users, channels, messages, members, workflows, git + // repos, active users/channels) are SET from GROUP BY queries — one per + // tick. In-memory gauges (ws_connections, subscriptions, users_online) + // are snapshotted from live in-memory state. Both avoid inc/dec drift. + // + // Multi-pod semantics: + // DB-derived: all pods export the same value → dashboard uses max() + // In-memory: each pod exports its partition → dashboard uses sum() + { + let usage_state = Arc::clone(&state); + let interval_secs = std::env::var("BUZZ_USAGE_METRICS_INTERVAL_SECS") + .ok() + .and_then(|v| v.parse::().ok()) + .unwrap_or(60) + .max(5); // 5s minimum: cheaper than pool poller's 1s minimum + tokio::spawn(async move { + let mut interval = tokio::time::interval(std::time::Duration::from_secs(interval_secs)); + loop { + interval.tick().await; + if let Err(e) = run_usage_metrics_tick(&usage_state).await { + error!(error = %e, "Usage metrics tick failed — skipping"); + } + } + }); + } + serve(router, health_router, Arc::clone(&state)).await?; // Signal the audit worker to stop accepting, flush buffered entries, and @@ -971,6 +1000,142 @@ 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. +async fn run_usage_metrics_tick(state: &AppState) -> anyhow::Result<()> { + // --- community id → host label map (one query, cached for this tick) --- + let hosts = state.db.usage_community_hosts().await?; + let host_map: std::collections::HashMap = + hosts.into_iter().map(|c| (c.id, c.host)).collect(); + + // Helper: resolve community UUID → host label string. + let host = + |id: Uuid| -> String { host_map.get(&id).cloned().unwrap_or_else(|| id.to_string()) }; + + // --- A. Adoption stocks (DB-polled) --- + + // buzz_communities_total (no tag — fleet-wide count) + let total = state.db.usage_community_count().await?; + metrics::gauge!("buzz_communities_total").set(total as f64); + + // buzz_community_users{community, type:human|agent} + for row in state.db.usage_user_counts().await? { + let community = host(row.community_id); + metrics::gauge!("buzz_community_users", "community" => community.clone(), "type" => "human") + .set(row.human as f64); + metrics::gauge!("buzz_community_users", "community" => community, "type" => "agent") + .set(row.agent as f64); + } + + // buzz_community_channels{community, type} + for row in state.db.usage_channel_counts().await? { + let community = host(row.community_id); + metrics::gauge!( + "buzz_community_channels", + "community" => community, + "type" => row.channel_type + ) + .set(row.count as f64); + } + + // buzz_community_messages{community} + for row in state.db.usage_message_counts().await? { + let community = host(row.community_id); + metrics::gauge!("buzz_community_messages", "community" => community).set(row.count as f64); + } + + // buzz_community_relay_members{community, role} + for row in state.db.usage_relay_member_counts().await? { + let community = host(row.community_id); + metrics::gauge!( + "buzz_community_relay_members", + "community" => community, + "role" => row.role + ) + .set(row.count as f64); + } + + // buzz_community_workflows{community, status} + for row in state.db.usage_workflow_counts().await? { + let community = host(row.community_id); + metrics::gauge!( + "buzz_community_workflows", + "community" => community, + "status" => row.status + ) + .set(row.count as f64); + } + + // buzz_community_git_repos{community} + for row in state.db.usage_git_repo_counts().await? { + let community = host(row.community_id); + metrics::gauge!("buzz_community_git_repos", "community" => community).set(row.count as f64); + } + + // --- C. Engagement — windowed DAU/WAU/MAU + active channels --- + + for (interval, label) in [("1 day", "1d"), ("7 days", "7d"), ("30 days", "30d")] { + for row in state.db.usage_active_user_counts(interval).await? { + let community = host(row.community_id); + metrics::gauge!( + "buzz_community_active_users", + "community" => community.clone(), + "window" => label, + "type" => "human" + ) + .set(row.human as f64); + metrics::gauge!( + "buzz_community_active_users", + "community" => community, + "window" => label, + "type" => "agent" + ) + .set(row.agent as f64); + } + } + + for (interval, label) in [("1 day", "1d"), ("7 days", "7d")] { + for row in state.db.usage_active_channel_counts(interval).await? { + let community = host(row.community_id); + metrics::gauge!( + "buzz_community_active_channels", + "community" => community, + "window" => label + ) + .set(row.count as f64); + } + } + + // --- D. Realtime — in-memory snapshots --- + + // buzz_community_ws_connections{community} + for (community_id, count) in state.conn_manager.per_community_ws_connections() { + let community = host(*community_id.as_uuid()); + metrics::gauge!("buzz_community_ws_connections", "community" => community) + .set(count as f64); + } + + // buzz_community_users_online{community} + for (community_id, count) in state.conn_manager.per_community_users_online() { + let community = host(*community_id.as_uuid()); + metrics::gauge!("buzz_community_users_online", "community" => community).set(count as f64); + } + + // buzz_community_subscriptions{community} + for (community_id, count) in state.sub_registry.per_community_subscriptions() { + let community = host(*community_id.as_uuid()); + metrics::gauge!("buzz_community_subscriptions", "community" => community).set(count as f64); + } + + Ok(()) +} + #[cfg(test)] mod tests { use super::buzz_auto_migrate_enabled; diff --git a/crates/buzz-relay/src/state.rs b/crates/buzz-relay/src/state.rs index 299b599aec..8e6b96bb88 100644 --- a/crates/buzz-relay/src/state.rs +++ b/crates/buzz-relay/src/state.rs @@ -1,5 +1,6 @@ //! Shared application state — Arc-wrapped, shared across all connections. +use std::collections::{HashMap, HashSet}; use std::sync::atomic::{AtomicBool, AtomicU8, Ordering}; use std::sync::Arc; use std::time::Instant; @@ -208,6 +209,41 @@ impl ConnectionManager { .map(|entry| Arc::clone(&entry.subscriptions)) } + /// Snapshot the number of live WebSocket connections per community. + /// + /// Returns a map from community UUID to connection count. Used by the + /// usage poller; snapshotting avoids per-community gauge drift from + /// mismatched inc/dec across async boundaries. + pub fn per_community_ws_connections(&self) -> HashMap { + let mut counts: HashMap = HashMap::new(); + for entry in self.connections.iter() { + *counts.entry(entry.community_id).or_default() += 1; + } + counts + } + + /// Snapshot the number of distinct authenticated pubkeys online per community. + /// + /// A pubkey connected to multiple pods will be counted once per pod — the + /// dashboard sums across pods, so per-pod partial counts are correct. + /// A pubkey connected twice on the same pod is counted once (distinct set). + pub fn per_community_users_online(&self) -> HashMap { + // community_id → set of pubkey bytes + let mut seen: HashMap>> = HashMap::new(); + for entry in self.connections.iter() { + if let Ok(lock) = entry.authenticated_pubkey.read() { + if let Some(pk) = lock.as_ref() { + seen.entry(entry.community_id) + .or_default() + .insert(pk.clone()); + } + } + } + seen.into_iter() + .map(|(cid, set)| (cid, set.len() as u64)) + .collect() + } + /// Return the authenticated pubkey for a connection, if any. pub fn pubkey_for(&self, conn_id: Uuid) -> Option> { self.connections diff --git a/crates/buzz-relay/src/subscription.rs b/crates/buzz-relay/src/subscription.rs index 419cf503cb..003cdd51e3 100644 --- a/crates/buzz-relay/src/subscription.rs +++ b/crates/buzz-relay/src/subscription.rs @@ -351,6 +351,21 @@ impl SubscriptionRegistry { self.subs.len() } + /// Snapshot the number of active subscriptions per community. + /// + /// Used by the usage poller to emit `buzz_community_subscriptions{community}`. + /// Snapshotting avoids gauge drift from mismatched inc/dec across communities. + pub fn per_community_subscriptions(&self) -> std::collections::HashMap { + let mut counts: std::collections::HashMap = + std::collections::HashMap::new(); + for conn_entry in self.subs.iter() { + for (_, community_id, _) in conn_entry.value().values() { + *counts.entry(*community_id).or_default() += 1; + } + } + counts + } + fn push_match( &self, conn_id: ConnId, @@ -1463,4 +1478,49 @@ mod tests { vec![(conn_b, "b".to_string())] ); } + + #[test] + fn per_community_subscriptions_snapshot_is_correctly_scoped() { + // Verify that per_community_subscriptions() returns the correct + // per-community counts and that removal keeps the snapshot accurate. + let registry = SubscriptionRegistry::new(); + let community_a = CommunityId::from_uuid(Uuid::from_u128(0xaaaa)); + let community_b = CommunityId::from_uuid(Uuid::from_u128(0xbbbb)); + let conn_a = Uuid::new_v4(); + let conn_b = Uuid::new_v4(); + let channel = Uuid::new_v4(); + + // 2 subs in community A, 1 sub in community B. + registry.register_scoped( + community_a, + conn_a, + "a1".to_string(), + vec![Filter::new().kind(Kind::TextNote)], + Some(channel), + ); + registry.register_scoped( + community_a, + conn_a, + "a2".to_string(), + vec![Filter::new().kind(Kind::Metadata)], + None, + ); + registry.register_scoped( + community_b, + conn_b, + "b1".to_string(), + vec![Filter::new().kind(Kind::TextNote)], + Some(channel), + ); + + let snap = registry.per_community_subscriptions(); + assert_eq!(snap.get(&community_a), Some(&2), "community A: 2 subs"); + assert_eq!(snap.get(&community_b), Some(&1), "community B: 1 sub"); + + // Remove one sub from community A — snapshot should reflect it. + registry.remove_subscription(conn_a, "a1"); + let snap2 = registry.per_community_subscriptions(); + assert_eq!(snap2.get(&community_a), Some(&1), "community A: 1 sub after removal"); + assert_eq!(snap2.get(&community_b), Some(&1), "community B: unchanged"); + } } From 8c4d4a8bd8576d61bab129897999e9fd2f8f437c Mon Sep 17 00:00:00 2001 From: npub1mn7jgtj4w2pd0g0zeuhxsa6jy6p0rewxz4kujt98my82ahfmp72sxjexk7 Date: Fri, 10 Jul 2026 15:04:44 -0400 Subject: [PATCH 2/6] fix(relay): address Pass-1 review findings for usage metrics (F1-F6) F1: Zero-fill buzz_community_channels, buzz_community_relay_members, and buzz_community_workflows across bounded label domains so counts going to zero emit 0 instead of retaining stale last-nonzero values. buzz_community_ users, messages, git_repos, active_users, active_channels, ws_connections, users_online, and subscriptions were already zero-filled. Remove the now- unused host() closure. Add two regression tests: - subscription.rs: per_community_subscriptions_drops_to_zero_when_all_ subs_removed (in-memory gauge) - usage.rs: test_channel_counts_drops_to_zero_after_last_channel_deleted (DB gauge) F2: Default interval 60s -> 300s (env override unchanged, min 5s). Add comment naming rollup-table escape hatch at event-rollup queries. F3: Increment buzz_channels_created_total{type="dm"} at open_dm call sites in command_executor.rs (two DM paths) and moderation_notices.rs. F4: Run cargo fmt; split long assertion in subscription.rs per rustfmt. F5: Fix test_user_counts_scoped_per_community: original insert_user(is_ agent=true) inserted the agent's owner as a separate human, making the assertion wrong (3 humans, not 2). Rewrite to share owner pubkey so agent and human count correctly. All 7 ignored usage tests pass against local PG. F6: FROM git_repo_names (not git_repos which does not exist). Add test test_git_repo_counts_scoped_per_community covering the corrected relation. Co-authored-by: Will Pfleger Signed-off-by: Will Pfleger --- crates/buzz-db/src/usage.rs | 134 +++++++++- .../src/handlers/command_executor.rs | 14 + .../src/handlers/moderation_notices.rs | 13 +- crates/buzz-relay/src/main.rs | 250 +++++++++++++----- crates/buzz-relay/src/subscription.rs | 37 ++- 5 files changed, 374 insertions(+), 74 deletions(-) diff --git a/crates/buzz-db/src/usage.rs b/crates/buzz-db/src/usage.rs index 0bed64c03b..dc6737be7f 100644 --- a/crates/buzz-db/src/usage.rs +++ b/crates/buzz-db/src/usage.rs @@ -205,7 +205,7 @@ pub async fn git_repo_counts(pool: &PgPool) -> Result let rows = sqlx::query_as::<_, (Uuid, i64)>( r#" SELECT community_id, COUNT(*) AS count - FROM git_repos + FROM git_repo_names GROUP BY community_id "#, ) @@ -405,13 +405,39 @@ mod tests { let (comm_a_uuid, _, _) = make_community(&pool).await; let (comm_b_uuid, _, _) = make_community(&pool).await; - // Community A: 2 human, 1 agent - insert_user(&pool, comm_a_uuid, &random_pubkey(), false).await; - insert_user(&pool, comm_a_uuid, &random_pubkey(), false).await; - insert_user(&pool, comm_a_uuid, &random_pubkey(), true).await; + // Community A: insert 2 humans first, then 1 agent whose owner is one + // of those humans (reuses existing pubkey — no extra human row). + let human1 = random_pubkey(); + let human2 = random_pubkey(); + let agent_pk = random_pubkey(); + insert_user(&pool, comm_a_uuid, &human1, false).await; + insert_user(&pool, comm_a_uuid, &human2, false).await; + // Insert agent with human1 as owner (human1 is already in users). + sqlx::query( + "INSERT INTO users (community_id, pubkey, agent_owner_pubkey) + VALUES ($1, $2, $3) ON CONFLICT DO NOTHING", + ) + .bind(comm_a_uuid) + .bind(&agent_pk) + .bind(&human1) + .execute(&pool) + .await + .expect("insert agent user"); - // Community B: 0 human, 1 agent - insert_user(&pool, comm_b_uuid, &random_pubkey(), true).await; + // Community B: 0 human, 1 agent (owner is a fresh human in comm_b). + let owner_b = random_pubkey(); + insert_user(&pool, comm_b_uuid, &owner_b, false).await; + let agent_b = random_pubkey(); + sqlx::query( + "INSERT INTO users (community_id, pubkey, agent_owner_pubkey) + VALUES ($1, $2, $3) ON CONFLICT DO NOTHING", + ) + .bind(comm_b_uuid) + .bind(&agent_b) + .bind(&owner_b) + .execute(&pool) + .await + .expect("insert agent user b"); let counts = user_counts(&pool).await.expect("user_counts"); @@ -423,7 +449,7 @@ mod tests { assert_eq!(a.agent, 1, "community A: 1 agent"); let b = b.expect("community B row"); - assert_eq!(b.human, 0, "community B: 0 humans"); + assert_eq!(b.human, 1, "community B: 1 human (the agent owner)"); assert_eq!(b.agent, 1, "community B: 1 agent"); } @@ -468,7 +494,7 @@ mod tests { // Insert a stream and a DM channel. sqlx::query( - "INSERT INTO channels (id, community_id, name, channel_type, visibility, owner_pubkey) + "INSERT INTO channels (id, community_id, name, channel_type, visibility, created_by) VALUES ($1, $2, 'test-stream', 'stream', 'open', $3)", ) .bind(uuid::Uuid::new_v4()) @@ -480,7 +506,7 @@ mod tests { let dm_id = uuid::Uuid::new_v4(); sqlx::query( - "INSERT INTO channels (id, community_id, name, channel_type, visibility, owner_pubkey) + "INSERT INTO channels (id, community_id, name, channel_type, visibility, created_by) VALUES ($1, $2, 'test-dm', 'dm', 'private', $3)", ) .bind(dm_id) @@ -535,4 +561,92 @@ mod tests { let after = community_count(&pool).await.expect("count after"); assert!(after > before, "count should increase after insert"); } + + /// git_repo_counts queries git_repo_names (not git_repos) and is scoped per community. + #[tokio::test] + #[ignore = "requires Postgres"] + async fn test_git_repo_counts_scoped_per_community() { + let pool = get_pool().await; + let (comm_uuid, _, _) = make_community(&pool).await; + let owner = random_pubkey(); + insert_user(&pool, comm_uuid, &owner, false).await; + let owner_hex = hex::encode(&owner); + + // Insert two repos for this community. + for repo_id in &["repo-alpha", "repo-beta"] { + sqlx::query( + "INSERT INTO git_repo_names (community_id, repo_id, owner_pubkey) + VALUES ($1, $2, $3) + ON CONFLICT DO NOTHING", + ) + .bind(comm_uuid) + .bind(repo_id) + .bind(&owner_hex) + .execute(&pool) + .await + .expect("insert git repo"); + } + + let counts = git_repo_counts(&pool).await.expect("git_repo_counts"); + let comm_counts: Vec<_> = counts + .iter() + .filter(|r| r.community_id == comm_uuid) + .collect(); + + assert_eq!(comm_counts.len(), 1, "one row per community"); + assert_eq!(comm_counts[0].count, 2, "two repos"); + } + + /// Regression: channel_counts returns no row for a community once all + /// channels of a type are soft-deleted. The poller zero-fills from + /// host_map, so absence from this query is the correct "zero" signal. + #[tokio::test] + #[ignore = "requires Postgres"] + async fn test_channel_counts_drops_to_zero_after_last_channel_deleted() { + let pool = get_pool().await; + let (comm_uuid, _, _) = make_community(&pool).await; + let owner = random_pubkey(); + insert_user(&pool, comm_uuid, &owner, false).await; + + // Insert one stream channel. + let ch_id = uuid::Uuid::new_v4(); + sqlx::query( + "INSERT INTO channels (id, community_id, name, channel_type, visibility, created_by) + VALUES ($1, $2, 'only-stream', 'stream', 'open', $3)", + ) + .bind(ch_id) + .bind(comm_uuid) + .bind(&owner) + .execute(&pool) + .await + .expect("insert channel"); + + // Sanity: row present before deletion. + let before = channel_counts(&pool).await.expect("channel_counts before"); + let before_row = before + .iter() + .find(|r| r.community_id == comm_uuid && r.channel_type == "stream"); + assert_eq!( + before_row.map(|r| r.count), + Some(1), + "1 stream channel before deletion" + ); + + // Soft-delete the channel. + sqlx::query("UPDATE channels SET deleted_at = NOW() WHERE id = $1") + .bind(ch_id) + .execute(&pool) + .await + .expect("soft-delete channel"); + + // After deletion: no row for this community+type — query returns nothing. + let after = channel_counts(&pool).await.expect("channel_counts after"); + let after_row = after + .iter() + .find(|r| r.community_id == comm_uuid && r.channel_type == "stream"); + assert!( + after_row.is_none(), + "no stream row after last channel deleted — poller will zero-fill" + ); + } } diff --git a/crates/buzz-relay/src/handlers/command_executor.rs b/crates/buzz-relay/src/handlers/command_executor.rs index 4cb88a570a..bd3eff4de9 100644 --- a/crates/buzz-relay/src/handlers/command_executor.rs +++ b/crates/buzz-relay/src/handlers/command_executor.rs @@ -372,6 +372,13 @@ async fn handle_dm_open( // 5. Side effects if newly created (post-commit, best-effort) if was_created { + metrics::counter!( + "buzz_channels_created_total", + "community" => tenant.host().to_owned(), + "type" => "dm" + ) + .increment(1); + // Invalidate caches for all participants for pk in &all_bytes { state.invalidate_membership(tenant, channel.id, pk); @@ -526,6 +533,13 @@ async fn handle_dm_add_member( // 7. Cache invalidation + notifications for new DM (post-commit, best-effort) if was_created { + metrics::counter!( + "buzz_channels_created_total", + "community" => tenant.host().to_owned(), + "type" => "dm" + ) + .increment(1); + for pk in &all_bytes { state.invalidate_membership(tenant, new_channel.id, pk); } diff --git a/crates/buzz-relay/src/handlers/moderation_notices.rs b/crates/buzz-relay/src/handlers/moderation_notices.rs index 0e5b51ef2f..8f57eea71f 100644 --- a/crates/buzz-relay/src/handlers/moderation_notices.rs +++ b/crates/buzz-relay/src/handlers/moderation_notices.rs @@ -97,7 +97,7 @@ pub async fn send_moderation_notice( // 1. Create/reuse the two-party DM channel {relay mod key, recipient}. // `open_dm` is participant-hash idempotent, so re-delivery to the same // user reuses the one thread per (community, user). - let (dm_channel, _was_created) = state + let (dm_channel, was_created) = state .db .open_dm( tenant.community(), @@ -107,6 +107,17 @@ pub async fn send_moderation_notice( .await?; let dm_channel_id = dm_channel.id; + // Count new DM creation; side-effect gates below intentionally do not + // gate on was_created (see comment at step 2). + if was_created { + metrics::counter!( + "buzz_channels_created_total", + "community" => tenant.host().to_owned(), + "type" => "dm" + ) + .increment(1); + } + // Resurface the moderation DM for the recipient. `open_dm` only clears // `hidden_at` for `created_by` (the relay key), so a user who hid the // "{host} Moderation" thread would never see a later ban/resolution notice. diff --git a/crates/buzz-relay/src/main.rs b/crates/buzz-relay/src/main.rs index cba53bf9bb..bc50116845 100644 --- a/crates/buzz-relay/src/main.rs +++ b/crates/buzz-relay/src/main.rs @@ -1,3 +1,4 @@ +use std::collections::HashMap; use std::sync::atomic::Ordering; use std::sync::Arc; @@ -7,6 +8,7 @@ use uuid::Uuid; use buzz_audit::AuditService; use buzz_auth::AuthService; +use buzz_core::CommunityId; use buzz_db::{Db, DbConfig}; use buzz_pubsub::PubSubManager; use buzz_search::SearchService; @@ -822,7 +824,10 @@ async fn main() -> anyhow::Result<()> { let interval_secs = std::env::var("BUZZ_USAGE_METRICS_INTERVAL_SECS") .ok() .and_then(|v| v.parse::().ok()) - .unwrap_or(60) + .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 tokio::spawn(async move { let mut interval = tokio::time::interval(std::time::Duration::from_secs(interval_secs)); @@ -1011,12 +1016,7 @@ fn reminder_to_event(reminder: &buzz_db::event::DueReminder) -> nostr::Event { async fn run_usage_metrics_tick(state: &AppState) -> anyhow::Result<()> { // --- community id → host label map (one query, cached for this tick) --- let hosts = state.db.usage_community_hosts().await?; - let host_map: std::collections::HashMap = - hosts.into_iter().map(|c| (c.id, c.host)).collect(); - - // Helper: resolve community UUID → host label string. - let host = - |id: Uuid| -> String { host_map.get(&id).cloned().unwrap_or_else(|| id.to_string()) }; + let host_map: HashMap = hosts.into_iter().map(|c| (c.id, c.host)).collect(); // --- A. Adoption stocks (DB-polled) --- @@ -1025,112 +1025,238 @@ async fn run_usage_metrics_tick(state: &AppState) -> anyhow::Result<()> { metrics::gauge!("buzz_communities_total").set(total as f64); // buzz_community_users{community, type:human|agent} - for row in state.db.usage_user_counts().await? { - let community = host(row.community_id); - metrics::gauge!("buzz_community_users", "community" => community.clone(), "type" => "human") - .set(row.human as f64); - metrics::gauge!("buzz_community_users", "community" => community, "type" => "agent") - .set(row.agent as f64); + // Emit from host_map so communities that have zero users still get a 0 + // rather than keeping the last nonzero value until process restart. + { + let rows: HashMap = state + .db + .usage_user_counts() + .await? + .into_iter() + .map(|r| (r.community_id, r)) + .collect(); + for (&id, community) in &host_map { + let (human, agent) = rows.get(&id).map(|r| (r.human, r.agent)).unwrap_or((0, 0)); + metrics::gauge!("buzz_community_users", "community" => community.clone(), "type" => "human") + .set(human as f64); + metrics::gauge!("buzz_community_users", "community" => community.clone(), "type" => "agent") + .set(agent as f64); + } } // buzz_community_channels{community, type} - for row in state.db.usage_channel_counts().await? { - let community = host(row.community_id); - metrics::gauge!( - "buzz_community_channels", - "community" => community, - "type" => row.channel_type - ) - .set(row.count as f64); + // Zero-fill across all (community, channel_type) pairs so a type that + // drops to zero emits 0 rather than retaining its last nonzero value. + { + const CHANNEL_TYPES: &[&str] = &["stream", "forum", "dm", "workflow"]; + let rows: HashMap<(Uuid, &str), i64> = state + .db + .usage_channel_counts() + .await? + .into_iter() + .filter_map(|r| { + // Map the DB string to a known static str so the key is &str. + CHANNEL_TYPES + .iter() + .find(|&&t| t == r.channel_type.as_str()) + .map(|&t| ((r.community_id, t), r.count)) + }) + .collect(); + for (&id, community) in &host_map { + for &ct in CHANNEL_TYPES { + let count = rows.get(&(id, ct)).copied().unwrap_or(0); + metrics::gauge!( + "buzz_community_channels", + "community" => community.clone(), + "type" => ct + ) + .set(count as f64); + } + } } // buzz_community_messages{community} - for row in state.db.usage_message_counts().await? { - let community = host(row.community_id); - metrics::gauge!("buzz_community_messages", "community" => community).set(row.count as f64); + // Emit 0 for communities with no messages so dashboards don't stale-read. + { + let rows: HashMap = state + .db + .usage_message_counts() + .await? + .into_iter() + .map(|r| (r.community_id, r.count)) + .collect(); + for (&id, community) in &host_map { + let count = rows.get(&id).copied().unwrap_or(0); + metrics::gauge!("buzz_community_messages", "community" => community.clone()) + .set(count as f64); + } } // buzz_community_relay_members{community, role} - for row in state.db.usage_relay_member_counts().await? { - let community = host(row.community_id); - metrics::gauge!( - "buzz_community_relay_members", - "community" => community, - "role" => row.role - ) - .set(row.count as f64); + // Zero-fill across all (community, role) pairs; relay_members.role is a + // CHECK constraint over {'owner', 'admin', 'member'}. + { + const RELAY_ROLES: &[&str] = &["owner", "admin", "member"]; + let rows: HashMap<(Uuid, &str), i64> = state + .db + .usage_relay_member_counts() + .await? + .into_iter() + .filter_map(|r| { + RELAY_ROLES + .iter() + .find(|&&role| role == r.role.as_str()) + .map(|&role| ((r.community_id, role), r.count)) + }) + .collect(); + for (&id, community) in &host_map { + for &role in RELAY_ROLES { + let count = rows.get(&(id, role)).copied().unwrap_or(0); + metrics::gauge!( + "buzz_community_relay_members", + "community" => community.clone(), + "role" => role + ) + .set(count as f64); + } + } } // buzz_community_workflows{community, status} - for row in state.db.usage_workflow_counts().await? { - let community = host(row.community_id); - metrics::gauge!( - "buzz_community_workflows", - "community" => community, - "status" => row.status - ) - .set(row.count as f64); + // Zero-fill across all (community, status) pairs; workflow_status is a + // DB enum: {'active', 'disabled', 'archived'}. + { + const WORKFLOW_STATUSES: &[&str] = &["active", "disabled", "archived"]; + let rows: HashMap<(Uuid, &str), i64> = state + .db + .usage_workflow_counts() + .await? + .into_iter() + .filter_map(|r| { + WORKFLOW_STATUSES + .iter() + .find(|&&s| s == r.status.as_str()) + .map(|&s| ((r.community_id, s), r.count)) + }) + .collect(); + for (&id, community) in &host_map { + for &status in WORKFLOW_STATUSES { + let count = rows.get(&(id, status)).copied().unwrap_or(0); + metrics::gauge!( + "buzz_community_workflows", + "community" => community.clone(), + "status" => status + ) + .set(count as f64); + } + } } // buzz_community_git_repos{community} - for row in state.db.usage_git_repo_counts().await? { - let community = host(row.community_id); - metrics::gauge!("buzz_community_git_repos", "community" => community).set(row.count as f64); + // Emit 0 for communities with no repos. + { + let rows: HashMap = state + .db + .usage_git_repo_counts() + .await? + .into_iter() + .map(|r| (r.community_id, r.count)) + .collect(); + for (&id, community) in &host_map { + let count = rows.get(&id).copied().unwrap_or(0); + metrics::gauge!("buzz_community_git_repos", "community" => community.clone()) + .set(count as f64); + } } // --- C. Engagement — windowed DAU/WAU/MAU + active channels --- + // Emit 0 for window/type/community combos that had no activity; this + // ensures a community that was active last tick but quiet this tick reads + // 0 rather than retaining its last nonzero value. for (interval, label) in [("1 day", "1d"), ("7 days", "7d"), ("30 days", "30d")] { - for row in state.db.usage_active_user_counts(interval).await? { - let community = host(row.community_id); + let rows: HashMap = state + .db + .usage_active_user_counts(interval) + .await? + .into_iter() + .map(|r| (r.community_id, r)) + .collect(); + for (&id, community) in &host_map { + let (human, agent) = rows.get(&id).map(|r| (r.human, r.agent)).unwrap_or((0, 0)); metrics::gauge!( "buzz_community_active_users", "community" => community.clone(), "window" => label, "type" => "human" ) - .set(row.human as f64); + .set(human as f64); metrics::gauge!( "buzz_community_active_users", - "community" => community, + "community" => community.clone(), "window" => label, "type" => "agent" ) - .set(row.agent as f64); + .set(agent as f64); } } for (interval, label) in [("1 day", "1d"), ("7 days", "7d")] { - for row in state.db.usage_active_channel_counts(interval).await? { - let community = host(row.community_id); + let rows: HashMap = state + .db + .usage_active_channel_counts(interval) + .await? + .into_iter() + .map(|r| (r.community_id, r.count)) + .collect(); + for (&id, community) in &host_map { + let count = rows.get(&id).copied().unwrap_or(0); metrics::gauge!( "buzz_community_active_channels", - "community" => community, + "community" => community.clone(), "window" => label ) - .set(row.count as f64); + .set(count as f64); } } // --- 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} - for (community_id, count) in state.conn_manager.per_community_ws_connections() { - let community = host(*community_id.as_uuid()); - metrics::gauge!("buzz_community_ws_connections", "community" => community) - .set(count as f64); + { + let conns = state.conn_manager.per_community_ws_connections(); + for (&id, community) in &host_map { + let count = conns.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{community} - for (community_id, count) in state.conn_manager.per_community_users_online() { - let community = host(*community_id.as_uuid()); - metrics::gauge!("buzz_community_users_online", "community" => community).set(count as f64); + { + let online = state.conn_manager.per_community_users_online(); + for (&id, community) in &host_map { + let count = online + .get(&CommunityId::from_uuid(id)) + .copied() + .unwrap_or(0); + metrics::gauge!("buzz_community_users_online", "community" => community.clone()) + .set(count as f64); + } } // buzz_community_subscriptions{community} - for (community_id, count) in state.sub_registry.per_community_subscriptions() { - let community = host(*community_id.as_uuid()); - metrics::gauge!("buzz_community_subscriptions", "community" => community).set(count as f64); + { + let subs = state.sub_registry.per_community_subscriptions(); + for (&id, community) in &host_map { + let count = subs.get(&CommunityId::from_uuid(id)).copied().unwrap_or(0); + metrics::gauge!("buzz_community_subscriptions", "community" => community.clone()) + .set(count as f64); + } } Ok(()) diff --git a/crates/buzz-relay/src/subscription.rs b/crates/buzz-relay/src/subscription.rs index 003cdd51e3..9831e8eee4 100644 --- a/crates/buzz-relay/src/subscription.rs +++ b/crates/buzz-relay/src/subscription.rs @@ -1520,7 +1520,42 @@ mod tests { // Remove one sub from community A — snapshot should reflect it. registry.remove_subscription(conn_a, "a1"); let snap2 = registry.per_community_subscriptions(); - assert_eq!(snap2.get(&community_a), Some(&1), "community A: 1 sub after removal"); + assert_eq!( + snap2.get(&community_a), + Some(&1), + "community A: 1 sub after removal" + ); assert_eq!(snap2.get(&community_b), Some(&1), "community B: unchanged"); } + + #[test] + 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. + let registry = SubscriptionRegistry::new(); + let community = CommunityId::from_uuid(Uuid::from_u128(0xcccc)); + let conn = Uuid::new_v4(); + + registry.register_scoped( + community, + conn, + "c1".to_string(), + vec![Filter::new().kind(Kind::TextNote)], + None, + ); + + // Verify nonzero before removal. + let snap1 = registry.per_community_subscriptions(); + assert_eq!(snap1.get(&community), Some(&1), "1 sub before removal"); + + // Remove the only sub — community should no longer appear in snapshot. + registry.remove_subscription(conn, "c1"); + let snap2 = registry.per_community_subscriptions(); + assert!( + snap2.get(&community).copied().unwrap_or(0) == 0, + "community must have 0 subs after last removal; got {:?}", + snap2.get(&community) + ); + } } From b423c8a96e4388a0b88924596d5923295625464d Mon Sep 17 00:00:00 2001 From: npub1mn7jgtj4w2pd0g0zeuhxsa6jy6p0rewxz4kujt98my82ahfmp72sxjexk7 Date: Fri, 10 Jul 2026 15:10:15 -0400 Subject: [PATCH 3/6] =?UTF-8?q?docs(buzz-db):=20correct=20usage=20module?= =?UTF-8?q?=20doc=20=E2=80=94=20event=20queries=20are=20exact=20aggregates?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The module header claimed all queries use indexed columns with no full-table scans. That is false for message_counts, active_user_counts, and active_channel_counts, which are exact event-table aggregates (accepted tradeoff from Pass-1 review F2). Split the doc to accurately describe the two query classes and call out the rollup-table escape hatch. Co-authored-by: Will Pfleger Signed-off-by: Will Pfleger --- crates/buzz-db/src/usage.rs | 10 ++++++++-- 1 file changed, 8 insertions(+), 2 deletions(-) diff --git a/crates/buzz-db/src/usage.rs b/crates/buzz-db/src/usage.rs index dc6737be7f..b62031ac89 100644 --- a/crates/buzz-db/src/usage.rs +++ b/crates/buzz-db/src/usage.rs @@ -1,7 +1,13 @@ //! Per-community usage rollup queries for Prometheus gauges. //! -//! All queries use `GROUP BY community_id` against indexed columns — -//! no per-community loops, no full-table scans. +//! Stock queries (`user_counts`, `channel_counts`, `relay_member_counts`, +//! `workflow_counts`, `git_repo_counts`) use `GROUP BY community_id` against +//! indexed columns — no per-community loops, no full-table scans. +//! +//! Event-derived queries (`message_counts`, `active_user_counts`, +//! `active_channel_counts`) are exact aggregates over the `events` table. +//! At scale these can become recurring partition scans; if that becomes a +//! problem, move them to a maintained rollup table and drop the interval. //! //! Returned structs are plain data; the caller (relay poller) maps them //! to Prometheus labels and calls `metrics::gauge!(...).set(...)`. From 59b94a1c2e5c9f864f46b32d827f27c7e6a531d5 Mon Sep 17 00:00:00 2001 From: npub1mn7jgtj4w2pd0g0zeuhxsa6jy6p0rewxz4kujt98my82ahfmp72sxjexk7 Date: Fri, 10 Jul 2026 16:35:36 -0400 Subject: [PATCH 4/6] =?UTF-8?q?fix(relay):=20address=20Phase-1=20review=20?= =?UTF-8?q?findings=20(C2=E2=80=93C5,=20N1=E2=80=93N3)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit C2: add unknown bucket to active_user_counts - Old SQL: LEFT JOIN + FILTER (WHERE u.agent_owner_pubkey IS NULL) counted pubkeys with no users row as human (NULL passes IS NULL check). - Fix: three-way FILTER on u.pubkey IS NOT NULL/IS NULL — profileless posters and agents with missing rows now land in a new 'unknown' gauge type rather than inflating the human count. - CommunityActiveUsers gains an 'unknown: i64' field; callers updated to emit buzz_community_active_users{type=unknown}. - New ignored PG test: test_active_user_counts_unknown_bucket_for_profileless_poster. C3: rename buzz_community_users_online → buzz_community_users_online_pod - The gauge is pod-local (distinct within a pod, not fleet-distinct). The _pod suffix makes the aggregation semantics explicit at the metric name level so dashboard authors don't sum and expect global-distinct counts. - Added comment explaining fleet-wide query pattern (sum across pods). C4: collect-before-publish in run_usage_metrics_tick - All DB .await? calls are now grouped at the top of the function. If any query fails the function returns early with no metrics emitted, preventing a mixed fresh/stale snapshot. In-memory snapshots (infallible) are taken once before the publish phase. C5: document double-count analysis in handle_create_group - No code change needed: the Err(_) fallback path only fires when the prior ingest DB lookup returns an error, not when ingest succeeded. Added a comment explaining this so the invariant is auditable. N1: fix PR body — default interval was documented as 60s, is actually 300s. N2: jitter first tick + MissedTickBehavior::Skip - PID-seeded Fibonacci hash gives each pod a start delay in [0, interval_secs), preventing a rolling-deploy thundering herd. - MissedTickBehavior::Skip prevents catch-up burst if a tick runs long. N3: warn! on unmatched enum values in filter_map - channel_type, relay_member role, and workflow status filter_maps now emit a tracing warn! for unrecognised values instead of silently dropping them, making schema drift immediately observable in logs. Co-authored-by: Will Pfleger Signed-off-by: Will Pfleger --- crates/buzz-db/src/usage.rs | 90 ++++++++-- .../buzz-relay/src/handlers/side_effects.rs | 8 + crates/buzz-relay/src/main.rs | 164 +++++++++++------- 3 files changed, 191 insertions(+), 71 deletions(-) diff --git a/crates/buzz-db/src/usage.rs b/crates/buzz-db/src/usage.rs index b62031ac89..f009dc6e05 100644 --- a/crates/buzz-db/src/usage.rs +++ b/crates/buzz-db/src/usage.rs @@ -234,13 +234,20 @@ pub struct CommunityActiveUsers { /// The UUID of the community. pub community_id: Uuid, /// Distinct human pubkeys that published at least one event in the window. + /// A pubkey is human when its `users` row exists and `agent_owner_pubkey IS NULL`. pub human: i64, /// Distinct agent pubkeys that published at least one event in the window. + /// A pubkey is an agent when its `users` row exists and `agent_owner_pubkey IS NOT NULL`. pub agent: i64, + /// Distinct pubkeys that published at least one event but have no `users` row. + /// Ingest does not guarantee a `users` row for every pubkey (profileless posters, + /// agents with missing rows). These are not classified and must not be folded into + /// `human` to avoid inflating the human count. + pub unknown: i64, } /// Return distinct-publisher counts for events in `[now - interval, now]` -/// per community, split by human/agent. +/// per community, split by human/agent/unknown. /// /// `interval_sql` must be a trusted literal (e.g. `"1 day"`, `"7 days"`) — /// it is not user-controlled; callers are in the relay process. @@ -248,16 +255,21 @@ pub async fn active_user_counts( pool: &PgPool, interval_sql: &'static str, ) -> Result> { - // JOIN users so we can discriminate human vs. agent. Events are authored - // by pubkeys; the users.agent_owner_pubkey column is the discriminator. - // We use encode(e.pubkey, 'hex') and encode(u.pubkey, 'hex') to join on - // bytea — the columns are the same type so direct equality works. + // LEFT JOIN users: pubkeys with no row have u.* = NULL. + // Three-way classification: + // human — row exists (u.pubkey IS NOT NULL) and agent_owner_pubkey IS NULL + // agent — row exists and agent_owner_pubkey IS NOT NULL + // unknown — no row (u.pubkey IS NULL); not classified, reported separately let sql = format!( r#" SELECT e.community_id, - COUNT(DISTINCT e.pubkey) FILTER (WHERE u.agent_owner_pubkey IS NULL) AS human, - COUNT(DISTINCT e.pubkey) FILTER (WHERE u.agent_owner_pubkey IS NOT NULL) AS agent + COUNT(DISTINCT e.pubkey) + FILTER (WHERE u.pubkey IS NOT NULL AND u.agent_owner_pubkey IS NULL) AS human, + COUNT(DISTINCT e.pubkey) + FILTER (WHERE u.pubkey IS NOT NULL AND u.agent_owner_pubkey IS NOT NULL) AS agent, + COUNT(DISTINCT e.pubkey) + FILTER (WHERE u.pubkey IS NULL) AS unknown FROM events e LEFT JOIN users u ON u.community_id = e.community_id AND u.pubkey = e.pubkey @@ -266,17 +278,20 @@ pub async fn active_user_counts( GROUP BY e.community_id "# ); - let rows = sqlx::query_as::<_, (Uuid, i64, i64)>(sqlx::AssertSqlSafe(sql)) + let rows = sqlx::query_as::<_, (Uuid, i64, i64, i64)>(sqlx::AssertSqlSafe(sql)) .fetch_all(pool) .await?; Ok(rows .into_iter() - .map(|(community_id, human, agent)| CommunityActiveUsers { - community_id, - human, - agent, - }) + .map( + |(community_id, human, agent, unknown)| CommunityActiveUsers { + community_id, + human, + agent, + unknown, + }, + ) .collect()) } @@ -603,6 +618,55 @@ mod tests { assert_eq!(comm_counts[0].count, 2, "two repos"); } + /// active_user_counts classifies pubkeys with no users row as "unknown", + /// not "human" — the old LEFT JOIN treated NULL.agent_owner_pubkey as human. + #[tokio::test] + #[ignore = "requires Postgres"] + async fn test_active_user_counts_unknown_bucket_for_profileless_poster() { + let pool = get_pool().await; + let (comm_uuid, _, _) = make_community(&pool).await; + + // One known human (has a users row). + let human_pk = random_pubkey(); + insert_user(&pool, comm_uuid, &human_pk, false).await; + + // One profileless poster (no users row at all). + let profileless_pk = random_pubkey(); + + // Insert events for both pubkeys in this community. + let event_id1 = random_pubkey(); // 32-byte id + let event_id2 = random_pubkey(); + let sig = vec![0u8; 64]; + for (pk, eid) in [(&human_pk, &event_id1), (&profileless_pk, &event_id2)] { + sqlx::query( + "INSERT INTO events \ + (community_id, id, pubkey, created_at, kind, tags, content, sig, received_at) \ + VALUES ($1, $2, $3, NOW(), 9, '[]', '', $4, NOW()) \ + ON CONFLICT DO NOTHING", + ) + .bind(comm_uuid) + .bind(eid) + .bind(pk) + .bind(&sig) + .execute(&pool) + .await + .expect("insert event"); + } + + let counts = active_user_counts(&pool, "1 day") + .await + .expect("active_user_counts"); + let row = counts.iter().find(|r| r.community_id == comm_uuid); + assert!(row.is_some(), "row for community must exist"); + let row = row.unwrap(); + assert_eq!(row.human, 1, "known human poster counts as human"); + assert_eq!(row.agent, 0, "no agents"); + assert_eq!( + row.unknown, 1, + "profileless poster must land in unknown, not human" + ); + } + /// Regression: channel_counts returns no row for a community once all /// channels of a type are soft-deleted. The poller zero-fills from /// host_map, so absence from this query is the correct "zero" signal. diff --git a/crates/buzz-relay/src/handlers/side_effects.rs b/crates/buzz-relay/src/handlers/side_effects.rs index a6a06ebd59..33a95ee64e 100644 --- a/crates/buzz-relay/src/handlers/side_effects.rs +++ b/crates/buzz-relay/src/handlers/side_effects.rs @@ -1667,6 +1667,14 @@ async fn handle_create_group( // If the event has an h-tag UUID, ingest_event() already created the channel // via create_channel_with_id(). Fetch it rather than creating a duplicate. // If no h-tag, fall back to the original auto-UUID creation path. + // + // Double-count analysis (C5): the counter increments below do NOT + // double-count vs. ingest.rs. For the h-tag path, ingest increments on + // was_created=true and this handler only reaches create_channel() on a DB + // lookup Err — an error recovery path where ingest's channel is + // inaccessible, so the counter correctly records a new creation. For the + // no-h-tag path, ingest never creates the channel, so this is the sole + // increment. let channel = if let Some(client_uuid) = extract_h_tag_channel(event) { match state.db.get_channel(tenant.community(), client_uuid).await { Ok(ch) => ch, diff --git a/crates/buzz-relay/src/main.rs b/crates/buzz-relay/src/main.rs index bc50116845..8c6b32f18f 100644 --- a/crates/buzz-relay/src/main.rs +++ b/crates/buzz-relay/src/main.rs @@ -830,7 +830,18 @@ async fn main() -> anyhow::Result<()> { // the interval back to 60s. .max(5); // 5s minimum: cheaper than pool poller's 1s minimum 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 + // simultaneously at boot. Each pod picks a start delay in + // [0, interval_secs) using its PID as the seed. + let jitter_secs = + (std::process::id() as u64).wrapping_mul(2_654_435_761) % interval_secs; + tokio::time::sleep(std::time::Duration::from_secs(jitter_secs)).await; + let mut interval = tokio::time::interval(std::time::Duration::from_secs(interval_secs)); + // Skip a tick rather than scheduling a burst of catch-up ticks if + // the system falls behind (e.g. the previous tick took > interval). + interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip); loop { interval.tick().await; if let Err(e) = run_usage_metrics_tick(&usage_state).await { @@ -1018,23 +1029,43 @@ async fn run_usage_metrics_tick(state: &AppState) -> anyhow::Result<()> { let hosts = state.db.usage_community_hosts().await?; let host_map: HashMap = hosts.into_iter().map(|c| (c.id, c.host)).collect(); + // --- Collect all DB results before emitting any metrics (C4) --- + // + // All `.await?` calls happen here. If any query fails the function returns + // early — no metrics are emitted for this tick — preventing a mixed + // fresh/stale snapshot where later gauges retain their last value while + // earlier ones are updated. + + let community_total = state.db.usage_community_count().await?; + let user_rows = state.db.usage_user_counts().await?; + let channel_rows = state.db.usage_channel_counts().await?; + let message_rows = state.db.usage_message_counts().await?; + let relay_member_rows = state.db.usage_relay_member_counts().await?; + let workflow_rows = state.db.usage_workflow_counts().await?; + let git_repo_rows = state.db.usage_git_repo_counts().await?; + let active_users_1d = state.db.usage_active_user_counts("1 day").await?; + let active_users_7d = state.db.usage_active_user_counts("7 days").await?; + let active_users_30d = state.db.usage_active_user_counts("30 days").await?; + 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(); + + // --- Publish phase: emit all metrics now that every query succeeded --- + // --- A. Adoption stocks (DB-polled) --- // buzz_communities_total (no tag — fleet-wide count) - let total = state.db.usage_community_count().await?; - metrics::gauge!("buzz_communities_total").set(total as f64); + metrics::gauge!("buzz_communities_total").set(community_total as f64); // buzz_community_users{community, type:human|agent} // Emit from host_map so communities that have zero users still get a 0 // rather than keeping the last nonzero value until process restart. { - let rows: HashMap = state - .db - .usage_user_counts() - .await? - .into_iter() - .map(|r| (r.community_id, r)) - .collect(); + let rows: HashMap = user_rows.into_iter().map(|r| (r.community_id, r)).collect(); for (&id, community) in &host_map { let (human, agent) = rows.get(&id).map(|r| (r.human, r.agent)).unwrap_or((0, 0)); metrics::gauge!("buzz_community_users", "community" => community.clone(), "type" => "human") @@ -1049,17 +1080,20 @@ async fn run_usage_metrics_tick(state: &AppState) -> anyhow::Result<()> { // drops to zero emits 0 rather than retaining its last nonzero value. { const CHANNEL_TYPES: &[&str] = &["stream", "forum", "dm", "workflow"]; - let rows: HashMap<(Uuid, &str), i64> = state - .db - .usage_channel_counts() - .await? + let rows: HashMap<(Uuid, &str), i64> = channel_rows .into_iter() .filter_map(|r| { - // Map the DB string to a known static str so the key is &str. - CHANNEL_TYPES + let matched = CHANNEL_TYPES .iter() .find(|&&t| t == r.channel_type.as_str()) - .map(|&t| ((r.community_id, t), r.count)) + .map(|&t| ((r.community_id, t), r.count)); + if matched.is_none() { + warn!( + channel_type = %r.channel_type, + "usage_channel_counts: unrecognised channel_type — row skipped" + ); + } + matched }) .collect(); for (&id, community) in &host_map { @@ -1078,10 +1112,7 @@ async fn run_usage_metrics_tick(state: &AppState) -> anyhow::Result<()> { // buzz_community_messages{community} // Emit 0 for communities with no messages so dashboards don't stale-read. { - let rows: HashMap = state - .db - .usage_message_counts() - .await? + let rows: HashMap = message_rows .into_iter() .map(|r| (r.community_id, r.count)) .collect(); @@ -1097,16 +1128,20 @@ async fn run_usage_metrics_tick(state: &AppState) -> anyhow::Result<()> { // CHECK constraint over {'owner', 'admin', 'member'}. { const RELAY_ROLES: &[&str] = &["owner", "admin", "member"]; - let rows: HashMap<(Uuid, &str), i64> = state - .db - .usage_relay_member_counts() - .await? + let rows: HashMap<(Uuid, &str), i64> = relay_member_rows .into_iter() .filter_map(|r| { - RELAY_ROLES + let matched = RELAY_ROLES .iter() .find(|&&role| role == r.role.as_str()) - .map(|&role| ((r.community_id, role), r.count)) + .map(|&role| ((r.community_id, role), r.count)); + if matched.is_none() { + warn!( + role = %r.role, + "usage_relay_member_counts: unrecognised role — row skipped" + ); + } + matched }) .collect(); for (&id, community) in &host_map { @@ -1127,16 +1162,20 @@ async fn run_usage_metrics_tick(state: &AppState) -> anyhow::Result<()> { // DB enum: {'active', 'disabled', 'archived'}. { const WORKFLOW_STATUSES: &[&str] = &["active", "disabled", "archived"]; - let rows: HashMap<(Uuid, &str), i64> = state - .db - .usage_workflow_counts() - .await? + let rows: HashMap<(Uuid, &str), i64> = workflow_rows .into_iter() .filter_map(|r| { - WORKFLOW_STATUSES + let matched = WORKFLOW_STATUSES .iter() .find(|&&s| s == r.status.as_str()) - .map(|&s| ((r.community_id, s), r.count)) + .map(|&s| ((r.community_id, s), r.count)); + if matched.is_none() { + warn!( + status = %r.status, + "usage_workflow_counts: unrecognised workflow status — row skipped" + ); + } + matched }) .collect(); for (&id, community) in &host_map { @@ -1155,10 +1194,7 @@ async fn run_usage_metrics_tick(state: &AppState) -> anyhow::Result<()> { // buzz_community_git_repos{community} // Emit 0 for communities with no repos. { - let rows: HashMap = state - .db - .usage_git_repo_counts() - .await? + let rows: HashMap = git_repo_rows .into_iter() .map(|r| (r.community_id, r.count)) .collect(); @@ -1174,16 +1210,17 @@ async fn run_usage_metrics_tick(state: &AppState) -> anyhow::Result<()> { // ensures a community that was active last tick but quiet this tick reads // 0 rather than retaining its last nonzero value. - for (interval, label) in [("1 day", "1d"), ("7 days", "7d"), ("30 days", "30d")] { - let rows: HashMap = state - .db - .usage_active_user_counts(interval) - .await? - .into_iter() - .map(|r| (r.community_id, r)) - .collect(); + for (data, label) in [ + (active_users_1d, "1d"), + (active_users_7d, "7d"), + (active_users_30d, "30d"), + ] { + let rows: HashMap = data.into_iter().map(|r| (r.community_id, r)).collect(); for (&id, community) in &host_map { - let (human, agent) = rows.get(&id).map(|r| (r.human, r.agent)).unwrap_or((0, 0)); + let (human, agent, unknown) = rows + .get(&id) + .map(|r| (r.human, r.agent, r.unknown)) + .unwrap_or((0, 0, 0)); metrics::gauge!( "buzz_community_active_users", "community" => community.clone(), @@ -1198,14 +1235,18 @@ async fn run_usage_metrics_tick(state: &AppState) -> anyhow::Result<()> { "type" => "agent" ) .set(agent as f64); + metrics::gauge!( + "buzz_community_active_users", + "community" => community.clone(), + "window" => label, + "type" => "unknown" + ) + .set(unknown as f64); } } - for (interval, label) in [("1 day", "1d"), ("7 days", "7d")] { - let rows: HashMap = state - .db - .usage_active_channel_counts(interval) - .await? + for (data, label) in [(active_channels_1d, "1d"), (active_channels_7d, "7d")] { + let rows: HashMap = data .into_iter() .map(|r| (r.community_id, r.count)) .collect(); @@ -1228,32 +1269,39 @@ async fn run_usage_metrics_tick(state: &AppState) -> anyhow::Result<()> { // buzz_community_ws_connections{community} { - let conns = state.conn_manager.per_community_ws_connections(); for (&id, community) in &host_map { - let count = conns.get(&CommunityId::from_uuid(id)).copied().unwrap_or(0); + 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{community} + // 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. { - let online = state.conn_manager.per_community_users_online(); for (&id, community) in &host_map { - let count = online + let count = online_snapshot .get(&CommunityId::from_uuid(id)) .copied() .unwrap_or(0); - metrics::gauge!("buzz_community_users_online", "community" => community.clone()) + metrics::gauge!("buzz_community_users_online_pod", "community" => community.clone()) .set(count as f64); } } // buzz_community_subscriptions{community} { - let subs = state.sub_registry.per_community_subscriptions(); for (&id, community) in &host_map { - let count = subs.get(&CommunityId::from_uuid(id)).copied().unwrap_or(0); + 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); } From e8d7224e950dad5ff16eb5c39e2cfc6acd9c660d Mon Sep 17 00:00:00 2001 From: npub1mn7jgtj4w2pd0g0zeuhxsa6jy6p0rewxz4kujt98my82ahfmp72sxjexk7 Date: Fri, 10 Jul 2026 16:47:29 -0400 Subject: [PATCH 5/6] =?UTF-8?q?fix(relay):=20address=20C1=20(kind=C3=97com?= =?UTF-8?q?munity=20cardinality)=20and=20K1=20(per-community=20knob)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit C1 — cardinality fix for buzz_events_received_total and buzz_events_stored_total: - bounded_kind_label passes through all 10k values in 20000..=29999 (client-controlled ephemeral range). Crossing kind × community with per-community tags produces up to millions of series — this PR had added 'community' to both counters. - Fix: revert both counters to {kind}-only (fleet-wide, pre-PR shape). - Add buzz_community_events_received_total{community} — a new counter with community-only label for per-community throughput graphs. Never cross kind × community. K1 — BUZZ_USAGE_METRICS_PER_COMMUNITY cardinality knob: - New PerCommunityMode enum parsed from env at startup. - Three modes: 'all' (default), 'off' (fleet totals only), 'top:' (per-community series for the k communities with most messages — reuses message_rows already computed each tick, no extra query). - Fleet-wide totals (buzz_total_*) always emit regardless of mode, preserving aggregate visibility at zero extra cardinality cost: buzz_total_users, buzz_total_channels, buzz_total_messages, buzz_total_relay_members, buzz_total_workflows, buzz_total_git_repos, buzz_total_active_users, buzz_total_active_channels, buzz_total_ws_connections, buzz_total_users_online_pod, buzz_total_subscriptions. - Malformed 'top:' values warn! and fall back to 'all'. PR description updated to reflect final state (C1-K1 + earlier C2-N3). Co-authored-by: Will Pfleger Signed-off-by: Will Pfleger --- crates/buzz-relay/src/handlers/event.rs | 20 ++- crates/buzz-relay/src/main.rs | 190 +++++++++++++++++++++++- 2 files changed, 200 insertions(+), 10 deletions(-) diff --git a/crates/buzz-relay/src/handlers/event.rs b/crates/buzz-relay/src/handlers/event.rs index ad70edcf4a..9e3bb05d86 100644 --- a/crates/buzz-relay/src/handlers/event.rs +++ b/crates/buzz-relay/src/handlers/event.rs @@ -591,9 +591,16 @@ pub async fn handle_event(event: Event, conn: Arc, state: Arc kind_str.clone()).increment(1); + // Per-community volume counter: community-only, no kind tag. + // Use this for per-community throughput graphs; the fleet counter above + // for per-kind breakdowns. metrics::counter!( - "buzz_events_received_total", - "kind" => kind_str.clone(), + "buzz_community_events_received_total", "community" => conn.tenant.host().to_owned() ) .increment(1); @@ -695,12 +702,9 @@ pub async fn handle_event(event: Event, conn: Arc, state: Arc { if result.accepted { - metrics::counter!( - "buzz_events_stored_total", - "kind" => kind_str, - "community" => conn.tenant.host().to_owned() - ) - .increment(1); + // Fleet-wide stored counter: kind-only, no community tag. + // Same cardinality rationale as buzz_events_received_total above. + metrics::counter!("buzz_events_stored_total", "kind" => kind_str).increment(1); info!( event_id = %result.event_id, kind = kind_u32, diff --git a/crates/buzz-relay/src/main.rs b/crates/buzz-relay/src/main.rs index 8c6b32f18f..f6d367b8fb 100644 --- a/crates/buzz-relay/src/main.rs +++ b/crates/buzz-relay/src/main.rs @@ -29,6 +29,58 @@ fn buzz_auto_migrate_enabled(value: Option<&str>) -> bool { }) } +/// Controls how many per-community gauge series the usage poller emits. +/// +/// Datadog cost is proportional to the number of unique time-series. With ~25 +/// gauge label combinations per community, a relay hosting thousands of +/// communities would incur five-figure monthly costs if every community always +/// gets a full set of series. This knob is the cost lever. +/// +/// Fleet-wide totals (`buzz_total_*`) always emit regardless of mode. +/// +/// Set via `BUZZ_USAGE_METRICS_PER_COMMUNITY`: +/// - `all` — emit per-community series for every community (default) +/// - `off` — suppress all per-community series; fleet totals only +/// - `top:` — emit per-community series for the k communities with the +/// most stored messages (uses message_counts already computed each tick — +/// no extra query) +#[derive(Debug, Clone)] +enum PerCommunityMode { + All, + Off, + Top(usize), +} + +impl PerCommunityMode { + 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, + s if s.starts_with("top:") => { + let k: usize = s["top:".len()..].parse().unwrap_or_else(|_| { + warn!( + value = s, + "BUZZ_USAGE_METRICS_PER_COMMUNITY: invalid top: — defaulting to all" + ); + usize::MAX + }); + PerCommunityMode::Top(k) + } + other => { + warn!( + value = other, + "BUZZ_USAGE_METRICS_PER_COMMUNITY: unknown value — defaulting to all" + ); + PerCommunityMode::All + } + } + } +} + #[tokio::main] async fn main() -> anyhow::Result<()> { // Install the ring CryptoProvider for rustls. Required before any rustls @@ -821,6 +873,7 @@ 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()) @@ -844,7 +897,7 @@ 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).await { + if let Err(e) = run_usage_metrics_tick(&usage_state, &per_community_mode).await { error!(error = %e, "Usage metrics tick failed — skipping"); } } @@ -1024,7 +1077,10 @@ fn reminder_to_event(reminder: &buzz_db::event::DueReminder) -> nostr::Event { /// /// All DB-derived gauges use absolute SET (not increment), so they self-heal /// after a missed tick or a pod restart. -async fn run_usage_metrics_tick(state: &AppState) -> anyhow::Result<()> { +async fn run_usage_metrics_tick( + state: &AppState, + per_community_mode: &PerCommunityMode, +) -> 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(); @@ -1054,6 +1110,33 @@ async fn run_usage_metrics_tick(state: &AppState) -> anyhow::Result<()> { 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. + // + // top: ranking reuses message_rows (already fetched above — no extra + // query) as the activity proxy: communities with more stored messages are + // the ones operators most want to dashboard per-community. + let message_by_id: HashMap = message_rows + .iter() + .map(|r| (r.community_id, r.count)) + .collect(); + + let active_set: std::collections::HashSet = match per_community_mode { + PerCommunityMode::All => host_map.keys().copied().collect(), + PerCommunityMode::Off => std::collections::HashSet::new(), + PerCommunityMode::Top(k) => { + let mut ranked: Vec = host_map.keys().copied().collect(); + ranked.sort_by(|a, b| { + let ma = message_by_id.get(a).copied().unwrap_or(0); + let mb = message_by_id.get(b).copied().unwrap_or(0); + mb.cmp(&ma) // descending: most messages first + }); + ranked.into_iter().take(*k).collect() + } + }; + // --- Publish phase: emit all metrics now that every query succeeded --- // --- A. Adoption stocks (DB-polled) --- @@ -1066,7 +1149,17 @@ async fn run_usage_metrics_tick(state: &AppState) -> anyhow::Result<()> { // rather than keeping the last nonzero value until process restart. { let rows: HashMap = user_rows.into_iter().map(|r| (r.community_id, r)).collect(); + // Fleet totals (always emitted). + let (total_human, total_agent): (i64, i64) = rows + .values() + .fold((0, 0), |(h, a), r| (h + r.human, a + r.agent)); + 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 { + if !active_set.contains(&id) { + continue; + } let (human, agent) = rows.get(&id).map(|r| (r.human, r.agent)).unwrap_or((0, 0)); metrics::gauge!("buzz_community_users", "community" => community.clone(), "type" => "human") .set(human as f64); @@ -1096,7 +1189,19 @@ async fn run_usage_metrics_tick(state: &AppState) -> anyhow::Result<()> { matched }) .collect(); + // Fleet totals (always emitted). + for &ct in CHANNEL_TYPES { + let total: i64 = host_map + .keys() + .map(|id| rows.get(&(*id, ct)).copied().unwrap_or(0)) + .sum(); + metrics::gauge!("buzz_total_channels", "type" => ct).set(total as f64); + } + // Per-community series (gated by active_set). for (&id, community) in &host_map { + if !active_set.contains(&id) { + continue; + } for &ct in CHANNEL_TYPES { let count = rows.get(&(id, ct)).copied().unwrap_or(0); metrics::gauge!( @@ -1116,7 +1221,14 @@ async fn run_usage_metrics_tick(state: &AppState) -> anyhow::Result<()> { .into_iter() .map(|r| (r.community_id, r.count)) .collect(); + // Fleet total (always emitted). + 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 { + if !active_set.contains(&id) { + continue; + } let count = rows.get(&id).copied().unwrap_or(0); metrics::gauge!("buzz_community_messages", "community" => community.clone()) .set(count as f64); @@ -1144,7 +1256,19 @@ async fn run_usage_metrics_tick(state: &AppState) -> anyhow::Result<()> { matched }) .collect(); + // Fleet totals (always emitted). + for &role in RELAY_ROLES { + let total: i64 = host_map + .keys() + .map(|id| rows.get(&(*id, role)).copied().unwrap_or(0)) + .sum(); + 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 { + if !active_set.contains(&id) { + continue; + } for &role in RELAY_ROLES { let count = rows.get(&(id, role)).copied().unwrap_or(0); metrics::gauge!( @@ -1178,7 +1302,19 @@ async fn run_usage_metrics_tick(state: &AppState) -> anyhow::Result<()> { matched }) .collect(); + // Fleet totals (always emitted). + for &status in WORKFLOW_STATUSES { + let total: i64 = host_map + .keys() + .map(|id| rows.get(&(*id, status)).copied().unwrap_or(0)) + .sum(); + metrics::gauge!("buzz_total_workflows", "status" => status).set(total as f64); + } + // Per-community series (gated by active_set). for (&id, community) in &host_map { + if !active_set.contains(&id) { + continue; + } for &status in WORKFLOW_STATUSES { let count = rows.get(&(id, status)).copied().unwrap_or(0); metrics::gauge!( @@ -1198,7 +1334,14 @@ async fn run_usage_metrics_tick(state: &AppState) -> anyhow::Result<()> { .into_iter() .map(|r| (r.community_id, r.count)) .collect(); + // Fleet total (always emitted). + 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 { + if !active_set.contains(&id) { + continue; + } let count = rows.get(&id).copied().unwrap_or(0); metrics::gauge!("buzz_community_git_repos", "community" => community.clone()) .set(count as f64); @@ -1216,7 +1359,22 @@ async fn run_usage_metrics_tick(state: &AppState) -> anyhow::Result<()> { (active_users_30d, "30d"), ] { let rows: HashMap = data.into_iter().map(|r| (r.community_id, r)).collect(); + // Fleet totals (always emitted). + let (total_human, total_agent, total_unknown): (i64, i64, i64) = + rows.values().fold((0, 0, 0), |(h, a, u), r| { + (h + r.human, a + r.agent, u + r.unknown) + }); + metrics::gauge!("buzz_total_active_users", "window" => label, "type" => "human") + .set(total_human as f64); + metrics::gauge!("buzz_total_active_users", "window" => label, "type" => "agent") + .set(total_agent as f64); + 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 { + if !active_set.contains(&id) { + continue; + } let (human, agent, unknown) = rows .get(&id) .map(|r| (r.human, r.agent, r.unknown)) @@ -1250,7 +1408,14 @@ async fn run_usage_metrics_tick(state: &AppState) -> anyhow::Result<()> { .into_iter() .map(|r| (r.community_id, r.count)) .collect(); + // Fleet total (always emitted). + 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 { + if !active_set.contains(&id) { + continue; + } let count = rows.get(&id).copied().unwrap_or(0); metrics::gauge!( "buzz_community_active_channels", @@ -1269,7 +1434,14 @@ async fn run_usage_metrics_tick(state: &AppState) -> anyhow::Result<()> { // 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() @@ -1285,7 +1457,14 @@ async fn run_usage_metrics_tick(state: &AppState) -> anyhow::Result<()> { // 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() @@ -1297,7 +1476,14 @@ async fn run_usage_metrics_tick(state: &AppState) -> anyhow::Result<()> { // 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() From 493d3cf6df66e9f3d85efb2ffd76e8cf53a933a1 Mon Sep 17 00:00:00 2001 From: npub1mn7jgtj4w2pd0g0zeuhxsa6jy6p0rewxz4kujt98my82ahfmp72sxjexk7 Date: Fri, 10 Jul 2026 17:21:24 -0400 Subject: [PATCH 6/6] fix(relay): use true randomness for jitter, drop top:k mode MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit F1: Replace PID-derived first-tick jitter with rand::random::() % interval_secs. In K8s the relay is typically PID 1 in every pod, so the old hash of std::process::id() produced the same delay on every pod — defeating the thundering-herd mitigation entirely. F2: Remove PerCommunityMode::Top(usize) from the BUZZ_USAGE_METRICS_PER_COMMUNITY knob. As designed, top: was not a real cardinality bound: HashMap iteration order made community selection non-deterministic across pods (union semantics, not k), and the zero-fill design provides no eviction path for communities that leave the active set, so their last gauge values persist until pod restart. The knob now accepts only all|off. A proper top: implementation (stable tie-breaking, gauge idle-timeout lifecycle) is tracked as a fast-follow PR. Co-authored-by: Will Pfleger Signed-off-by: Will Pfleger --- crates/buzz-relay/src/main.rs | 47 ++++++++--------------------------- 1 file changed, 10 insertions(+), 37 deletions(-) diff --git a/crates/buzz-relay/src/main.rs b/crates/buzz-relay/src/main.rs index f6d367b8fb..df68ec2700 100644 --- a/crates/buzz-relay/src/main.rs +++ b/crates/buzz-relay/src/main.rs @@ -39,16 +39,16 @@ fn buzz_auto_migrate_enabled(value: Option<&str>) -> bool { /// Fleet-wide totals (`buzz_total_*`) always emit regardless of mode. /// /// Set via `BUZZ_USAGE_METRICS_PER_COMMUNITY`: -/// - `all` — emit per-community series for every community (default) -/// - `off` — suppress all per-community series; fleet totals only -/// - `top:` — emit per-community series for the k communities with the -/// most stored messages (uses message_counts already computed each tick — -/// no extra query) +/// - `all` — emit per-community series for every community (default) +/// - `off` — suppress all per-community series; fleet totals only +/// +/// A `top:` mode (per-community series for the k most-active communities) +/// 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 { All, Off, - Top(usize), } impl PerCommunityMode { @@ -60,16 +60,6 @@ impl PerCommunityMode { match raw.as_str() { "" | "all" => PerCommunityMode::All, "off" => PerCommunityMode::Off, - s if s.starts_with("top:") => { - let k: usize = s["top:".len()..].parse().unwrap_or_else(|_| { - warn!( - value = s, - "BUZZ_USAGE_METRICS_PER_COMMUNITY: invalid top: — defaulting to all" - ); - usize::MAX - }); - PerCommunityMode::Top(k) - } other => { warn!( value = other, @@ -886,9 +876,10 @@ async fn main() -> anyhow::Result<()> { // Jitter the first tick by a random fraction of the interval so // that a rolling deploy with N pods doesn't hammer the DB // simultaneously at boot. Each pod picks a start delay in - // [0, interval_secs) using its PID as the seed. - let jitter_secs = - (std::process::id() as u64).wrapping_mul(2_654_435_761) % interval_secs; + // [0, interval_secs) using true per-process randomness (PID-derived + // seeds are unsafe in containers where the relay is typically PID 1 + // in every pod, which would make all pods compute the same delay). + let jitter_secs = rand::random::() % interval_secs; tokio::time::sleep(std::time::Duration::from_secs(jitter_secs)).await; let mut interval = tokio::time::interval(std::time::Duration::from_secs(interval_secs)); @@ -1114,27 +1105,9 @@ async fn run_usage_metrics_tick( // // `active_set` is the subset of host_map IDs that get per-community gauges // this tick. Fleet-wide totals (buzz_total_*) always emit regardless. - // - // top: ranking reuses message_rows (already fetched above — no extra - // query) as the activity proxy: communities with more stored messages are - // the ones operators most want to dashboard per-community. - let message_by_id: HashMap = message_rows - .iter() - .map(|r| (r.community_id, r.count)) - .collect(); - let active_set: std::collections::HashSet = match per_community_mode { PerCommunityMode::All => host_map.keys().copied().collect(), PerCommunityMode::Off => std::collections::HashSet::new(), - PerCommunityMode::Top(k) => { - let mut ranked: Vec = host_map.keys().copied().collect(); - ranked.sort_by(|a, b| { - let ma = message_by_id.get(a).copied().unwrap_or(0); - let mb = message_by_id.get(b).copied().unwrap_or(0); - mb.cmp(&ma) // descending: most messages first - }); - ranked.into_iter().take(*k).collect() - } }; // --- Publish phase: emit all metrics now that every query succeeded ---