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

relay_knowledge/storage/
mod.rs
