relay_knowledge/storage/
mod.rs

1//! Storage contracts and SQLite-backed graph state.
2//!
3//! Storage owns persisted graph facts, mutation log entries, derived index
4//! metadata, and health snapshots. Domain and interface modules must not depend
5//! on SQL or concrete database types.
6
7mod canvas;
8mod code;
9mod file_index;
10mod sqlite;
11
12use std::{error::Error, fmt, future::Future, pin::Pin};
13
14use serde::{Deserialize, Serialize};
15
16use crate::domain::{
17    AuditEventRecord, AuditStatus, CodeChunkRecord, CodeGraphBatch, CodeGraphCommitReceipt,
18    CodeParseStatusCounts, CodeReferenceRecord, CodeSymbolRecord, CommitReceipt,
19    GraphMutationBatch, GraphVersion, IndexKind, IndexModality, IndexStatus,
20    ProposalConflictRecord, ProposalConflictSeverity, ProposalKind, ProposalProvenance,
21    ProposalRecord, ProposalState, RetrievalHit, RetrieverSource, ServiceOperatorState,
22    ServiceOperatorStatus, WorkerKind, WorkerStatus, WorkerTaskRecord,
23};
24
25pub use canvas::{
26    GraphCanvasSelection, GraphCanvasStorageEdge, GraphCanvasStorageNode,
27    GraphCanvasStorageRequest, GraphCanvasStorageSnapshot,
28};
29pub use code::{
30    CODE_INDEX_TASK_LEASE_RECOVERY_UNAVAILABLE, CODE_INDEX_TASK_LEASE_RENEWAL_UNAVAILABLE,
31    CodeImpactChanges, CodeIndexTaskClaimRequest, CodeIndexTaskCompletion, CodeIndexTaskFailure,
32    CodeIndexTaskLeaseRenewal, CodeIndexTaskSeed, CodeRepositorySetMemberSeed,
33    CodeRepositorySetRefreshTaskClaimRequest, CodeRepositorySetRefreshTaskCompletion,
34    CodeRepositorySetRefreshTaskFailure, CodeRepositorySetRefreshTaskSeed, CodeRepositorySetSeed,
35    CodeRepositoryStore, CodeScopeRetentionRequest,
36};
37pub use file_index::{
38    FileIndexDiagnostics, FileIndexEntry, FileIndexRoot, FileIndexRootStatus, FileIndexRootUpdate,
39    FileIndexScanSummary, FileSearchHit, FileSearchRequest,
40};
41pub use sqlite::SqliteGraphStore;
42
43pub type StorageFuture<'a, T> = Pin<Box<dyn Future<Output = Result<T, StorageError>> + Send + 'a>>;
44
45/// Synthetic scope used for graph-wide index work that is not tied to evidence.
46pub const DEFAULT_INDEX_SOURCE_SCOPE: &str = "graph";
47
48/// Graph fact persistence and query contract.
49pub trait GraphStore: Send + Sync {
50    fn commit_mutation_batch(&self, batch: GraphMutationBatch) -> StorageFuture<'_, CommitReceipt>;
51
52    fn inspect_graph(&self) -> StorageFuture<'_, GraphInspection>;
53
54    fn health_snapshot(&self, _now_ms: u64) -> StorageFuture<'_, HealthStorageSnapshot> {
55        Box::pin(async {
56            Err(StorageError::InvalidInput(
57                "health snapshot storage is unavailable".to_owned(),
58            ))
59        })
60    }
61
62    fn graph_canvas(
63        &self,
64        _request: GraphCanvasStorageRequest,
65    ) -> StorageFuture<'_, GraphCanvasStorageSnapshot> {
66        Box::pin(async {
67            Err(StorageError::InvalidInput(
68                "graph canvas storage is unavailable".to_owned(),
69            ))
70        })
71    }
72
73    fn search(&self, request: GraphSearchRequest) -> StorageFuture<'_, Vec<RetrievalHit>>;
74
75    fn current_graph_version(&self) -> StorageFuture<'_, GraphVersion>;
76}
77
78/// Mutation log contract consumed by reconcilers and indexers.
79pub trait MutationLogStore: Send + Sync {
80    fn read_after(
81        &self,
82        graph_version: GraphVersion,
83        limit: usize,
84    ) -> StorageFuture<'_, Vec<MutationLogEntry>>;
85}
86
87/// Derived index metadata contract.
88pub trait IndexStore: Send + Sync {
89    fn index_statuses(&self) -> StorageFuture<'_, Vec<IndexStatus>>;
90
91    fn mark_refresh_complete(
92        &self,
93        kind: IndexKind,
94        graph_version: GraphVersion,
95    ) -> StorageFuture<'_, IndexStatus>;
96
97    fn index_cursors(&self) -> StorageFuture<'_, Vec<IndexCursor>> {
98        Box::pin(async {
99            Err(StorageError::InvalidInput(
100                "index cursor storage is unavailable".to_owned(),
101            ))
102        })
103    }
104
105    fn queue_index_refreshes(
106        &self,
107        _request: IndexRefreshQueueRequest,
108    ) -> StorageFuture<'_, IndexRefreshDiagnostics> {
109        Box::pin(async {
110            Err(StorageError::InvalidInput(
111                "index refresh task storage is unavailable".to_owned(),
112            ))
113        })
114    }
115
116    fn claim_index_refresh_task(
117        &self,
118        _request: IndexRefreshClaimRequest,
119    ) -> StorageFuture<'_, Option<IndexRefreshTask>> {
120        Box::pin(async {
121            Err(StorageError::InvalidInput(
122                "index refresh task storage is unavailable".to_owned(),
123            ))
124        })
125    }
126
127    fn complete_index_refresh_task(
128        &self,
129        _request: IndexRefreshCompletion,
130    ) -> StorageFuture<'_, IndexRefreshTask> {
131        Box::pin(async {
132            Err(StorageError::InvalidInput(
133                "index refresh task storage is unavailable".to_owned(),
134            ))
135        })
136    }
137
138    fn fail_index_refresh_task(
139        &self,
140        _request: IndexRefreshFailure,
141    ) -> StorageFuture<'_, IndexRefreshTask> {
142        Box::pin(async {
143            Err(StorageError::InvalidInput(
144                "index refresh task storage is unavailable".to_owned(),
145            ))
146        })
147    }
148
149    fn index_refresh_diagnostics(
150        &self,
151        _now_ms: u64,
152    ) -> StorageFuture<'_, IndexRefreshDiagnostics> {
153        Box::pin(async {
154            Err(StorageError::InvalidInput(
155                "index refresh diagnostics are unavailable".to_owned(),
156            ))
157        })
158    }
159
160    fn queue_worker_tasks(
161        &self,
162        _tasks: Vec<WorkerTaskSeed>,
163    ) -> StorageFuture<'_, Vec<WorkerTaskRecord>> {
164        Box::pin(async { Ok(Vec::new()) })
165    }
166
167    fn worker_statuses(&self) -> StorageFuture<'_, Vec<WorkerStatus>> {
168        Box::pin(async { Ok(Vec::new()) })
169    }
170
171    fn claim_worker_task(
172        &self,
173        _request: WorkerTaskClaimRequest,
174    ) -> StorageFuture<'_, Option<WorkerTaskRecord>> {
175        Box::pin(async { Ok(None) })
176    }
177
178    fn complete_worker_task(
179        &self,
180        _request: WorkerTaskCompletion,
181    ) -> StorageFuture<'_, WorkerTaskRecord> {
182        Box::pin(async {
183            Err(StorageError::InvalidInput(
184                "worker task storage is unavailable".to_owned(),
185            ))
186        })
187    }
188
189    fn fail_worker_task(&self, _request: WorkerTaskFailure) -> StorageFuture<'_, WorkerTaskRecord> {
190        Box::pin(async {
191            Err(StorageError::InvalidInput(
192                "worker task storage is unavailable".to_owned(),
193            ))
194        })
195    }
196
197    fn insert_proposal(&self, _proposal: NewProposal) -> StorageFuture<'_, ProposalRecord> {
198        Box::pin(async {
199            Err(StorageError::InvalidInput(
200                "proposal storage is unavailable".to_owned(),
201            ))
202        })
203    }
204
205    fn list_proposals(
206        &self,
207        _request: ProposalListRequest,
208    ) -> StorageFuture<'_, Vec<ProposalRecord>> {
209        Box::pin(async { Ok(Vec::new()) })
210    }
211
212    fn proposal_by_id(&self, _proposal_id: String) -> StorageFuture<'_, Option<ProposalRecord>> {
213        Box::pin(async { Ok(None) })
214    }
215
216    fn proposal_conflicts(
217        &self,
218        _proposal_id: String,
219    ) -> StorageFuture<'_, Vec<ProposalConflictRecord>> {
220        Box::pin(async { Ok(Vec::new()) })
221    }
222
223    fn decide_proposal(&self, _request: ProposalDecision) -> StorageFuture<'_, ProposalRecord> {
224        Box::pin(async {
225            Err(StorageError::InvalidInput(
226                "proposal storage is unavailable".to_owned(),
227            ))
228        })
229    }
230
231    fn insert_audit_event(&self, _event: NewAuditEvent) -> StorageFuture<'_, AuditEventRecord> {
232        Box::pin(async {
233            Err(StorageError::InvalidInput(
234                "audit storage is unavailable".to_owned(),
235            ))
236        })
237    }
238
239    fn query_audit_events(
240        &self,
241        _request: AuditQueryRequest,
242    ) -> StorageFuture<'_, Vec<AuditEventRecord>> {
243        Box::pin(async { Ok(Vec::new()) })
244    }
245
246    fn audit_event_count(&self) -> StorageFuture<'_, usize> {
247        Box::pin(async { Ok(0) })
248    }
249
250    fn service_operator_status(&self) -> StorageFuture<'_, ServiceOperatorStatus> {
251        Box::pin(async {
252            Ok(ServiceOperatorStatus {
253                state: ServiceOperatorState::Disabled,
254                silent_updates_enabled: false,
255                allowed_scopes: Vec::new(),
256                last_run_at_ms: None,
257                next_retry_at_ms: None,
258                last_error: None,
259                updated_at_ms: 0,
260            })
261        })
262    }
263
264    fn update_service_operator(
265        &self,
266        _request: ServiceOperatorUpdate,
267    ) -> StorageFuture<'_, ServiceOperatorStatus> {
268        Box::pin(async {
269            Err(StorageError::InvalidInput(
270                "service operator storage is unavailable".to_owned(),
271            ))
272        })
273    }
274
275    fn replace_file_index_root(
276        &self,
277        _update: FileIndexRootUpdate,
278    ) -> StorageFuture<'_, FileIndexRootStatus> {
279        unavailable_file_index_storage()
280    }
281
282    fn mark_file_index_roots_unconfigured(
283        &self,
284        _active_roots: Vec<FileIndexRoot>,
285        _now_ms: u64,
286    ) -> StorageFuture<'_, FileIndexDiagnostics> {
287        unavailable_file_index_storage()
288    }
289
290    fn search_files(&self, _request: FileSearchRequest) -> StorageFuture<'_, Vec<FileSearchHit>> {
291        unavailable_file_index_storage()
292    }
293
294    fn file_index_diagnostics(&self) -> StorageFuture<'_, FileIndexDiagnostics> {
295        unavailable_file_index_storage()
296    }
297}
298
299fn unavailable_file_index_storage<T>() -> StorageFuture<'static, T> {
300    Box::pin(async {
301        Err(StorageError::InvalidInput(
302            "file index storage is unavailable".to_owned(),
303        ))
304    })
305}
306
307/// Code graph fact persistence and query contract for tree-sitter output.
308pub trait CodeGraphStore: Send + Sync {
309    fn commit_code_graph_batch(
310        &self,
311        batch: CodeGraphBatch,
312    ) -> StorageFuture<'_, CodeGraphCommitReceipt>;
313
314    fn search_code_symbols(
315        &self,
316        request: CodeSymbolSearchRequest,
317    ) -> StorageFuture<'_, Vec<CodeSymbolRecord>>;
318
319    fn search_code_references(
320        &self,
321        request: CodeReferenceSearchRequest,
322    ) -> StorageFuture<'_, Vec<CodeReferenceRecord>>;
323
324    fn search_code_chunks(
325        &self,
326        request: CodeChunkSearchRequest,
327    ) -> StorageFuture<'_, Vec<CodeChunkRecord>>;
328}
329
330/// Combined storage facade used by the application service.
331pub trait KnowledgeStore:
332    GraphStore + MutationLogStore + IndexStore + CodeGraphStore + CodeRepositoryStore
333{
334}
335
336impl<T> KnowledgeStore for T where
337    T: GraphStore + MutationLogStore + IndexStore + CodeGraphStore + CodeRepositoryStore
338{
339}
340
341/// Bounded graph search request against an explicit graph snapshot.
342#[derive(Debug, Clone, PartialEq, Eq)]
343pub struct GraphSearchRequest {
344    pub query: String,
345    pub source_scope: Option<String>,
346    pub graph_version: GraphVersion,
347    pub limit: usize,
348    pub disabled_retriever_sources: Vec<RetrieverSource>,
349}
350
351impl GraphSearchRequest {
352    /// Returns whether storage may execute a retriever family for this request.
353    pub fn allows_retriever_source(&self, source: RetrieverSource) -> bool {
354        !self.disabled_retriever_sources.contains(&source)
355    }
356}
357
358/// Bounded code symbol search against an explicit graph snapshot.
359#[derive(Debug, Clone, PartialEq, Eq)]
360pub struct CodeSymbolSearchRequest {
361    pub source_scope: Option<String>,
362    pub path: Option<String>,
363    pub name: Option<String>,
364    pub graph_version: GraphVersion,
365    pub limit: usize,
366}
367
368/// Bounded code reference search against an explicit graph snapshot.
369#[derive(Debug, Clone, PartialEq, Eq)]
370pub struct CodeReferenceSearchRequest {
371    pub source_scope: Option<String>,
372    pub path: Option<String>,
373    pub symbol_text: Option<String>,
374    pub target_symbol_id: Option<String>,
375    pub graph_version: GraphVersion,
376    pub limit: usize,
377}
378
379/// Bounded code chunk search against an explicit graph snapshot.
380#[derive(Debug, Clone, PartialEq, Eq)]
381pub struct CodeChunkSearchRequest {
382    pub source_scope: Option<String>,
383    pub path: Option<String>,
384    pub query: Option<String>,
385    pub graph_version: GraphVersion,
386    pub limit: usize,
387}
388
389/// Worker task input inserted after graph changes or service reconciliation.
390#[derive(Debug, Clone, PartialEq, Eq)]
391pub struct WorkerTaskSeed {
392    pub kind: WorkerKind,
393    pub source_scope: String,
394    pub evidence_id: Option<String>,
395    pub target_graph_version: GraphVersion,
396    pub input_fingerprint: String,
397    pub payload_json: String,
398    pub now_ms: u64,
399}
400
401/// Worker lease acquisition request.
402#[derive(Debug, Clone, PartialEq, Eq)]
403pub struct WorkerTaskClaimRequest {
404    pub kind: Option<WorkerKind>,
405    pub lease_owner: String,
406    pub lease_duration_ms: u64,
407    pub max_attempts: u32,
408    pub now_ms: u64,
409}
410
411/// Worker completion guarded by the active lease.
412#[derive(Debug, Clone, PartialEq, Eq)]
413pub struct WorkerTaskCompletion {
414    pub task_id: String,
415    pub lease_owner: String,
416    pub attempt_count: u32,
417    pub now_ms: u64,
418}
419
420/// Worker failure report for retry and dead-letter handling.
421#[derive(Debug, Clone, PartialEq, Eq)]
422pub struct WorkerTaskFailure {
423    pub task_id: String,
424    pub lease_owner: String,
425    pub attempt_count: u32,
426    pub error_kind: String,
427    pub error_message: String,
428    pub retry_backoff_ms: u64,
429    pub max_attempts: u32,
430    pub now_ms: u64,
431}
432
433/// New proposal to persist before manual approval.
434#[derive(Debug, Clone, PartialEq, Eq)]
435pub struct NewProposal {
436    pub proposal_id: String,
437    pub source_scope: String,
438    pub kind: ProposalKind,
439    pub title: String,
440    pub summary: String,
441    pub payload_json: String,
442    pub origin: String,
443    pub provenance: ProposalProvenance,
444    pub confidence_basis_points: u16,
445    pub conflicts: Vec<NewProposalConflict>,
446    pub now_ms: u64,
447}
448
449/// New proposal conflict to persist with a proposal.
450#[derive(Debug, Clone, PartialEq, Eq)]
451pub struct NewProposalConflict {
452    pub conflict_id: String,
453    pub existing_fact_kind: String,
454    pub existing_fact_id: String,
455    pub severity: ProposalConflictSeverity,
456    pub reason: String,
457}
458
459/// Proposal list filter.
460#[derive(Debug, Clone, PartialEq, Eq)]
461pub struct ProposalListRequest {
462    pub state: Option<ProposalState>,
463    pub limit: usize,
464}
465
466/// Proposal decision request.
467#[derive(Debug, Clone, PartialEq, Eq)]
468pub struct ProposalDecision {
469    pub proposal_id: String,
470    pub next_state: ProposalState,
471    pub actor: String,
472    pub reason: Option<String>,
473    pub now_ms: u64,
474}
475
476/// New durable audit event.
477#[derive(Debug, Clone, PartialEq, Eq)]
478pub struct NewAuditEvent {
479    pub operation: String,
480    pub interface: String,
481    pub request_id: String,
482    pub trace_id: String,
483    pub status: AuditStatus,
484    pub actor: Option<String>,
485    pub source_scope: Option<String>,
486    pub graph_version: u64,
487    pub detail_json: String,
488    pub message: Option<String>,
489    pub now_ms: u64,
490}
491
492/// Audit query filter.
493#[derive(Debug, Clone, PartialEq, Eq)]
494pub struct AuditQueryRequest {
495    pub operation: Option<String>,
496    pub limit: usize,
497}
498
499/// Service operator state update.
500#[derive(Debug, Clone, PartialEq, Eq)]
501pub struct ServiceOperatorUpdate {
502    pub state: ServiceOperatorState,
503    pub silent_updates_enabled: bool,
504    pub allowed_scopes: Vec<String>,
505    pub last_error: Option<String>,
506    pub now_ms: u64,
507}
508
509/// Aggregated graph status for diagnostics.
510#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
511pub struct GraphInspection {
512    pub graph_version: GraphVersion,
513    pub entity_count: usize,
514    pub evidence_count: usize,
515    pub relation_count: usize,
516    pub claim_count: usize,
517    pub event_count: usize,
518    pub mutation_count: usize,
519    pub code_file_count: usize,
520    pub code_symbol_count: usize,
521    pub code_reference_count: usize,
522    pub code_chunk_count: usize,
523    pub code_parse_status_counts: CodeParseStatusCounts,
524}
525
526impl Default for GraphInspection {
527    fn default() -> Self {
528        Self {
529            graph_version: GraphVersion::ZERO,
530            entity_count: 0,
531            evidence_count: 0,
532            relation_count: 0,
533            claim_count: 0,
534            event_count: 0,
535            mutation_count: 0,
536            code_file_count: 0,
537            code_symbol_count: 0,
538            code_reference_count: 0,
539            code_chunk_count: 0,
540            code_parse_status_counts: CodeParseStatusCounts::default(),
541        }
542    }
543}
544
545/// Read-only storage view used by service health without mutating indexes.
546#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
547pub struct HealthStorageSnapshot {
548    pub graph: GraphInspection,
549    pub repository_code_totals: crate::domain::CodeRepositoryTotals,
550    pub indexes: Vec<IndexStatus>,
551    pub index_cursors: Vec<IndexCursor>,
552    pub index_refresh: IndexRefreshDiagnostics,
553    pub file_index: FileIndexDiagnostics,
554}
555
556/// Mutation log entry returned for replay and index refresh planning.
557#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
558pub struct MutationLogEntry {
559    pub graph_version: GraphVersion,
560    pub evidence_count: usize,
561    pub entity_count: usize,
562    pub relation_count: usize,
563    pub claim_count: usize,
564    pub event_count: usize,
565    pub affected_scopes: Vec<String>,
566    pub affected_entity_ids: Vec<String>,
567    pub evidence_ids: Vec<String>,
568    pub source_hashes: Vec<String>,
569}
570
571/// Scoped cursor for a derived index read model.
572#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
573pub struct IndexCursor {
574    pub kind: IndexKind,
575    pub source_scope: String,
576    pub modality: IndexModality,
577    pub index_version: u64,
578    pub indexed_graph_version: GraphVersion,
579    pub state: crate::domain::IndexState,
580    pub last_error: Option<String>,
581    #[serde(skip_serializing_if = "Option::is_none")]
582    pub source_hash: Option<String>,
583    #[serde(skip_serializing_if = "Option::is_none")]
584    pub backend_cursor: Option<String>,
585    #[serde(skip_serializing_if = "Option::is_none")]
586    pub model_name: Option<String>,
587    #[serde(skip_serializing_if = "Option::is_none")]
588    pub model_dimension: Option<u32>,
589}
590
591/// Persistent index refresh task lifecycle state.
592#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
593#[serde(rename_all = "snake_case")]
594pub enum IndexRefreshTaskState {
595    Queued,
596    Running,
597    Succeeded,
598    Retrying,
599    Failed,
600    DeadLetter,
601}
602
603impl IndexRefreshTaskState {
604    /// Stable storage and API representation.
605    pub const fn as_str(self) -> &'static str {
606        match self {
607            Self::Queued => "queued",
608            Self::Running => "running",
609            Self::Succeeded => "succeeded",
610            Self::Retrying => "retrying",
611            Self::Failed => "failed",
612            Self::DeadLetter => "dead_letter",
613        }
614    }
615}
616
617/// Persistent task used by foreground refresh and startup recovery.
618#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
619pub struct IndexRefreshTask {
620    pub task_id: String,
621    pub kind: IndexKind,
622    pub source_scope: String,
623    pub modality: IndexModality,
624    pub target_graph_version: GraphVersion,
625    pub state: IndexRefreshTaskState,
626    pub lease_owner: Option<String>,
627    pub lease_expires_at_ms: Option<u64>,
628    pub attempt_count: u32,
629    pub next_retry_at_ms: u64,
630    pub input_fingerprint: String,
631    pub cursor_before: GraphVersion,
632    pub cursor_after: Option<GraphVersion>,
633    pub last_error_kind: Option<String>,
634    pub last_error_message: Option<String>,
635    pub created_at_ms: u64,
636    pub updated_at_ms: u64,
637}
638
639/// Queue request created by refresh APIs or the reconciler.
640#[derive(Debug, Clone, PartialEq, Eq)]
641pub struct IndexRefreshQueueRequest {
642    pub kinds: Vec<IndexKind>,
643    pub target_graph_version: GraphVersion,
644    pub max_queue_depth: usize,
645    pub reset_dead_letter_tasks: bool,
646    pub now_ms: u64,
647}
648
649/// Lease acquisition request for bounded foreground workers.
650#[derive(Debug, Clone, PartialEq, Eq)]
651pub struct IndexRefreshClaimRequest {
652    pub lease_owner: String,
653    pub lease_duration_ms: u64,
654    pub max_attempts: u32,
655    pub now_ms: u64,
656}
657
658/// Completion report guarded by the active task lease and attempt token.
659#[derive(Debug, Clone, PartialEq, Eq)]
660pub struct IndexRefreshCompletion {
661    pub task_id: String,
662    pub lease_owner: String,
663    pub attempt_count: u32,
664    pub indexed_graph_version: GraphVersion,
665    pub model_name: Option<String>,
666    pub model_dimension: Option<u32>,
667    pub now_ms: u64,
668}
669
670/// Failure report for retry backoff and dead-letter isolation.
671#[derive(Debug, Clone, PartialEq, Eq)]
672pub struct IndexRefreshFailure {
673    pub task_id: String,
674    pub lease_owner: String,
675    pub attempt_count: u32,
676    pub error_kind: String,
677    pub error_message: String,
678    pub retry_backoff_ms: u64,
679    pub max_attempts: u32,
680    pub now_ms: u64,
681}
682
683/// Per-kind lag included in diagnostics snapshots.
684#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
685pub struct IndexLag {
686    pub kind: IndexKind,
687    pub lag_versions: u64,
688}
689
690/// Structured reason explaining why an index family or scoped cursor is stale.
691#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
692pub struct IndexStalenessReason {
693    pub kind: IndexKind,
694    #[serde(skip_serializing_if = "Option::is_none")]
695    pub source_scope: Option<String>,
696    #[serde(skip_serializing_if = "Option::is_none")]
697    pub modality: Option<IndexModality>,
698    pub reason: String,
699    pub lag_versions: u64,
700    #[serde(skip_serializing_if = "Option::is_none")]
701    pub last_error: Option<String>,
702}
703
704/// Snapshot for queue, dead-letter, and stale-index diagnostics.
705#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)]
706pub struct IndexRefreshDiagnostics {
707    pub queue_depth: usize,
708    pub running_count: usize,
709    pub retrying_count: usize,
710    pub dead_letter_count: usize,
711    pub oldest_unfinished_age_ms: Option<u64>,
712    pub index_lag_by_kind: Vec<IndexLag>,
713    pub max_index_lag_versions: u64,
714    pub stale_index_count: usize,
715    pub stale_reasons: Vec<IndexStalenessReason>,
716}
717
718/// Storage health surfaced to diagnostics.
719#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
720pub struct StorageHealth {
721    pub graph_version: GraphVersion,
722    pub healthy: bool,
723    pub detail: String,
724}
725
726/// Storage boundary failure.
727#[derive(Debug)]
728pub enum StorageError {
729    Io(std::io::Error),
730    Sqlite(rusqlite::Error),
731    Join(tokio::task::JoinError),
732    LockPoisoned,
733    Busy(String),
734    InvalidInput(String),
735}
736
737impl fmt::Display for StorageError {
738    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
739        match self {
740            Self::Io(error) => write!(formatter, "storage I/O failed: {error}"),
741            Self::Sqlite(error) => write!(formatter, "sqlite operation failed: {error}"),
742            Self::Join(error) => write!(formatter, "storage worker failed: {error}"),
743            Self::LockPoisoned => write!(formatter, "sqlite connection lock was poisoned"),
744            Self::Busy(message) => write!(formatter, "storage busy: {message}"),
745            Self::InvalidInput(message) => write!(formatter, "invalid storage input: {message}"),
746        }
747    }
748}
749
750impl Error for StorageError {}
751
752impl From<std::io::Error> for StorageError {
753    fn from(error: std::io::Error) -> Self {
754        Self::Io(error)
755    }
756}
757
758impl From<rusqlite::Error> for StorageError {
759    fn from(error: rusqlite::Error) -> Self {
760        Self::Sqlite(error)
761    }
762}
763
764impl From<tokio::task::JoinError> for StorageError {
765    fn from(error: tokio::task::JoinError) -> Self {
766        Self::Join(error)
767    }
768}
769
770#[cfg(test)]
771#[path = "mod_tests.rs"]
772mod tests;
relay_knowledge/storage/mod.rs

relay_knowledge/storage/
mod.rs
