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

relay_knowledge/storage/
mod.rs
