coven/sync/
sync_loop.rs

1//! Sync loop handle: runs the background sync loop on a dedicated OS thread.
2//!
3//! Owns the sync infrastructure (storage client, HLC, the owned [`Database`]
4//! handle, etc.) and runs sync cycles on a timer or manual trigger. The
5//! database access goes through the [`Database`] handle, so the loop holds
6//! nothing `!Send` — but it still runs on its own OS thread (a current-thread
7//! tokio runtime that `block_on`s the loop) for the *stack*: aws-sdk-s3's
8//! endpoint/auth resolution recurses deeply enough to overflow the default
9//! secondary-thread stack in debug builds. The thread is given a main-thread-
10//! sized stack so S3 sync doesn't `SIGBUS` in `resolve_endpoint`.
11//! Publishes the current [`SyncLoopStatus`] through a watch channel the
12//! [`CovenHandle`](crate::CovenHandle) owns — so a subscription survives a loop
13//! restart, and the loop only ever sends.
14
15use std::sync::atomic::{AtomicBool, Ordering};
16use std::sync::Arc;
17use std::time::Duration;
18
19use tokio::sync::mpsc::error::TrySendError;
20use tracing::{debug, error, info};
21
22use crate::blob::BlobTransitionObserver;
23use crate::clock::ClockRef;
24use crate::config::Config;
25use crate::coven::StoreOpenGuard;
26use crate::keys::MasterKeyCustody;
27use crate::store_dir::StoreDir;
28
29use super::cloud_storage::{BlobPathScheme, CloudSyncStorage};
30use super::cycle::SyncComponents;
31use super::hlc::Hlc;
32use super::loop_policy::{self, LoopWait, SyncLoopReport, SyncLoopSuccess};
33
34/// Why starting or stopping the background sync loop failed.
35#[derive(Debug, thiserror::Error)]
36pub enum SyncLoopError {
37    /// `start` was called on a handle whose channels a prior `start` already
38    /// took: a stopped loop's handle is not restartable.
39    #[error("sync loop cannot be restarted after its channels were taken")]
40    NotRestartable,
41    /// The dedicated sync-loop OS thread could not be spawned.
42    #[error("failed to spawn sync loop thread: {0}")]
43    ThreadSpawn(std::io::Error),
44    /// The sync-loop thread panicked; `stop` observed it on join.
45    #[error("sync loop thread panicked")]
46    ThreadPanicked,
47}
48
49/// A sync-loop status the host renders. The loop reports provider reachability,
50/// publication, and one terminal status. [`Blocked`](Self::Blocked) is a
51/// successful storage cycle with unresolved durable writes;
52/// [`Synchronized`](Self::Synchronized) has none, while
53/// [`Failed`](Self::Failed) means the cycle itself failed. The in-progress marker
54/// is the variant itself, so there is no separate "syncing" flag.
55///
56/// A whole-cycle failure is `Failed`; an otherwise-successful cycle carries its
57/// [`SyncLoopSuccess`] in `Synchronized` or `Blocked`. Warnings ride in
58/// [`SyncLoopSuccess::alerts`].
59///
60/// A subscription immediately exposes the current value. Intermediate values may
61/// be coalesced when the producer changes state faster than a receiver observes
62/// it. A `Synchronized` value's [`SyncLoopSuccess::row_changes`] therefore remains a
63/// refresh hint, not a complete change stream.
64#[derive(Debug, Clone)]
65pub enum SyncLoopStatus {
66    /// No provider operation has succeeded for the current connection.
67    Offline,
68    /// The loop is checking whether storage is reachable.
69    CheckingStorage,
70    /// Storage is reachable and the cycle may publish local state.
71    Publishing,
72    /// The cycle completed. Warnings, if any, ride in the success's `alerts`;
73    /// the observed device activity and applied row changes are on it too.
74    Synchronized(SyncLoopSuccess),
75    /// The cycle reached storage, but one or more writes cannot publish until
76    /// their named prerequisite is supplied or repaired.
77    Blocked {
78        success: SyncLoopSuccess,
79        writes: Vec<crate::PendingWrite>,
80    },
81    /// The cycle failed as a whole — no outcome to report, only the fault.
82    Failed { error: String },
83}
84
85/// Manages the background sync loop and provides access to sync components.
86pub(crate) struct SyncLoopHandle {
87    inner: Arc<SyncLoopInner>,
88    clock: ClockRef,
89    config: Config,
90    store_dir: StoreDir,
91    trigger_tx: tokio::sync::mpsc::Sender<()>,
92    trigger_rx: std::sync::Mutex<Option<tokio::sync::mpsc::Receiver<()>>>,
93    command_tx: tokio::sync::mpsc::Sender<SyncCommand>,
94    command_rx: std::sync::Mutex<Option<tokio::sync::mpsc::Receiver<SyncCommand>>>,
95    stop_tx: tokio::sync::watch::Sender<bool>,
96    stop_rx: std::sync::Mutex<Option<tokio::sync::watch::Receiver<bool>>>,
97    /// The current status value, owned by the [`CovenHandle`] and cloned into each
98    /// loop it starts, so a subscription survives a loop restart (a reconnect
99    /// builds a fresh loop but keeps this same sender). The loop only sends here.
100    status_tx: tokio::sync::watch::Sender<SyncLoopStatus>,
101    thread_handle: std::sync::Mutex<Option<std::thread::JoinHandle<()>>>,
102    running: Arc<AtomicBool>,
103}
104
105struct SyncLoopInner {
106    components: SyncComponents,
107    custody: Arc<dyn MasterKeyCustody>,
108    observer: Option<Arc<dyn BlobTransitionObserver>>,
109
110    /// The store-directory lock, held so it releases only when the loop's
111    /// thread exits. The running thread keeps a clone of this `SyncLoopInner`
112    /// alive across its whole cycle, so the last handle dropping never releases
113    /// `.coven-lock` while a mid-cycle pull or upload is still writing — a second
114    /// `open()` of the same store stays refused until this writer is gone.
115    _open_guard: Arc<StoreOpenGuard>,
116}
117
118type CircleReply<T> =
119    tokio::sync::oneshot::Sender<Result<T, crate::sync::store::CircleOperationError>>;
120
121enum SyncCommand {
122    CreateCircle {
123        name: String,
124        reply: CircleReply<crate::CircleId>,
125    },
126    RenameCircle {
127        circle_id: crate::CircleId,
128        name: String,
129        reply: CircleReply<()>,
130    },
131    AddCircleMember {
132        circle_id: crate::CircleId,
133        member_pubkey: String,
134        role: crate::CircleRole,
135        reply: CircleReply<()>,
136    },
137    RemoveCircleMember {
138        circle_id: crate::CircleId,
139        member_pubkey: String,
140        reply: CircleReply<crate::CircleOperationId>,
141    },
142    ResolveCircleControl {
143        circle_id: crate::CircleId,
144        chosen: crate::CircleControlCoord,
145        reply: CircleReply<()>,
146    },
147    CancelCircleEpochClose {
148        circle_id: crate::CircleId,
149        reply: CircleReply<crate::CircleOperationId>,
150    },
151    ExcludeCircleCloseDevice {
152        circle_id: crate::CircleId,
153        excluded_device_id: crate::StoreDeviceId,
154        reply: CircleReply<()>,
155    },
156    DeleteCircle {
157        circle_id: crate::CircleId,
158        reply: CircleReply<()>,
159    },
160    RetryCircleOperation {
161        operation_id: crate::CircleOperationId,
162        reply: CircleReply<()>,
163    },
164    DiscardCircleOperation {
165        operation_id: crate::CircleOperationId,
166        reply: CircleReply<()>,
167    },
168}
169
170impl SyncLoopHandle {
171    pub(crate) fn new(
172        components: SyncComponents,
173        custody: Arc<dyn MasterKeyCustody>,
174        clock: ClockRef,
175        config: Config,
176        observer: Option<Arc<dyn BlobTransitionObserver>>,
177        open_guard: Arc<StoreOpenGuard>,
178        status_tx: tokio::sync::watch::Sender<SyncLoopStatus>,
179    ) -> Self {
180        let (trigger_tx, trigger_rx) = tokio::sync::mpsc::channel(1);
181        let (command_tx, command_rx) = tokio::sync::mpsc::channel(16);
182        let (stop_tx, stop_rx) = tokio::sync::watch::channel(false);
183        let store_dir = config.store_dir.clone();
184
185        Self {
186            inner: Arc::new(SyncLoopInner {
187                components,
188                custody,
189                observer,
190                _open_guard: open_guard,
191            }),
192            clock,
193            config,
194            store_dir,
195            trigger_tx,
196            trigger_rx: std::sync::Mutex::new(Some(trigger_rx)),
197            command_tx,
198            command_rx: std::sync::Mutex::new(Some(command_rx)),
199            stop_tx,
200            stop_rx: std::sync::Mutex::new(Some(stop_rx)),
201            status_tx,
202            thread_handle: std::sync::Mutex::new(None),
203            running: Arc::new(AtomicBool::new(false)),
204        }
205    }
206
207    /// Start the background sync loop on a dedicated OS thread. No-op if already
208    /// running.
209    ///
210    /// The thread runs a current-thread tokio runtime that `block_on`s the loop.
211    /// A dedicated thread (rather than `tokio::spawn` on the host runtime) is for
212    /// the *stack*: aws-sdk-s3's endpoint/auth resolution recurses deeply enough
213    /// to overflow the default secondary-thread stack in debug builds (SIGBUS in
214    /// `resolve_endpoint`), so the thread is given a main-thread-sized stack.
215    /// Everything the loop holds is `Send`; database access goes through async
216    /// calls on the [`Database`] handle, so nothing here is bound to this thread
217    /// except by choice of stack size.
218    pub(crate) fn start(&self) -> Result<(), SyncLoopError> {
219        if self.running.swap(true, Ordering::AcqRel) {
220            return Ok(());
221        }
222
223        let mut trigger_rx = match self.trigger_rx.lock().unwrap().take() {
224            Some(rx) => rx,
225            None => {
226                self.running.store(false, Ordering::Release);
227                return Err(SyncLoopError::NotRestartable);
228            }
229        };
230        let mut stop_rx = match self.stop_rx.lock().unwrap().take() {
231            Some(rx) => rx,
232            None => {
233                self.running.store(false, Ordering::Release);
234                return Err(SyncLoopError::NotRestartable);
235            }
236        };
237        let mut command_rx = match self.command_rx.lock().unwrap().take() {
238            Some(rx) => rx,
239            None => {
240                self.running.store(false, Ordering::Release);
241                return Err(SyncLoopError::NotRestartable);
242            }
243        };
244
245        let inner = Arc::clone(&self.inner);
246        let status_tx = self.status_tx.clone();
247        let clock = self.clock.clone();
248        let store_dir = self.store_dir.clone();
249        let running = Arc::clone(&self.running);
250
251        let handle = std::thread::Builder::new()
252            .name("coven-sync-loop".to_string())
253            // aws-sdk-s3's endpoint/auth resolution recurses deeply enough to blow
254            // the ~2 MiB default secondary-thread stack in debug builds (SIGBUS in
255            // resolve_endpoint). Give this thread a main-thread-sized stack so S3
256            // sync doesn't overflow it.
257            .stack_size(8 * 1024 * 1024)
258            .spawn(move || {
259                let _running_guard = RunningGuard {
260                    running: Arc::clone(&running),
261                };
262                let rt = match tokio::runtime::Builder::new_current_thread()
263                    .enable_all()
264                    .build()
265                {
266                    Ok(rt) => rt,
267                    Err(e) => {
268                        let error = format!("failed to create sync loop runtime: {e}");
269                        error!("{error}");
270                        status_tx.send_replace(SyncLoopStatus::Failed { error });
271                        return;
272                    }
273                };
274
275                rt.block_on(async move {
276                    // Short delay to avoid racing with app startup.
277                    let startup_delay = tokio::time::sleep(Duration::from_secs(3));
278                    tokio::pin!(startup_delay);
279                    loop {
280                        tokio::select! {
281                            _ = &mut startup_delay => break,
282                            changed = stop_rx.changed() => {
283                                if changed.is_err() || *stop_rx.borrow() {
284                                    info!("Sync loop stopped before first cycle");
285                                    return;
286                                }
287                            }
288                            command = command_rx.recv() => {
289                                let Some(command) = command else {
290                                    info!("Sync command channel closed before first cycle");
291                                    return;
292                                };
293                                execute_command(&inner, &store_dir, command).await;
294                            }
295                        }
296                    }
297
298                    let mut consecutive_failures: u32 = 0;
299                    while running.load(Ordering::Acquire) && !*stop_rx.borrow() {
300                        status_tx.send_replace(SyncLoopStatus::CheckingStorage);
301                        let reachable = inner
302                            .components
303                            .storage()
304                            .cloud_home()
305                            .probe()
306                            .await
307                            .map_err(crate::sync::storage::StorageError::from);
308                        let (decision, status) = match reachable {
309                            Err(error) => {
310                                let status = storage_check_failure_status(&error);
311                                let decision = loop_policy::after_failure(
312                                    format!("check sync storage: {error}"),
313                                    consecutive_failures,
314                                    300,
315                                );
316                                (decision, status)
317                            }
318                            Ok(_) => {
319                                status_tx.send_replace(SyncLoopStatus::Publishing);
320                                let (decision, cycle_went_offline) = match run_single_cycle(&inner, clock.as_ref(), &store_dir).await {
321                                    Ok(result) => (loop_policy::after_success(result), false),
322                                    Err(error) => {
323                                        let offline = error.is_offline();
324                                        (
325                                            loop_policy::after_failure(
326                                                error.to_string(),
327                                                consecutive_failures,
328                                                300,
329                                            ),
330                                            offline,
331                                        )
332                                    }
333                                };
334                                let status = match &decision.report {
335                                    SyncLoopReport::Success(success) => {
336                                        match current_success_status(inner.components.database(), success.clone()).await {
337                                            Ok(status) => status,
338                                            Err(error) => SyncLoopStatus::Failed { error },
339                                        }
340                                    }
341                                    SyncLoopReport::Failure(_) if cycle_went_offline => {
342                                        SyncLoopStatus::Offline
343                                    }
344                                    SyncLoopReport::Failure(error) => SyncLoopStatus::Failed {
345                                        error: error.clone(),
346                                    },
347                                };
348                                (decision, status)
349                            }
350                        };
351                        consecutive_failures = decision.consecutive_failures;
352                        status_tx.send_replace(status);
353
354                        let wait = match decision.wait {
355                            LoopWait::Immediate => Duration::ZERO,
356                            LoopWait::Idle => Duration::from_secs(super::backoff::backoff_secs(0, 300)),
357                            LoopWait::BackoffSecs(secs) => Duration::from_secs(secs),
358                        };
359                        if matches!(decision.wait, LoopWait::BackoffSecs(_)) {
360                            debug!(
361                                "Backing off {wait:?} after {consecutive_failures} consecutive failure(s)",
362                            );
363                        }
364                        tokio::select! {
365                            _ = tokio::time::sleep(wait) => {}
366                            changed = stop_rx.changed() => {
367                                if changed.is_err() || *stop_rx.borrow() {
368                                    info!("Sync loop stop requested");
369                                    break;
370                                }
371                            }
372                            msg = trigger_rx.recv() => {
373                                if msg.is_none() {
374                                    info!("Sync trigger channel closed, stopping sync loop");
375                                    break;
376                                }
377                            }
378                            command = command_rx.recv() => {
379                                let Some(command) = command else {
380                                    info!("Sync command channel closed, stopping sync loop");
381                                    break;
382                                };
383                                execute_command(&inner, &store_dir, command).await;
384                            }
385                        }
386                    }
387                });
388            })
389            .map_err(|e| {
390                self.running.store(false, Ordering::Release);
391                SyncLoopError::ThreadSpawn(e)
392            })?;
393
394        *self.thread_handle.lock().unwrap() = Some(handle);
395        Ok(())
396    }
397
398    /// Whether the background sync thread is running.
399    pub(crate) fn is_running(&self) -> bool {
400        self.running.load(Ordering::Acquire)
401    }
402
403    /// Request loop shutdown and join the sync thread.
404    pub(crate) fn stop(&self) -> Result<(), SyncLoopError> {
405        let handle = {
406            let mut guard = self.thread_handle.lock().unwrap();
407            if guard.is_none() && !self.running.load(Ordering::Acquire) {
408                return Ok(());
409            }
410            if self.stop_tx.send(true).is_err() {
411                debug!("sync loop stop requested after stop receiver closed");
412            }
413            self.trigger();
414            guard.take()
415        };
416
417        if let Some(handle) = handle {
418            if handle.join().is_err() {
419                self.running.store(false, Ordering::Release);
420                return Err(SyncLoopError::ThreadPanicked);
421            }
422        }
423        self.running.store(false, Ordering::Release);
424        Ok(())
425    }
426
427    /// Signal the sync loop to run a cycle immediately.
428    ///
429    /// `Full` means a trigger is already pending — our request collapses into the
430    /// existing one, which is exactly what the capacity-1 channel is for.
431    /// `Closed` means the loop is gone, so the trigger is moot.
432    pub(crate) fn trigger(&self) {
433        match self.trigger_tx.try_send(()) {
434            Ok(()) | Err(TrySendError::Full(())) => {}
435            Err(TrySendError::Closed(())) => {
436                debug!("Sync trigger channel closed, loop is not running");
437            }
438        }
439    }
440
441    // -- Accessors for membership operations --
442
443    pub(crate) fn storage(&self) -> &Arc<CloudSyncStorage> {
444        self.inner.components.storage()
445    }
446
447    pub(crate) fn store_dir(&self) -> &StoreDir {
448        &self.store_dir
449    }
450
451    pub(crate) fn config(&self) -> &Config {
452        &self.config
453    }
454
455    pub(crate) fn blob_path_scheme(&self) -> BlobPathScheme {
456        self.inner.components.blob_path_scheme()
457    }
458
459    pub(crate) fn self_uploader(&self) -> String {
460        self.inner.components.self_uploader()
461    }
462
463    pub(crate) fn hlc(&self) -> &Arc<Hlc> {
464        self.inner.components.hlc()
465    }
466
467    pub(crate) fn current_encryption(&self) -> Option<crate::encryption::EncryptionService> {
468        self.inner.components.current_encryption()
469    }
470
471    pub(crate) async fn invite_member(
472        &self,
473        public_key_hex: &str,
474        invitee_email: Option<&str>,
475        role: super::membership::MemberRole,
476        store_name: &str,
477    ) -> Result<crate::join_code::InviteCode, super::store::MembershipOpsError> {
478        self.inner
479            .components
480            .invite_member(public_key_hex, invitee_email, role, store_name)
481            .await
482    }
483
484    pub(crate) async fn remove_member(
485        &self,
486        public_key_hex: &str,
487    ) -> Result<String, super::store::MembershipOpsError> {
488        self.inner
489            .components
490            .remove_member(public_key_hex, self.inner.custody.as_ref())
491            .await
492    }
493
494    pub(crate) async fn resolve_membership_conflict(
495        &self,
496        choice: &crate::MembershipConflictChoice,
497    ) -> Result<(), super::store::MembershipOpsError> {
498        self.inner
499            .components
500            .resolve_membership_conflict(choice)
501            .await
502    }
503
504    #[cfg(test)]
505    pub(crate) fn adopt_key_rotation_for_test(
506        &self,
507        encryption: crate::encryption::EncryptionService,
508    ) -> Result<String, crate::keys::KeyError> {
509        self.inner
510            .components
511            .adopt_key_rotation(encryption, self.inner.custody.as_ref())
512    }
513
514    pub(crate) async fn drain_uploads(
515        &self,
516    ) -> Result<crate::blob::upload::DrainOutcome, crate::database::DbError> {
517        self.inner
518            .components
519            .drain_uploads(
520                self.clock.as_ref(),
521                &self.store_dir,
522                self.inner.observer.as_deref(),
523            )
524            .await
525    }
526
527    /// Send a Circle write command to the loop thread and await its reply. Circle
528    /// writes run on the loop thread so they never interleave with a sync cycle.
529    async fn send_circle_command<T>(
530        &self,
531        command: impl FnOnce(CircleReply<T>) -> SyncCommand,
532    ) -> Result<T, crate::sync::store::CircleOperationError> {
533        let (reply, response) = tokio::sync::oneshot::channel();
534        self.command_tx
535            .send(command(reply))
536            .await
537            .map_err(|_| crate::sync::store::CircleOperationError::CommandChannelClosed)?;
538        response
539            .await
540            .map_err(|_| crate::sync::store::CircleOperationError::ReplyChannelClosed)?
541    }
542
543    pub(crate) async fn create_circle(
544        &self,
545        name: &str,
546    ) -> Result<crate::CircleId, crate::sync::store::CircleOperationError> {
547        let name = name.to_string();
548        self.send_circle_command(|reply| SyncCommand::CreateCircle { name, reply })
549            .await
550    }
551
552    pub(crate) async fn rename_circle(
553        &self,
554        circle_id: crate::CircleId,
555        name: &str,
556    ) -> Result<(), crate::sync::store::CircleOperationError> {
557        let name = name.to_string();
558        self.send_circle_command(|reply| SyncCommand::RenameCircle {
559            circle_id,
560            name,
561            reply,
562        })
563        .await
564    }
565
566    pub(crate) async fn add_circle_member(
567        &self,
568        circle_id: crate::CircleId,
569        member_pubkey: String,
570        role: crate::CircleRole,
571    ) -> Result<(), crate::sync::store::CircleOperationError> {
572        self.send_circle_command(|reply| SyncCommand::AddCircleMember {
573            circle_id,
574            member_pubkey,
575            role,
576            reply,
577        })
578        .await
579    }
580
581    pub(crate) async fn remove_circle_member(
582        &self,
583        circle_id: crate::CircleId,
584        member_pubkey: String,
585    ) -> Result<crate::CircleOperationId, crate::sync::store::CircleOperationError> {
586        self.send_circle_command(|reply| SyncCommand::RemoveCircleMember {
587            circle_id,
588            member_pubkey,
589            reply,
590        })
591        .await
592    }
593
594    pub(crate) async fn resolve_circle_control(
595        &self,
596        circle_id: crate::CircleId,
597        chosen: crate::CircleControlCoord,
598    ) -> Result<(), crate::sync::store::CircleOperationError> {
599        self.send_circle_command(|reply| SyncCommand::ResolveCircleControl {
600            circle_id,
601            chosen,
602            reply,
603        })
604        .await
605    }
606
607    pub(crate) async fn cancel_circle_epoch_close(
608        &self,
609        circle_id: crate::CircleId,
610    ) -> Result<crate::CircleOperationId, crate::sync::store::CircleOperationError> {
611        self.send_circle_command(|reply| SyncCommand::CancelCircleEpochClose { circle_id, reply })
612            .await
613    }
614
615    pub(crate) async fn exclude_circle_close_device(
616        &self,
617        circle_id: crate::CircleId,
618        excluded_device_id: crate::StoreDeviceId,
619    ) -> Result<(), crate::sync::store::CircleOperationError> {
620        self.send_circle_command(|reply| SyncCommand::ExcludeCircleCloseDevice {
621            circle_id,
622            excluded_device_id,
623            reply,
624        })
625        .await
626    }
627
628    pub(crate) async fn delete_circle(
629        &self,
630        circle_id: crate::CircleId,
631    ) -> Result<(), crate::sync::store::CircleOperationError> {
632        self.send_circle_command(|reply| SyncCommand::DeleteCircle { circle_id, reply })
633            .await
634    }
635
636    pub(crate) async fn retry_circle_operation(
637        &self,
638        operation_id: crate::CircleOperationId,
639    ) -> Result<(), crate::sync::store::CircleOperationError> {
640        self.send_circle_command(|reply| SyncCommand::RetryCircleOperation {
641            operation_id,
642            reply,
643        })
644        .await
645    }
646
647    pub(crate) async fn discard_circle_operation(
648        &self,
649        operation_id: crate::CircleOperationId,
650    ) -> Result<(), crate::sync::store::CircleOperationError> {
651        self.send_circle_command(|reply| SyncCommand::DiscardCircleOperation {
652            operation_id,
653            reply,
654        })
655        .await
656    }
657
658    /// Inspect a Circle's in-flight epoch close. A read, so it runs directly on the
659    /// components rather than serializing behind the write-command channel.
660    pub(crate) async fn circle_close_status(
661        &self,
662        circle_id: crate::CircleId,
663    ) -> Result<crate::CircleCloseStatus, crate::sync::store::CircleOperationError> {
664        self.inner.components.circle_close_status(circle_id).await
665    }
666}
667
668async fn execute_command(inner: &SyncLoopInner, store_dir: &StoreDir, command: SyncCommand) {
669    match command {
670        SyncCommand::CreateCircle { name, reply } => {
671            reply_circle_command(reply, inner.components.create_circle(&name).await);
672        }
673        SyncCommand::RenameCircle {
674            circle_id,
675            name,
676            reply,
677        } => {
678            reply_circle_command(
679                reply,
680                inner.components.rename_circle(circle_id, &name).await,
681            );
682        }
683        SyncCommand::AddCircleMember {
684            circle_id,
685            member_pubkey,
686            role,
687            reply,
688        } => {
689            reply_circle_command(
690                reply,
691                inner
692                    .components
693                    .add_circle_member(store_dir, circle_id, member_pubkey, role)
694                    .await,
695            );
696        }
697        SyncCommand::RemoveCircleMember {
698            circle_id,
699            member_pubkey,
700            reply,
701        } => {
702            reply_circle_command(
703                reply,
704                inner
705                    .components
706                    .remove_circle_member(circle_id, member_pubkey)
707                    .await,
708            );
709        }
710        SyncCommand::ResolveCircleControl {
711            circle_id,
712            chosen,
713            reply,
714        } => {
715            reply_circle_command(
716                reply,
717                inner
718                    .components
719                    .resolve_circle_control(circle_id, chosen)
720                    .await,
721            );
722        }
723        SyncCommand::CancelCircleEpochClose { circle_id, reply } => {
724            reply_circle_command(
725                reply,
726                inner.components.cancel_circle_epoch_close(circle_id).await,
727            );
728        }
729        SyncCommand::ExcludeCircleCloseDevice {
730            circle_id,
731            excluded_device_id,
732            reply,
733        } => {
734            reply_circle_command(
735                reply,
736                inner
737                    .components
738                    .exclude_circle_close_device(circle_id, excluded_device_id)
739                    .await,
740            );
741        }
742        SyncCommand::DeleteCircle { circle_id, reply } => {
743            reply_circle_command(reply, inner.components.delete_circle(circle_id).await);
744        }
745        SyncCommand::RetryCircleOperation {
746            operation_id,
747            reply,
748        } => {
749            reply_circle_command(
750                reply,
751                inner.components.retry_circle_operation(&operation_id).await,
752            );
753        }
754        SyncCommand::DiscardCircleOperation {
755            operation_id,
756            reply,
757        } => {
758            reply_circle_command(
759                reply,
760                inner
761                    .components
762                    .discard_circle_operation(&operation_id)
763                    .await,
764            );
765        }
766    }
767}
768
769fn reply_circle_command<T>(
770    reply: CircleReply<T>,
771    result: Result<T, crate::sync::store::CircleOperationError>,
772) {
773    if reply.send(result).is_err() {
774        debug!("Circle command caller dropped its reply receiver");
775    }
776}
777
778fn storage_check_failure_status(error: &crate::sync::storage::StorageError) -> SyncLoopStatus {
779    if error.is_transport() {
780        SyncLoopStatus::Offline
781    } else {
782        SyncLoopStatus::Failed {
783            error: format!("check sync storage: {error}"),
784        }
785    }
786}
787
788async fn current_success_status(
789    db: &crate::database::Database,
790    success: SyncLoopSuccess,
791) -> Result<SyncLoopStatus, String> {
792    let writes: Vec<_> = db
793        .pending_writes()
794        .await
795        .map_err(|error| format!("read pending writes after sync: {error}"))?
796        .into_iter()
797        .filter(|write| matches!(write.status, crate::WriteStatus::Blocked(_)))
798        .collect();
799    if !writes.is_empty() {
800        return Ok(SyncLoopStatus::Blocked { success, writes });
801    }
802    Ok(SyncLoopStatus::Synchronized(success))
803}
804
805struct RunningGuard {
806    running: Arc<AtomicBool>,
807}
808
809impl Drop for RunningGuard {
810    fn drop(&mut self) {
811        self.running.store(false, Ordering::Release);
812    }
813}
814
815/// Run a single sync cycle.
816async fn run_single_cycle(
817    inner: &SyncLoopInner,
818    clock: &dyn crate::clock::Clock,
819    store_dir: &StoreDir,
820) -> Result<super::cycle::SyncCycleResult, super::cycle::SyncCycleFailure> {
821    inner
822        .components
823        .run_cycle(
824            clock,
825            Some(inner.custody.as_ref()),
826            store_dir,
827            inner.observer.as_deref(),
828        )
829        .await
830}
831
832#[cfg(test)]
833mod tests {
834    use super::*;
835
836    fn success() -> SyncLoopSuccess {
837        SyncLoopSuccess {
838            last_sync_time: "2026-07-14T00:00:00Z".to_string(),
839            device_count: 1,
840            device_activity: Vec::new(),
841            data_changed: false,
842            row_changes: None,
843            alerts: crate::SyncLoopAlerts {
844                rotation_pending: None,
845                held_positions: Vec::new(),
846                asset_downloads_failed: false,
847                local_blob_cleanup_pending: false,
848            },
849        }
850    }
851
852    #[test]
853    fn storage_configuration_failure_is_terminal() {
854        let status = storage_check_failure_status(
855            &crate::sync::storage::StorageError::Configuration("missing bucket".to_string()),
856        );
857
858        assert!(matches!(status, SyncLoopStatus::Failed { .. }));
859    }
860
861    fn database() -> crate::database::Database {
862        coven_core::database::Database::open(
863            std::path::Path::new(":memory:"),
864            Vec::new(),
865            chrono::Duration::days(30),
866            coven_core::blob::TransferLimits::one_at_a_time(),
867            "status-test".to_string(),
868            &[],
869        )
870        .expect("open status test database")
871        .0
872    }
873
874    async fn insert_write_status(
875        db: &crate::database::Database,
876        write_id: &'static str,
877        status: crate::WriteStatus,
878    ) {
879        let base = serde_json::json!({ "dependencies": {} }).to_string();
880        let status = serde_json::to_string(&status).expect("serialize durable write status");
881        db.call(move |conn| {
882            conn.execute(
883                r#"INSERT INTO store_writes
884                 (write_id, status, affected_rows, changeset, inverse_changeset, base, blob_facts)
885                 VALUES (?1, ?2, '[]', X'', X'', ?3, '{"blobs":[]}')"#,
886                (write_id, status, base),
887            )
888            .map(|_| ())
889            .map_err(crate::DbError::from)
890        })
891        .await
892        .expect("insert durable write status");
893    }
894
895    #[tokio::test]
896    async fn successful_cycle_projects_durable_blocked_state() {
897        let db = database();
898        insert_write_status(
899            &db,
900            "blocked-write",
901            crate::WriteStatus::Blocked(crate::WriteBlock::MissingBlob {
902                namespace: "audio".to_string(),
903                id: "missing".to_string(),
904            }),
905        )
906        .await;
907        let blocked = current_success_status(&db, success())
908            .await
909            .expect("project blocked state");
910        assert!(matches!(
911            blocked,
912            SyncLoopStatus::Blocked { writes, .. }
913                if writes.len() == 1 && writes[0].write_id.as_str() == "blocked-write"
914        ));
915        db.call(|conn| {
916            conn.execute(
917                "DELETE FROM store_writes WHERE write_id = 'blocked-write'",
918                [],
919            )
920            .map(|_| ())
921            .map_err(crate::DbError::from)
922        })
923        .await
924        .expect("remove blocked projection fixture");
925
926        assert!(matches!(
927            current_success_status(&db, success())
928                .await
929                .expect("project synchronized state"),
930            SyncLoopStatus::Synchronized(_)
931        ));
932    }
933}
coven/sync/sync_loop.rs

coven/sync/
sync_loop.rs
