coven_replication/sync/
sync_loop.rs

1//! Sync loop handle: runs the background sync loop on a prepared OS thread.
2//!
3//! Owns the sync infrastructure (storage client, HLC, the owned [`Database`](coven_database::Database)
4//! handle, etc.) and runs sync cycles on a timer or manual trigger. Setup
5//! prepares that thread and its current-thread Tokio runtime before Store
6//! publication, so installing a connected Store does not construct a runtime
7//! or depend on a host-provided one.
8//! Publishes the current [`SyncLoopStatus`] through a watch channel the
9//! host handle owns — so a subscription survives a loop
10//! restart, and the loop only ever sends.
11
12use std::sync::atomic::{AtomicBool, Ordering};
13use std::sync::Arc;
14
15use tokio::sync::mpsc::error::TrySendError;
16use tracing::debug;
17
18use coven_foundation::clock::ClockRef;
19use coven_foundation::config::Config;
20#[cfg(any(test, feature = "test-utils"))]
21use coven_foundation::store_dir::StoreDir;
22use coven_foundation::store_dir::StoreOpenGuard;
23use coven_protocol::blob::BlobTransitionObserver;
24
25use super::cycle::SyncComponents;
26use super::loop_policy::SyncLoopSuccess;
27use coven_storage::BlobPathScheme;
28
29mod thread;
30#[cfg(test)]
31pub(crate) use thread::current_success_status;
32#[cfg(test)]
33use thread::storage_check_failure_status;
34pub use thread::PreparedSyncLoopRuntime;
35
36/// Why preparing the background sync loop failed.
37#[derive(Debug, thiserror::Error)]
38pub enum SyncLoopError {
39    /// The dedicated sync-loop OS thread could not be spawned.
40    #[error("failed to spawn sync loop thread: {0}")]
41    ThreadSpawn(std::io::Error),
42    /// The dedicated sync-loop thread could not construct its Tokio runtime.
43    #[error("failed to create sync loop runtime: {0}")]
44    Runtime(Arc<std::io::Error>),
45    /// The sync-loop thread panicked; `stop` observed it on join.
46    #[error("sync loop thread panicked")]
47    ThreadPanicked,
48}
49
50#[derive(Debug, Clone, thiserror::Error)]
51pub enum SyncLoopFailure {
52    #[error("check sync storage: {0}")]
53    Storage(Arc<coven_protocol::objects::StorageError>),
54    #[error("sync cycle: {0}")]
55    Cycle(Arc<crate::sync::cycle::SyncCycleFailure>),
56    #[error("read blocked operations after sync: {0}")]
57    BlockedOperations(Arc<coven_database::DbError>),
58    #[error("sync loop panicked")]
59    Panicked,
60}
61
62/// Creates a ready sync-loop thread and runtime before Store publication.
63pub trait SyncLoopRuntimeFactory: Send + Sync {
64    /// Prepare the runtime without attaching an initialized Store session.
65    fn prepare(&self) -> Result<PreparedSyncLoopRuntime, SyncLoopError>;
66}
67
68/// The production sync-loop runtime factory.
69pub struct SystemSyncLoopRuntimeFactory;
70
71impl SyncLoopRuntimeFactory for SystemSyncLoopRuntimeFactory {
72    fn prepare(&self) -> Result<PreparedSyncLoopRuntime, SyncLoopError> {
73        PreparedSyncLoopRuntime::prepare()
74    }
75}
76
77/// A sync-loop status the host renders. The loop reports provider reachability,
78/// publication, and one terminal status. [`Blocked`](Self::Blocked) is a
79/// successful storage cycle with durable operations waiting on a person;
80/// [`Synchronized`](Self::Synchronized) has none, while
81/// [`Failed`](Self::Failed) means the cycle itself failed. The in-progress marker
82/// is the variant itself, so there is no separate "syncing" flag.
83///
84/// A whole-cycle failure is `Failed`; an otherwise-successful cycle carries its
85/// [`SyncLoopSuccess`] in `Synchronized` or `Blocked`. Warnings ride in
86/// [`SyncLoopSuccess::alerts`].
87///
88/// A subscription immediately exposes the current value. Intermediate values may
89/// be coalesced when the producer changes state faster than a receiver observes
90/// it. A `Synchronized` value's [`SyncLoopSuccess::row_changes`] therefore remains a
91/// refresh hint, not a complete change stream.
92#[derive(Debug, Clone)]
93pub enum SyncLoopStatus {
94    /// No provider operation has succeeded for the current connection.
95    Offline,
96    /// The loop is checking whether storage is reachable.
97    CheckingStorage,
98    /// Storage is reachable and the cycle may publish local state.
99    Publishing,
100    /// The cycle completed. Warnings, if any, ride in the success's `alerts`;
101    /// the observed device activity and applied row changes are on it too.
102    Synchronized(SyncLoopSuccess),
103    /// The cycle reached storage, but one or more durable operations cannot
104    /// proceed until their named prerequisite is supplied or repaired.
105    Blocked {
106        success: SyncLoopSuccess,
107        operations: Vec<BlockedOperation>,
108    },
109    /// The cycle failed as a whole — no outcome to report, only the fault.
110    Failed { error: SyncLoopFailure },
111}
112
113/// One durable operation a successful cycle left waiting on a person.
114///
115/// Each kind is stopped for its own reason and returns to work through its own
116/// path, but the host shows them as one list with one button, so they travel as
117/// one value.
118#[derive(Debug, Clone)]
119pub enum BlockedOperation {
120    /// A write stopped by a semantic publication fault.
121    Write(coven_protocol::write::PendingWrite),
122    /// A Circle operation whose author lost the write authority or the stream
123    /// position it was prepared against.
124    CircleOperation(coven_protocol::circle::CircleOperationInfo),
125    /// A reclaim operation that failed with an error running it again cannot
126    /// change.
127    Reclaim(coven_database::StuckReclaimOperation),
128}
129
130impl BlockedOperation {
131    /// Which operation a retry names.
132    pub fn id(&self) -> BlockedOperationId {
133        match self {
134            Self::Write(write) => BlockedOperationId::Write(write.write_id.clone()),
135            Self::CircleOperation(operation) => {
136                BlockedOperationId::CircleOperation(operation.operation_id.clone())
137            }
138            Self::Reclaim(operation) => BlockedOperationId::Reclaim(operation.operation_id),
139        }
140    }
141}
142
143/// Names one blocked operation for a retry, whichever kind it is.
144#[derive(Debug, Clone, PartialEq, Eq)]
145pub enum BlockedOperationId {
146    Write(coven_protocol::write::WriteId),
147    CircleOperation(coven_protocol::circle::CircleOperationId),
148    Reclaim(coven_protocol::store_commit::ObjectHash),
149}
150
151/// Why the sync loop could not return a stuck reclaim operation to its journal.
152#[derive(Debug, thiserror::Error)]
153pub enum RetryStuckReclaimError {
154    #[error("the sync loop is not accepting commands")]
155    CommandChannelClosed,
156    #[error("the sync loop dropped its reply")]
157    ReplyChannelClosed,
158    #[error("{0}")]
159    Database(#[source] Box<coven_database::DbError>),
160}
161
162/// Manages the background sync loop and provides access to sync components.
163pub struct SyncLoopHandle {
164    inner: Arc<SyncLoopHandleInner>,
165    trigger_tx: tokio::sync::mpsc::Sender<()>,
166    command_tx: tokio::sync::mpsc::Sender<SyncCommand>,
167    stop_tx: tokio::sync::watch::Sender<bool>,
168    eager_cache_cancel_tx: tokio::sync::watch::Sender<bool>,
169    activate_tx: tokio::sync::watch::Sender<bool>,
170    /// The current status value, owned by the [`CovenHandle`] and cloned into each
171    /// loop it starts, so a subscription survives a loop restart (a reconnect
172    /// builds a fresh loop but keeps this same sender). The loop only sends here.
173    status_tx: tokio::sync::watch::Sender<SyncLoopStatus>,
174    thread_handle: std::sync::Mutex<Option<std::thread::JoinHandle<()>>>,
175    running: Arc<AtomicBool>,
176}
177
178struct SyncLoopHandleInner {
179    components: SyncComponents,
180    clock: ClockRef,
181    config: Config,
182    observer: Option<Arc<dyn BlobTransitionObserver>>,
183
184    /// The store-directory lock, held so it releases only when the loop's
185    /// thread exits. The running thread keeps a clone of this `SyncLoopHandleInner`
186    /// alive across its whole cycle, so the last handle dropping never releases
187    /// `.coven-lock` while a mid-cycle pull or upload is still writing — a second
188    /// `open()` of the same store stays refused until this writer is gone.
189    _open_guard: Arc<StoreOpenGuard>,
190}
191
192type CircleReply<T> =
193    tokio::sync::oneshot::Sender<Result<T, crate::sync::store::CircleOperationError>>;
194
195enum SyncCommand {
196    CreateCircle {
197        name: String,
198        reply: CircleReply<coven_protocol::CircleId>,
199    },
200    RenameCircle {
201        circle_id: coven_protocol::CircleId,
202        name: String,
203        reply: CircleReply<()>,
204    },
205    AddCircleMember {
206        circle_id: coven_protocol::CircleId,
207        member_pubkey: String,
208        role: coven_protocol::CircleRole,
209        reply: CircleReply<()>,
210    },
211    RemoveCircleMember {
212        circle_id: coven_protocol::CircleId,
213        member_pubkey: String,
214        reply: CircleReply<coven_protocol::CircleOperationId>,
215    },
216    ResolveCircleControl {
217        circle_id: coven_protocol::CircleId,
218        chosen: coven_protocol::CircleControlCoord,
219        reply: CircleReply<()>,
220    },
221    CancelCircleEpochClose {
222        circle_id: coven_protocol::CircleId,
223        reply: CircleReply<coven_protocol::CircleOperationId>,
224    },
225    ExcludeCircleCloseDevice {
226        circle_id: coven_protocol::CircleId,
227        excluded_device_id: coven_protocol::StoreDeviceId,
228        reply: CircleReply<()>,
229    },
230    DeleteCircle {
231        circle_id: coven_protocol::CircleId,
232        reply: CircleReply<()>,
233    },
234    RetryCircleOperation {
235        operation_id: coven_protocol::CircleOperationId,
236        reply: CircleReply<()>,
237    },
238    DiscardCircleOperation {
239        operation_id: coven_protocol::CircleOperationId,
240        reply: CircleReply<()>,
241    },
242    RetryStuckReclaim {
243        operation_id: coven_protocol::store_commit::ObjectHash,
244        reply: tokio::sync::oneshot::Sender<Result<(), RetryStuckReclaimError>>,
245    },
246}
247
248impl SyncLoopHandle {
249    pub fn new(
250        components: SyncComponents,
251        clock: ClockRef,
252        config: Config,
253        observer: Option<Arc<dyn BlobTransitionObserver>>,
254        open_guard: Arc<StoreOpenGuard>,
255        status_tx: tokio::sync::watch::Sender<SyncLoopStatus>,
256        eager_cache_status_tx: tokio::sync::watch::Sender<super::store::EagerCacheFillStatus>,
257        runtime: Option<PreparedSyncLoopRuntime>,
258    ) -> Self {
259        let (trigger_tx, trigger_rx) = tokio::sync::mpsc::channel(1);
260        let (command_tx, command_rx) = tokio::sync::mpsc::channel(16);
261        let (stop_tx, stop_rx) = tokio::sync::watch::channel(false);
262        let (eager_cache_cancel_tx, eager_cache_cancel_rx) = tokio::sync::watch::channel(false);
263        let (activate_tx, activate_rx) = tokio::sync::watch::channel(false);
264        let inner = Arc::new(SyncLoopHandleInner {
265            components,
266            clock,
267            config,
268            observer,
269            _open_guard: open_guard,
270        });
271        let running = Arc::new(AtomicBool::new(runtime.is_some()));
272        let thread_handle = runtime.map(|runtime| {
273            runtime.install(thread::SyncLoopThread::new(
274                Arc::clone(&inner),
275                trigger_rx,
276                command_rx,
277                stop_rx,
278                eager_cache_cancel_rx,
279                activate_rx,
280                status_tx.clone(),
281                eager_cache_status_tx,
282                Arc::clone(&running),
283            ))
284        });
285        Self {
286            inner,
287            trigger_tx,
288            command_tx,
289            stop_tx,
290            eager_cache_cancel_tx,
291            activate_tx,
292            status_tx,
293            thread_handle: std::sync::Mutex::new(thread_handle),
294            running,
295        }
296    }
297
298    /// Release a prepared loop to begin its normal startup delay and cycles.
299    pub fn activate(&self) {
300        self.activate_tx.send_replace(true);
301    }
302
303    /// The provider-operation counter of the home this loop works through, so
304    /// a run driven from outside the loop — an owner-side device-join step —
305    /// can report each stage's count beside its wall time.
306    pub fn provider_requests(
307        &self,
308    ) -> Option<Arc<dyn coven_foundation::stage_timing::ProviderRequests>> {
309        self.inner.components.provider_requests()
310    }
311
312    /// Whether the background sync thread is running.
313    pub fn is_running(&self) -> bool {
314        self.running.load(Ordering::Acquire)
315    }
316
317    /// Request loop shutdown and join the sync thread.
318    pub fn stop(&self) {
319        let handle = {
320            let mut guard = self.thread_handle.lock().unwrap();
321            if guard.is_none() && !self.running.load(Ordering::Acquire) {
322                return;
323            }
324            if self.stop_tx.send(true).is_err() {
325                debug!("sync loop stop requested after stop receiver closed");
326            }
327            self.eager_cache_cancel_tx.send_replace(true);
328            self.trigger();
329            guard.take()
330        };
331
332        if let Some(handle) = handle {
333            if handle.join().is_err() {
334                self.running.store(false, Ordering::Release);
335                let failure = SyncLoopFailure::Panicked;
336                self.status_tx
337                    .send_replace(SyncLoopStatus::Failed { error: failure });
338            }
339        }
340        self.running.store(false, Ordering::Release);
341    }
342
343    /// Signal the sync loop to run a cycle immediately.
344    ///
345    /// `Full` means a trigger is already pending — our request collapses into the
346    /// existing one, which is exactly what the capacity-1 channel is for.
347    /// `Closed` means the loop is gone, so the trigger is moot.
348    pub fn trigger(&self) {
349        match self.trigger_tx.try_send(()) {
350            Ok(()) | Err(TrySendError::Full(())) => {}
351            Err(TrySendError::Closed(())) => {
352                debug!("Sync trigger channel closed, loop is not running");
353            }
354        }
355    }
356
357    /// Stop the post-open CacheEager fill without stopping cloud sync.
358    pub fn cancel_eager_cache_fill(&self) {
359        self.eager_cache_cancel_tx.send_replace(true);
360    }
361
362    pub async fn discard_blocked_write(
363        &self,
364        write_id: coven_protocol::write::WriteId,
365    ) -> Result<Vec<coven_protocol::write::WriteId>, crate::sync::store::StoreError> {
366        self.inner.components.discard_blocked_write(write_id).await
367    }
368
369    pub async fn members(
370        &self,
371    ) -> Result<Vec<coven_protocol::membership::MemberInfo>, super::store::MembershipOpsError> {
372        self.inner.components.members().await
373    }
374
375    pub async fn restore_membership(
376        &self,
377    ) -> Result<super::store::authorization::StoreRestoreMembership, super::store::MembershipOpsError>
378    {
379        self.inner.components.restore_membership().await
380    }
381
382    pub fn host_write_blob_staging(
383        &self,
384        runtime: tokio::runtime::Handle,
385    ) -> crate::sync::store::HostWriteBlobStaging {
386        self.inner.components.host_write_blob_staging(runtime)
387    }
388
389    pub async fn propose_device_exclusion(
390        &self,
391        device_id: coven_protocol::StoreDeviceId,
392    ) -> Result<
393        coven_protocol::store_commit::StoreDeviceExclusionProposalRef,
394        crate::sync::store::StoreDeviceExclusionError,
395    > {
396        self.inner
397            .components
398            .propose_device_exclusion(device_id)
399            .await
400    }
401
402    pub async fn cancel_device_exclusion(
403        &self,
404        proposal: &coven_protocol::store_commit::StoreDeviceExclusionProposalRef,
405    ) -> Result<(), crate::sync::store::StoreDeviceExclusionError> {
406        self.inner
407            .components
408            .cancel_device_exclusion(proposal)
409            .await
410    }
411
412    pub async fn finalize_device_exclusion(
413        &self,
414        proposal: &coven_protocol::store_commit::StoreDeviceExclusionProposalRef,
415    ) -> Result<(), crate::sync::store::StoreDeviceExclusionError> {
416        self.inner
417            .components
418            .finalize_device_exclusion(proposal)
419            .await
420    }
421
422    pub async fn begin_owner_promotion(
423        &self,
424        device_id: coven_protocol::StoreDeviceId,
425    ) -> Result<
426        coven_protocol::store_commit::OwnerPromotionRequest,
427        crate::sync::store::OwnerPromotionError,
428    > {
429        self.inner.components.begin_owner_promotion(device_id).await
430    }
431
432    pub async fn accept_owner_promotion(
433        &self,
434        request: coven_protocol::store_commit::OwnerPromotionRequest,
435    ) -> Result<
436        coven_protocol::store_commit::OwnerPromotionAcceptance,
437        crate::sync::store::OwnerPromotionError,
438    > {
439        self.inner.components.accept_owner_promotion(request).await
440    }
441
442    pub async fn finalize_owner_promotion(
443        &self,
444        acceptance: coven_protocol::store_commit::OwnerPromotionAcceptance,
445    ) -> Result<(), crate::sync::store::OwnerPromotionError> {
446        self.inner
447            .components
448            .finalize_owner_promotion(acceptance)
449            .await
450    }
451
452    pub async fn begin_device_join_bundle(
453        &self,
454        member_pubkey: &str,
455    ) -> Result<crate::sync::DeviceJoinOfferBundle, crate::sync::store::DeviceJoinTransportError>
456    {
457        self.inner
458            .components
459            .begin_device_join_bundle(member_pubkey)
460            .await
461    }
462
463    pub async fn drive_device_join(
464        &self,
465        bundle: &crate::sync::DeviceJoinOfferBundle,
466        policy: crate::sync::DeviceJoinApprovalPolicy<'_>,
467        access_administrator: Option<&dyn crate::sync::DeviceProviderAccessAdministrator>,
468        on_progress: &(dyn Fn(crate::sync::AdmittingDeviceJoinProgress) + Send + Sync),
469        timing: crate::sync::DeviceJoinTransportTiming,
470    ) -> Result<crate::sync::DeviceJoinDriveOutcome, crate::sync::store::DeviceJoinTransportError>
471    {
472        self.inner
473            .components
474            .drive_device_join(bundle, policy, access_administrator, on_progress, timing)
475            .await
476    }
477
478    pub async fn abandon_device_join_transport(
479        &self,
480        bundle: &crate::sync::DeviceJoinOfferBundle,
481    ) -> Result<crate::sync::DeviceJoinAbandonment, crate::sync::store::DeviceJoinTransportError>
482    {
483        self.inner
484            .components
485            .abandon_device_join_transport(bundle)
486            .await
487    }
488
489    pub async fn abort_device_join_transport(
490        &self,
491        bundle: &crate::sync::DeviceJoinOfferBundle,
492    ) -> Result<(), crate::sync::store::DeviceJoinTransportError> {
493        self.inner
494            .components
495            .abort_device_join_transport(bundle)
496            .await
497    }
498
499    pub async fn begin_device_join(
500        &self,
501        member_pubkey: &str,
502    ) -> Result<crate::sync::DeviceJoinOffer, crate::sync::DeviceJoinError> {
503        self.inner.components.begin_device_join(member_pubkey).await
504    }
505
506    pub async fn abandon_device_join(
507        &self,
508        offer: crate::sync::DeviceJoinOffer,
509    ) -> Result<crate::sync::DeviceJoinAbandonment, crate::sync::DeviceJoinError> {
510        self.inner.components.abandon_device_join(offer).await
511    }
512
513    pub async fn authorize_device_provider_access(
514        &self,
515        request: crate::sync::DeviceProviderAccessRequest,
516        access_administrator: Option<&dyn crate::sync::DeviceProviderAccessAdministrator>,
517    ) -> Result<crate::sync::DeviceProviderAdmissionApproval, crate::sync::DeviceJoinError> {
518        self.inner
519            .components
520            .authorize_device_provider_access(request, access_administrator)
521            .await
522    }
523
524    pub async fn accept_device_registration(
525        &self,
526        request: crate::sync::DeviceRegistrationRequest,
527    ) -> Result<crate::sync::ProvisionalDeviceBootstrap, crate::sync::DeviceJoinError> {
528        self.inner
529            .components
530            .accept_device_registration(request)
531            .await
532    }
533
534    pub async fn publish_device_provider_challenge(
535        &self,
536        bootstrap: crate::sync::ProvisionalDeviceBootstrap,
537    ) -> Result<crate::sync::ProviderReadyDeviceBootstrap, crate::sync::DeviceJoinError> {
538        self.inner
539            .components
540            .publish_device_provider_challenge(bootstrap)
541            .await
542    }
543
544    pub async fn complete_device_provider_admission(
545        &self,
546        readiness: crate::sync::DeviceJoinReadiness,
547    ) -> Result<crate::sync::DeviceProviderAdmissionCompletion, crate::sync::DeviceJoinError> {
548        self.inner
549            .components
550            .complete_device_provider_admission(readiness)
551            .await
552    }
553
554    pub async fn finalize_device_join(
555        &self,
556        completion: crate::sync::DeviceProviderAdmissionCompletion,
557    ) -> Result<crate::sync::DeviceJoinActivation, crate::sync::DeviceJoinError> {
558        self.inner.components.finalize_device_join(completion).await
559    }
560
561    pub fn config(&self) -> &Config {
562        &self.inner.config
563    }
564
565    pub fn blob_path_scheme(&self) -> BlobPathScheme {
566        self.inner.components.blob_path_scheme()
567    }
568
569    pub fn is_encrypted(&self) -> bool {
570        self.inner.components.is_encrypted()
571    }
572
573    pub async fn admit_member(
574        &self,
575        public_key_hex: &str,
576        member_email: Option<&str>,
577        role: coven_protocol::membership::MemberRole,
578        store_name: &str,
579    ) -> Result<crate::sync::store::MemberAdmission, super::store::MembershipOpsError> {
580        self.inner
581            .components
582            .admit_member(public_key_hex, member_email, role, store_name)
583            .await
584    }
585
586    pub async fn remove_member(
587        &self,
588        public_key_hex: &str,
589    ) -> Result<String, super::store::MembershipOpsError> {
590        self.inner.components.remove_member(public_key_hex).await
591    }
592
593    pub async fn drain_uploads(
594        &self,
595    ) -> Result<crate::blob::DrainOutcome, super::store::StoreError> {
596        self.inner
597            .components
598            .drain_uploads(self.inner.clock.as_ref(), self.inner.observer.as_deref())
599            .await
600    }
601
602    pub async fn make_remote(
603        &self,
604        root_table: &str,
605        root_id: &str,
606        root_label: &str,
607        pin: bool,
608        refs: Vec<coven_protocol::blob::RowBlobRef>,
609    ) -> Result<(), crate::blob::transition::MakeRemoteError> {
610        self.inner
611            .components
612            .make_remote(root_table, root_id, root_label, pin, refs)
613            .await
614    }
615
616    pub async fn make_remote_batch(
617        &self,
618        root_table: &str,
619        roots: Vec<crate::blob::MakeRemoteRoot>,
620        pin: bool,
621    ) -> Result<(), crate::blob::transition::MakeRemoteError> {
622        self.inner
623            .components
624            .make_remote_batch(root_table, roots, pin)
625            .await
626    }
627
628    pub async fn cancel_make_remote(
629        &self,
630        root_table: &str,
631        root_id: &str,
632    ) -> Result<(), crate::blob::transition::MakeRemoteError> {
633        self.inner
634            .components
635            .cancel_make_remote(root_table, root_id)
636            .await
637    }
638
639    pub async fn make_local(
640        &self,
641        root_table: &str,
642        root_id: &str,
643        dest: &std::collections::HashMap<String, std::path::PathBuf>,
644        cancel: &tokio::sync::watch::Receiver<bool>,
645    ) -> Result<(), crate::blob::transition::MakeLocalError> {
646        self.inner
647            .components
648            .make_local(root_table, root_id, dest, cancel)
649            .await
650    }
651
652    /// Send a Circle write command to the loop thread and await its reply. Circle
653    /// writes run on the loop thread so they never interleave with a sync cycle.
654    async fn send_circle_command<T>(
655        &self,
656        command: impl FnOnce(CircleReply<T>) -> SyncCommand,
657    ) -> Result<T, crate::sync::store::CircleOperationError> {
658        let (reply, response) = tokio::sync::oneshot::channel();
659        self.command_tx
660            .send(command(reply))
661            .await
662            .map_err(|_| crate::sync::store::CircleOperationError::CommandChannelClosed)?;
663        response
664            .await
665            .map_err(|_| crate::sync::store::CircleOperationError::ReplyChannelClosed)?
666    }
667
668    pub async fn create_circle(
669        &self,
670        name: &str,
671    ) -> Result<coven_protocol::CircleId, crate::sync::store::CircleOperationError> {
672        let name = name.to_string();
673        self.send_circle_command(|reply| SyncCommand::CreateCircle { name, reply })
674            .await
675    }
676
677    pub async fn rename_circle(
678        &self,
679        circle_id: coven_protocol::CircleId,
680        name: &str,
681    ) -> Result<(), crate::sync::store::CircleOperationError> {
682        let name = name.to_string();
683        self.send_circle_command(|reply| SyncCommand::RenameCircle {
684            circle_id,
685            name,
686            reply,
687        })
688        .await
689    }
690
691    pub async fn add_circle_member(
692        &self,
693        circle_id: coven_protocol::CircleId,
694        member_pubkey: String,
695        role: coven_protocol::CircleRole,
696    ) -> Result<(), crate::sync::store::CircleOperationError> {
697        self.send_circle_command(|reply| SyncCommand::AddCircleMember {
698            circle_id,
699            member_pubkey,
700            role,
701            reply,
702        })
703        .await
704    }
705
706    pub async fn remove_circle_member(
707        &self,
708        circle_id: coven_protocol::CircleId,
709        member_pubkey: String,
710    ) -> Result<coven_protocol::CircleOperationId, crate::sync::store::CircleOperationError> {
711        self.send_circle_command(|reply| SyncCommand::RemoveCircleMember {
712            circle_id,
713            member_pubkey,
714            reply,
715        })
716        .await
717    }
718
719    pub async fn resolve_circle_control(
720        &self,
721        circle_id: coven_protocol::CircleId,
722        chosen: coven_protocol::CircleControlCoord,
723    ) -> Result<(), crate::sync::store::CircleOperationError> {
724        self.send_circle_command(|reply| SyncCommand::ResolveCircleControl {
725            circle_id,
726            chosen,
727            reply,
728        })
729        .await
730    }
731
732    pub async fn cancel_circle_epoch_close(
733        &self,
734        circle_id: coven_protocol::CircleId,
735    ) -> Result<coven_protocol::CircleOperationId, crate::sync::store::CircleOperationError> {
736        self.send_circle_command(|reply| SyncCommand::CancelCircleEpochClose { circle_id, reply })
737            .await
738    }
739
740    pub async fn exclude_circle_close_device(
741        &self,
742        circle_id: coven_protocol::CircleId,
743        excluded_device_id: coven_protocol::StoreDeviceId,
744    ) -> Result<(), crate::sync::store::CircleOperationError> {
745        self.send_circle_command(|reply| SyncCommand::ExcludeCircleCloseDevice {
746            circle_id,
747            excluded_device_id,
748            reply,
749        })
750        .await
751    }
752
753    pub async fn delete_circle(
754        &self,
755        circle_id: coven_protocol::CircleId,
756    ) -> Result<(), crate::sync::store::CircleOperationError> {
757        self.send_circle_command(|reply| SyncCommand::DeleteCircle { circle_id, reply })
758            .await
759    }
760
761    pub async fn retry_circle_operation(
762        &self,
763        operation_id: coven_protocol::CircleOperationId,
764    ) -> Result<(), crate::sync::store::CircleOperationError> {
765        self.send_circle_command(|reply| SyncCommand::RetryCircleOperation {
766            operation_id,
767            reply,
768        })
769        .await
770    }
771
772    pub async fn discard_circle_operation(
773        &self,
774        operation_id: coven_protocol::CircleOperationId,
775    ) -> Result<(), crate::sync::store::CircleOperationError> {
776        self.send_circle_command(|reply| SyncCommand::DiscardCircleOperation {
777            operation_id,
778            reply,
779        })
780        .await
781    }
782
783    /// Clear one reclaim operation's stuck mark, so the journal runs it again.
784    ///
785    /// Runs on the loop thread, so it never lands in the middle of a pass that
786    /// has already read the journal; delivering it also ends the loop's wait,
787    /// so the cycle that re-runs the operation begins straight after.
788    pub async fn retry_stuck_reclaim(
789        &self,
790        operation_id: coven_protocol::store_commit::ObjectHash,
791    ) -> Result<(), RetryStuckReclaimError> {
792        let (reply, response) = tokio::sync::oneshot::channel();
793        self.command_tx
794            .send(SyncCommand::RetryStuckReclaim {
795                operation_id,
796                reply,
797            })
798            .await
799            .map_err(|_| RetryStuckReclaimError::CommandChannelClosed)?;
800        response
801            .await
802            .map_err(|_| RetryStuckReclaimError::ReplyChannelClosed)?
803    }
804
805    /// Inspect a Circle's in-flight epoch close. A read, so it runs directly on the
806    /// components rather than serializing behind the write-command channel.
807    pub async fn circle_close_status(
808        &self,
809        circle_id: coven_protocol::CircleId,
810    ) -> Result<coven_protocol::CircleCloseStatus, crate::sync::store::CircleOperationError> {
811        self.inner.components.circle_close_status(circle_id).await
812    }
813
814    #[cfg(any(test, feature = "test-utils"))]
815    pub fn uses_storage_for_test(
816        &self,
817        expected: &Arc<dyn coven_storage::CloudSyncObjectStorage>,
818    ) -> bool {
819        self.inner.components.uses_storage_for_test(expected)
820    }
821
822    #[cfg(any(test, feature = "test-utils"))]
823    pub fn uses_store_dir_for_test(&self, expected: &StoreDir) -> bool {
824        self.inner.components.uses_store_dir_for_test(expected)
825    }
826
827    #[cfg(any(test, feature = "test-utils"))]
828    pub fn encryption_generation_for_test(&self) -> Option<u64> {
829        self.inner.components.encryption_generation_for_test()
830    }
831
832    #[cfg(any(test, feature = "test-utils"))]
833    pub fn open_sealed_blob_for_test(
834        &self,
835        stored: &[u8],
836        aad_context: &[u8],
837    ) -> Result<
838        (coven_keys::encryption::KeyFingerprint, Vec<u8>),
839        coven_keys::encryption::EncryptionError,
840    > {
841        self.inner
842            .components
843            .open_sealed_blob_for_test(stored, aad_context)
844    }
845
846    #[cfg(any(test, feature = "test-utils"))]
847    pub fn adopt_key_rotation_for_test(
848        &self,
849        encryption: coven_keys::encryption::EncryptionService,
850    ) -> Result<String, coven_keys::keys::KeyError> {
851        self.inner.components.adopt_key_rotation(encryption)
852    }
853}
854
855impl SyncLoopHandleInner {
856    async fn execute_command(&self, command: SyncCommand) {
857        match command {
858            SyncCommand::CreateCircle { name, reply } => {
859                reply_circle_command(reply, self.components.create_circle(&name).await);
860            }
861            SyncCommand::RenameCircle {
862                circle_id,
863                name,
864                reply,
865            } => {
866                reply_circle_command(reply, self.components.rename_circle(circle_id, &name).await);
867            }
868            SyncCommand::AddCircleMember {
869                circle_id,
870                member_pubkey,
871                role,
872                reply,
873            } => {
874                reply_circle_command(
875                    reply,
876                    self.components
877                        .add_circle_member(circle_id, member_pubkey, role)
878                        .await,
879                );
880            }
881            SyncCommand::RemoveCircleMember {
882                circle_id,
883                member_pubkey,
884                reply,
885            } => {
886                reply_circle_command(
887                    reply,
888                    self.components
889                        .remove_circle_member(circle_id, member_pubkey)
890                        .await,
891                );
892            }
893            SyncCommand::ResolveCircleControl {
894                circle_id,
895                chosen,
896                reply,
897            } => {
898                reply_circle_command(
899                    reply,
900                    self.components
901                        .resolve_circle_control(circle_id, chosen)
902                        .await,
903                );
904            }
905            SyncCommand::CancelCircleEpochClose { circle_id, reply } => {
906                reply_circle_command(
907                    reply,
908                    self.components.cancel_circle_epoch_close(circle_id).await,
909                );
910            }
911            SyncCommand::ExcludeCircleCloseDevice {
912                circle_id,
913                excluded_device_id,
914                reply,
915            } => {
916                reply_circle_command(
917                    reply,
918                    self.components
919                        .exclude_circle_close_device(circle_id, excluded_device_id)
920                        .await,
921                );
922            }
923            SyncCommand::DeleteCircle { circle_id, reply } => {
924                reply_circle_command(reply, self.components.delete_circle(circle_id).await);
925            }
926            SyncCommand::RetryCircleOperation {
927                operation_id,
928                reply,
929            } => {
930                reply_circle_command(
931                    reply,
932                    self.components.retry_circle_operation(&operation_id).await,
933                );
934            }
935            SyncCommand::DiscardCircleOperation {
936                operation_id,
937                reply,
938            } => {
939                reply_circle_command(
940                    reply,
941                    self.components
942                        .discard_circle_operation(&operation_id)
943                        .await,
944                );
945            }
946            SyncCommand::RetryStuckReclaim {
947                operation_id,
948                reply,
949            } => {
950                let result = self
951                    .components
952                    .retry_stuck_reclaim(operation_id)
953                    .await
954                    .map_err(|error| RetryStuckReclaimError::Database(Box::new(error)));
955                if reply.send(result).is_err() {
956                    debug!("stuck reclaim retry caller dropped its reply receiver");
957                }
958            }
959        }
960    }
961
962    async fn run_single_cycle(
963        &self,
964    ) -> Result<super::cycle::SyncCycleResult, super::cycle::SyncCycleFailure> {
965        self.components
966            .run_cycle(
967                self.clock.as_ref(),
968                self.observer.as_deref(),
969                self.config.snapshot_commit_threshold,
970            )
971            .await
972    }
973}
974
975fn reply_circle_command<T>(
976    reply: CircleReply<T>,
977    result: Result<T, crate::sync::store::CircleOperationError>,
978) {
979    if reply.send(result).is_err() {
980        debug!("Circle command caller dropped its reply receiver");
981    }
982}
983
984#[cfg(test)]
985#[path = "sync_loop_tests.rs"]
986mod tests;
coven_replication/sync/sync_loop.rs

coven_replication/sync/
sync_loop.rs
