relay_knowledge/storage/contracts/
index.rs

1use serde::{Deserialize, Serialize};
2
3use crate::domain::{
4    AuditEventRecord, GraphVersion, IndexKind, IndexModality, IndexStatus, ProposalConflictRecord,
5    ProposalRecord, ProposalState, ServiceOperatorState, ServiceOperatorStatus, WorkerStatus,
6    WorkerTaskRecord,
7};
8
9use super::{
10    AuditQueryRequest, FileContentSearchHit, FileContentSearchRequest, FileIndexDiagnostics,
11    FileIndexRoot, FileIndexRootStatus, FileIndexRootUpdate, FileSearchHit, FileSearchRequest,
12    NewAuditEvent, NewProposal, ProposalDecision, ProposalListRequest, ServiceOperatorUpdate,
13    StorageError, StorageFuture, WorkerTaskClaimRequest, WorkerTaskCompletion, WorkerTaskFailure,
14    WorkerTaskSeed,
15};
16
17/// Synthetic scope used for graph-wide index work that is not tied to evidence.
18pub const DEFAULT_INDEX_SOURCE_SCOPE: &str = "graph";
19
20/// Derived index metadata and operational persistence contract.
21pub trait IndexStore: Send + Sync {
22    fn index_statuses(&self) -> StorageFuture<'_, Vec<IndexStatus>>;
23
24    fn mark_refresh_complete(
25        &self,
26        kind: IndexKind,
27        graph_version: GraphVersion,
28    ) -> StorageFuture<'_, IndexStatus>;
29
30    fn index_cursors(&self) -> StorageFuture<'_, Vec<IndexCursor>> {
31        Box::pin(async {
32            Err(StorageError::InvalidInput(
33                "index cursor storage is unavailable".to_owned(),
34            ))
35        })
36    }
37
38    fn queue_index_refreshes(
39        &self,
40        _request: IndexRefreshQueueRequest,
41    ) -> StorageFuture<'_, IndexRefreshDiagnostics> {
42        Box::pin(async {
43            Err(StorageError::InvalidInput(
44                "index refresh task storage is unavailable".to_owned(),
45            ))
46        })
47    }
48
49    fn claim_index_refresh_task(
50        &self,
51        _request: IndexRefreshClaimRequest,
52    ) -> StorageFuture<'_, Option<IndexRefreshTask>> {
53        Box::pin(async {
54            Err(StorageError::InvalidInput(
55                "index refresh task storage is unavailable".to_owned(),
56            ))
57        })
58    }
59
60    fn complete_index_refresh_task(
61        &self,
62        _request: IndexRefreshCompletion,
63    ) -> StorageFuture<'_, IndexRefreshTask> {
64        Box::pin(async {
65            Err(StorageError::InvalidInput(
66                "index refresh task storage is unavailable".to_owned(),
67            ))
68        })
69    }
70
71    fn fail_index_refresh_task(
72        &self,
73        _request: IndexRefreshFailure,
74    ) -> StorageFuture<'_, IndexRefreshTask> {
75        Box::pin(async {
76            Err(StorageError::InvalidInput(
77                "index refresh task storage is unavailable".to_owned(),
78            ))
79        })
80    }
81
82    fn index_refresh_diagnostics(
83        &self,
84        _now_ms: u64,
85    ) -> StorageFuture<'_, IndexRefreshDiagnostics> {
86        Box::pin(async {
87            Err(StorageError::InvalidInput(
88                "index refresh diagnostics are unavailable".to_owned(),
89            ))
90        })
91    }
92
93    fn queue_worker_tasks(
94        &self,
95        _tasks: Vec<WorkerTaskSeed>,
96    ) -> StorageFuture<'_, Vec<WorkerTaskRecord>> {
97        Box::pin(async { Ok(Vec::new()) })
98    }
99
100    fn worker_statuses(&self) -> StorageFuture<'_, Vec<WorkerStatus>> {
101        Box::pin(async { Ok(Vec::new()) })
102    }
103
104    fn claim_worker_task(
105        &self,
106        _request: WorkerTaskClaimRequest,
107    ) -> StorageFuture<'_, Option<WorkerTaskRecord>> {
108        Box::pin(async { Ok(None) })
109    }
110
111    fn complete_worker_task(
112        &self,
113        _request: WorkerTaskCompletion,
114    ) -> StorageFuture<'_, WorkerTaskRecord> {
115        Box::pin(async {
116            Err(StorageError::InvalidInput(
117                "worker task storage is unavailable".to_owned(),
118            ))
119        })
120    }
121
122    fn fail_worker_task(&self, _request: WorkerTaskFailure) -> StorageFuture<'_, WorkerTaskRecord> {
123        Box::pin(async {
124            Err(StorageError::InvalidInput(
125                "worker task storage is unavailable".to_owned(),
126            ))
127        })
128    }
129
130    fn insert_proposal(&self, _proposal: NewProposal) -> StorageFuture<'_, ProposalRecord> {
131        Box::pin(async {
132            Err(StorageError::InvalidInput(
133                "proposal storage is unavailable".to_owned(),
134            ))
135        })
136    }
137
138    fn list_proposals(
139        &self,
140        _request: ProposalListRequest,
141    ) -> StorageFuture<'_, Vec<ProposalRecord>> {
142        Box::pin(async { Ok(Vec::new()) })
143    }
144
145    fn proposal_count(&self, _state: Option<ProposalState>) -> StorageFuture<'_, usize> {
146        Box::pin(async { Ok(0) })
147    }
148
149    fn proposal_by_id(&self, _proposal_id: String) -> StorageFuture<'_, Option<ProposalRecord>> {
150        Box::pin(async { Ok(None) })
151    }
152
153    fn proposal_conflicts(
154        &self,
155        _proposal_id: String,
156    ) -> StorageFuture<'_, Vec<ProposalConflictRecord>> {
157        Box::pin(async { Ok(Vec::new()) })
158    }
159
160    fn decide_proposal(&self, _request: ProposalDecision) -> StorageFuture<'_, ProposalRecord> {
161        Box::pin(async {
162            Err(StorageError::InvalidInput(
163                "proposal storage is unavailable".to_owned(),
164            ))
165        })
166    }
167
168    fn insert_audit_event(&self, _event: NewAuditEvent) -> StorageFuture<'_, AuditEventRecord> {
169        Box::pin(async {
170            Err(StorageError::InvalidInput(
171                "audit storage is unavailable".to_owned(),
172            ))
173        })
174    }
175
176    fn query_audit_events(
177        &self,
178        _request: AuditQueryRequest,
179    ) -> StorageFuture<'_, Vec<AuditEventRecord>> {
180        Box::pin(async { Ok(Vec::new()) })
181    }
182
183    fn audit_event_count(&self) -> StorageFuture<'_, usize> {
184        Box::pin(async { Ok(0) })
185    }
186
187    fn service_operator_status(&self) -> StorageFuture<'_, ServiceOperatorStatus> {
188        Box::pin(async {
189            Ok(ServiceOperatorStatus {
190                state: ServiceOperatorState::Disabled,
191                silent_updates_enabled: false,
192                allowed_scopes: Vec::new(),
193                last_run_at_ms: None,
194                next_retry_at_ms: None,
195                last_error: None,
196                updated_at_ms: 0,
197            })
198        })
199    }
200
201    fn update_service_operator(
202        &self,
203        _request: ServiceOperatorUpdate,
204    ) -> StorageFuture<'_, ServiceOperatorStatus> {
205        Box::pin(async {
206            Err(StorageError::InvalidInput(
207                "service operator storage is unavailable".to_owned(),
208            ))
209        })
210    }
211
212    fn replace_file_index_root(
213        &self,
214        _update: FileIndexRootUpdate,
215    ) -> StorageFuture<'_, FileIndexRootStatus> {
216        unavailable_file_index_storage()
217    }
218
219    fn mark_file_index_roots_unconfigured(
220        &self,
221        _active_roots: Vec<FileIndexRoot>,
222        _now_ms: u64,
223    ) -> StorageFuture<'_, FileIndexDiagnostics> {
224        unavailable_file_index_storage()
225    }
226
227    fn search_files(&self, _request: FileSearchRequest) -> StorageFuture<'_, Vec<FileSearchHit>> {
228        unavailable_file_index_storage()
229    }
230
231    fn search_file_content(
232        &self,
233        _request: FileContentSearchRequest,
234    ) -> StorageFuture<'_, Vec<FileContentSearchHit>> {
235        unavailable_file_index_storage()
236    }
237
238    fn file_index_diagnostics(&self) -> StorageFuture<'_, FileIndexDiagnostics> {
239        unavailable_file_index_storage()
240    }
241}
242
243fn unavailable_file_index_storage<T>() -> StorageFuture<'static, T> {
244    Box::pin(async {
245        Err(StorageError::InvalidInput(
246            "file index storage is unavailable".to_owned(),
247        ))
248    })
249}
250
251/// Scoped cursor for a derived index read model.
252#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
253pub struct IndexCursor {
254    pub kind: IndexKind,
255    pub source_scope: String,
256    pub modality: IndexModality,
257    pub index_version: u64,
258    pub indexed_graph_version: GraphVersion,
259    pub state: crate::domain::IndexState,
260    pub last_error: Option<String>,
261    #[serde(skip_serializing_if = "Option::is_none")]
262    pub source_hash: Option<String>,
263    #[serde(skip_serializing_if = "Option::is_none")]
264    pub backend_cursor: Option<String>,
265    #[serde(skip_serializing_if = "Option::is_none")]
266    pub model_name: Option<String>,
267    #[serde(skip_serializing_if = "Option::is_none")]
268    pub model_dimension: Option<u32>,
269}
270
271/// Persistent index refresh task lifecycle state.
272#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
273#[serde(rename_all = "snake_case")]
274pub enum IndexRefreshTaskState {
275    Queued,
276    Running,
277    Succeeded,
278    Retrying,
279    Failed,
280    DeadLetter,
281}
282
283impl IndexRefreshTaskState {
284    /// Stable storage and API representation.
285    pub const fn as_str(self) -> &'static str {
286        match self {
287            Self::Queued => "queued",
288            Self::Running => "running",
289            Self::Succeeded => "succeeded",
290            Self::Retrying => "retrying",
291            Self::Failed => "failed",
292            Self::DeadLetter => "dead_letter",
293        }
294    }
295}
296
297/// Persistent task used by foreground refresh and startup recovery.
298#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
299pub struct IndexRefreshTask {
300    pub task_id: String,
301    pub kind: IndexKind,
302    pub source_scope: String,
303    pub modality: IndexModality,
304    pub target_graph_version: GraphVersion,
305    pub state: IndexRefreshTaskState,
306    pub lease_owner: Option<String>,
307    pub lease_expires_at_ms: Option<u64>,
308    pub attempt_count: u32,
309    pub next_retry_at_ms: u64,
310    pub input_fingerprint: String,
311    pub cursor_before: GraphVersion,
312    pub cursor_after: Option<GraphVersion>,
313    pub last_error_kind: Option<String>,
314    pub last_error_message: Option<String>,
315    pub created_at_ms: u64,
316    pub updated_at_ms: u64,
317}
318
319/// Queue request created by refresh APIs or the reconciler.
320#[derive(Debug, Clone, PartialEq, Eq)]
321pub struct IndexRefreshQueueRequest {
322    pub kinds: Vec<IndexKind>,
323    pub target_graph_version: GraphVersion,
324    pub max_queue_depth: usize,
325    pub reset_dead_letter_tasks: bool,
326    pub now_ms: u64,
327}
328
329/// Lease acquisition request for bounded foreground workers.
330#[derive(Debug, Clone, PartialEq, Eq)]
331pub struct IndexRefreshClaimRequest {
332    pub lease_owner: String,
333    pub lease_duration_ms: u64,
334    pub max_attempts: u32,
335    pub now_ms: u64,
336}
337
338/// Completion report guarded by the active task lease and attempt token.
339#[derive(Debug, Clone, PartialEq, Eq)]
340pub struct IndexRefreshCompletion {
341    pub task_id: String,
342    pub lease_owner: String,
343    pub attempt_count: u32,
344    pub indexed_graph_version: GraphVersion,
345    pub model_name: Option<String>,
346    pub model_dimension: Option<u32>,
347    pub now_ms: u64,
348}
349
350/// Failure report for retry backoff and dead-letter isolation.
351#[derive(Debug, Clone, PartialEq, Eq)]
352pub struct IndexRefreshFailure {
353    pub task_id: String,
354    pub lease_owner: String,
355    pub attempt_count: u32,
356    pub error_kind: String,
357    pub error_message: String,
358    pub retry_backoff_ms: u64,
359    pub max_attempts: u32,
360    pub now_ms: u64,
361}
362
363/// Per-kind lag included in diagnostics snapshots.
364#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
365pub struct IndexLag {
366    pub kind: IndexKind,
367    pub lag_versions: u64,
368}
369
370/// Structured reason explaining why an index family or scoped cursor is stale.
371#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
372pub struct IndexStalenessReason {
373    pub kind: IndexKind,
374    #[serde(skip_serializing_if = "Option::is_none")]
375    pub source_scope: Option<String>,
376    #[serde(skip_serializing_if = "Option::is_none")]
377    pub modality: Option<IndexModality>,
378    pub reason: String,
379    pub lag_versions: u64,
380    #[serde(skip_serializing_if = "Option::is_none")]
381    pub last_error: Option<String>,
382}
383
384/// Snapshot for queue, dead-letter, and stale-index diagnostics.
385#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)]
386pub struct IndexRefreshDiagnostics {
387    pub queue_depth: usize,
388    pub running_count: usize,
389    pub retrying_count: usize,
390    pub dead_letter_count: usize,
391    pub oldest_unfinished_age_ms: Option<u64>,
392    pub index_lag_by_kind: Vec<IndexLag>,
393    pub max_index_lag_versions: u64,
394    pub stale_index_count: usize,
395    pub stale_reasons: Vec<IndexStalenessReason>,
396}
397
398#[cfg(test)]
399#[path = "index_tests.rs"]
400mod tests;
relay_knowledge/storage/contracts/index.rs

relay_knowledge/storage/contracts/
index.rs
