relay_knowledge/application/service/
storage_diagnostics.rs

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

relay_knowledge/application/service/
storage_diagnostics.rs
