linkrs_api/api_core/
sync_api.rs

1//! Sync Management API – Core Layer
2//!
3//! Provides transport layer independent sync system management operations.
4
5use crate::api_core::error::{CoreError, CoreResult};
6use linkrs_sync::SyncManager;
7use std::sync::Arc;
8
9/// Sync Management API – Core Layer
10pub struct SyncApi {
11    sync_manager: Arc<SyncManager>,
12}
13
14impl SyncApi {
15    /// Create a new SyncApi instance
16    pub fn new(sync_manager: Arc<SyncManager>) -> Self {
17        Self { sync_manager }
18    }
19
20    /// Get the sync manager
21    pub fn sync_manager(&self) -> &Arc<SyncManager> {
22        &self.sync_manager
23    }
24
25    /// Check if sync is running
26    pub fn is_running(&self) -> bool {
27        self.sync_manager.is_running()
28    }
29
30    /// Get dead letter queue size
31    pub fn get_dlq_size(&self) -> usize {
32        self.sync_manager.get_dlq_size()
33    }
34
35    /// Get unrecovered dead letter queue size
36    pub fn get_unrecovered_dlq_size(&self) -> usize {
37        self.sync_manager.get_unrecovered_dlq_size()
38    }
39
40    /// Get durable outbox delivery statistics.
41    pub fn outbox_stats(&self) -> linkrs_sync::OutboxStats {
42        self.sync_manager.outbox_stats()
43    }
44
45    /// Retry delivery of all pending durable outbox entries.
46    pub fn retry_outbox_projection(&self) -> CoreResult<usize> {
47        self.sync_manager
48            .retry_outbox_sync()
49            .map_err(|e| CoreError::SyncError(e.to_string()))
50    }
51
52    /// Get vector coordinator
53    #[cfg(feature = "vector")]
54    pub fn vector_coordinator(&self) -> Option<&Arc<linkrs_sync::VectorSyncCoordinator>> {
55        self.sync_manager.vector_coordinator()
56    }
57
58    /// Get fulltext sync coordinator. Returns `None` when the fulltext feature
59    /// is not enabled.
60    #[cfg(feature = "fulltext")]
61    pub fn sync_coordinator(&self) -> Option<&Arc<linkrs_sync::FulltextSyncCoordinator>> {
62        self.sync_manager.sync_coordinator()
63    }
64
65    pub fn sync_diagnostics(&self) -> CoreResult<linkrs_sync::SyncDiagnostics> {
66        self.sync_manager
67            .sync_diagnostics()
68            .map_err(|e| CoreError::SyncError(e.to_string()))
69    }
70
71    /// Fulltext indexes whose engine is in the `Inconsistent` state. These
72    /// reject writes and need operator attention (online rebuild or drop and
73    /// recreate). Empty when fulltext is not configured.
74    #[cfg(feature = "fulltext")]
75    pub fn inconsistent_fulltext_indexes(&self) -> Vec<linkrs_fulltext::IndexMetadata> {
76        self.sync_manager.inconsistent_fulltext_indexes()
77    }
78
79    /// Vector indexes whose engine is in the `Inconsistent` state. These
80    /// reject writes and need operator attention. Empty when vector is not configured.
81    #[cfg(feature = "vector")]
82    pub fn inconsistent_vector_indexes(&self) -> Vec<linkrs_core::index_events::RebuildProgress> {
83        self.sync_manager.inconsistent_vector_indexes()
84    }
85
86    /// Fail stranded rebuild generations and collect rebuild scratch state.
87    /// Idempotent; safe to run at startup before serving traffic.
88    pub fn recover_stale_rebuilds(&self) -> CoreResult<usize> {
89        self.sync_manager
90            .recover_stale_rebuilds_sync()
91            .map_err(|e| CoreError::SyncError(e.to_string()))
92    }
93
94    pub fn list_dead_letters(
95        &self,
96        target: Option<&linkrs_core::types::TargetId>,
97        index_id: Option<u64>,
98        generation: Option<u64>,
99        limit: usize,
100        offset: usize,
101    ) -> CoreResult<Vec<linkrs_sync::DeadLetterRow>> {
102        self.sync_manager
103            .list_dead_letters(target, index_id, generation, limit, offset)
104            .map_err(|e| CoreError::SyncError(e.to_string()))
105    }
106
107    pub fn requeue_dead_letter(&self, event_id: i64) -> CoreResult<bool> {
108        self.sync_manager
109            .requeue_dead_letter(event_id)
110            .map_err(|e| CoreError::SyncError(e.to_string()))
111    }
112
113    pub fn requeue_dead_letters_batch(
114        &self,
115        target: Option<&linkrs_core::types::TargetId>,
116        index_id: Option<u64>,
117        generation: Option<u64>,
118        limit: usize,
119    ) -> CoreResult<usize> {
120        self.sync_manager
121            .requeue_dead_letters_batch(target, index_id, generation, limit)
122            .map_err(|e| CoreError::SyncError(e.to_string()))
123    }
124
125    pub fn list_degraded_ranges(
126        &self,
127        target: Option<&linkrs_core::types::TargetId>,
128        index_id: Option<u64>,
129        generation: Option<u64>,
130    ) -> CoreResult<Vec<linkrs_sync::DegradedRangeRow>> {
131        self.sync_manager
132            .list_degraded_ranges(target, index_id, generation)
133            .map_err(|e| CoreError::SyncError(e.to_string()))
134    }
135
136    pub fn clear_degraded_range(
137        &self,
138        target: &linkrs_core::types::TargetId,
139        index_id: u64,
140        generation: u64,
141        start_lsn: linkrs_core::types::CommitLsn,
142        end_lsn: linkrs_core::types::CommitLsn,
143    ) -> CoreResult<bool> {
144        self.sync_manager
145            .clear_degraded_range(target, index_id, generation, start_lsn, end_lsn)
146            .map_err(|e| CoreError::SyncError(e.to_string()))
147    }
148
149    pub fn retention_lsn(&self) -> CoreResult<linkrs_core::types::CommitLsn> {
150        self.sync_manager
151            .retention_lsn()
152            .map_err(|e| CoreError::SyncError(e.to_string()))
153    }
154
155    pub fn prune_applied_events(
156        &self,
157        retention_lsn: linkrs_core::types::CommitLsn,
158    ) -> CoreResult<u64> {
159        self.sync_manager
160            .prune_applied_events(retention_lsn)
161            .map_err(|e| CoreError::SyncError(e.to_string()))
162    }
163
164    pub fn run_retention_once(
165        &self,
166        grace_lsn_distance: u64,
167        max_age_ms: u64,
168    ) -> CoreResult<(u64, u64, u64)> {
169        self.sync_manager
170            .run_retention_once(grace_lsn_distance, max_age_ms)
171            .map_err(|e| CoreError::SyncError(e.to_string()))
172    }
173}
linkrs_api/api_core/sync_api.rs

linkrs_api/api_core/
sync_api.rs
