relay_knowledge/storage/partitioned/diagnostics/
mod.rs

1//! Bounded aggregation of control and repository-shard SQLite diagnostics.
2
3use std::collections::BTreeSet;
4use std::path::Path;
5
6use crate::paths::RuntimePaths;
7use crate::storage::sqlite::read_only_database_diagnostics;
8use crate::storage::{
9    GraphInspection, GraphStore, HealthStorageSnapshot, SqliteStorageDiagnostics, StorageError,
10    StorageTopologySnapshot,
11};
12
13use super::{
14    PartitionedSqliteKnowledgeStore,
15    catalog::{catalog_has_active_repositories, catalog_topology_snapshot},
16};
17
18impl PartitionedSqliteKnowledgeStore {
19    pub fn has_active_catalog(control_path: impl AsRef<Path>) -> Result<bool, StorageError> {
20        catalog_has_active_repositories(control_path.as_ref())
21    }
22
23    pub async fn topology_snapshot(&self) -> Result<StorageTopologySnapshot, StorageError> {
24        self.catalog.topology_snapshot().await
25    }
26
27    pub fn topology_snapshot_from_catalog(
28        control_path: impl AsRef<Path>,
29        paths: &RuntimePaths,
30    ) -> Result<StorageTopologySnapshot, StorageError> {
31        catalog_topology_snapshot(control_path.as_ref(), paths)
32    }
33}
34
35pub(super) async fn inspect_graph(
36    store: &PartitionedSqliteKnowledgeStore,
37) -> Result<GraphInspection, StorageError> {
38    let mut graph = store.control.inspect_graph().await?;
39    graph.sqlite = aggregate_sqlite_diagnostics(store, graph.sqlite).await?;
40    Ok(graph)
41}
42
43pub(super) async fn health_snapshot(
44    store: &PartitionedSqliteKnowledgeStore,
45    now_ms: u64,
46) -> Result<HealthStorageSnapshot, StorageError> {
47    let mut snapshot = store.control.health_snapshot(now_ms).await?;
48    snapshot.graph.sqlite = aggregate_sqlite_diagnostics(store, snapshot.graph.sqlite).await?;
49    Ok(snapshot)
50}
51
52async fn aggregate_sqlite_diagnostics(
53    store: &PartitionedSqliteKnowledgeStore,
54    control_sqlite: SqliteStorageDiagnostics,
55) -> Result<SqliteStorageDiagnostics, StorageError> {
56    let mut aggregate = SqliteDiagnosticsAggregate::new();
57    aggregate.push("control", control_sqlite);
58    for (repository_id, shard_path) in store.catalog.active_repository_database_paths().await? {
59        let label = format!("shard {repository_id}");
60        let diagnostics =
61            tokio::task::spawn_blocking(move || shard_sqlite_diagnostics(&shard_path)).await?;
62        match diagnostics {
63            Ok(diagnostics) => aggregate.push(format!("shard {repository_id}"), diagnostics),
64            Err(error) => aggregate.push_error(label, error),
65        }
66    }
67    Ok(aggregate.finish())
68}
69
70fn shard_sqlite_diagnostics(shard_path: &Path) -> Result<SqliteStorageDiagnostics, StorageError> {
71    match std::fs::metadata(shard_path) {
72        Ok(_) => read_only_database_diagnostics(shard_path),
73        Err(error) if error.kind() == std::io::ErrorKind::NotFound => {
74            Err(StorageError::InvalidInput(format!(
75                "repository shard '{}' is missing",
76                shard_path.display()
77            )))
78        }
79        Err(error) => Err(error.into()),
80    }
81}
82
83struct SqliteDiagnosticsAggregate {
84    journal_modes: BTreeSet<String>,
85    wal_size_bytes: Option<u64>,
86    last_maintenance_at_ms: Option<u64>,
87    maintenance_errors: Vec<String>,
88}
89
90impl SqliteDiagnosticsAggregate {
91    fn new() -> Self {
92        Self {
93            journal_modes: BTreeSet::new(),
94            wal_size_bytes: Some(0),
95            last_maintenance_at_ms: None,
96            maintenance_errors: Vec::new(),
97        }
98    }
99
100    fn push(&mut self, label: impl AsRef<str>, diagnostics: SqliteStorageDiagnostics) {
101        if !diagnostics.journal_mode.is_empty() {
102            self.journal_modes.insert(diagnostics.journal_mode);
103        }
104        self.wal_size_bytes = match (self.wal_size_bytes, diagnostics.wal_size_bytes) {
105            (Some(left), Some(right)) => Some(left.saturating_add(right)),
106            _ => None,
107        };
108        self.last_maintenance_at_ms = self
109            .last_maintenance_at_ms
110            .max(diagnostics.last_maintenance_at_ms);
111        if let Some(error) = diagnostics.last_maintenance_error {
112            self.maintenance_errors
113                .push(format!("{}: {error}", label.as_ref()));
114        }
115    }
116
117    fn push_error(&mut self, label: impl AsRef<str>, error: StorageError) {
118        self.wal_size_bytes = None;
119        self.maintenance_errors
120            .push(format!("{}: {error}", label.as_ref()));
121    }
122
123    fn finish(self) -> SqliteStorageDiagnostics {
124        SqliteStorageDiagnostics {
125            journal_mode: match self.journal_modes.len() {
126                0 => String::new(),
127                1 => self
128                    .journal_modes
129                    .into_iter()
130                    .next()
131                    .expect("one journal mode should exist"),
132                _ => "mixed".to_owned(),
133            },
134            wal_size_bytes: self.wal_size_bytes,
135            last_maintenance_at_ms: self.last_maintenance_at_ms,
136            last_maintenance_error: (!self.maintenance_errors.is_empty())
137                .then(|| self.maintenance_errors.join("; ")),
138        }
139    }
140}
141
142#[cfg(test)]
143#[path = "mod_tests.rs"]
144mod tests;
relay_knowledge/storage/partitioned/diagnostics/mod.rs

relay_knowledge/storage/partitioned/diagnostics/
mod.rs
