relay_knowledge/application/service/storage_diagnostics/
mod.rs

1//! Storage-topology diagnostics and API projection.
2
3use std::time::Duration;
4
5use crate::{
6    api::{
7        ApiError, ApiMetadata, RequestContext, StorageShardDiagnostics, StorageTopologyDiagnostics,
8        StorageTopologyResponse,
9    },
10    application::service::{RelayKnowledgeService, storage_api_error},
11    domain::GraphVersion,
12    storage::StorageTopologySnapshot,
13};
14
15const STORAGE_TOPOLOGY_BUDGET: Duration = Duration::from_millis(500);
16
17impl RelayKnowledgeService {
18    /// Returns storage topology diagnostics without exposing concrete storage handles.
19    pub async fn storage_topology_diagnostics(&self) -> StorageTopologyDiagnostics {
20        self.storage_topology_diagnostics_with_budget(STORAGE_TOPOLOGY_BUDGET)
21            .await
22    }
23
24    pub(super) async fn storage_topology_diagnostics_with_budget(
25        &self,
26        budget: Duration,
27    ) -> StorageTopologyDiagnostics {
28        match tokio::time::timeout(budget, self.storage.topology_snapshot()).await {
29            Ok(Ok(snapshot)) => self.storage_diagnostics_from_snapshot(snapshot, None),
30            Ok(Err(error)) => self.storage_diagnostics_from_snapshot(
31                StorageTopologySnapshot::default(),
32                Some(error.to_string()),
33            ),
34            Err(_) => self.storage_diagnostics_from_snapshot(
35                StorageTopologySnapshot::default(),
36                Some("storage_topology_busy: topology snapshot timed out".to_owned()),
37            ),
38        }
39    }
40
41    pub async fn storage_topology_status(
42        &self,
43        context: RequestContext,
44    ) -> Result<StorageTopologyResponse, ApiError> {
45        let graph_version = match self.storage.ready_store() {
46            Some(store) => store
47                .current_graph_version()
48                .await
49                .map_err(storage_api_error)?,
50            None => GraphVersion::ZERO,
51        };
52
53        Ok(StorageTopologyResponse {
54            metadata: ApiMetadata::graph_only(&context, graph_version),
55            storage: self.storage_topology_diagnostics().await,
56        })
57    }
58
59    pub(super) fn storage_diagnostics_from_snapshot(
60        &self,
61        snapshot: StorageTopologySnapshot,
62        snapshot_error: Option<String>,
63    ) -> StorageTopologyDiagnostics {
64        let topology = self.runtime.storage.topology.as_str().to_owned();
65        let control_database_path = self.runtime.paths.database_file().display().to_string();
66        let has_partitioned_catalog = !snapshot.shards.is_empty();
67        let configured_partitioned =
68            self.runtime.storage.topology == crate::storage::StorageTopology::PartitionedSqlite;
69        let repository_shards_dir =
70            (configured_partitioned || has_partitioned_catalog).then(|| {
71                self.runtime
72                    .paths
73                    .repository_shards_dir()
74                    .display()
75                    .to_string()
76            });
77        let mut runtime_state_paths = vec![control_database_path.clone()];
78        if let Some(path) = repository_shards_dir.clone() {
79            runtime_state_paths.push(path);
80        }
81
82        let shards = snapshot
83            .shards
84            .into_iter()
85            .map(|entry| StorageShardDiagnostics {
86                repository_id: entry.repository_id,
87                state: entry.state,
88                shard_locator: entry.shard_locator,
89                resolved_path: entry.resolved_path,
90                source_scope_count: entry.source_scope_count,
91                exists: entry.exists,
92                updated_at_ms: entry.updated_at_ms,
93                degraded_reason: (!entry.exists)
94                    .then(|| "repository shard file is missing".to_owned()),
95            })
96            .collect::<Vec<_>>();
97        let active_shard_count = shards
98            .iter()
99            .filter(|shard| shard.state == "active")
100            .count();
101        let staged_shard_count = shards
102            .iter()
103            .filter(|shard| shard.state == "staged")
104            .count();
105        let missing_shard_count = shards.iter().filter(|shard| !shard.exists).count();
106        let degraded_reason = snapshot_error.or_else(|| {
107            if !configured_partitioned && active_shard_count > 0 {
108                Some(
109                    "single_sqlite configuration found active partitioned_sqlite shard catalog"
110                        .to_owned(),
111                )
112            } else {
113                (missing_shard_count > 0).then(|| {
114                    "partitioned_sqlite shard catalog references missing shard files".to_owned()
115                })
116            }
117        });
118
119        StorageTopologyDiagnostics {
120            topology,
121            control_database_path,
122            repository_shards_dir,
123            shard_catalog_active: active_shard_count > 0,
124            active_shard_count,
125            staged_shard_count,
126            missing_shard_count,
127            runtime_state_paths,
128            shards,
129            degraded_reason,
130        }
131    }
132}
relay_knowledge/application/service/storage_diagnostics/mod.rs

relay_knowledge/application/service/storage_diagnostics/
mod.rs
