relay_knowledge/application/service/
health.rs

1use std::{sync::Arc, time::Duration};
2
3use crate::{
4    api::{ApiError, ApiMetadata, HealthResponse, RequestContext},
5    domain::{CodeRepositoryTotals, GraphVersion},
6    storage::{
7        FileIndexDiagnostics, GraphInspection, HealthStorageSnapshot, IndexRefreshDiagnostics,
8        KnowledgeStore, StorageError,
9    },
10};
11
12use super::{
13    RelayKnowledgeService, current_time_millis, graph_with_repository_code_totals,
14    storage_api_error,
15};
16use crate::application::{
17    knowledge::index_refresh::{
18        IndexRefreshOutcome, filter_outcome_to_read_models, metadata_for_indexes,
19    },
20    status::runtime_status_with_model_profiles,
21};
22
23const HEALTH_STORAGE_BUDGET: Duration = Duration::from_millis(500);
24
25impl RelayKnowledgeService {
26    /// Returns liveness-safe service and data health diagnostics.
27    pub async fn health(&self, context: RequestContext) -> Result<HealthResponse, ApiError> {
28        let store = self.storage.get().await.map_err(storage_api_error)?;
29        match tokio::time::timeout(HEALTH_STORAGE_BUDGET, self.storage_health_snapshot(&store))
30            .await
31        {
32            Ok(Ok(snapshot)) => {
33                let response = self
34                    .health_from_storage_snapshot(context, snapshot, None)
35                    .await;
36                *self.health_cache.write().await = Some(response.clone());
37                Ok(response)
38            }
39            Ok(Err(StorageError::Busy(message))) => Ok(self
40                .degraded_cached_health(context, format!("storage_busy: {message}"))
41                .await),
42            Ok(Err(error)) => Err(storage_api_error(error)),
43            Err(_) => Ok(self
44                .degraded_cached_health(context, "storage_busy: health snapshot timed out")
45                .await),
46        }
47    }
48
49    async fn storage_health_snapshot(
50        &self,
51        store: &Arc<dyn KnowledgeStore>,
52    ) -> Result<HealthStorageSnapshot, StorageError> {
53        match store.health_snapshot(current_time_millis()).await {
54            Ok(snapshot) => Ok(snapshot),
55            Err(StorageError::InvalidInput(message))
56                if message == "health snapshot storage is unavailable" =>
57            {
58                self.legacy_health_snapshot(store).await
59            }
60            Err(error) => Err(error),
61        }
62    }
63
64    async fn health_from_storage_snapshot(
65        &self,
66        context: RequestContext,
67        snapshot: HealthStorageSnapshot,
68        degraded_reason: Option<String>,
69    ) -> HealthResponse {
70        let HealthStorageSnapshot {
71            graph,
72            repository_code_totals,
73            indexes,
74            index_cursors,
75            index_refresh,
76            file_index,
77        } = snapshot;
78        let graph = graph_with_repository_code_totals(graph, &repository_code_totals);
79        let outcome = filter_outcome_to_read_models(
80            IndexRefreshOutcome {
81                indexes,
82                cursors: index_cursors,
83                diagnostics: index_refresh,
84            },
85            &self.runtime.retrieval,
86        );
87        let healthy = degraded_reason.is_none()
88            && outcome
89                .indexes
90                .iter()
91                .all(|status| !status.is_stale_for(graph.graph_version));
92
93        HealthResponse {
94            metadata: metadata_for_indexes(&context, graph.graph_version, &outcome.indexes),
95            healthy,
96            degraded_reason,
97            graph,
98            repository_code_totals,
99            indexes: outcome.indexes,
100            index_cursors: outcome.cursors,
101            index_refresh: outcome.diagnostics,
102            file_index,
103            runtime: runtime_status_with_model_profiles(
104                &self.runtime,
105                self.model_provider_config()
106                    .profile_summary(&self.runtime.retrieval)
107                    .await,
108            ),
109        }
110    }
111
112    async fn legacy_health_snapshot(
113        &self,
114        store: &Arc<dyn KnowledgeStore>,
115    ) -> Result<HealthStorageSnapshot, StorageError> {
116        Ok(HealthStorageSnapshot {
117            graph: store.inspect_graph().await?,
118            repository_code_totals: store.code_repository_totals().await?,
119            indexes: store.index_statuses().await?,
120            index_cursors: store.index_cursors().await?,
121            index_refresh: store
122                .index_refresh_diagnostics(current_time_millis())
123                .await?,
124            file_index: legacy_file_index_diagnostics_or_default(store).await?,
125        })
126    }
127
128    async fn degraded_cached_health(
129        &self,
130        context: RequestContext,
131        degraded_reason: impl Into<String>,
132    ) -> HealthResponse {
133        let degraded_reason = degraded_reason.into();
134        if let Some(cached) = self.health_cache.read().await.clone() {
135            let mut response = cached;
136            response.metadata.trace_id = context.trace_id;
137            response.metadata.request_id = context.request_id;
138            response.metadata.stale = true;
139            response.healthy = false;
140            response.degraded_reason = Some(degraded_reason);
141            return response;
142        }
143
144        HealthResponse {
145            metadata: ApiMetadata::indexed(&context, GraphVersion::ZERO, None, None, true),
146            healthy: false,
147            degraded_reason: Some(degraded_reason),
148            graph: GraphInspection::default(),
149            repository_code_totals: CodeRepositoryTotals::default(),
150            indexes: Vec::new(),
151            index_cursors: Vec::new(),
152            index_refresh: IndexRefreshDiagnostics::default(),
153            file_index: FileIndexDiagnostics::default(),
154            runtime: runtime_status_with_model_profiles(
155                &self.runtime,
156                self.model_provider_config()
157                    .profile_summary(&self.runtime.retrieval)
158                    .await,
159            ),
160        }
161    }
162}
163
164async fn legacy_file_index_diagnostics_or_default(
165    store: &Arc<dyn KnowledgeStore>,
166) -> Result<FileIndexDiagnostics, StorageError> {
167    match store.file_index_diagnostics().await {
168        Ok(diagnostics) => Ok(diagnostics),
169        Err(StorageError::InvalidInput(message))
170            if message == "file index storage is unavailable" =>
171        {
172            Ok(FileIndexDiagnostics::default())
173        }
174        Err(error) => Err(error),
175    }
176}
relay_knowledge/application/service/health.rs

relay_knowledge/application/service/
health.rs
