linkrs_sync/manager/
lifecycle.rs

1//! Sync manager construction, builder-style configuration and runtime lifecycle.
2use super::*;
3#[cfg(feature = "fulltext")]
4use linkrs_fulltext::SyncConfig;
5use linkrs_metrics::StatsManager;
6#[cfg_attr(
7    not(any(feature = "fulltext", feature = "vector")),
8    allow(unused_variables)
9)]
10impl super::SyncManager {
11    fn new_common() -> Self {
12        Self {
13            #[cfg(feature = "fulltext")]
14            sync_coordinator: None,
15            #[cfg(feature = "vector")]
16            vector_coordinator: None,
17            pending_intents: DashMap::new(),
18            running: Arc::new(std::sync::atomic::AtomicBool::new(false)),
19            dead_letter_queue: None,
20            sqlite_outbox: None,
21            #[cfg(feature = "vector")]
22            vector_receiver: None,
23            #[cfg(feature = "vector")]
24            auto_embed_text: Arc::new(std::sync::atomic::AtomicBool::new(false)),
25            outbox_consumer: Arc::new(OutboxConsumerConfig::default()),
26            #[cfg(feature = "vector")]
27            backend_policy: None,
28            backpressure: Arc::new(OutboxBackpressureConfig::default()),
29            auth_paused_until_ms: Arc::new(std::sync::atomic::AtomicU64::new(0)),
30            stats_manager: None,
31            handle: Mutex::new(None),
32            rebuild_locks: DashMap::new(),
33        }
34    }
35
36    #[cfg(feature = "fulltext")]
37    pub fn new(sync_coordinator: Arc<SyncCoordinator>) -> Self {
38        Self {
39            sync_coordinator: Some(sync_coordinator),
40            ..Self::new_common()
41        }
42    }
43
44    pub fn new_without_fulltext() -> Self {
45        Self::new_common()
46    }
47
48    #[cfg(feature = "vector")]
49    pub fn with_vector_coordinator(
50        mut self,
51        vector_coordinator: Arc<VectorSyncCoordinator>,
52    ) -> Self {
53        #[cfg(any(feature = "embedding", feature = "rerank"))]
54        if let Some(stats) = self.stats_manager.clone() {
55            vector_coordinator.set_stats_manager(stats);
56        }
57        self.vector_coordinator = Some(vector_coordinator);
58        self
59    }
60
61    #[cfg(feature = "fulltext")]
62    pub fn with_sync_config(
63        sync_coordinator: Arc<SyncCoordinator>,
64        _sync_config: SyncConfig,
65    ) -> Self {
66        Self::new(sync_coordinator)
67    }
68
69    pub fn with_dead_letter_queue(
70        mut self,
71        dead_letter_queue: Arc<crate::DeadLetterQueue>,
72    ) -> Self {
73        self.dead_letter_queue = Some(dead_letter_queue);
74        self
75    }
76
77    pub fn with_stats_manager(mut self, stats_manager: Arc<StatsManager>) -> Self {
78        #[cfg(all(feature = "vector", any(feature = "embedding", feature = "rerank")))]
79        if let Some(coordinator) = self.vector_coordinator.as_ref() {
80            coordinator.set_stats_manager(stats_manager.clone());
81        }
82        self.stats_manager = Some(stats_manager);
83        self
84    }
85
86    #[cfg(feature = "vector")]
87    pub fn set_auto_embed_text(&self, enabled: bool) {
88        self.auto_embed_text
89            .store(enabled, std::sync::atomic::Ordering::Relaxed);
90    }
91
92    #[cfg(feature = "vector")]
93    pub fn auto_embed_text(&self) -> bool {
94        self.auto_embed_text
95            .load(std::sync::atomic::Ordering::Relaxed)
96    }
97
98    pub fn set_stats_manager(&mut self, stats_manager: Arc<StatsManager>) {
99        #[cfg(all(feature = "vector", any(feature = "embedding", feature = "rerank")))]
100        if let Some(coordinator) = self.vector_coordinator.as_ref() {
101            coordinator.set_stats_manager(stats_manager.clone());
102        }
103        self.stats_manager = Some(stats_manager);
104    }
105
106    pub async fn start(&self) -> Result<(), SyncError> {
107        if self.running.load(std::sync::atomic::Ordering::SeqCst) {
108            return Ok(());
109        }
110
111        self.running
112            .store(true, std::sync::atomic::Ordering::SeqCst);
113
114        if self.sqlite_outbox.is_some() {
115            let mut handle = self.handle.lock().await;
116            if handle.is_none() {
117                let manager = self.clone();
118                *handle = Some(tokio::spawn(async move {
119                    let retry_interval = std::time::Duration::from_secs(5);
120                    while manager.running.load(std::sync::atomic::Ordering::SeqCst) {
121                        if let Err(error) = manager.retry_outbox_sync() {
122                            tracing::warn!("Outbox delivery attempt failed: {}", error);
123                        }
124                        tokio::time::sleep(retry_interval).await;
125                    }
126                }));
127            }
128        }
129
130        Ok(())
131    }
132
133    pub async fn stop(&self) {
134        self.running
135            .store(false, std::sync::atomic::Ordering::SeqCst);
136
137        if let Some(handle) = self.handle.lock().await.take() {
138            let _ = handle.await;
139        }
140    }
141
142    pub fn is_running(&self) -> bool {
143        self.running.load(std::sync::atomic::Ordering::SeqCst)
144    }
145}
linkrs_sync/manager/lifecycle.rs

linkrs_sync/manager/
lifecycle.rs
