nodedb_lite/nodedb/
health.rs

1//! Health API — structured status report for NodeDB-Lite.
2//!
3//! `db.health()` returns a `HealthStatus` covering:
4//! - **Storage**: accessible, approximate size
5//! - **Memory**: governor pressure per engine
6//! - **Engines**: HNSW collection count, CSR node/edge count, CRDT doc count, text indices
7//! - **Sync**: connection state, pending delta count/bytes (if sync client available)
8//!
9//! The response is JSON-serializable for HTTP health endpoints.
10
11use serde::Serialize;
12
13use nodedb_mem::{EngineId, PressureLevel};
14use nodedb_types::error::NodeDbResult;
15
16use crate::storage::engine::StorageEngine;
17
18use super::core::NodeDbLite;
19use super::lock_ext::LockExt;
20
21/// Overall health status.
22#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)]
23#[serde(rename_all = "snake_case")]
24pub enum OverallStatus {
25    /// All subsystems healthy.
26    Healthy,
27    /// Some subsystems under pressure but functional.
28    Degraded,
29    /// Critical issues — immediate attention needed.
30    Unhealthy,
31}
32
33/// Structured health report for NodeDB-Lite.
34#[derive(Debug, Serialize)]
35pub struct HealthStatus {
36    /// Overall status.
37    pub status: OverallStatus,
38    /// Storage subsystem.
39    pub storage: StorageHealth,
40    /// Memory governor.
41    pub memory: MemoryHealth,
42    /// Engine-specific health.
43    pub engines: EnginesHealth,
44}
45
46/// Storage subsystem health.
47#[derive(Debug, Serialize)]
48pub struct StorageHealth {
49    /// Whether the storage backend is accessible (can read/write).
50    pub accessible: bool,
51}
52
53/// Memory governor health.
54#[derive(Debug, Serialize)]
55pub struct MemoryHealth {
56    /// Total budget in bytes.
57    pub budget_bytes: usize,
58    /// Total used in bytes.
59    pub used_bytes: usize,
60    /// Usage ratio (0.0–1.0+).
61    pub usage_ratio: f64,
62    /// Overall pressure level.
63    pub pressure: &'static str,
64    /// Per-engine breakdown.
65    pub engines: EngineMemoryBreakdown,
66}
67
68/// Per-engine memory breakdown.
69#[derive(Debug, Serialize)]
70pub struct EngineMemoryBreakdown {
71    pub hnsw: EngineMemory,
72    pub csr: EngineMemory,
73    pub loro: EngineMemory,
74    pub query: EngineMemory,
75}
76
77/// Single engine memory stats.
78#[derive(Debug, Serialize)]
79pub struct EngineMemory {
80    pub budget_bytes: usize,
81    pub used_bytes: usize,
82    pub pressure: &'static str,
83}
84
85/// Engine-specific health summary.
86#[derive(Debug, Serialize)]
87pub struct EnginesHealth {
88    /// Number of loaded HNSW collections.
89    pub hnsw_collection_count: usize,
90    /// Total vectors across all HNSW collections.
91    pub hnsw_total_vectors: usize,
92    /// CSR graph node count.
93    pub csr_node_count: usize,
94    /// CSR graph edge count.
95    pub csr_edge_count: usize,
96    /// Number of CRDT collections with data.
97    pub crdt_collection_count: usize,
98    /// Number of text-indexed collections.
99    pub text_index_count: usize,
100    /// Total pending CRDT deltas awaiting sync.
101    pub pending_deltas: usize,
102}
103
104fn pressure_str(p: PressureLevel) -> &'static str {
105    match p {
106        PressureLevel::Normal => "normal",
107        PressureLevel::Warning => "warning",
108        PressureLevel::Critical => "critical",
109        PressureLevel::Emergency => "emergency",
110    }
111}
112
113fn engine_memory(gov: &nodedb_mem::MemoryGovernor, id: EngineId) -> EngineMemory {
114    let budget = gov.budget(id);
115    EngineMemory {
116        budget_bytes: budget.limit(),
117        used_bytes: budget.allocated(),
118        pressure: pressure_str(gov.engine_pressure(id)),
119    }
120}
121
122impl<S: StorageEngine> NodeDbLite<S> {
123    /// Borrow the underlying storage engine.
124    ///
125    /// Public so benchmark code can call backend-specific methods like
126    /// backend-specific methods (e.g. size reporting) for compression-ratio measurement.
127    pub fn storage(&self) -> &S {
128        &self.storage
129    }
130
131    /// Compact the backing storage engine, reclaiming dead pages and
132    /// truncating the file to bound on-disk growth.
133    ///
134    /// Forwards to [`StorageEngine::compact`]. For the pagedb-backed engine this
135    /// drains the deferred-free list and truncates `main.db`; for in-memory or
136    /// test engines it is a no-op returning a zero
137    /// [`CompactionOutcome`](crate::storage::engine::CompactionOutcome).
138    pub async fn compact(&self) -> NodeDbResult<crate::storage::engine::CompactionOutcome> {
139        Ok(self.storage.compact().await?)
140    }
141
142    /// Get a structured health report.
143    ///
144    /// This is a cheap, non-blocking call — reads atomic counters and lock-free state.
145    /// Safe to call frequently from health check endpoints.
146    pub fn health(&self) -> HealthStatus {
147        // Refresh memory stats before reporting.
148        self.update_memory_stats();
149
150        let gov = &self.governor;
151
152        let memory = MemoryHealth {
153            budget_bytes: gov.global_ceiling(),
154            used_bytes: gov.total_allocated(),
155            usage_ratio: gov.global_utilization_percent() as f64 / 100.0,
156            pressure: pressure_str(gov.global_pressure()),
157            engines: EngineMemoryBreakdown {
158                hnsw: engine_memory(gov, EngineId::Vector),
159                csr: engine_memory(gov, EngineId::Graph),
160                loro: engine_memory(gov, EngineId::Crdt),
161                query: engine_memory(gov, EngineId::Query),
162            },
163        };
164
165        let (hnsw_count, hnsw_vectors) = {
166            let indices = self.vector_state.hnsw_indices.lock_or_recover();
167            let count = indices.len();
168            let vectors: usize = indices.values().map(|idx| idx.len()).sum();
169            (count, vectors)
170        };
171
172        let (csr_nodes, csr_edges) = {
173            let csr_map = self.csr.lock_or_recover();
174            let nodes: usize = csr_map.values().map(|c| c.node_count()).sum();
175            let edges: usize = csr_map.values().map(|c| c.edge_count()).sum();
176            (nodes, edges)
177        };
178
179        let (crdt_collections, pending_deltas) = {
180            let crdt = self.crdt.lock_or_recover();
181            (crdt.collection_names().len(), crdt.pending_count())
182        };
183
184        let text_count = {
185            let fts = self.fts_state.manager.lock_or_recover();
186            fts.collection_count()
187        };
188
189        let engines = EnginesHealth {
190            hnsw_collection_count: hnsw_count,
191            hnsw_total_vectors: hnsw_vectors,
192            csr_node_count: csr_nodes,
193            csr_edge_count: csr_edges,
194            crdt_collection_count: crdt_collections,
195            text_index_count: text_count,
196            pending_deltas,
197        };
198
199        // Determine overall status. Matches the old thresholds: nodedb_mem's
200        // Critical (85-95%) is where Lite's own Warning used to start, and
201        // Emergency (>95%) is where Lite's own Critical used to start.
202        let overall = match gov.global_pressure() {
203            PressureLevel::Emergency => OverallStatus::Unhealthy,
204            PressureLevel::Critical => OverallStatus::Degraded,
205            PressureLevel::Warning => OverallStatus::Healthy,
206            PressureLevel::Normal => OverallStatus::Healthy,
207        };
208
209        HealthStatus {
210            status: overall,
211            storage: StorageHealth { accessible: true },
212            memory,
213            engines,
214        }
215    }
216}
217
218#[cfg(test)]
219mod tests {
220    use super::*;
221    use crate::PagedbStorageMem;
222
223    async fn make_db() -> std::sync::Arc<NodeDbLite<PagedbStorageMem>> {
224        let storage = PagedbStorageMem::open_in_memory().await.unwrap();
225        NodeDbLite::open(storage).await.unwrap()
226    }
227
228    #[tokio::test]
229    async fn health_empty_db() {
230        let db = make_db().await;
231        let h = db.health();
232        assert_eq!(h.status, OverallStatus::Healthy);
233        assert!(h.storage.accessible);
234        assert_eq!(h.memory.pressure, "normal");
235        assert_eq!(h.engines.hnsw_collection_count, 0);
236        assert_eq!(h.engines.pending_deltas, 0);
237    }
238
239    #[tokio::test]
240    async fn health_with_data() {
241        use nodedb_client::NodeDb;
242
243        let db = make_db().await;
244        db.vector_insert("vecs", "v1", &[1.0, 0.0, 0.0], None)
245            .await
246            .unwrap();
247        db.graph_insert_edge(
248            "test",
249            &nodedb_types::id::NodeId::from_validated("a".to_string()),
250            &nodedb_types::id::NodeId::from_validated("b".to_string()),
251            "REL",
252            None,
253        )
254        .await
255        .unwrap();
256
257        let h = db.health();
258        assert_eq!(h.engines.hnsw_collection_count, 1);
259        assert_eq!(h.engines.hnsw_total_vectors, 1);
260        assert!(h.engines.csr_edge_count >= 1);
261    }
262
263    #[tokio::test]
264    async fn health_serializes_to_json() {
265        let db = make_db().await;
266        let h = db.health();
267        let json = sonic_rs::to_string_pretty(&h).unwrap();
268        assert!(json.contains("\"status\""));
269        assert!(json.contains("\"storage\""));
270        assert!(json.contains("\"memory\""));
271        assert!(json.contains("\"engines\""));
272    }
273
274    #[tokio::test]
275    async fn health_pending_deltas_counted() {
276        use nodedb_client::NodeDb;
277        use nodedb_types::document::Document;
278
279        let db = make_db().await;
280        let doc = Document::new("d1");
281        db.document_put("docs", doc).await.unwrap();
282
283        let h = db.health();
284        assert!(h.engines.pending_deltas > 0);
285    }
286}
nodedb_lite/nodedb/health.rs

nodedb_lite/nodedb/
health.rs
