Instrument database pool roles

ARCHITECTURE.md

@@ -229,15 +229,14 @@ When the relay receives `["EVENT", <event>]`, the handler in `handlers/event.rs`
 4. EPHEMERAL ROUTE   — kind 20000–29999 → ephemeral sub-pipeline (see below)
 5. VERIFY            — spawn_blocking(verify_event) — Schnorr sig + ID hash
 6. MEMBERSHIP        — channel_id in event tags? → check_channel_membership
-7. DB INSERT         — db.insert_event (ON CONFLICT DO NOTHING — idempotent)
+7. DB INSERT         — db.insert_event (idempotent; search_tsv generated synchronously)
 8. REDIS PUBLISH     — pubsub.publish_event (if channel-scoped)
 9. FAN-OUT           — sub_registry.fan_out → conn_manager.send_to
-10. SEARCH INDEX     — search_index_tx.send (bounded worker queue, non-blocking)
-11. AUDIT LOG        — audit.log (spawned async, non-blocking)
-12. WORKFLOW TRIGGER — wf.on_event (spawned async, excludes kinds 46001–46012)
+10. AUDIT LOG        — audit.log (spawned async, non-blocking)
+11. WORKFLOW TRIGGER — wf.on_event (spawned async, excludes kinds 46001–46012)
 ```
 
-Steps 10–12 are fire-and-forget. Search indexing is sent to a bounded worker queue (`search_index_tx`, capacity 1000); audit and workflow triggers are spawned as independent async tasks. A failure in any of these does not fail the event submission. The client receives `["OK", <id>, true, ""]` at the end of the pipeline, not immediately after DB insert.
+Steps 10–11 are fire-and-forget: audit and workflow triggers are spawned as independent async tasks. A failure in either does not fail the event submission. Search has no asynchronous indexing step: Postgres maintains the generated `events.search_tsv` column as part of step 7, and `buzz-search` only queries it. The client receives `["OK", <id>, true, ""]` at the end of the pipeline, not immediately after DB insert.
 
 Step 9 (fan-out) explicitly **excludes** global subscriptions (no `channel_id` constraint) from channel-scoped events — global subscriptions do NOT receive events from private channels, regardless of filter match. This is a deliberate security boundary: only subscriptions scoped to an accessible `channel_id` receive those events.
 
@@ -600,11 +599,22 @@ pub struct AppState {
     pub handler_semaphore: Arc<Semaphore>,    // 1024 concurrent handlers
     pub relay_keypair: nostr::Keys,           // relay identity
     pub local_event_ids: moka::sync::Cache,   // local-echo dedup
-    pub search_index_tx: mpsc::Sender,        // bounded search worker queue
     // + config, redis_pool, membership_cache, media_storage, shutdown state
 }
 ```
 
+Postgres pools use the closed physical role vocabulary `writer`, `reader`,
+`audit`, and `search`. The periodic sampler retains cheap SQLx pool handles and
+exports `buzz_db_pool_connections{pool_role,state}` for the bounded states
+`idle` and `active`, `buzz_db_pool_max_connections{pool_role}` for capacity,
+and `buzz_db_pool_configured{pool_role}`. All four roles are always present (16
+raw gauge series total); absent optional pools report zero. Current pool size is
+exactly the sum of its idle and active connection gauges. The legacy
+`buzz_db_pool_*` writer gauges and `buzz_db_read_pool_*` reader gauges remain
+for dashboard compatibility. Pool sizing and aggregate deployment connection
+budgets remain configuration/deployment concerns rather than a relay pool
+manager.
+
 **`ConnectionState`** (per-connection):
 
 ```rust

crates/buzz-db/src/lib.rs

@@ -31,8 +31,8 @@ pub mod error;
 mod test_support;
 
 pub use runtime::{
-    insert_mentions, migration, replica_fence, Db, DbConfig, DbPoolStats, DbReadinessOutcome,
-    ReadSession,
+    insert_mentions, migration, replica_fence, Db, DbConfig, DbPoolRole, DbPoolStats,
+    DbReadinessOutcome, ReadSession,
 };
 
 /// Valid low-cardinality `(pool_role, operation)` pairs for pool-acquisition telemetry.

crates/buzz-db/src/runtime/mod.rs

@@ -427,6 +427,52 @@ pub struct DbPoolStats {
     pub max: u32,
 }
 
+impl DbPoolStats {
+    /// Read a utilization snapshot from a physical SQLx pool handle.
+    pub fn from_pool(pool: &sqlx::PgPool) -> Self {
+        Self {
+            size: pool.size(),
+            idle: pool.num_idle() as u32,
+            max: pool.options().get_max_connections(),
+        }
+    }
+
+    /// Connections currently checked out from the pool.
+    pub const fn active(self) -> u32 {
+        self.size.saturating_sub(self.idle)
+    }
+}
+
+/// Physical role of a Postgres connection pool owned by the relay.
+///
+/// This vocabulary is intentionally closed so metrics labels remain bounded.
+#[derive(Debug, Clone, Copy, Eq, PartialEq)]
+pub enum DbPoolRole {
+    /// Primary pool used for authoritative writes and consistency-sensitive reads.
+    Writer,
+    /// Optional read-replica pool used for eligible lag-tolerant reads.
+    Reader,
+    /// Optional independent pool used by the hash-chain audit service.
+    Audit,
+    /// Independent pool used for Postgres full-text search queries.
+    Search,
+}
+
+impl DbPoolRole {
+    /// Every physical pool role, in stable metrics-contract order.
+    pub const ALL: [Self; 4] = [Self::Writer, Self::Reader, Self::Audit, Self::Search];
+
+    /// Stable low-cardinality metrics label for this role.
+    pub const fn as_str(self) -> &'static str {
+        match self {
+            Self::Writer => "writer",
+            Self::Reader => "reader",
+            Self::Audit => "audit",
+            Self::Search => "search",
+        }
+    }
+}
+
 /// Bounded outcome of the Postgres portion of a relay readiness check.
 ///
 /// The variants deliberately separate waiting for a pooled connection from
@@ -1073,9 +1119,9 @@ impl Db {
     /// exactly the ratio of the two pool sizes — in the direction that hides
     /// the problem.
     pub fn read_pool_stats(&self) -> Option<DbPoolStats> {
-        self.read_pool.as_ref().map(|p| DbPoolStats {
-            size: p.size(),
-            idle: p.num_idle() as u32,
+        self.read_pool.as_ref().map(|pool| DbPoolStats {
+            size: pool.size(),
+            idle: pool.num_idle() as u32,
             max: self.read_max_connections,
         })
     }
@@ -1245,6 +1291,57 @@ impl Db {
     }
 }
 
+#[cfg(test)]
+mod pool_role_tests {
+    use super::{DbPoolRole, DbPoolStats};
+
+    #[test]
+    fn database_pool_role_vocabulary_is_exact() {
+        assert_eq!(
+            DbPoolRole::ALL.map(DbPoolRole::as_str),
+            ["writer", "reader", "audit", "search"]
+        );
+    }
+
+    #[test]
+    fn database_pool_active_connections_use_saturating_arithmetic() {
+        assert_eq!(
+            DbPoolStats {
+                size: 12,
+                idle: 5,
+                max: 20,
+            }
+            .active(),
+            7
+        );
+        assert_eq!(
+            DbPoolStats {
+                size: 2,
+                idle: 3,
+                max: 20,
+            }
+            .active(),
+            0
+        );
+    }
+
+    /// `from_pool` must read the live SQLx handle rather than a cached copy.
+    /// A lazy pool never opens a connection, so this stays infrastructure-free.
+    #[tokio::test]
+    async fn database_pool_stats_are_read_from_the_physical_pool_handle() {
+        let pool = sqlx::postgres::PgPoolOptions::new()
+            .max_connections(7)
+            .connect_lazy(&crate::test_support::database_url())
+            .expect("construct lazy pool");
+
+        let stats = DbPoolStats::from_pool(&pool);
+        assert_eq!(stats.size, 0);
+        assert_eq!(stats.idle, 0);
+        assert_eq!(stats.active(), 0);
+        assert_eq!(stats.max, 7);
+    }
+}
+
 #[cfg(test)]
 #[path = "tests.rs"]
 mod postgres_tests;

crates/buzz-db/src/runtime/observability.rs

@@ -7,6 +7,8 @@ use std::future::Future;
 use std::sync::Mutex;
 use std::time::{Duration, Instant};
 
+use super::DbPoolRole;
+
 /// One valid pool/operation acquisition family.
 ///
 /// Keeping role and operation in one enum makes invalid combinations
@@ -111,9 +113,9 @@ impl PoolOperation {
     pub(crate) const fn pool_role(self) -> &'static str {
         match self {
             Self::ReaderBootstrap | Self::ReaderAuthorization | Self::ReaderSubscriptionHistory => {
-                "reader"
+                DbPoolRole::Reader.as_str()
             }
-            _ => "writer",
+            _ => DbPoolRole::Writer.as_str(),
         }
     }
 
@@ -138,17 +140,17 @@ impl PoolOperation {
 }
 
 pub(crate) const POOL_ACQUIRE_VALID_PAIRS: [(&str, &str); 11] = [
-    ("writer", "bootstrap"),
-    ("reader", "bootstrap"),
-    ("writer", "readiness"),
-    ("writer", "tenant_resolution"),
-    ("writer", "authentication"),
-    ("writer", "authorization"),
-    ("reader", "authorization"),
-    ("writer", "subscription_history"),
-    ("reader", "subscription_history"),
-    ("writer", "event_write"),
-    ("writer", "maintenance"),
+    (DbPoolRole::Writer.as_str(), "bootstrap"),
+    (DbPoolRole::Reader.as_str(), "bootstrap"),
+    (DbPoolRole::Writer.as_str(), "readiness"),
+    (DbPoolRole::Writer.as_str(), "tenant_resolution"),
+    (DbPoolRole::Writer.as_str(), "authentication"),
+    (DbPoolRole::Writer.as_str(), "authorization"),
+    (DbPoolRole::Reader.as_str(), "authorization"),
+    (DbPoolRole::Writer.as_str(), "subscription_history"),
+    (DbPoolRole::Reader.as_str(), "subscription_history"),
+    (DbPoolRole::Writer.as_str(), "event_write"),
+    (DbPoolRole::Writer.as_str(), "maintenance"),
 ];
 
 /// Eleven valid pairs × (12 histogram series + 4 outcome counters + 1 gauge).
@@ -355,7 +357,7 @@ fn publish_waiters(pair: PoolOperation, value: u64) {
 /// idle eviction cannot turn an expected zero into ambiguous missing data.
 pub(crate) fn refresh_pool_waiters(include_reader: bool) {
     for pair in PoolOperation::ALL {
-        if pair.pool_role() == "reader" && !include_reader {
+        if pair.pool_role() == DbPoolRole::Reader.as_str() && !include_reader {
             continue;
         }
         let waiters = POOL_WAITERS[pair.index()]

crates/buzz-relay/src/main.rs

@@ -437,15 +437,16 @@ async fn run_relay_main(boot: BootTracker) -> anyhow::Result<()> {
         Err(e) => error!("Failed to backfill d_tags: {e}"),
     }
 
-    let audit = if config.audit_enabled {
+    let (audit, audit_metrics_pool) = if config.audit_enabled {
         let audit_pool = connect_audit_pool(&db_config)
             .await
             .map_err(|e| anyhow::anyhow!("Audit DB connection failed: {e}"))?;
         info!("Audit service ready");
-        Some(AuditService::new(audit_pool))
+        let metrics_pool = audit_pool.clone();
+        (Some(AuditService::new(audit_pool)), Some(metrics_pool))
     } else {
         info!("Audit logging disabled by BUZZ_AUDIT_ENABLED");
-        None
+        (None, None)
     };
 
     let redis_pool = {
@@ -495,6 +496,7 @@ async fn run_relay_main(boot: BootTracker) -> anyhow::Result<()> {
         .connect(search_db_url)
         .await
         .map_err(|e| anyhow::anyhow!("Search DB connection failed: {e}"))?;
+    let search_metrics_pool = search_pool.clone();
     let search = SearchService::new(search_pool);
     info!(
         replica = config.read_database_url.is_some(),
@@ -1101,20 +1103,18 @@ async fn run_relay_main(boot: BootTracker) -> anyhow::Result<()> {
             loop {
                 interval.tick().await;
                 let db_stats = pool_state.db.pool_stats();
-                let active = db_stats.size.saturating_sub(db_stats.idle);
-                metrics::gauge!("buzz_db_pool_size").set(db_stats.size as f64);
-                metrics::gauge!("buzz_db_pool_idle").set(db_stats.idle as f64);
-                metrics::gauge!("buzz_db_pool_active").set(active as f64);
-                metrics::gauge!("buzz_db_pool_max").set(db_stats.max as f64);
+                let read_stats = pool_state.db.read_pool_stats();
+                relay_metrics::record_db_pool_metrics(relay_metrics::DbPoolMetricsInput {
+                    writer: db_stats,
+                    reader: read_stats,
+                    audit: audit_metrics_pool
+                        .as_ref()
+                        .map(buzz_db::DbPoolStats::from_pool),
+                    search: buzz_db::DbPoolStats::from_pool(&search_metrics_pool),
+                });
                 pool_state.db.refresh_pool_waiter_metrics();
 
-                if let Some(read_stats) = pool_state.db.read_pool_stats() {
-                    let read_active = read_stats.size.saturating_sub(read_stats.idle);
-                    metrics::gauge!("buzz_db_read_pool_size").set(read_stats.size as f64);
-                    metrics::gauge!("buzz_db_read_pool_idle").set(read_stats.idle as f64);
-                    metrics::gauge!("buzz_db_read_pool_active").set(read_active as f64);
-                    metrics::gauge!("buzz_db_read_pool_max").set(read_stats.max as f64);
-
+                if read_stats.is_some() {
                     // Fence observability: 1 when replica routing is
                     // eligible, and the verified-freshness lag in seconds.
                     // Closed/stale fence reports open=0 with lag untouched.