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

relay_knowledge/storage/
mod.rs
