Skip to main content

max / audiofiles

11.5 KB · 317 lines History Blame Raw
1 //! Background sync scheduler: periodic auto-sync with exponential backoff on failure.
2 //! SSE push notifications trigger immediate sync when another device pushes changes.
3
4 use std::path::{Path, PathBuf};
5 use std::sync::Arc;
6
7 use parking_lot::Mutex;
8 use synckit_client::SyncKitClient;
9 use tokio::sync::mpsc;
10
11 use tracing::instrument;
12
13 use crate::error::SyncError;
14 use crate::service;
15 use crate::SyncStatus;
16
17 /// Commands the GUI can send to the scheduler.
18 pub enum SyncCommand {
19 /// Trigger an immediate sync cycle.
20 SyncNow,
21 /// Download a single sample blob by hash. Used by the "Download" row-level
22 /// context menu so the user can pull one cloud-only sample without
23 /// triggering a full sync cycle.
24 DownloadOne { hash: String },
25 /// Stop the scheduler loop.
26 Stop,
27 }
28
29 /// Delay before reconnecting the SSE stream after a disconnect (seconds).
30 const SSE_RECONNECT_DELAY_SECS: u64 = 5;
31
32 /// Run the background sync scheduler loop.
33 ///
34 /// Uses `db_path` to open short-lived connections inside `spawn_blocking`.
35 #[instrument(skip_all)]
36 pub async fn run_scheduler(
37 client: Arc<SyncKitClient>,
38 db_path: PathBuf,
39 content_dir: PathBuf,
40 status: Arc<Mutex<SyncStatus>>,
41 mut commands: mpsc::UnboundedReceiver<SyncCommand>,
42 ) {
43 let mut interval = tokio::time::interval(std::time::Duration::from_secs(60));
44 let mut consecutive_failures: u32 = 0;
45 let mut backoff_until: Option<chrono::DateTime<chrono::Utc>> = None;
46 let mut sse_stream: Option<synckit_client::SyncNotifyStream> = None;
47
48 loop {
49 tokio::select! {
50 _ = interval.tick() => {
51 // Check backoff
52 if let Some(until) = backoff_until {
53 if chrono::Utc::now() < until {
54 continue;
55 }
56 }
57
58 // Check if auto-sync is enabled and client is ready
59 if !should_auto_sync(&db_path, &client) {
60 continue;
61 }
62
63 // Check interval elapsed
64 if !interval_elapsed(&db_path) {
65 continue;
66 }
67
68 match run_sync_cycle(&db_path, &content_dir, &client, &status).await {
69 Ok(pulled) => {
70 consecutive_failures = 0;
71 backoff_until = None;
72 if pulled > 0 {
73 status.lock().needs_refresh = true;
74 }
75 }
76 Err(e) => {
77 consecutive_failures += 1;
78 let backoff_minutes = std::cmp::min(
79 2u64.saturating_pow(consecutive_failures),
80 15,
81 );
82 backoff_until = Some(
83 chrono::Utc::now()
84 + chrono::Duration::minutes(backoff_minutes as i64),
85 );
86 tracing::warn!(
87 "Auto-sync failed (attempt {consecutive_failures}, backoff {backoff_minutes}m): {e}"
88 );
89 status.lock().last_error = Some(e.to_string());
90 }
91 }
92 }
93 result = async {
94 if let Some(ref mut stream) = sse_stream {
95 stream.next_change().await
96 } else {
97 std::future::pending::<Option<()>>().await
98 }
99 } => {
100 match result {
101 Some(()) => {
102 tracing::debug!("SSE: received change notification, triggering immediate sync");
103 if should_auto_sync(&db_path, &client) {
104 match run_sync_cycle(&db_path, &content_dir, &client, &status).await {
105 Ok(pulled) => {
106 consecutive_failures = 0;
107 backoff_until = None;
108 if pulled > 0 {
109 status.lock().needs_refresh = true;
110 }
111 }
112 Err(e) => {
113 tracing::warn!("SSE-triggered sync failed: {e}");
114 status.lock().last_error = Some(e.to_string());
115 }
116 }
117 }
118 }
119 None => {
120 tracing::debug!("SSE: stream disconnected, will reconnect");
121 sse_stream = None;
122 tokio::time::sleep(std::time::Duration::from_secs(SSE_RECONNECT_DELAY_SECS)).await;
123 }
124 }
125 }
126 cmd = commands.recv() => {
127 match cmd {
128 Some(SyncCommand::SyncNow) => {
129 match run_sync_cycle(&db_path, &content_dir, &client, &status).await {
130 Ok(pulled) => {
131 consecutive_failures = 0;
132 backoff_until = None;
133 if pulled > 0 {
134 status.lock().needs_refresh = true;
135 }
136 }
137 Err(e) => {
138 tracing::error!("Manual sync failed: {e}");
139 status.lock().last_error = Some(e.to_string());
140 }
141 }
142 }
143 Some(SyncCommand::DownloadOne { hash }) => {
144 match service::download_one_blob(&db_path, &content_dir, &client, &hash).await {
145 Ok(()) => {
146 status.lock().needs_refresh = true;
147 }
148 Err(e) => {
149 tracing::warn!("Single-blob download failed for {hash}: {e}");
150 status.lock().last_error = Some(e.to_string());
151 }
152 }
153 }
154 Some(SyncCommand::Stop) | None => {
155 tracing::info!("Sync scheduler stopping");
156 break;
157 }
158 }
159 }
160 }
161
162 // Try to establish SSE connection if not connected and client is ready
163 if sse_stream.is_none()
164 && client.session_info().is_some()
165 && client.has_master_key()
166 {
167 match client.subscribe().await {
168 Ok(stream) => {
169 tracing::debug!("SSE: connected to push notification stream");
170 sse_stream = Some(stream);
171 }
172 Err(e) => {
173 tracing::debug!("SSE: failed to connect (will retry): {e}");
174 }
175 }
176 }
177 }
178 }
179
180 /// Check if auto-sync is enabled and the client is ready (authenticated + has key).
181 #[instrument(skip_all)]
182 fn should_auto_sync(db_path: &std::path::Path, client: &SyncKitClient) -> bool {
183 if client.session_info().is_none()
184 || !client.has_master_key()
185 {
186 return false;
187 }
188 let conn = match rusqlite::Connection::open(db_path) {
189 Ok(c) => c,
190 Err(e) => {
191 tracing::warn!("Failed to open DB for auto_sync check: {e}");
192 return false;
193 }
194 };
195 service::get_sync_state(&conn, "auto_sync_enabled").unwrap_or_default() == "1"
196 }
197
198 /// Check if enough time has elapsed since the last sync.
199 #[instrument(skip_all)]
200 fn interval_elapsed(db_path: &std::path::Path) -> bool {
201 let conn = match rusqlite::Connection::open(db_path) {
202 Ok(c) => c,
203 Err(e) => {
204 tracing::warn!("Failed to open DB for interval check: {e}");
205 return true;
206 }
207 };
208 let last_sync = service::get_sync_state(&conn, "last_sync_at").unwrap_or_default();
209 let interval_str = service::get_sync_state(&conn, "sync_interval_minutes")
210 .unwrap_or_else(|_| "15".to_string());
211 let interval_minutes: i64 = interval_str.parse().unwrap_or(15);
212
213 if last_sync.is_empty() {
214 return true;
215 }
216
217 match chrono::DateTime::parse_from_rfc3339(&last_sync) {
218 Ok(last) => {
219 let elapsed = chrono::Utc::now() - last.with_timezone(&chrono::Utc);
220 elapsed.num_minutes() >= interval_minutes
221 }
222 Err(_) => true,
223 }
224 }
225
226 /// Run a single sync cycle: snapshot if needed, sync, blob sync, cleanup.
227 #[instrument(skip_all)]
228 async fn run_sync_cycle(
229 db_path: &Path,
230 content_dir: &Path,
231 client: &SyncKitClient,
232 status: &Arc<Mutex<SyncStatus>>,
233 ) -> std::result::Result<i64, SyncError> {
234 {
235 let mut s = status.lock();
236 s.state = crate::SyncState::Syncing;
237 s.last_error = None;
238 }
239
240 // Create initial snapshot if first sync
241 let p = db_path.to_path_buf();
242 tokio::task::spawn_blocking(move || {
243 let conn = rusqlite::Connection::open(&p)?;
244 service::create_initial_snapshot(&conn)
245 })
246 .await
247 .map_err(|e| SyncError::Other(e.to_string()))??;
248
249 let result = service::perform_sync(db_path, client).await?;
250
251 // Mark samples as cloud_only when blob doesn't exist locally
252 let p = db_path.to_path_buf();
253 let cd = content_dir.to_path_buf();
254 if let Err(e) = tokio::task::spawn_blocking(move || {
255 let conn = rusqlite::Connection::open(&p)?;
256 service::mark_cloud_only_samples(&conn, &cd)
257 })
258 .await
259 .map_err(|e| SyncError::Other(e.to_string()))? {
260 tracing::warn!("Cloud-only marking failed (non-fatal): {e}");
261 }
262
263 // Blob sync: upload pending, then download missing
264 if let Err(e) = service::upload_pending_blobs(db_path, content_dir, client).await {
265 tracing::warn!("Blob upload failed (non-fatal): {e}");
266 }
267 if let Err(e) = service::download_missing_blobs(db_path, content_dir, client).await {
268 tracing::warn!("Blob download failed (non-fatal): {e}");
269 }
270
271 // Cleanup old entries + enforce retention cap
272 let p = db_path.to_path_buf();
273 tokio::task::spawn_blocking(move || {
274 let conn = rusqlite::Connection::open(&p)?;
275 service::cleanup_changelog(&conn)?;
276 service::enforce_changelog_retention(&conn)
277 })
278 .await
279 .map_err(|e| SyncError::Other(e.to_string()))??;
280
281 // Update status
282 {
283 let p = db_path.to_path_buf();
284 let last_sync = tokio::task::spawn_blocking(move || {
285 match rusqlite::Connection::open(&p) {
286 Ok(c) => service::get_sync_state(&c, "last_sync_at").unwrap_or_default(),
287 Err(e) => {
288 tracing::warn!("Failed to open DB for last_sync_at: {e}");
289 String::new()
290 }
291 }
292 })
293 .await
294 .unwrap_or_default();
295
296 let p2 = db_path.to_path_buf();
297 let pending = tokio::task::spawn_blocking(move || {
298 match rusqlite::Connection::open(&p2) {
299 Ok(c) => service::count_pending_changes(&c).unwrap_or(0),
300 Err(e) => {
301 tracing::warn!("Failed to open DB for pending changes: {e}");
302 0
303 }
304 }
305 })
306 .await
307 .unwrap_or(0);
308
309 let mut s = status.lock();
310 s.state = crate::SyncState::Ready;
311 s.last_sync_at = if last_sync.is_empty() { None } else { Some(last_sync) };
312 s.pending_changes = pending;
313 }
314
315 Ok(result.pulled)
316 }
317