diff --git a/nodedb-mem/src/governor/metrics.rs b/nodedb-mem/src/governor/metrics.rs
index 15a1ba696..ddd470c13 100644
--- a/nodedb-mem/src/governor/metrics.rs
+++ b/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);
+    }
 }
diff --git a/nodedb-mem/src/governor/reserve.rs b/nodedb-mem/src/governor/reserve.rs
index 1b3cd5dfb..ac4267e83 100644
--- a/nodedb-mem/src/governor/reserve.rs
+++ b/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.
@@ -78,6 +79,24 @@ impl MemoryGovernor {
                         });
                     }
                 }
+            } else {
+                drop(map);
+                let mut map = self
+                    .database_budgets
+                    .write()
+                    .unwrap_or_else(|p| p.into_inner());
+                let budget = map.entry(db).or_insert_with(|| ScopedBudget::new(None));
+                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,
+                        });
+                    }
+                }
             }
         }
 
@@ -144,6 +163,14 @@ impl MemoryGovernor {
                 .unwrap_or_else(|p| p.into_inner());
             if let Some(budget) = map.get(&db) {
                 scope.credit_database(budget.credit(size));
+            } else {
+                drop(map);
+                let mut map = self
+                    .database_budgets
+                    .write()
+                    .unwrap_or_else(|p| p.into_inner());
+                let budget = map.entry(db).or_insert_with(|| ScopedBudget::new(None));
+                scope.credit_database(budget.credit(size));
             }
         }
 
@@ -493,4 +520,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);
+    }
 }
diff --git a/nodedb-mem/src/scoped_budget.rs b/nodedb-mem/src/scoped_budget.rs
index d756898ed..f1034c6e6 100644
--- a/nodedb-mem/src/scoped_budget.rs
+++ b/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)),
diff --git a/nodedb/src/bootstrap/background_loops/spawn.rs b/nodedb/src/bootstrap/background_loops/spawn.rs
index 11118701d..1b21aa717 100644
--- a/nodedb/src/bootstrap/background_loops/spawn.rs
+++ b/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
diff --git a/nodedb/src/bootstrap/database_metrics_sampler.rs b/nodedb/src/bootstrap/database_metrics_sampler.rs
new file mode 100644
index 000000000..6644167cc
--- /dev/null
+++ b/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"
+        )));
+    }
+}
diff --git a/nodedb/src/bootstrap/mod.rs b/nodedb/src/bootstrap/mod.rs
index d57dfdff0..336901d85 100644
--- a/nodedb/src/bootstrap/mod.rs
+++ b/nodedb/src/bootstrap/mod.rs
@@ -1,7 +1,6 @@
 // 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;
@@ -9,6 +8,7 @@ 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;
diff --git a/nodedb/src/bridge/dispatch/dispatcher.rs b/nodedb/src/bridge/dispatch/dispatcher.rs
index 08f4e6324..0e3826c47 100644
--- a/nodedb/src/bridge/dispatch/dispatcher.rs
+++ b/nodedb/src/bridge/dispatch/dispatcher.rs
@@ -242,6 +242,14 @@ impl Dispatcher {
             .unwrap_or(PressureState::Normal)
     }
 
+    /// Sum of bridge virtual-queue depths for a database across all cores.
+    pub fn virtual_queue_depth(&self, database_id: u64) -> u64 {
+        self.cores
+            .iter()
+            .map(|c| c.wfq.depth_for(database_id) as u64)
+            .sum()
+    }
+
     /// Number of Data Plane cores.
     pub fn num_cores(&self) -> usize {
         self.cores.len()
@@ -259,3 +267,55 @@ impl Dispatcher {
         &self.router
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::bridge::dispatch::test_requests::make_request_for_db;
+
+    #[test]
+    fn virtual_queue_depth_reporting() {
+        let (mut dispatcher, data_sides) = Dispatcher::new(2, 2);
+        assert_eq!(dispatcher.virtual_queue_depth(1), 0);
+
+        // Fill physical rings on core 0 and core 1 so subsequent requests park in WFQ.
+        dispatcher
+            .dispatch_to_core(0, make_request_for_db(0, 1, 1))
+            .unwrap();
+        dispatcher
+            .dispatch_to_core(0, make_request_for_db(0, 1, 2))
+            .unwrap();
+        assert_eq!(data_sides[0].request_rx.len(), 2);
+
+        dispatcher
+            .dispatch_to_core(1, make_request_for_db(1, 1, 10))
+            .unwrap();
+        dispatcher
+            .dispatch_to_core(1, make_request_for_db(1, 1, 11))
+            .unwrap();
+        assert_eq!(data_sides[1].request_rx.len(), 2);
+
+        // Park 2 requests in core 0's WFQ for db 1.
+        dispatcher
+            .dispatch_to_core(0, make_request_for_db(0, 1, 3))
+            .unwrap();
+        dispatcher
+            .dispatch_to_core(0, make_request_for_db(0, 1, 4))
+            .unwrap();
+
+        // Park 3 requests in core 1's WFQ for db 1.
+        dispatcher
+            .dispatch_to_core(1, make_request_for_db(1, 1, 12))
+            .unwrap();
+        dispatcher
+            .dispatch_to_core(1, make_request_for_db(1, 1, 13))
+            .unwrap();
+        dispatcher
+            .dispatch_to_core(1, make_request_for_db(1, 1, 14))
+            .unwrap();
+
+        // Virtual queue depth must sum across all cores.
+        assert_eq!(dispatcher.virtual_queue_depth(1), 5);
+        assert_eq!(dispatcher.virtual_queue_depth(2), 0);
+    }
+}
diff --git a/nodedb/src/control/catalog_entry/post_apply/database.rs b/nodedb/src/control/catalog_entry/post_apply/database.rs
index 16a8879cf..764d7ddec 100644
--- a/nodedb/src/control/catalog_entry/post_apply/database.rs
+++ b/nodedb/src/control/catalog_entry/post_apply/database.rs
@@ -1,10 +1,10 @@
 // SPDX-License-Identifier: BUSL-1.1
 
-//! Post-apply side effects for database catalog entries.
+//! Synchronous post-apply side effects for database catalog entries.
 //!
 //! Database descriptors and grants are read directly from redb on the hot
 //! path (no separate in-memory registry), so most arms are no-ops. The
-//! synchronous `DeleteDatabase` arm releases the dropped scope's quota caps.
+//! `DeleteDatabase` arm releases the dropped scope's quota caps on every node.
 
 use std::sync::Arc;
 
@@ -16,10 +16,14 @@ use crate::control::state::SharedState;
 /// Post-apply for `PutDatabase` — no in-memory cache to update.
 pub fn put(_descriptor: DatabaseDescriptor, _shared: Arc<SharedState>) {}
 
-/// Synchronous post-apply for `DeleteDatabase` — release the quota caps of
-/// the dropped scope, its tenants' caps included.
+/// Post-apply for `DeleteDatabase` — release the quota caps of the dropped
+/// scope, its tenants' caps included, unregister from maintenance budget,
+/// and remove any metrics handles from `DatabaseMetricsRegistry`.
 pub fn delete(db_id: u64, shared: Arc<SharedState>) {
-    super::quota::release_database_scope(DatabaseId::new(db_id), &shared);
+    let db = DatabaseId::new(db_id);
+    shared.database_metrics.remove(db);
+    shared.maintenance_budget.remove_database(db);
+    super::quota::release_database_scope(db, &shared);
 }
 
 /// Post-apply for `PutDatabaseGrant`.
diff --git a/nodedb/src/control/catalog_entry/post_apply/quota.rs b/nodedb/src/control/catalog_entry/post_apply/quota.rs
index 2e81debd4..594ce2bb8 100644
--- a/nodedb/src/control/catalog_entry/post_apply/quota.rs
+++ b/nodedb/src/control/catalog_entry/post_apply/quota.rs
@@ -18,9 +18,15 @@ use crate::control::state::SharedState;
 
 /// Install a database quota into live enforcement.
 pub fn put_database(db_id: DatabaseId, record: &QuotaRecord, shared: &SharedState) {
-    shared
-        .maintenance_budget
-        .set_cap(db_id, record.maintenance_cpu_pct);
+    if let Ok(Some(name)) = shared.credentials.catalog().get_database_name_by_id(db_id) {
+        shared
+            .maintenance_budget
+            .set_cap_named(db_id, name, record.maintenance_cpu_pct);
+    } else {
+        shared
+            .maintenance_budget
+            .set_cap(db_id, record.maintenance_cpu_pct);
+    }
     if record.max_memory_bytes > 0 {
         shared
             .governor
diff --git a/nodedb/src/control/maintenance/budget.rs b/nodedb/src/control/maintenance/budget.rs
index 697164948..df7c69148 100644
--- a/nodedb/src/control/maintenance/budget.rs
+++ b/nodedb/src/control/maintenance/budget.rs
@@ -31,6 +31,8 @@ struct DbWindow {
     buckets: [Bucket; 60],
     /// UNIX-style second index of the last bucket written (wraps at 60).
     last_bucket_secs: u64,
+    /// Cumulative CPU-seconds consumed since process start.
+    cumulative_cpu_secs: f64,
 }
 
 impl DbWindow {
@@ -38,6 +40,7 @@ impl DbWindow {
         Self {
             buckets: std::array::from_fn(|_| Bucket::default()),
             last_bucket_secs: 0,
+            cumulative_cpu_secs: 0.0,
         }
     }
 
@@ -64,6 +67,9 @@ impl DbWindow {
         self.advance(now_secs);
         let idx = (now_secs % 60) as usize;
         self.buckets[idx].consumed_secs += secs;
+        if secs.is_finite() && secs > 0.0 {
+            self.cumulative_cpu_secs += secs;
+        }
     }
 }
 
@@ -115,6 +121,23 @@ impl MaintenanceBudgetTracker {
         }
     }
 
+    /// Cumulative maintenance CPU-seconds consumed by `db`.
+    pub fn cumulative_cpu_secs(&self, db: DatabaseId) -> f64 {
+        let inner = self.inner.lock().unwrap_or_else(|p| p.into_inner());
+        inner
+            .windows
+            .get(&db)
+            .map(|w| w.cumulative_cpu_secs)
+            .unwrap_or(0.0)
+    }
+
+    /// Remove a database's budget window and cap (e.g. on database drop).
+    pub fn remove_database(&self, db: DatabaseId) {
+        let mut inner = self.inner.lock().unwrap_or_else(|p| p.into_inner());
+        inner.caps.remove(&db);
+        inner.windows.remove(&db);
+    }
+
     /// Install or replace the maintenance CPU cap for `db`.
     ///
     /// `maintenance_cpu_pct` is the `QuotaRecord` field (0–100).
@@ -129,6 +152,11 @@ impl MaintenanceBudgetTracker {
         inner.caps.insert(db, cap);
     }
 
+    /// Install or replace the maintenance CPU cap and register the database name for `db`.
+    pub fn set_cap_named(&self, db: DatabaseId, _name: impl Into<String>, maintenance_cpu_pct: u8) {
+        self.set_cap(db, maintenance_cpu_pct);
+    }
+
     /// Attempt to acquire a maintenance lease for `db`.
     ///
     /// Returns `Some(MaintenanceLease)` when `consumed + estimated_secs ≤ cap`
@@ -171,13 +199,20 @@ impl Default for MaintenanceBudgetTracker {
 /// RAII lease returned by [`MaintenanceBudgetTracker::try_acquire`].
 ///
 /// On drop, the actual elapsed wall-clock seconds are recorded into the
-/// sliding window for the database.
+/// sliding window and cumulative total for the database.
 pub struct MaintenanceLease {
     tracker: Arc<MaintenanceBudgetTracker>,
     db: DatabaseId,
     start: Instant,
 }
 
+impl MaintenanceLease {
+    /// Database id for which the lease was acquired.
+    pub fn db(&self) -> DatabaseId {
+        self.db
+    }
+}
+
 impl Drop for MaintenanceLease {
     fn drop(&mut self) {
         let elapsed = self.start.elapsed().as_secs_f64();
@@ -270,6 +305,7 @@ mod tests {
             assert!(total > 0.0, "lease drop should have recorded elapsed time");
             assert!(total < 1.0, "elapsed should be under 1s");
         }
+        assert!(t.cumulative_cpu_secs(db) > 0.0);
     }
 
     #[test]
@@ -294,4 +330,21 @@ mod tests {
             "old consumption should have expired out of the window"
         );
     }
+
+    #[test]
+    fn cumulative_cpu_secs_increments() {
+        let t = tracker();
+        let db = DatabaseId::new(42);
+        t.set_cap(db, 100);
+
+        assert_eq!(t.cumulative_cpu_secs(db), 0.0);
+        {
+            let _lease = t.try_acquire(db, 1.0).expect("acquire should succeed");
+            std::thread::sleep(std::time::Duration::from_millis(15));
+        }
+        assert!(
+            t.cumulative_cpu_secs(db) > 0.0,
+            "cumulative maintenance CPU time should be > 0 after lease drop"
+        );
+    }
 }
diff --git a/nodedb/src/control/metrics/database.rs b/nodedb/src/control/metrics/database.rs
index 75abd4a5d..2851e8f44 100644
--- a/nodedb/src/control/metrics/database.rs
+++ b/nodedb/src/control/metrics/database.rs
@@ -30,9 +30,10 @@ pub struct DatabaseCounters {
     pub connections: AtomicU64,
     /// SPSC bridge virtual-queue depth snapshot.
     pub bridge_queue_depth: AtomicU64,
-    /// WAL commit latency P99 in microseconds (updated by WAL group-commit path).
+    /// WAL commit latency P99 in microseconds. Node-wide metric sampled from the node's
+    /// single WalManager; every database reports the same node-wide value.
     pub wal_commit_latency_p99_us: AtomicU64,
-    /// Cumulative maintenance CPU-seconds consumed by this database.
+    /// Cumulative maintenance CPU-seconds consumed by this database (in microseconds).
     pub maintenance_cpu_seconds_total: AtomicU64,
     /// Mirror replication lag in milliseconds (`now_ms - last_apply_ms`).
     /// Zero for non-mirror databases or a promoted mirror.
@@ -46,11 +47,10 @@ pub struct DatabaseCounters {
 /// control-plane-side collection point for Prometheus scrapes.
 #[derive(Debug, Default)]
 pub struct DatabaseMetricsRegistry {
-    /// Map from database display name to its live counter handle.
+    /// Map from `DatabaseId` to its display name and live counter handle.
     ///
-    /// Display names are resolved from `DatabaseId` via the catalog name cache.
-    /// The registry is write-locked only on first registration of a new database.
-    counters: RwLock<HashMap<String, Arc<DatabaseCounters>>>,
+    /// The registry is write-locked only on first registration or removal of a database.
+    counters: RwLock<HashMap<DatabaseId, (String, Arc<DatabaseCounters>)>>,
 }
 
 impl DatabaseMetricsRegistry {
@@ -59,94 +59,101 @@ impl DatabaseMetricsRegistry {
         Self::default()
     }
 
-    /// Get or create the counter handle for `db_name`.
+    /// Get or create the counter handle for `(db_id, db_name)`.
     ///
     /// The returned `Arc<DatabaseCounters>` is stable for the process lifetime.
-    pub fn get_or_create(&self, db_name: &str) -> Arc<DatabaseCounters> {
+    pub fn get_or_create(&self, db_id: DatabaseId, db_name: &str) -> Arc<DatabaseCounters> {
         // Fast path: read lock.
         {
             let r = self.counters.read().unwrap_or_else(|p| p.into_inner());
-            if let Some(c) = r.get(db_name) {
+            if let Some((_, c)) = r.get(&db_id) {
                 return Arc::clone(c);
             }
         }
         // Slow path: write lock.
         let mut w = self.counters.write().unwrap_or_else(|p| p.into_inner());
-        w.entry(db_name.to_string())
-            .or_insert_with(|| Arc::new(DatabaseCounters::default()))
+        w.entry(db_id)
+            .or_insert_with(|| (db_name.to_string(), Arc::new(DatabaseCounters::default())))
+            .1
             .clone()
     }
 
-    /// Increment the QPS counter for `db_name` by 1.
-    pub fn record_qps(&self, db_name: &str) {
-        self.get_or_create(db_name)
+    /// Remove all metrics for `db_id` from the registry (e.g. on DROP DATABASE).
+    pub fn remove(&self, db_id: DatabaseId) {
+        let mut w = self.counters.write().unwrap_or_else(|p| p.into_inner());
+        w.remove(&db_id);
+    }
+
+    /// Increment the QPS counter for `db_id` by 1.
+    pub fn record_qps(&self, db_id: DatabaseId, db_name: &str) {
+        self.get_or_create(db_id, db_name)
             .qps_total
             .fetch_add(1, Ordering::Relaxed);
     }
 
-    /// Set the memory-bytes gauge for `db_name`.
-    pub fn set_memory_bytes(&self, db_name: &str, bytes: u64) {
-        self.get_or_create(db_name)
+    /// Set the memory-bytes gauge for `db_id`.
+    pub fn set_memory_bytes(&self, db_id: DatabaseId, db_name: &str, bytes: u64) {
+        self.get_or_create(db_id, db_name)
             .memory_bytes
             .store(bytes, Ordering::Relaxed);
     }
 
-    /// Set the storage-bytes gauge for `db_name`.
-    pub fn set_storage_bytes(&self, db_name: &str, bytes: u64) {
-        self.get_or_create(db_name)
+    /// Set the storage-bytes gauge for `db_id`.
+    pub fn set_storage_bytes(&self, db_id: DatabaseId, db_name: &str, bytes: u64) {
+        self.get_or_create(db_id, db_name)
             .storage_bytes
             .store(bytes, Ordering::Relaxed);
     }
 
-    /// Set the active-connections gauge for `db_name`.
-    pub fn set_connections(&self, db_name: &str, count: u64) {
-        self.get_or_create(db_name)
+    /// Set the active-connections gauge for `db_id`.
+    pub fn set_connections(&self, db_id: DatabaseId, db_name: &str, count: u64) {
+        self.get_or_create(db_id, db_name)
             .connections
             .store(count, Ordering::Relaxed);
     }
 
-    /// Set the bridge queue-depth snapshot for `db_name`.
-    pub fn set_bridge_queue_depth(&self, db_name: &str, depth: u64) {
-        self.get_or_create(db_name)
+    /// Set the bridge queue-depth snapshot for `db_id`.
+    pub fn set_bridge_queue_depth(&self, db_id: DatabaseId, db_name: &str, depth: u64) {
+        self.get_or_create(db_id, db_name)
             .bridge_queue_depth
             .store(depth, Ordering::Relaxed);
     }
 
-    /// Set the WAL commit latency P99 (microseconds) for `db_name`.
-    pub fn set_wal_latency_p99(&self, db_name: &str, us: u64) {
-        self.get_or_create(db_name)
+    /// Set the WAL commit latency P99 (microseconds) for `db_id`.
+    ///
+    /// Note: WAL latency is a node-wide metric (one WalManager per node); all databases
+    /// report the same node-wide latency.
+    pub fn set_wal_latency_p99(&self, db_id: DatabaseId, db_name: &str, us: u64) {
+        self.get_or_create(db_id, db_name)
             .wal_commit_latency_p99_us
             .store(us, Ordering::Relaxed);
     }
 
-    /// Set the mirror replication lag in milliseconds for `db_name`.
+    /// Set the mirror replication lag in milliseconds for `db_id`.
     ///
     /// Computed as `now_ms - last_apply_ms` using wall-clock time. The
     /// bounded-staleness read rejection path uses the same wall-clock basis,
     /// so the metric and the rejection gate are always consistent.
-    pub fn set_mirror_lag_ms(&self, db_name: &str, lag_ms: u64) {
-        self.get_or_create(db_name)
+    pub fn set_mirror_lag_ms(&self, db_id: DatabaseId, db_name: &str, lag_ms: u64) {
+        self.get_or_create(db_id, db_name)
             .mirror_lag_ms
             .store(lag_ms, Ordering::Relaxed);
     }
 
-    /// Add `secs` fractional CPU-seconds to the maintenance counter for `db_name`.
+    /// Set cumulative maintenance CPU-seconds for `db_id`.
     ///
     /// Converts to integer microseconds (rounded) to avoid `AtomicF64` complexity.
     /// Negative or non-finite inputs are clamped to zero so accidental underflow
     /// in upstream timing arithmetic cannot subtract from the cumulative counter.
-    pub fn add_maintenance_cpu_secs(&self, db_name: &str, secs: f64) {
+    pub fn set_maintenance_cpu_secs(&self, db_id: DatabaseId, db_name: &str, secs: f64) {
         let us = if secs.is_finite() && secs > 0.0 {
             (secs * 1_000_000.0).round() as u64
         } else {
             0
         };
-        if us == 0 {
-            return;
-        }
-        self.get_or_create(db_name)
+        self.get_or_create(db_id, db_name)
             .maintenance_cpu_seconds_total
-            .fetch_add(us, Ordering::Relaxed);
+            .store(us, Ordering::Relaxed);
     }
 
     /// Render all registered databases as Prometheus text format 0.0.4.
@@ -160,7 +167,7 @@ impl DatabaseMetricsRegistry {
             return;
         }
 
-        let mut pairs: Vec<_> = r.iter().collect();
+        let mut pairs: Vec<_> = r.values().map(|(name, c)| (name.as_str(), c)).collect();
         pairs.sort_by(|a, b| a.0.cmp(b.0));
 
         macro_rules! emit_counter {
@@ -226,7 +233,7 @@ impl DatabaseMetricsRegistry {
         );
         emit_gauge!(
             "nodedb_database_wal_commit_latency_p99_us",
-            "WAL commit latency P99 in microseconds per database",
+            "WAL commit latency P99 in microseconds (node-wide; shared across all databases)",
             wal_commit_latency_p99_us
         );
         emit_counter!(
@@ -280,8 +287,9 @@ mod tests {
     #[test]
     fn registry_get_or_create_idempotent() {
         let reg = DatabaseMetricsRegistry::new();
-        let a = reg.get_or_create("mydb");
-        let b = reg.get_or_create("mydb");
+        let db = DatabaseId::new(1);
+        let a = reg.get_or_create(db, "mydb");
+        let b = reg.get_or_create(db, "mydb");
         // Both arcs point to the same allocation.
         assert!(Arc::ptr_eq(&a, &b));
     }
@@ -289,17 +297,18 @@ mod tests {
     #[test]
     fn record_qps_increments() {
         let reg = DatabaseMetricsRegistry::new();
-        reg.record_qps("db1");
-        reg.record_qps("db1");
-        let c = reg.get_or_create("db1");
+        let db = DatabaseId::new(1);
+        reg.record_qps(db, "db1");
+        reg.record_qps(db, "db1");
+        let c = reg.get_or_create(db, "db1");
         assert_eq!(c.qps_total.load(Ordering::Relaxed), 2);
     }
 
     #[test]
     fn render_prometheus_contains_labels() {
         let reg = DatabaseMetricsRegistry::new();
-        reg.record_qps("alpha");
-        reg.record_qps("beta");
+        reg.record_qps(DatabaseId::new(1), "alpha");
+        reg.record_qps(DatabaseId::new(2), "beta");
         let mut out = String::new();
         reg.render_prometheus(&mut out);
         assert!(out.contains(r#"database="alpha""#));
@@ -326,4 +335,39 @@ mod tests {
         };
         assert!(m.is_over_quota());
     }
+
+    #[test]
+    fn gauge_setters_update_counters() {
+        let reg = DatabaseMetricsRegistry::new();
+        let db = DatabaseId::new(1);
+        reg.set_connections(db, "db1", 42);
+        reg.set_memory_bytes(db, "db1", 1024 * 1024);
+        reg.set_storage_bytes(db, "db1", 10 * 1024 * 1024);
+        reg.set_bridge_queue_depth(db, "db1", 7);
+        reg.set_wal_latency_p99(db, "db1", 1500);
+
+        let c = reg.get_or_create(db, "db1");
+        assert_eq!(c.connections.load(Ordering::Relaxed), 42);
+        assert_eq!(c.memory_bytes.load(Ordering::Relaxed), 1024 * 1024);
+        assert_eq!(c.storage_bytes.load(Ordering::Relaxed), 10 * 1024 * 1024);
+        assert_eq!(c.bridge_queue_depth.load(Ordering::Relaxed), 7);
+        assert_eq!(c.wal_commit_latency_p99_us.load(Ordering::Relaxed), 1500);
+    }
+
+    #[test]
+    fn remove_clears_database_metrics() {
+        let reg = DatabaseMetricsRegistry::new();
+        let db = DatabaseId::new(99);
+        reg.set_connections(db, "db_to_drop", 10);
+        assert_eq!(
+            reg.get_or_create(db, "db_to_drop")
+                .connections
+                .load(Ordering::Relaxed),
+            10
+        );
+        reg.remove(db);
+        let mut out = String::new();
+        reg.render_prometheus(&mut out);
+        assert!(!out.contains(r#"database="db_to_drop""#));
+    }
 }
diff --git a/nodedb/src/control/metrics/histogram.rs b/nodedb/src/control/metrics/histogram.rs
index 48a228efa..66ebb45e7 100644
--- a/nodedb/src/control/metrics/histogram.rs
+++ b/nodedb/src/control/metrics/histogram.rs
@@ -168,6 +168,19 @@ impl AtomicHistogram {
         snap
     }
 
+    /// Create a point-in-time snapshot of bucket counts for rolling delta calculations.
+    pub fn snapshot_counts(&self) -> HistogramSnapshot {
+        HistogramSnapshot {
+            boundaries: self.boundaries,
+            buckets: self
+                .buckets
+                .iter()
+                .map(|b| b.load(Ordering::Relaxed))
+                .collect(),
+            count: self.count.load(Ordering::Relaxed),
+        }
+    }
+
     /// Merge another histogram's counts into this one.
     ///
     /// Both histograms must share the same bucket boundaries — if they do
@@ -186,6 +199,71 @@ impl AtomicHistogram {
     }
 }
 
+/// Point-in-time snapshot of bucket counts for rolling-window percentile calculations.
+#[derive(Debug, Clone, Default)]
+pub struct HistogramSnapshot {
+    /// Upper bounds in microseconds.
+    pub boundaries: &'static [u64],
+    /// Bucket counters at snapshot time.
+    pub buckets: Vec<u64>,
+    /// Total observations at snapshot time.
+    pub count: u64,
+}
+
+impl HistogramSnapshot {
+    /// Compute percentile from the bucket count delta between `self` (earlier) and `newer`.
+    ///
+    /// `p` is a fraction between `0.0` and `1.0` (e.g. `0.99` for P99).
+    /// Returns `0` if no new observations occurred in the interval.
+    pub fn delta_percentile(&self, newer: &HistogramSnapshot, p: f64) -> u64 {
+        let is_empty_baseline = self.boundaries.is_empty() && self.count == 0;
+        if !is_empty_baseline && (self.boundaries != newer.boundaries || newer.count <= self.count)
+        {
+            return 0;
+        }
+        if newer.count == 0 {
+            return 0;
+        }
+        let delta_count = if is_empty_baseline {
+            newer.count
+        } else {
+            newer.count - self.count
+        };
+        if delta_count == 0 {
+            return 0;
+        }
+        let target = (p * delta_count as f64) as u64;
+        let mut cumulative = 0u64;
+        let mut prev_boundary = 0u64;
+
+        for (i, &boundary) in newer.boundaries.iter().enumerate() {
+            let older_bucket = if is_empty_baseline {
+                0
+            } else {
+                self.buckets.get(i).copied().unwrap_or(0)
+            };
+            let newer_bucket = newer.buckets.get(i).copied().unwrap_or(0);
+            let bucket_delta = newer_bucket.saturating_sub(older_bucket);
+            cumulative += bucket_delta;
+            if cumulative >= target {
+                let bucket_start = prev_boundary;
+                let bucket_width = boundary - bucket_start;
+                if bucket_delta == 0 {
+                    return boundary;
+                }
+                let fraction = if cumulative > target {
+                    (bucket_delta - (cumulative - target)) as f64 / bucket_delta as f64
+                } else {
+                    1.0
+                };
+                return bucket_start + (fraction * bucket_width as f64) as u64;
+            }
+            prev_boundary = boundary;
+        }
+        newer.boundaries.last().copied().unwrap_or(0)
+    }
+}
+
 impl Default for AtomicHistogram {
     fn default() -> Self {
         Self::new()
@@ -229,6 +307,67 @@ mod tests {
         assert!((500..=1000).contains(&p50), "p50={p50}");
     }
 
+    #[test]
+    fn percentile_p99_numerically_correct() {
+        // WAL buckets: [100, 500, 1000, 5000, 10000, 50000, 100000, 500000, 1000000]
+        let h = AtomicHistogram::with_buckets(WAL_FSYNC_BUCKETS_US);
+        // Observe 90 items at 80us (falls in <=100us bucket)
+        for _ in 0..90 {
+            h.observe(80);
+        }
+        // Observe 9 items at 400us (falls in <=500us bucket)
+        for _ in 0..9 {
+            h.observe(400);
+        }
+        // Observe 1 item at 800us (falls in <=1000us bucket)
+        h.observe(800);
+
+        assert_eq!(h.count(), 100);
+
+        // p50 is rank 50 (within first bucket 0..100us)
+        let p50 = h.percentile(0.50);
+        assert!(p50 <= 100, "expected p50 <= 100, got {p50}");
+
+        // p99 is rank 99 (90 in bucket0 + 9 in bucket1 = 99 -> top of <=500us bucket)
+        let p99 = h.percentile(0.99);
+        assert!(
+            (100..=500).contains(&p99),
+            "expected p99 in [100, 500], got {p99}"
+        );
+    }
+
+    #[test]
+    fn rolling_window_delta_percentile() {
+        let h = AtomicHistogram::with_buckets(WAL_FSYNC_BUCKETS_US);
+
+        // Window 1: 100 observations at 80us (<=100us)
+        for _ in 0..100 {
+            h.observe(80);
+        }
+        let snap1 = h.snapshot_counts();
+
+        // Window 2: 90 observations at 80us, 10 observations at 4000us (<=5000us)
+        for _ in 0..90 {
+            h.observe(80);
+        }
+        for _ in 0..10 {
+            h.observe(4000);
+        }
+        let snap2 = h.snapshot_counts();
+
+        // Delta p99 in Window 2 alone (100 new observations: 90 at 80us, 10 at 4000us)
+        let delta_p99 = snap1.delta_percentile(&snap2, 0.99);
+        assert!(
+            (1000..=5000).contains(&delta_p99),
+            "expected delta p99 in [1000, 5000], got {delta_p99}"
+        );
+
+        // Window 3: No new observations
+        let snap3 = h.snapshot_counts();
+        let delta_p99_empty = snap2.delta_percentile(&snap3, 0.99);
+        assert_eq!(delta_p99_empty, 0);
+    }
+
     #[test]
     fn prometheus_output() {
         let h = AtomicHistogram::new();
diff --git a/nodedb/src/control/mirror/observer.rs b/nodedb/src/control/mirror/observer.rs
index 86699e444..cf9f11d2a 100644
--- a/nodedb/src/control/mirror/observer.rs
+++ b/nodedb/src/control/mirror/observer.rs
@@ -183,7 +183,7 @@ pub fn update_lag_status(
     let lag_ms = now.saturating_sub(lag_record.last_apply_ms);
 
     // Publish the metric: now_ms - last_apply_ms.
-    db_metrics.set_mirror_lag_ms(db_name, lag_ms);
+    db_metrics.set_mirror_lag_ms(db_id, db_name, lag_ms);
 
     // Fall back to apply-time when the registry has no entry.
     let last_received = last_entry_received_ms.unwrap_or(lag_record.last_apply_ms);
diff --git a/nodedb/src/control/server/admission/registry.rs b/nodedb/src/control/server/admission/registry.rs
index 3e6257888..5a8f06658 100644
--- a/nodedb/src/control/server/admission/registry.rs
+++ b/nodedb/src/control/server/admission/registry.rs
@@ -139,7 +139,7 @@ impl LimitEntry {
         let owed = u32::try_from(owed).unwrap_or(u32::MAX);
         let _ = self
             .pending_shrink
-            .try_update(Ordering::AcqRel, Ordering::Acquire, |current| {
+            .fetch_update(Ordering::AcqRel, Ordering::Acquire, |current| {
                 Some(current.saturating_add(owed))
             });
     }
@@ -273,13 +273,23 @@ impl AdmissionRegistry {
 
     /// Attempt to acquire a database-level permit for a new connection.
     ///
-    /// Returns `Ok(Some(permit))` if an entry exists and a slot was acquired,
-    /// `Ok(None)` if the database has no entry, or `Err(AdmissionError)` if the
-    /// database is at capacity.
+    /// Returns `Ok(Some(permit))` when a slot was acquired, or `Err(AdmissionError)`
+    /// if the database is at capacity.
     pub fn try_acquire_database(
         &self,
         db: DatabaseId,
     ) -> Result<Option<OwnedSemaphorePermit>, AdmissionError> {
+        let has_entry = {
+            let map = self.db_semaphores.read().unwrap_or_else(|p| p.into_inner());
+            map.contains_key(&db)
+        };
+        if !has_entry {
+            let mut map = self
+                .db_semaphores
+                .write()
+                .unwrap_or_else(|p| p.into_inner());
+            map.entry(db).or_insert_with(|| LimitEntry::new(None));
+        }
         let map = self.db_semaphores.read().unwrap_or_else(|p| p.into_inner());
         let permit = try_acquire_entry(map.get(&db), |limit| {
             AdmissionError::DatabaseCapExhausted { db, limit }
@@ -372,10 +382,12 @@ mod tests {
     #[test]
     fn no_database_cap_allows_unlimited() {
         let reg = AdmissionRegistry::new();
-        // No entry configured → Ok(None).
-        let r = reg.try_acquire_database(db(0));
-        assert!(r.unwrap().is_none());
-        assert_eq!(reg.database_live_connections(db(0)), None);
+        // An uncapped database admits and tracks live connections.
+        let p = reg.try_acquire_database(db(0)).unwrap();
+        assert!(p.is_some());
+        assert_eq!(reg.database_live_connections(db(0)), Some(1));
+        drop(p);
+        assert_eq!(reg.database_live_connections(db(0)), Some(0));
     }
 
     #[test]
@@ -563,7 +575,7 @@ mod tests {
         reg.set_database_limit(db(5), 0);
 
         assert_eq!(reg.database_live_connections(db(5)), None);
-        assert!(reg.try_acquire_database(db(5)).unwrap().is_none());
+        assert!(reg.try_acquire_database(db(5)).unwrap().is_some());
     }
 
     #[test]
diff --git a/nodedb/src/control/server/pgwire/handler/sql_exec.rs b/nodedb/src/control/server/pgwire/handler/sql_exec.rs
index 429660efa..93e9d4a45 100644
--- a/nodedb/src/control/server/pgwire/handler/sql_exec.rs
+++ b/nodedb/src/control/server/pgwire/handler/sql_exec.rs
@@ -11,7 +11,6 @@ use std::sync::Arc;
 
 use nodedb_sql::parser::preprocess::lex::find_ascii_case_insensitive;
 use nodedb_types::strip_prefix_ascii_case_insensitive;
-use pgwire::api::portal::Format;
 use pgwire::api::results::{DataRowEncoder, QueryResponse, Response, Tag};
 use pgwire::error::{ErrorInfo, PgWireError, PgWireResult};
 
@@ -23,7 +22,6 @@ use super::connection_admin;
 use super::core::NodeDbPgHandler;
 use super::sql_split::split_sql_statements;
 use crate::control::server::shared::session::{SessionId, TransactionState};
-use crate::control::server::shared::txn_control::{TxnControl, classify as classify_txn_control};
 
 impl NodeDbPgHandler {
     /// Execute a SQL query: session state → identity → DDL check → quota → plan → perms → dispatch.
@@ -31,30 +29,23 @@ impl NodeDbPgHandler {
     /// Handles multi-statement queries (e.g. psql heredoc sends all statements in one message).
     /// Splits at top-level semicolons before dispatching so that `parts[2]` is never polluted
     /// with trailing `;` characters.
-    ///
-    /// `requested` is the client's result-format request: `TEXT_RESULTS`
-    /// for the simple-query protocol, the portal's formats for an
-    /// extended-query Execute. Typed DDL and planned results honour it.
     pub(super) async fn execute_sql(
         &self,
         identity: &AuthenticatedIdentity,
         session_id: SessionId,
         sql: &str,
-        requested: &Format,
     ) -> PgWireResult<Vec<Response>> {
         let statements = split_sql_statements(sql);
         match statements.len() {
             0 => Ok(vec![Response::EmptyQuery]),
             1 => {
-                self.execute_single_sql(identity, session_id, &statements[0], requested)
+                self.execute_single_sql(identity, session_id, &statements[0])
                     .await
             }
             _ => {
                 let mut all = Vec::new();
                 for stmt in statements {
-                    let mut resp = self
-                        .execute_single_sql(identity, session_id, &stmt, requested)
-                        .await?;
+                    let mut resp = self.execute_single_sql(identity, session_id, &stmt).await?;
                     all.append(&mut resp);
                 }
                 Ok(all)
@@ -68,7 +59,6 @@ impl NodeDbPgHandler {
         identity: &AuthenticatedIdentity,
         session_id: SessionId,
         sql: &str,
-        requested: &Format,
     ) -> PgWireResult<Vec<Response>> {
         use super::super::types::error_to_sqlstate;
 
@@ -80,35 +70,17 @@ impl NodeDbPgHandler {
         }
 
         // ── Transaction commands ──────────────────────────────────────
-        // Their session handlers own the aborted-block state, so they run
-        // before the failed-transaction guard.
-
-        if let Some(control) = classify_txn_control(sql_trimmed) {
-            return match control {
-                TxnControl::Begin(modes) => self.handle_begin(session_id, &modes),
-                TxnControl::Commit => self.handle_commit(identity, session_id).await,
-                TxnControl::Rollback => self.handle_rollback(identity, session_id).await,
-                TxnControl::Savepoint(name) => {
-                    self.handle_savepoint(identity, session_id, &name).await
-                }
-                TxnControl::Release(name) => self.handle_release_savepoint(session_id, &name),
-                TxnControl::RollbackTo(name) => {
-                    self.handle_rollback_to_savepoint(identity, session_id, &name)
-                        .await
-                }
-            };
+
+        if upper == "BEGIN" || upper == "BEGIN TRANSACTION" || upper == "START TRANSACTION" {
+            return self.handle_begin(session_id);
         }
 
-        // ── Failed transaction guard ──────────────────────────────────
-        // An aborted block refuses every other statement until it ends.
+        if upper == "COMMIT" || upper == "END" || upper == "END TRANSACTION" {
+            return self.handle_commit(identity, session_id).await;
+        }
 
-        if self.sessions.transaction_state(session_id) == TransactionState::Failed {
-            return Err(PgWireError::UserError(Box::new(ErrorInfo::new(
-                "ERROR".to_owned(),
-                "25P02".to_owned(),
-                "current transaction is aborted, commands ignored until end of transaction block"
-                    .to_owned(),
-            ))));
+        if upper == "ROLLBACK" || upper == "ABORT" {
+            return self.handle_rollback(identity, session_id).await;
         }
 
         if let Some(result) =
@@ -125,6 +97,22 @@ impl NodeDbPgHandler {
                 .map(|r| vec![r]);
         }
 
+        if upper.starts_with("SAVEPOINT ") {
+            return self
+                .handle_savepoint(identity, session_id, sql_trimmed)
+                .await;
+        }
+
+        if upper.starts_with("RELEASE SAVEPOINT ") || upper.starts_with("RELEASE ") {
+            return self.handle_release_savepoint(session_id, sql_trimmed);
+        }
+
+        if upper.starts_with("ROLLBACK TO ") {
+            return self
+                .handle_rollback_to_savepoint(identity, session_id, sql_trimmed)
+                .await;
+        }
+
         // ── Cursor commands ───────────────────────────────────────────
 
         if upper.starts_with("DECLARE ") && upper.contains(" CURSOR ") {
@@ -187,6 +175,17 @@ impl NodeDbPgHandler {
             return Ok(vec![Response::Execution(Tag::new("CLOSE CURSOR"))]);
         }
 
+        // ── Failed transaction guard ──────────────────────────────────
+
+        if self.sessions.transaction_state(session_id) == TransactionState::Failed {
+            return Err(PgWireError::UserError(Box::new(ErrorInfo::new(
+                "ERROR".to_owned(),
+                "25P02".to_owned(),
+                "current transaction is aborted, commands ignored until end of transaction block"
+                    .to_owned(),
+            ))));
+        }
+
         // ── Session commands ──────────────────────────────────────────
 
         if upper.starts_with("SET ") {
@@ -322,9 +321,7 @@ impl NodeDbPgHandler {
             return self.handle_prepare(session_id, sql_trimmed);
         }
         if upper.starts_with("EXECUTE ") {
-            return self
-                .handle_execute(identity, session_id, sql_trimmed, requested)
-                .await;
+            return self.handle_execute(identity, session_id, sql_trimmed).await;
         }
         if upper.starts_with("DEALLOCATE ") {
             return self.handle_deallocate(session_id, sql_trimmed);
@@ -414,7 +411,7 @@ impl NodeDbPgHandler {
             if let Some(ref m) = self.state.system_metrics {
                 m.record_database_query(&desc.name);
             }
-            self.state.database_metrics.record_qps(&desc.name);
+            self.state.database_metrics.record_qps(desc.id, &desc.name);
         }
 
         // Request-admission gate: internal-service exemption, blacklist,
@@ -443,9 +440,7 @@ impl NodeDbPgHandler {
             )
             .await
         {
-            return crate::control::server::pgwire::ddl_encode::ddl_results_to_pgwire(
-                result, requested,
-            );
+            return crate::control::server::pgwire::ddl_encode::ddl_results_to_pgwire(result);
         }
 
         if let Some(result) = crate::control::server::shared::ddl::dispatch(
@@ -457,9 +452,7 @@ impl NodeDbPgHandler {
         )
         .await
         {
-            return crate::control::server::pgwire::ddl_encode::ddl_results_to_pgwire(
-                result, requested,
-            );
+            return crate::control::server::pgwire::ddl_encode::ddl_results_to_pgwire(result);
         }
 
         // SHOW commands the DDL / AST router did not claim are PG
@@ -485,7 +478,7 @@ impl NodeDbPgHandler {
 
         let _request = self.state.tenant_request_guard(tenant_id);
         let result = self
-            .execute_planned_sql(identity, sql_trimmed, tenant_id, session_id, requested)
+            .execute_planned_sql(identity, sql_trimmed, tenant_id, session_id)
             .await;
 
         if result.is_err() {
diff --git a/nodedb/src/control/server/shared/ddl/neutral/observability.rs b/nodedb/src/control/server/shared/ddl/neutral/observability.rs
index e5e9e47f2..c7fa02d3d 100644
--- a/nodedb/src/control/server/shared/ddl/neutral/observability.rs
+++ b/nodedb/src/control/server/shared/ddl/neutral/observability.rs
@@ -14,8 +14,10 @@
 //! `SHOW MEMORY` reports per-engine memory budgets and current
 //! utilisation from `nodedb_mem::MemoryGovernor`.
 //!
-//! The metric source reads, ordering, and the tenant-admin gate run here. The
-//! result is the protocol-neutral `DdlResult` over `ShapedRows`.
+//! Ported from the pgwire `ddl::observability` handlers. The metric source
+//! reads, ordering, and the tenant-admin gate are preserved verbatim; only the
+//! result construction changed from pgwire `Response` / `QueryResponse` to the
+//! protocol-neutral `DdlResult` over `ShapedRows`.
 
 use serde_json::{Map, Value as JsonValue};
 
@@ -169,19 +171,19 @@ pub fn show_metrics(
     if let Some(sys) = state.system_metrics.as_ref() {
         rows.push((
             "wal_fsync_p50_us".into(),
-            sys.wal_fsync_seconds.percentile(50.0).to_string(),
+            sys.wal_fsync_seconds.percentile(0.5).to_string(),
         ));
         rows.push((
             "wal_fsync_p99_us".into(),
-            sys.wal_fsync_seconds.percentile(99.0).to_string(),
+            sys.wal_fsync_seconds.percentile(0.99).to_string(),
         ));
         rows.push((
             "query_latency_p50_us".into(),
-            sys.query_latency.percentile(50.0).to_string(),
+            sys.query_latency.percentile(0.5).to_string(),
         ));
         rows.push((
             "query_latency_p99_us".into(),
-            sys.query_latency.percentile(99.0).to_string(),
+            sys.query_latency.percentile(0.99).to_string(),
         ));
     }
 
diff --git a/nodedb/src/control/state/init.rs b/nodedb/src/control/state/init.rs
index e24b1a411..9d578e970 100644
--- a/nodedb/src/control/state/init.rs
+++ b/nodedb/src/control/state/init.rs
@@ -37,7 +37,7 @@ impl SharedState {
     /// Construct the test state every constructor in [`super::init_variants`]
     /// builds on.
     pub(super) fn new_inner(
-        mut dispatcher: Dispatcher,
+        dispatcher: Dispatcher,
         wal: Arc<WalManager>,
     ) -> crate::Result<Arc<Self>> {
         let shutdown = Arc::new(crate::control::shutdown::ShutdownWatch::new());
@@ -102,21 +102,6 @@ impl SharedState {
         // can't silently diverge from it.
         let rate_limit_config = RateLimitConfig::default();
 
-        // The CDC router matches events against the registry DDL writes to,
-        // as the production constructor wires it.
-        let stream_registry = Arc::new(crate::event::cdc::StreamRegistry::new());
-
-        let redo_chunks = Arc::new(
-            crate::control::wal_replication::transaction_redo::RedoChunkStore::new(
-                Arc::clone(&wal),
-                crate::control::wal_replication::transaction_redo::chunks::RedoChunkLimits::from_tuning(
-                    &TuningConfig::default().calvin,
-                ),
-            ),
-        );
-        let hlc_clock = wal.hlc_clock();
-        let system_metrics = Arc::new(crate::control::metrics::SystemMetrics::new());
-        dispatcher.adopt_capacity_metrics(Arc::clone(&system_metrics.dispatch_capacity));
         let state = Arc::new(Self {
             outcome_floor: dispatcher.outcome_floor(),
             dispatcher: Mutex::new(dispatcher),
@@ -140,7 +125,10 @@ impl SharedState {
             )
             .0,
             group_watchers: Arc::new(nodedb_cluster::GroupAppliedWatchers::new()),
-            metadata_ddl: super::MetadataDdlState::new(),
+            metadata_ddl_lock: std::sync::Mutex::new(()),
+            metadata_ddl_owner: std::sync::Mutex::new(None),
+            metadata_ddl_applied_token: std::sync::atomic::AtomicU64::new(0),
+            metadata_ddl_token_seq: std::sync::atomic::AtomicU64::new(1),
             pending_ddl: crate::control::pending_ddl::PendingDdlTable::new(),
             metadata_apply_wedge: std::sync::Arc::default(),
             sequencer_halt: std::sync::Arc::default(),
@@ -154,12 +142,9 @@ impl SharedState {
             ),
             raft_compactor: std::sync::OnceLock::new(),
             raft_applied_index_sink: std::sync::OnceLock::new(),
-            raft_apply_gates: std::sync::OnceLock::new(),
             raft_read_gate: std::sync::OnceLock::new(),
-            multi_raft: std::sync::OnceLock::new(),
             cluster_epoch: std::sync::OnceLock::new(),
             raft_status_fn: std::sync::OnceLock::new(),
-            data_groups_replayed: std::sync::atomic::AtomicBool::new(false),
             cluster_observer: std::sync::OnceLock::new(),
             loop_metrics_registry: nodedb_cluster::LoopMetricsRegistry::new(),
             per_vshard_metrics: crate::control::metrics::PerVShardMetricsRegistry::new(),
@@ -263,8 +248,10 @@ impl SharedState {
             block_cache: crate::control::planner::procedural::executor::ProcedureBlockCache::new(
                 4096,
             ),
-            stream_registry: Arc::clone(&stream_registry),
-            cdc_router: Arc::new(crate::event::cdc::CdcRouter::new(stream_registry)),
+            stream_registry: Arc::new(crate::event::cdc::StreamRegistry::new()),
+            cdc_router: Arc::new(crate::event::cdc::CdcRouter::new(Arc::new(
+                crate::event::cdc::StreamRegistry::new(),
+            ))),
             group_registry: crate::event::cdc::GroupRegistry::new(),
             offset_store: {
                 let dir = test_state_dir.path().join("offsets");
@@ -300,8 +287,7 @@ impl SharedState {
             cross_shard_dispatcher: None,
             cross_shard_dlq: None,
             cross_shard_metrics: None,
-            cross_shard_dedup: std::sync::OnceLock::new(),
-            redo_chunks,
+            hwm_store: None,
             kafka_manager: crate::event::kafka::KafkaManager::new(shutdown.raw_receiver()),
             definition_sync_fanout: std::sync::Arc::new(
                 crate::control::server::sync::definition_fanout::DefinitionSyncFanout::new(),
@@ -323,7 +309,7 @@ impl SharedState {
             raft_propose_leader_change_retries: AtomicU64::new(0),
             request_id_counter: AtomicU64::new(1),
             shuffle_id_counter: AtomicU64::new(1),
-            system_metrics: Some(system_metrics),
+            system_metrics: Some(Arc::new(crate::control::metrics::SystemMetrics::new())),
             database_metrics: Arc::new(crate::control::metrics::DatabaseMetricsRegistry::new()),
             quota_ceiling: Arc::new(std::sync::RwLock::new(
                 crate::control::security::catalog::GlobalQuotaCeiling::default(),
@@ -339,16 +325,14 @@ impl SharedState {
             ts_partition_registries: Some(Mutex::new(std::collections::HashMap::new())),
             cold_storage: None,
             snapshot_storage: Arc::new(object_store::memory::InMemory::new()),
-            pitr: Default::default(),
             quarantine_storage: Arc::new(object_store::memory::InMemory::new()),
-            hlc_clock,
+            hlc_clock: Arc::new(nodedb_types::HlcClock::new()),
             tenant_write_hlc: Arc::new(std::sync::Mutex::new(std::collections::HashMap::new())),
             tenant_marks: super::tenant_marks::TenantMarks::load(test_credentials.catalog())?,
             lease_admission_gate: Mutex::new(()),
-            lease_grant_gate: Arc::new(tokio::sync::Mutex::new(())),
+            lease_grant_gate: Arc::new(Mutex::new(())),
             lease_drain: Arc::new(crate::control::lease::DescriptorDrainTracker::new()),
             lease_refcount: Arc::new(crate::control::lease::LeaseRefCount::new()),
-            lease_runtime: crate::control::lease::LeaseRuntime::new(),
             sequencer_inbox: std::sync::OnceLock::new(),
             reservation_inbox: std::sync::OnceLock::new(),
             sequencer_metrics: std::sync::OnceLock::new(),
@@ -373,8 +357,6 @@ impl SharedState {
             write_order_locks: Arc::new(
                 crate::control::server::shared::write_admission::KeyedWriteOrderLock::new(),
             ),
-            write_order_fence:
-                crate::control::server::shared::write_admission::WriteOrderFence::new(),
             presence: Arc::new(tokio::sync::RwLock::new(
                 crate::control::server::sync::presence::PresenceManager::new(
                     crate::control::server::sync::presence::PresenceConfig::default(),
@@ -385,13 +367,11 @@ impl SharedState {
             gateway_invalidator: std::sync::OnceLock::new(),
             gateway: std::sync::OnceLock::new(),
             backup_kek: None,
-            backup_storage: None,
-            cut_captures: Arc::new(crate::control::backup::cut_capture::CutCaptures::new()),
-            backup_schedules: Vec::new(),
             quarantine_registry: Arc::new(crate::storage::quarantine::QuarantineRegistry::new()),
             admission_registry: Arc::new(
                 crate::control::server::admission::AdmissionRegistry::new(),
             ),
+            lsn_ms_map: Arc::new(Mutex::new(nodedb_types::temporal::LsnMsMap::new())),
             audit_dml_cache: Arc::new(crate::control::state::audit_dml_cache::AuditDmlCache::new()),
             idle_timeout_cache: Arc::new(
                 crate::control::state::idle_timeout_cache::IdleTimeoutCache::new(),
@@ -399,6 +379,7 @@ impl SharedState {
             collection_to_database: Arc::new(
                 crate::control::state::collection_to_database::CollectionToDatabase::new(),
             ),
+            materialize_freeze: crate::control::clone::MaterializeFreezeRegistry::new(),
             shuffle_registry: Arc::new(
                 // Test path: no catalog data dir, so stage under a process- and
                 // test-unique temp subdir to keep concurrent test inboxes
@@ -414,6 +395,7 @@ impl SharedState {
             startup: Arc::clone(&startup_gate),
         });
         Self::wire_session_handle_audit(&state);
+        state.wire_wal_and_metrics();
         Ok(state)
     }
 }
diff --git a/nodedb/src/control/state/init_prod/post_init.rs b/nodedb/src/control/state/init_prod/post_init.rs
index cbf37066b..aacdd9743 100644
--- a/nodedb/src/control/state/init_prod/post_init.rs
+++ b/nodedb/src/control/state/init_prod/post_init.rs
@@ -16,7 +16,24 @@ use super::super::SharedState;
 /// the first write after startup, then wire the session-handle audit sink.
 pub(super) fn hydrate_caches(state: &Arc<SharedState>) {
     SharedState::wire_session_handle_audit(state);
-    super::catalog_registries::load_catalog_caches(state, state.credentials.catalog());
+
+    let catalog = state.credentials.catalog();
+    if let Err(e) = state.audit_dml_cache.load_from_catalog(catalog) {
+        tracing::warn!(error = %e, "boot: failed to populate audit_dml_cache from catalog");
+    }
+    if let Err(e) = state.collection_to_database.load_from_catalog(catalog) {
+        tracing::warn!(
+            error = %e,
+            "boot: failed to populate collection_to_database cache from catalog"
+        );
+    }
+    if let Err(e) = state.idle_timeout_cache.load_from_catalog(catalog) {
+        tracing::warn!(
+            error = %e,
+            "boot: failed to populate idle_timeout_cache from catalog"
+        );
+    }
+    state.wire_wal_and_metrics();
 }
 
 /// Spawn the array GC background task. The handle is stored by the caller
@@ -32,6 +49,6 @@ pub(super) fn spawn_array_gc(state: &Arc<SharedState>) {
         Arc::clone(&state.shutdown),
         crate::control::array_sync::gc_task::DEFAULT_GC_INTERVAL,
     );
-    // `array_gc_handle` in SharedState stays None; main.rs can install the
+    // `array_gc_handle` in SharedState stays None; main.rs may install the
     // handle via Arc::get_mut after open() returns (before cloning the Arc).
 }
diff --git a/nodedb/src/control/state/methods.rs b/nodedb/src/control/state/methods.rs
index fb48065f9..d788198ee 100644
--- a/nodedb/src/control/state/methods.rs
+++ b/nodedb/src/control/state/methods.rs
@@ -13,6 +13,13 @@ use crate::types::TenantId;
 use super::SharedState;
 
 impl SharedState {
+    /// Wire system metrics to the WAL manager if system metrics are present.
+    pub(super) fn wire_wal_and_metrics(&self) {
+        if let Some(metrics) = &self.system_metrics {
+            self.wal.set_metrics(Arc::clone(metrics));
+        }
+    }
+
     /// Bundle of the auth-adjacent stores every `RequestAuthScope`
     /// construction needs (`scope_grants`, `quota_manager`), for callers
     /// that build a scope from `&SharedState` directly rather than pulling
@@ -21,81 +28,31 @@ impl SharedState {
         AuthStores::new(&self.scope_grants, &self.quota_manager, &self.risk_scorer)
     }
 
-    /// Sequenced Raft proposer for ordinary writes. `start_raft` installs it
-    /// before any listener opens.
+    /// Sequenced Raft proposer for ordinary writes; absent outside cluster mode.
     pub(crate) fn async_raft_proposer(
         &self,
-    ) -> crate::Result<&Arc<crate::control::wal_replication::AsyncRaftProposer>> {
+    ) -> Option<&Arc<crate::control::wal_replication::AsyncRaftProposer>> {
         self.async_raft_proposer_pair
             .get()
             .map(|pair| &pair.sequenced)
-            .ok_or_else(|| raft_not_started("the async raft proposer"))
     }
 
     /// Raw proposer used only while a CRDT admission holds the vShard sequence.
     /// Ordinary writes must use [`Self::async_raft_proposer`].
     pub(in crate::control) fn raw_async_raft_proposer(
         &self,
-    ) -> crate::Result<&Arc<crate::control::wal_replication::AsyncRaftProposer>> {
-        self.async_raft_proposer_pair
-            .get()
-            .map(|pair| &pair.raw)
-            .ok_or_else(|| raft_not_started("the raw async raft proposer"))
-    }
-
-    /// The propose phase alone: it returns once the data-group leader holds
-    /// the entry in its log, with the wait for this node's apply of it. It
-    /// takes no vShard admission slot. Calvin bookkeeping entries, which no
-    /// write admission orders, propose through it.
-    pub(in crate::control) fn async_raft_submit(
-        &self,
-    ) -> crate::Result<&Arc<crate::control::wal_replication::AsyncRaftSubmit>> {
-        self.async_raft_proposer_pair
-            .get()
-            .map(|pair| &pair.submit)
-            .ok_or_else(|| raft_not_started("the async raft submit"))
-    }
-
-    /// Handle for proposing to the metadata Raft group. `start_raft`
-    /// installs it before any listener opens.
-    pub fn metadata_raft_handle(
-        &self,
-    ) -> crate::Result<&Arc<dyn crate::control::metadata_proposer::MetadataRaftHandle>> {
-        self.metadata_raft
-            .get()
-            .ok_or_else(|| raft_not_started("the metadata raft handle"))
+    ) -> Option<&Arc<crate::control::wal_replication::AsyncRaftProposer>> {
+        self.async_raft_proposer_pair.get().map(|pair| &pair.raw)
     }
 
-    /// Synchronous data-group Raft proposer. `start_raft` installs it before
-    /// any listener opens.
-    pub(crate) fn sync_raft_proposer(
-        &self,
-    ) -> crate::Result<&Arc<crate::control::wal_replication::RaftProposer>> {
-        self.raft_proposer
-            .get()
-            .ok_or_else(|| raft_not_started("the raft proposer"))
-    }
-
-    /// The gateway `install_gateway` installs before any listener opens.
-    pub fn installed_gateway(&self) -> crate::Result<&Arc<crate::control::gateway::Gateway>> {
-        self.gateway.get().ok_or_else(|| crate::Error::Internal {
-            detail: "the gateway is not installed: install_gateway has not run on this node".into(),
-        })
-    }
-
-    /// Install the Raft proposal handles atomically during cluster startup.
+    /// Install both Raft proposal handles atomically during cluster startup.
     pub(in crate::control) fn install_async_raft_proposer_pair(
         &self,
         sequenced: Arc<crate::control::wal_replication::AsyncRaftProposer>,
         raw: Arc<crate::control::wal_replication::AsyncRaftProposer>,
-        submit: Arc<crate::control::wal_replication::AsyncRaftSubmit>,
     ) -> crate::Result<()> {
         self.async_raft_proposer_pair
-            .set(super::proposer_pair::AsyncRaftProposerPair {
-                sequenced,
-                raw,
-                submit,
-            })
+            .set(super::fields::AsyncRaftProposerPair { sequenced, raw })
             .map_err(|_| crate::Error::Internal {
                 detail: "async raft proposer already installed".into(),
             })
@@ -103,18 +60,24 @@ impl SharedState {
 
     /// Recover a strong `Arc<SharedState>` from `&self` by upgrading the
     /// gateway's `Weak` back-reference. Always succeeds on a booted node;
-    /// returns a typed error while racing teardown, or before `install_gateway`
-    /// runs.
+    /// returns a typed error while racing teardown, or on a state built
+    /// without `state_wiring` (unit fixtures).
     pub(crate) fn self_arc(&self) -> crate::Result<Arc<SharedState>> {
-        self.installed_gateway()?.shared()
+        let gateway = self.gateway.get().ok_or_else(|| crate::Error::Internal {
+            detail: "SharedState::self_arc: gateway back-reference is not installed; \
+                     this state was built without `bootstrap::state_wiring`"
+                .into(),
+        })?;
+        gateway.shared()
     }
 
     /// Whether this node is the leader of the metadata Raft group.
     ///
     /// Reuses the installed `raft_status_fn` snapshot (set by `start_raft`),
     /// looks up the metadata group (`METADATA_GROUP_ID == 0`), and reports
-    /// whether its role string is `"Leader"`. Returns `false` before
-    /// `start_raft` installs `raft_status_fn`.
+    /// whether its role string is `"Leader"`. Returns `false` in single-node
+    /// mode, where `raft_status_fn` is never installed — there is no metadata
+    /// Raft group to lead.
     ///
     /// Not unit-tested in isolation: it requires a live `raft_status_fn`, so
     /// it is exercised by the cluster-level constraint-reconcile test.
@@ -127,11 +90,15 @@ impl SharedState {
             .any(|g| g.group_id == nodedb_cluster::METADATA_GROUP_ID && g.role == "Leader")
     }
 
-    /// Whether this node runs work that must happen exactly once
-    /// cluster-wide: the metadata-group leader does it. A one-node cluster
-    /// leads its own metadata group.
+    /// Whether this node should perform work that must happen exactly once
+    /// cluster-wide.
+    ///
+    /// Standalone deployments never install a metadata Raft group, so
+    /// `is_metadata_leader` is permanently false there. Gating a singleton job
+    /// on leadership alone therefore disables it entirely on single-node
+    /// deployments — the job must run when there is no group to elect from.
     pub fn is_singleton_worker(&self) -> bool {
-        self.is_metadata_leader()
+        self.metadata_raft.get().is_none() || self.is_metadata_leader()
     }
 
     /// Snapshot the configured global quota ceiling.
@@ -150,7 +117,7 @@ impl SharedState {
     }
 
     /// Replace the global quota ceiling. Called once at startup after the
-    /// server config is parsed; future `ALTER SYSTEM` paths can also call this.
+    /// server config is parsed; future `ALTER SYSTEM` paths may also call this.
     pub fn set_quota_ceiling(
         &self,
         ceiling: crate::control::security::catalog::GlobalQuotaCeiling,
@@ -189,28 +156,95 @@ impl SharedState {
             .fetch_add(1, std::sync::atomic::Ordering::Relaxed)
     }
 
-    /// Commit time, in wall milliseconds, of the WAL state below `lsn`: every
-    /// record under `lsn` committed by then.
+    /// Convert an LSN back to wall-clock milliseconds using the anchor map.
+    ///
+    /// Used by the clone resolver to compute `effective_source_ms` for the
+    /// engine's `system_as_of_ms` field. Returns `None` when the anchor map
+    /// has no entries that bracket the requested LSN (engine then uses
+    /// latest-version behaviour — safe for tests where anchors are absent).
     ///
-    /// `lsn` is an exclusive bound, as `wal.next_lsn()` and [`Self::ms_to_lsn`]
-    /// return. `None` when no time anchor covers `lsn - 1` yet.
+    /// A poisoned lock is recovered via `into_inner()` (consistent with the
+    /// other `lsn_ms_map` accessors below) and logged so that bitemporal
+    /// reads do not silently degrade to "latest" on the back of a panic in
+    /// an unrelated writer.
     pub fn ms_to_lsn_inverse(&self, lsn: nodedb_types::Lsn) -> Option<i64> {
-        let last = lsn.as_u64().checked_sub(1)?;
-        let ns = self.wal.time_anchors().commit_ns_of(last)?;
-        i64::try_from(ns / 1_000_000).ok()
+        let map = match self.lsn_ms_map.lock() {
+            Ok(g) => g,
+            Err(poisoned) => {
+                warn!(
+                    lsn = %lsn,
+                    "lsn_ms_map poisoned in ms_to_lsn_inverse — recovering inner state; \
+                     bitemporal reads may use a stale anchor view until the next anchor is recorded"
+                );
+                poisoned.into_inner()
+            }
+        };
+        nodedb_types::temporal::lsn_to_ms(&map, lsn).ok()
+    }
+
+    /// Convert wall-clock milliseconds to the nearest LSN using the anchor map.
+    ///
+    /// When the map is populated (WAL anchor records have been replayed), this
+    /// performs linear interpolation between surrounding anchors.  When the map
+    /// is empty (no anchors yet, or testing) the current WAL frontier is
+    /// returned as the best available approximation.
+    pub fn ms_to_lsn(&self, wall_ms: i64) -> nodedb_types::Lsn {
+        let map = match self.lsn_ms_map.lock() {
+            Ok(g) => g,
+            Err(poisoned) => {
+                warn!(
+                    wall_ms = wall_ms,
+                    "lsn_ms_map poisoned in ms_to_lsn — recovering inner state"
+                );
+                poisoned.into_inner()
+            }
+        };
+        // Walk anchors to find the pair that brackets wall_ms.
+        let anchors = map.anchors();
+        if anchors.is_empty() {
+            // No anchor data yet.  A very small wall_ms (well before any
+            // plausible server-start epoch) predates all recorded LSNs;
+            // return LSN 0 so the clone predation check fires correctly.
+            // Any other value maps to the current frontier.
+            const EPOCH_THRESHOLD_MS: i64 = 1_000_000; // ~16 minutes after Unix epoch
+            if wall_ms < EPOCH_THRESHOLD_MS {
+                return nodedb_types::Lsn::new(0);
+            }
+            return self.wal.next_lsn();
+        }
+        // Binary search by wall_ms to find the bracketing pair.
+        match anchors.binary_search_by_key(&wall_ms, |a| a.wall_ms) {
+            Ok(idx) => nodedb_types::Lsn::new(anchors[idx].lsn),
+            Err(0) => nodedb_types::Lsn::new(anchors[0].lsn),
+            Err(idx) if idx >= anchors.len() => {
+                nodedb_types::Lsn::new(anchors[anchors.len() - 1].lsn)
+            }
+            Err(idx) => {
+                // Interpolate between anchors[idx-1] and anchors[idx].
+                let lo = anchors[idx - 1];
+                let hi = anchors[idx];
+                let ms_span = (hi.wall_ms - lo.wall_ms).max(1) as u64;
+                let lsn_span = hi.lsn.saturating_sub(lo.lsn);
+                let ms_delta = (wall_ms - lo.wall_ms).max(0) as u64;
+                let lsn = lo.lsn + (lsn_span * ms_delta / ms_span);
+                nodedb_types::Lsn::new(lsn)
+            }
+        }
     }
 
-    /// The exclusive LSN bound of the WAL state committed by the end of
-    /// millisecond `wall_ms`: every record under it committed by then.
+    /// Record a new WAL anchor into the LSN↔ms map.
     ///
-    /// A time before the oldest retained anchor is an error, never the WAL
-    /// frontier. A time past the newest anchor resolves to that anchor, since
-    /// the records after it have not committed.
-    pub fn ms_to_lsn(&self, wall_ms: i64) -> Result<nodedb_types::Lsn, nodedb_types::LsnTimeError> {
-        self.wal
-            .time_anchors()
-            .lsn_at_or_before_ms(wall_ms)
-            .map(|last| nodedb_types::Lsn::new(last.saturating_add(1)))
+    /// Called when an `LsnMsAnchor` WAL record is replayed at startup or
+    /// emitted by the WAL writer.  Non-monotonic anchors are silently ignored
+    /// (the WAL guarantees monotonicity; a violation here means a partially
+    /// replayed or corrupt segment and must not panic).
+    pub fn push_lsn_ms_anchor(&self, lsn: u64, wall_ms: i64) {
+        if let Ok(mut map) = self.lsn_ms_map.lock() {
+            let anchor = nodedb_types::temporal::LsnMsAnchor::new(lsn, wall_ms);
+            // Silently ignore non-monotonic anchors — they arrive during WAL
+            // replay of partially-written segments and must not crash the server.
+            let _ = map.push(anchor);
+        }
     }
 
     /// Shared HTTP client reused by every outbound emitter. Cloning the
@@ -226,9 +260,11 @@ impl SharedState {
     /// live topology under a short read guard, so version updates
     /// from joins / leaves are observed immediately.
     ///
-    /// Returns `ClusterVersionView::single_node()` when no topology
-    /// handle is wired: callers that gate on a cluster-wide minimum
-    /// treat this as "all nodes run the local build".
+    /// Returns `ClusterVersionView::single_node()` when no
+    /// topology handle is installed (single-node mode): callers
+    /// that gate on a cluster-wide minimum treat this as "all
+    /// nodes run the local build", which is the correct behavior
+    /// for a solo node.
     pub fn cluster_version_view(&self) -> crate::control::rolling_upgrade::ClusterVersionView {
         let Some(topology) = &self.cluster_topology else {
             return crate::control::rolling_upgrade::ClusterVersionView::single_node();
@@ -242,7 +278,7 @@ impl SharedState {
     /// Lazily creates the watcher if it does not yet exist so a
     /// proposer can register its waiter before the first apply on a
     /// brand-new group. Used by
-    /// [`crate::control::metadata_proposer::propose_catalog_entry_async`]
+    /// [`crate::control::metadata_proposer::propose_catalog_entry`]
     /// (with `nodedb_cluster::METADATA_GROUP_ID`) and by the
     /// descriptor-lease drain path. Distributed-write commit
     /// waiting goes through `propose_tracker` directly because it
@@ -404,13 +440,6 @@ impl SharedState {
     }
 }
 
-/// The error for a Raft handle read before `start_raft` installed it.
-fn raft_not_started(what: &str) -> crate::Error {
-    crate::Error::Internal {
-        detail: format!("{what} is not installed: start_raft has not run on this node"),
-    }
-}
-
 #[cfg(test)]
 mod tests {
     use std::collections::HashMap;
@@ -421,7 +450,7 @@ mod tests {
     /// Reading `0` from a poisoned lock silently disables the restore
     /// staleness gate: the gate refuses an envelope whose watermark is older
     /// than this value, and every watermark clears a mark of `0`. A restore
-    /// then overwrites newer writes with no error.
+    /// would then overwrite newer writes with no error.
     #[test]
     fn a_poisoned_write_hlc_map_still_reports_the_recorded_mark() {
         let map: Arc<Mutex<HashMap<u64, u64>>> = Arc::new(Mutex::new(HashMap::new()));
diff --git a/nodedb/src/wal/manager/core.rs b/nodedb/src/wal/manager/core.rs
index 03ad79e25..c410842fa 100644
--- a/nodedb/src/wal/manager/core.rs
+++ b/nodedb/src/wal/manager/core.rs
@@ -7,7 +7,6 @@ use std::sync::{Arc, Mutex};
 use tracing::info;
 
 use nodedb_types::config::tuning::WalTuning;
-use nodedb_wal::TimeAnchors;
 use nodedb_wal::segmented::{SegmentedWal, SegmentedWalConfig};
 use nodedb_wal::writer::WalWriterConfig;
 
@@ -46,11 +45,8 @@ pub struct WalManager {
     /// Wakes `wait_durable` followers when `durable_lsn` advances (or a leader's
     /// fsync fails, so they re-attempt and observe the same error).
     pub(super) durable_notify: tokio::sync::Notify,
-    /// Commit-time anchors of this WAL, stamped from the node's HLC. The writer
-    /// records one per fsynced batch. Replay records the persisted ones.
-    pub(super) time_anchors: Arc<TimeAnchors>,
-    /// Floors truncation keeps for holders that read records back at boot.
-    pub(super) floor_holds: super::floor_holds::WalFloorHolds,
+    /// System metrics registry handle for recording group-commit fsync latency.
+    pub(super) metrics: std::sync::RwLock<Option<Arc<crate::control::metrics::SystemMetrics>>>,
 }
 
 impl WalManager {
@@ -69,6 +65,20 @@ impl WalManager {
         self.durable_lsn.load(std::sync::atomic::Ordering::Acquire)
     }
 
+    /// Wire or replace the system metrics handle used to record group-commit fsync latency.
+    pub fn set_metrics(&self, metrics: Arc<crate::control::metrics::SystemMetrics>) {
+        let mut w = self.metrics.write().unwrap_or_else(|p| p.into_inner());
+        *w = Some(metrics);
+    }
+
+    /// Read the optional system metrics handle.
+    pub fn metrics(&self) -> Option<Arc<crate::control::metrics::SystemMetrics>> {
+        self.metrics
+            .read()
+            .unwrap_or_else(|p| p.into_inner())
+            .clone()
+    }
+
     /// Return the stable in-memory root for per-user CRDT signing keys.
     /// The root is persisted only as WAL-key-wrapped ciphertext and is
     /// rewrapped on rotation, so offline signatures survive chained key
@@ -78,22 +88,6 @@ impl WalManager {
         Ok(self.crdt_signing_root)
     }
 
-    /// The floors truncation of this WAL keeps.
-    pub fn floor_holds(&self) -> &super::floor_holds::WalFloorHolds {
-        &self.floor_holds
-    }
-
-    /// This WAL's commit-time anchors.
-    pub fn time_anchors(&self) -> &Arc<TimeAnchors> {
-        &self.time_anchors
-    }
-
-    /// The node's HLC. Anchors are stamped from it, so every commit stamp it
-    /// issues orders against anchor times.
-    pub fn hlc_clock(&self) -> Arc<nodedb_types::HlcClock> {
-        Arc::clone(self.time_anchors.clock())
-    }
-
     /// Open or create a segmented WAL at the given path.
     ///
     /// The `path` argument is the WAL directory for the segmented format.
@@ -136,8 +130,6 @@ impl WalManager {
                 alignment: tuning.alignment,
                 use_direct_io: tuning.direct_io,
                 dwb_mode: None,
-                // `open_internal` installs the manager's anchors.
-                time_anchors: None,
             },
         )
     }
@@ -146,11 +138,9 @@ impl WalManager {
     pub(super) fn open_internal(
         path: &Path,
         segment_target_size: u64,
-        mut writer_config: WalWriterConfig,
+        writer_config: WalWriterConfig,
     ) -> crate::Result<Self> {
         let wal_dir = path.to_path_buf();
-        let time_anchors = Arc::new(TimeAnchors::new(Arc::new(nodedb_types::HlcClock::new())));
-        writer_config.time_anchors = Some(Arc::clone(&time_anchors));
 
         let effective_target = if segment_target_size > 0 {
             segment_target_size
@@ -166,11 +156,6 @@ impl WalManager {
         };
 
         let wal = SegmentedWal::open(config).map_err(crate::Error::Wal)?;
-        // A new log has nothing to replay. Anchor its empty state now, so a
-        // time after open resolves to LSN 0 instead of an error.
-        if wal.next_lsn() == 1 {
-            time_anchors.cover_recovered(0);
-        }
 
         // `direct_io` is logged because a WAL running buffered is a weaker
         // durability posture than the default, and the log is the only place
@@ -196,15 +181,14 @@ impl WalManager {
 
         Ok(Self {
             wal: Arc::new(Mutex::new(wal)),
-            wal_dir,
+            wal_dir: wal_dir.to_path_buf(),
             encryption_ring: None,
             crdt_signing_root: None,
             audit_wal,
             durable_lsn: AtomicU64::new(0),
             commit_lock: tokio::sync::Mutex::new(()),
             durable_notify: tokio::sync::Notify::new(),
-            time_anchors,
-            floor_holds: super::floor_holds::WalFloorHolds::default(),
+            metrics: std::sync::RwLock::new(None),
         })
     }
 
diff --git a/nodedb/src/wal/manager/durable_commit.rs b/nodedb/src/wal/manager/durable_commit.rs
index feb92eb38..f563e00d0 100644
--- a/nodedb/src/wal/manager/durable_commit.rs
+++ b/nodedb/src/wal/manager/durable_commit.rs
@@ -63,9 +63,15 @@ impl WalManager {
                     // advances to exactly what the fsync made durable, never past
                     // it.
                     let wal = std::sync::Arc::clone(&self.wal);
+                    let metrics = self.metrics();
                     let join = tokio::task::spawn_blocking(move || -> crate::Result<u64> {
                         let mut guard = wal.lock().unwrap_or_else(|p| p.into_inner());
+                        let start = std::time::Instant::now();
                         guard.sync().map_err(crate::Error::Wal)?;
+                        let elapsed_us = start.elapsed().as_micros() as u64;
+                        if let Some(m) = &metrics {
+                            m.record_wal_fsync(elapsed_us);
+                        }
                         // `next_lsn()` is the next LSN to assign; the highest LSN
                         // this sync made durable is one below it.
                         Ok(guard.next_lsn().saturating_sub(1))
@@ -113,6 +119,8 @@ mod tests {
     use super::*;
     use crate::types::{DatabaseId, TenantId, VShardId};
     use crate::wal::manager::NO_APPLY_KEY;
+    use std::sync::Arc;
+    use std::sync::atomic::Ordering;
 
     fn open_wal(dir: &std::path::Path) -> WalManager {
         WalManager::open_for_testing(&dir.join("test.wal")).expect("open wal")
@@ -140,6 +148,8 @@ mod tests {
     async fn wait_durable_fast_path_when_already_durable() {
         let dir = tempfile::tempdir().expect("tempdir");
         let wal = open_wal(dir.path());
+        let metrics = Arc::new(crate::control::metrics::SystemMetrics::new());
+        wal.set_metrics(Arc::clone(&metrics));
         let lsn = wal
             .appender(NO_APPLY_KEY)
             .with_event_source(crate::event::EventSource::User)
@@ -151,8 +161,11 @@ mod tests {
             )
             .expect("append");
         wal.wait_durable(lsn).await.expect("first");
+        assert_eq!(metrics.wal_fsync_count.load(Ordering::Relaxed), 1);
+        assert_eq!(metrics.wal_fsync_seconds.count(), 1);
         // Second call is the fast path — no further fsync required.
         wal.wait_durable(lsn).await.expect("second");
+        assert_eq!(metrics.wal_fsync_count.load(Ordering::Relaxed), 1);
     }
 
     #[tokio::test(flavor = "multi_thread", worker_threads = 4)]
diff --git a/nodedb/tests/inproc/cases/quota_drop_cleanup.rs b/nodedb/tests/inproc/cases/quota_drop_cleanup.rs
index 63a7b7d57..f834b3e61 100644
--- a/nodedb/tests/inproc/cases/quota_drop_cleanup.rs
+++ b/nodedb/tests/inproc/cases/quota_drop_cleanup.rs
@@ -245,11 +245,11 @@ async fn dropped_database_id_does_not_inherit_cap() {
         None,
         "the dead cap must not survive for a database reusing this id"
     );
+    let permit = registry
+        .try_acquire_database(db)
+        .expect("an uncapped database admits");
     assert!(
-        registry
-            .try_acquire_database(db)
-            .expect("an uncapped database admits")
-            .is_none(),
-        "an uncapped database hands out no permit"
+        permit.is_some(),
+        "an uncapped database hands out an unconstrained permit"
     );
 }
diff --git a/nodedb/tests/inproc/cases/quota_live_enforcement_apply.rs b/nodedb/tests/inproc/cases/quota_live_enforcement_apply.rs
index 49bcfff32..29f3a3eb2 100644
--- a/nodedb/tests/inproc/cases/quota_live_enforcement_apply.rs
+++ b/nodedb/tests/inproc/cases/quota_live_enforcement_apply.rs
@@ -90,12 +90,12 @@ async fn alter_database_quota_zero_clears_connection_cap() {
         None,
         "zero drops the entry, so the database is uncapped again"
     );
+    let permit = registry
+        .try_acquire_database(db)
+        .expect("an uncapped database admits");
     assert!(
-        registry
-            .try_acquire_database(db)
-            .expect("an uncapped database admits")
-            .is_none(),
-        "an uncapped database hands out no permit"
+        permit.is_some(),
+        "an uncapped database hands out an unconstrained permit"
     );
 }
 
diff --git a/nodedb/tests/inproc/cases/quota_three_level_denial.rs b/nodedb/tests/inproc/cases/quota_three_level_denial.rs
index dc927dd0a..967379d09 100644
--- a/nodedb/tests/inproc/cases/quota_three_level_denial.rs
+++ b/nodedb/tests/inproc/cases/quota_three_level_denial.rs
@@ -216,15 +216,23 @@ fn maintenance_budget_tracker_three_level() {
 #[test]
 fn database_metrics_registry_independent_counters() {
     let reg = DatabaseMetricsRegistry::new();
+    let db_alpha = nodedb_types::DatabaseId::new(1);
+    let db_beta = nodedb_types::DatabaseId::new(2);
     for _ in 0..5 {
-        reg.record_qps("alpha");
+        reg.record_qps(db_alpha, "alpha");
     }
     for _ in 0..3 {
-        reg.record_qps("beta");
+        reg.record_qps(db_beta, "beta");
     }
     use std::sync::atomic::Ordering;
-    let alpha = reg.get_or_create("alpha").qps_total.load(Ordering::Relaxed);
-    let beta = reg.get_or_create("beta").qps_total.load(Ordering::Relaxed);
+    let alpha = reg
+        .get_or_create(db_alpha, "alpha")
+        .qps_total
+        .load(Ordering::Relaxed);
+    let beta = reg
+        .get_or_create(db_beta, "beta")
+        .qps_total
+        .load(Ordering::Relaxed);
     assert_eq!(alpha, 5);
     assert_eq!(beta, 3);
 
diff --git a/nodedb/tests/wire/cases/database_metrics.rs b/nodedb/tests/wire/cases/database_metrics.rs
index 3f9b0ca5b..44d09fc8e 100644
--- a/nodedb/tests/wire/cases/database_metrics.rs
+++ b/nodedb/tests/wire/cases/database_metrics.rs
@@ -168,3 +168,38 @@ async fn tenant_quota_metrics_non_zero_under_load() {
         "at least one tenant should have tracked requests: {body}"
     );
 }
+
+#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
+async fn drop_database_removes_metrics_series() {
+    let (server, db_a) = TestServer::with_database("metrics_drop_test").await;
+
+    // Issue a query so db_a creates a series in DatabaseMetricsRegistry
+    server
+        .exec("CREATE COLLECTION col_drop (id STRING PRIMARY KEY) WITH (engine='kv')")
+        .await
+        .unwrap();
+
+    let body = fetch_metrics_body(server.http_port).await;
+    assert!(
+        body.contains(&format!("database=\"{db_a}\"")),
+        "database '{db_a}' must appear in /metrics after queries"
+    );
+
+    // Drop database via SQL through real post-apply path
+    server
+        .client
+        .simple_query("USE DATABASE default")
+        .await
+        .unwrap();
+    server
+        .client
+        .simple_query(&format!("DROP DATABASE {db_a}"))
+        .await
+        .unwrap();
+
+    let body_after = fetch_metrics_body(server.http_port).await;
+    assert!(
+        !body_after.contains(&format!("database=\"{db_a}\"")),
+        "database '{db_a}' series must be removed from /metrics after DROP DATABASE: {body_after}"
+    );
+}