fix(obsv): wire live per-database metrics sampler (fixes #386)

nodedb-mem/src/governor/metrics.rs

@@ -5,11 +5,25 @@
 
 use std::sync::atomic::Ordering;
 
+use nodedb_types::DatabaseId;
+
 use super::core::MemoryGovernor;
 use crate::engine::EngineId;
 use crate::pressure::{PressureLevel, PressureThresholds};
 
 impl MemoryGovernor {
+    /// Current allocated memory in bytes for a database.
+    ///
+    /// Returns 0 if the database has no active allocations.
+    pub fn database_usage_bytes(&self, db: DatabaseId) -> usize {
+        self.database_budgets
+            .read()
+            .unwrap_or_else(|p| p.into_inner())
+            .get(&db)
+            .map(|b| b.allocated.load(Ordering::Relaxed))
+            .unwrap_or(0)
+    }
+
     /// Total memory allocated across all engines (engine-layer sum). A
     /// separate aggregate from [`global_utilization_percent`](Self::global_utilization_percent),
     /// which reads the global counter admission enforces the ceiling against.
@@ -164,4 +178,16 @@ mod tests {
         assert_eq!(gov.engine_pressure(EngineId::Vector), PressureLevel::Normal);
         assert_eq!(gov.worst_engine_pressure(), PressureLevel::Critical);
     }
+
+    #[test]
+    fn database_usage_bytes_tracks_allocation() {
+        let gov = MemoryGovernor::new(test_config()).unwrap();
+        gov.set_database_budget(db(), 10_000);
+        assert_eq!(gov.database_usage_bytes(db()), 0);
+
+        let _tok = gov
+            .try_reserve(db(), tenant(), EngineId::Vector, 512)
+            .unwrap();
+        assert_eq!(gov.database_usage_bytes(db()), 512);
+    }
 }

nodedb-mem/src/governor/reserve.rs

@@ -13,6 +13,7 @@ use crate::error::{MemError, Result};
 use crate::over_release::ReleaseIdentity;
 use crate::reservation_token::{ReservationParams, ReservationToken};
 use crate::reserve_scope::{ReserveScope, ReservedLayers};
+use crate::scoped_budget::ScopedBudget;
 
 /// Build the token both entry points return. They differ in how they reach
 /// a committed [`ReservedLayers`], never in what they build from one.
@@ -36,7 +37,28 @@ fn token_from_layers(
 }
 
 impl MemoryGovernor {
-    // ── 4-arity reservation ───────────────────────────────────────────────────
+    /// Run `f` with the database's `ScopedBudget`, inserting an uncapped one if absent.
+    ///
+    /// Takes a read lock for existing entries to avoid contending with writer locks.
+    /// Takes a write lock only on missing entries.
+    fn with_database_budget<R>(&self, db: DatabaseId, f: impl FnOnce(&ScopedBudget) -> R) -> R {
+        let read_map = self
+            .database_budgets
+            .read()
+            .unwrap_or_else(|p| p.into_inner());
+        if let Some(budget) = read_map.get(&db) {
+            return f(budget);
+        }
+        drop(read_map);
+        let mut write_map = self
+            .database_budgets
+            .write()
+            .unwrap_or_else(|p| p.into_inner());
+        let budget = write_map
+            .entry(db)
+            .or_insert_with(|| ScopedBudget::new(None));
+        f(budget)
+    }
 
     /// Reserve `size` bytes for one (database, tenant, engine) triple.
     ///
@@ -61,25 +83,18 @@ impl MemoryGovernor {
 
         scope.try_credit_global()?;
 
-        {
-            let map = self
-                .database_budgets
-                .read()
-                .unwrap_or_else(|p| p.into_inner());
-            if let Some(budget) = map.get(&db) {
-                match budget.try_reserve(size) {
-                    Ok(arc) => scope.credit_database(arc),
-                    Err(denied) => {
-                        return Err(MemError::DatabaseBudgetExhausted {
-                            db,
-                            requested: size,
-                            available: budget.available(),
-                            limit: denied.limit,
-                        });
-                    }
-                }
+        self.with_database_budget(db, |budget| match budget.try_reserve(size) {
+            Ok(arc) => {
+                scope.credit_database(arc);
+                Ok(())
             }
-        }
+            Err(denied) => Err(MemError::DatabaseBudgetExhausted {
+                db,
+                requested: size,
+                available: budget.available(),
+                limit: denied.limit,
+            }),
+        })?;
 
         {
             let map = self
@@ -137,15 +152,9 @@ impl MemoryGovernor {
 
         scope.credit_global_unchecked();
 
-        {
-            let map = self
-                .database_budgets
-                .read()
-                .unwrap_or_else(|p| p.into_inner());
-            if let Some(budget) = map.get(&db) {
-                scope.credit_database(budget.credit(size));
-            }
-        }
+        self.with_database_budget(db, |budget| {
+            scope.credit_database(budget.credit(size));
+        });
 
         {
             let map = self
@@ -493,4 +502,18 @@ mod tests {
             tok.err()
         );
     }
+
+    #[test]
+    fn uncapped_database_usage_tracks_allocation() {
+        let gov = MemoryGovernor::new(test_config()).unwrap();
+        assert_eq!(gov.database_usage_bytes(db()), 0);
+
+        let tok = gov
+            .try_reserve(db(), tenant(), EngineId::Vector, 512)
+            .unwrap();
+        assert_eq!(gov.database_usage_bytes(db()), 512);
+
+        drop(tok);
+        assert_eq!(gov.database_usage_bytes(db()), 0);
+    }
 }

nodedb-mem/src/scoped_budget.rs

@@ -25,7 +25,7 @@ pub(crate) struct BudgetDenied {
 }
 
 impl ScopedBudget {
-    fn new(limit: Option<usize>) -> Self {
+    pub(crate) fn new(limit: Option<usize>) -> Self {
         Self {
             limit,
             allocated: Arc::new(AtomicUsize::new(0)),

nodedb/src/bootstrap/background_loops/spawn.rs

@@ -60,6 +60,7 @@ pub fn spawn_background_loops(
     maintenance::spawn_pending_leave_cleanup(shared);
     maintenance::spawn_orphaned_drain_sweep(shared);
     timers::spawn_tenant_timers(shared);
+    crate::bootstrap::database_metrics_sampler::spawn_database_metrics_sampler(shared);
     spawn_security_loops(shared);
 
     // Checkpoint manager. It is handed the Event Plane's watermark store

nodedb/src/bootstrap/database_metrics_sampler.rs

@@ -0,0 +1,200 @@
+// SPDX-License-Identifier: BUSL-1.1
+
+//! Background sampler for per-database Prometheus metrics.
+//!
+//! Samples live gauges every 10 seconds across memory, admission, storage,
+//! virtual queue depths, and rolling-window WAL fsync latency.
+
+use std::collections::HashMap;
+use std::sync::Arc;
+use std::time::Duration;
+
+use tracing::info;
+
+use crate::control::metrics::histogram::HistogramSnapshot;
+use crate::control::state::SharedState;
+
+/// Spawn the 10-second periodic database metrics sampler.
+pub fn spawn_database_metrics_sampler(shared: &Arc<SharedState>) {
+    let shared_sampler = Arc::clone(shared);
+    crate::control::shutdown::spawn_loop(
+        &shared.loop_registry,
+        &shared.shutdown,
+        "database_metrics_sampler",
+        crate::control::shutdown::ShutdownPhase::DrainingControlPlane,
+        move |mut shutdown| async move {
+            let mut tick = tokio::time::interval(Duration::from_secs(10));
+            let mut last_fsync_snapshot: HistogramSnapshot = shared_sampler
+                .system_metrics
+                .as_ref()
+                .map(|s| s.wal_fsync_seconds.snapshot_counts())
+                .unwrap_or_default();
+
+            loop {
+                tokio::select! {
+                    _ = shutdown.wait_cancelled() => break,
+                    _ = tick.tick() => {}
+                }
+                if shutdown.is_cancelled() {
+                    break;
+                }
+                let catalog = shared_sampler.credentials.catalog();
+                let databases = match catalog.list_databases() {
+                    Ok(d) => d,
+                    Err(e) => {
+                        tracing::warn!(error = %e, "database_metrics_sampler: catalog list error");
+                        continue;
+                    }
+                };
+
+                // Compute rolling P99 fsync latency across the 10s tick window.
+                let curr_fsync_snapshot = shared_sampler
+                    .system_metrics
+                    .as_ref()
+                    .map(|s| s.wal_fsync_seconds.snapshot_counts())
+                    .unwrap_or_default();
+                let wal_latency_p99 =
+                    last_fsync_snapshot.delta_percentile(&curr_fsync_snapshot, 0.99);
+                last_fsync_snapshot = curr_fsync_snapshot;
+
+                // Lock dispatcher ONCE per sampler tick to read virtual-queue depths for all databases.
+                let queue_depths: HashMap<u64, u64> = {
+                    let d = shared_sampler
+                        .dispatcher
+                        .lock()
+                        .unwrap_or_else(|p| p.into_inner());
+                    databases
+                        .iter()
+                        .map(|db| (db.id.as_u64(), d.virtual_queue_depth(db.id.as_u64())))
+                        .collect()
+                };
+
+                for db in databases {
+                    let depth = queue_depths.get(&db.id.as_u64()).copied().unwrap_or(0);
+                    sample_database(&shared_sampler, db.id, &db.name, depth, wal_latency_p99);
+                }
+            }
+        },
+    );
+    info!("database metrics sampler running");
+}
+
+/// Sample and record gauges for a single database (internal helper, not pub).
+fn sample_database(
+    shared: &SharedState,
+    db_id: nodedb_types::DatabaseId,
+    db_name: &str,
+    bridge_queue_depth: u64,
+    wal_latency_p99: u64,
+) {
+    // 1. connections: active connections holding an admission permit for this database
+    let connections = shared
+        .admission_registry
+        .database_live_connections(db_id)
+        .map(u64::from)
+        .unwrap_or(0);
+    shared
+        .database_metrics
+        .set_connections(db_id, db_name, connections);
+
+    // 2. memory: resident memory in bytes allocated by this database in governor
+    let memory_bytes = shared.governor.database_usage_bytes(db_id) as u64;
+    shared
+        .database_metrics
+        .set_memory_bytes(db_id, db_name, memory_bytes);
+
+    // 3. storage: real on-disk storage usage from system metrics if present
+    if let Some(m) = &shared.system_metrics {
+        let storage_bytes = m.database_storage_bytes(db_name);
+        shared
+            .database_metrics
+            .set_storage_bytes(db_id, db_name, storage_bytes);
+    }
+
+    // 4. bridge_queue_depth: sum of SPSC bridge virtual-queue depths across cores
+    shared
+        .database_metrics
+        .set_bridge_queue_depth(db_id, db_name, bridge_queue_depth);
+
+    // 5. wal_latency_p99: node-wide rolling 10s P99 group-commit fsync latency
+    shared
+        .database_metrics
+        .set_wal_latency_p99(db_id, db_name, wal_latency_p99);
+
+    // 6. maintenance: cumulative maintenance CPU seconds tracked by MaintenanceBudgetTracker
+    let maintenance_cpu = shared.maintenance_budget.cumulative_cpu_secs(db_id);
+    shared
+        .database_metrics
+        .set_maintenance_cpu_secs(db_id, db_name, maintenance_cpu);
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::bridge::dispatch::Dispatcher;
+    use crate::control::metrics::histogram::HistogramSnapshot;
+    use crate::control::state::SharedState;
+    use crate::wal::WalManager;
+    use nodedb_mem::engine::EngineId;
+    use nodedb_types::{DatabaseId, TenantId};
+
+    #[tokio::test]
+    async fn sampler_records_memory_connections_and_wal_fsync() {
+        let dir = tempfile::tempdir().expect("tempdir");
+        let wal = Arc::new(
+            WalManager::open_for_testing(&dir.path().join("sampler-test.wal")).expect("open wal"),
+        );
+        let (dispatcher, _data_sides) = Dispatcher::new(1, 64);
+        let shared = SharedState::new(dispatcher, wal).expect("construct state");
+
+        if let Some(metrics) = shared.system_metrics.as_ref() {
+            for _ in 0..90 {
+                metrics.record_wal_fsync(100);
+            }
+            for _ in 0..10 {
+                metrics.record_wal_fsync(500);
+            }
+        }
+
+        let db_id = DatabaseId::new(42);
+        let db_name = "test_analytics";
+
+        // Reserve memory in governor for db_id.
+        let _token = shared
+            .governor
+            .try_reserve(db_id, TenantId::new(1), EngineId::Vector, 8192)
+            .expect("reserve memory");
+
+        // Acquire connection permit for db_id.
+        let _permit = shared
+            .admission_registry
+            .try_acquire_database(db_id)
+            .expect("acquire permit");
+
+        let curr_snapshot = shared
+            .system_metrics
+            .as_ref()
+            .map(|s| s.wal_fsync_seconds.snapshot_counts())
+            .unwrap_or_default();
+        let delta_p99 = HistogramSnapshot::default().delta_percentile(&curr_snapshot, 0.99);
+
+        // Run sample_database.
+        sample_database(&shared, db_id, db_name, 5, delta_p99);
+
+        // Verify gauge outputs in DatabaseMetricsRegistry.
+        let mut text = String::new();
+        shared.database_metrics.render_prometheus(&mut text);
+        assert!(text.contains(&format!(
+            "nodedb_database_connections{{database=\"{db_name}\"}} 1"
+        )));
+        assert!(text.contains(&format!(
+            "nodedb_database_memory_used_bytes{{database=\"{db_name}\"}} 8192"
+        )));
+        assert!(text.contains(&format!(
+            "nodedb_database_bridge_queue_depth{{database=\"{db_name}\"}} 5"
+        )));
+        assert!(text.contains(&format!(
+            "nodedb_database_wal_commit_latency_p99_us{{database=\"{db_name}\"}} 500"
+        )));
+    }
+}

nodedb/src/bootstrap/mod.rs

@@ -1,14 +1,14 @@
 // SPDX-License-Identifier: BUSL-1.1
 
 pub mod background_loops;
-pub mod calvin_catch_up;
 pub mod catalog_open;
 pub mod cluster_ready;
 pub mod constraint_reconcile;
 pub mod core_stall_monitor;
 pub mod credentials;
 pub mod data_group_recovery;
 pub mod data_plane;
+pub mod database_metrics_sampler;
 pub mod diagnostics;
 pub mod index_registry_seed;
 pub mod listeners;
@@ -21,4 +21,3 @@ pub mod state_wiring;
 pub mod tls;
 pub mod tracing_init;
 pub mod wal_init;
-pub mod write_group_settle;