relay_knowledge/application/service/health/
mod.rs

1//! Health snapshot assembly and bounded storage fallback policy.
2
3use std::{sync::Arc, time::Duration};
4
5use crate::{
6    api::{ApiError, ApiMetadata, HealthResponse, RequestContext, StorageTopologyDiagnostics},
7    domain::{CodeRepositoryTotals, GraphVersion},
8    storage::{
9        FileIndexDiagnostics, GraphInspection, HealthStorageSnapshot, IndexRefreshDiagnostics,
10        KnowledgeStore, StorageError,
11    },
12};
13
14use super::{
15    RelayKnowledgeService, current_time_millis, graph_with_repository_code_totals,
16    storage_api_error,
17};
18use crate::application::{
19    knowledge::index_refresh::{
20        IndexRefreshOutcome, filter_outcome_to_read_models, metadata_for_indexes,
21    },
22    runtime::runtime_status_with_model_profiles,
23};
24
25const HEALTH_STORAGE_BUDGET: Duration = Duration::from_millis(500);
26
27struct HealthStorageReport {
28    snapshot: HealthStorageSnapshot,
29    storage: StorageTopologyDiagnostics,
30    degraded_reason: Option<String>,
31}
32
33impl RelayKnowledgeService {
34    /// Returns liveness-safe service and data health diagnostics.
35    pub async fn health(&self, context: RequestContext) -> Result<HealthResponse, ApiError> {
36        let store = self.storage.get().await.map_err(storage_api_error)?;
37        match tokio::time::timeout(HEALTH_STORAGE_BUDGET, self.health_storage_report(&store)).await
38        {
39            Ok(Ok(report)) => {
40                let response = self.health_from_storage_report(context, report).await;
41                *self.health_cache.write().await = Some(response.clone());
42                Ok(response)
43            }
44            Ok(Err(StorageError::Busy(message))) => Ok(self
45                .degraded_cached_health(context, format!("storage_busy: {message}"))
46                .await),
47            Ok(Err(error)) => Err(storage_api_error(error)),
48            Err(_) => Ok(self
49                .degraded_cached_health(context, "storage_busy: health snapshot timed out")
50                .await),
51        }
52    }
53
54    /// Returns control-plane health without opening cold graph storage.
55    pub async fn read_only_health(
56        &self,
57        context: RequestContext,
58    ) -> Result<HealthResponse, ApiError> {
59        if self.storage.ready_store().is_some() {
60            return self.health(context).await;
61        }
62
63        match tokio::time::timeout(
64            HEALTH_STORAGE_BUDGET,
65            self.storage_free_health(context.clone()),
66        )
67        .await
68        {
69            Ok(response) => Ok(response),
70            Err(_) => Ok(self
71                .degraded_cached_health(context, "storage_busy: cold health snapshot timed out")
72                .await),
73        }
74    }
75
76    async fn storage_free_health(&self, context: RequestContext) -> HealthResponse {
77        let storage = self
78            .storage_topology_diagnostics_with_budget(HEALTH_STORAGE_BUDGET)
79            .await;
80        let degraded_reason = storage.degraded_reason.clone();
81
82        HealthResponse {
83            metadata: ApiMetadata::graph_only(&context, GraphVersion::ZERO),
84            healthy: degraded_reason.is_none(),
85            degraded_reason,
86            storage,
87            graph: GraphInspection::default(),
88            repository_code_totals: CodeRepositoryTotals::default(),
89            indexes: Vec::new(),
90            index_cursors: Vec::new(),
91            index_refresh: IndexRefreshDiagnostics::default(),
92            file_index: FileIndexDiagnostics::default(),
93            runtime: runtime_status_with_model_profiles(
94                &self.runtime,
95                self.model_provider_config()
96                    .profile_summary(&self.runtime.retrieval)
97                    .await,
98            ),
99        }
100    }
101
102    async fn health_storage_report(
103        &self,
104        store: &Arc<dyn KnowledgeStore>,
105    ) -> Result<HealthStorageReport, StorageError> {
106        let snapshot = match self.storage_health_snapshot(store).await {
107            Ok(snapshot) => snapshot,
108            Err(error) => {
109                let storage = self.storage_topology_diagnostics().await;
110                if storage.missing_shard_count == 0 {
111                    return Err(error);
112                }
113                return Ok(HealthStorageReport {
114                    snapshot: self
115                        .degraded_health_snapshot_without_repository_totals(store)
116                        .await?,
117                    degraded_reason: storage
118                        .degraded_reason
119                        .clone()
120                        .or_else(|| Some(error.to_string())),
121                    storage,
122                });
123            }
124        };
125        let storage = self.storage_topology_diagnostics().await;
126
127        Ok(HealthStorageReport {
128            snapshot,
129            storage,
130            degraded_reason: None,
131        })
132    }
133
134    async fn storage_health_snapshot(
135        &self,
136        store: &Arc<dyn KnowledgeStore>,
137    ) -> Result<HealthStorageSnapshot, StorageError> {
138        match store.health_snapshot(current_time_millis()).await {
139            Ok(snapshot) => Ok(snapshot),
140            Err(StorageError::InvalidInput(message))
141                if message == "health snapshot storage is unavailable" =>
142            {
143                self.legacy_health_snapshot(store).await
144            }
145            Err(error) => Err(error),
146        }
147    }
148
149    async fn health_from_storage_report(
150        &self,
151        context: RequestContext,
152        report: HealthStorageReport,
153    ) -> HealthResponse {
154        let HealthStorageReport {
155            snapshot,
156            storage,
157            degraded_reason,
158        } = report;
159        let HealthStorageSnapshot {
160            graph,
161            repository_code_totals,
162            indexes,
163            index_cursors,
164            index_refresh,
165            file_index,
166        } = snapshot;
167        let graph = graph_with_repository_code_totals(graph, &repository_code_totals);
168        let degraded_reason = degraded_reason.or_else(|| storage.degraded_reason.clone());
169        let outcome = filter_outcome_to_read_models(
170            IndexRefreshOutcome {
171                indexes,
172                cursors: index_cursors,
173                diagnostics: index_refresh,
174            },
175            &self.runtime.retrieval,
176        );
177        let healthy = degraded_reason.is_none()
178            && outcome
179                .indexes
180                .iter()
181                .all(|status| !status.is_stale_for(graph.graph_version));
182
183        HealthResponse {
184            metadata: metadata_for_indexes(&context, graph.graph_version, &outcome.indexes),
185            healthy,
186            degraded_reason,
187            storage,
188            graph,
189            repository_code_totals,
190            indexes: outcome.indexes,
191            index_cursors: outcome.cursors,
192            index_refresh: outcome.diagnostics,
193            file_index,
194            runtime: runtime_status_with_model_profiles(
195                &self.runtime,
196                self.model_provider_config()
197                    .profile_summary(&self.runtime.retrieval)
198                    .await,
199            ),
200        }
201    }
202
203    async fn legacy_health_snapshot(
204        &self,
205        store: &Arc<dyn KnowledgeStore>,
206    ) -> Result<HealthStorageSnapshot, StorageError> {
207        Ok(HealthStorageSnapshot {
208            graph: store.inspect_graph().await?,
209            repository_code_totals: store.code_repository_totals().await?,
210            indexes: store.index_statuses().await?,
211            index_cursors: store.index_cursors().await?,
212            index_refresh: store
213                .index_refresh_diagnostics(current_time_millis())
214                .await?,
215            file_index: legacy_file_index_diagnostics_or_default(store).await?,
216        })
217    }
218
219    async fn degraded_health_snapshot_without_repository_totals(
220        &self,
221        store: &Arc<dyn KnowledgeStore>,
222    ) -> Result<HealthStorageSnapshot, StorageError> {
223        Ok(HealthStorageSnapshot {
224            graph: store.inspect_graph().await?,
225            repository_code_totals: CodeRepositoryTotals::default(),
226            indexes: store.index_statuses().await?,
227            index_cursors: store.index_cursors().await?,
228            index_refresh: store
229                .index_refresh_diagnostics(current_time_millis())
230                .await?,
231            file_index: legacy_file_index_diagnostics_or_default(store).await?,
232        })
233    }
234
235    async fn degraded_cached_health(
236        &self,
237        context: RequestContext,
238        degraded_reason: impl Into<String>,
239    ) -> HealthResponse {
240        let degraded_reason = degraded_reason.into();
241        if let Some(cached) = self.health_cache.read().await.clone() {
242            let mut response = cached;
243            response.metadata.trace_id = context.trace_id;
244            response.metadata.request_id = context.request_id;
245            response.metadata.stale = true;
246            response.healthy = false;
247            response.degraded_reason = Some(degraded_reason);
248            return response;
249        }
250
251        HealthResponse {
252            metadata: ApiMetadata::indexed(&context, GraphVersion::ZERO, None, None, true),
253            healthy: false,
254            degraded_reason: Some(degraded_reason),
255            storage: self.storage_diagnostics_from_snapshot(Default::default(), None),
256            graph: GraphInspection::default(),
257            repository_code_totals: CodeRepositoryTotals::default(),
258            indexes: Vec::new(),
259            index_cursors: Vec::new(),
260            index_refresh: IndexRefreshDiagnostics::default(),
261            file_index: FileIndexDiagnostics::default(),
262            runtime: runtime_status_with_model_profiles(
263                &self.runtime,
264                self.model_provider_config()
265                    .profile_summary(&self.runtime.retrieval)
266                    .await,
267            ),
268        }
269    }
270}
271
272async fn legacy_file_index_diagnostics_or_default(
273    store: &Arc<dyn KnowledgeStore>,
274) -> Result<FileIndexDiagnostics, StorageError> {
275    match store.file_index_diagnostics().await {
276        Ok(diagnostics) => Ok(diagnostics),
277        Err(StorageError::InvalidInput(message))
278            if message == "file index storage is unavailable" =>
279        {
280            Ok(FileIndexDiagnostics::default())
281        }
282        Err(error) => Err(error),
283    }
284}
relay_knowledge/application/service/health/mod.rs

relay_knowledge/application/service/health/
mod.rs
