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 membership_conflict(
376        &self,
377    ) -> Result<Option<coven_protocol::MembershipConflictInfo>, super::store::MembershipOpsError>
378    {
379        self.inner.components.membership_conflict().await
380    }
381
382    pub async fn restore_membership(
383        &self,
384    ) -> Result<super::store::authorization::StoreRestoreMembership, super::store::MembershipOpsError>
385    {
386        self.inner.components.restore_membership().await
387    }
388
389    pub fn host_write_blob_staging(
390        &self,
391        runtime: tokio::runtime::Handle,
392    ) -> crate::sync::store::HostWriteBlobStaging {
393        self.inner.components.host_write_blob_staging(runtime)
394    }
395
396    pub async fn propose_device_exclusion(
397        &self,
398        device_id: coven_protocol::StoreDeviceId,
399    ) -> Result<
400        coven_protocol::store_commit::StoreDeviceExclusionProposalRef,
401        crate::sync::store::StoreDeviceExclusionError,
402    > {
403        self.inner
404            .components
405            .propose_device_exclusion(device_id)
406            .await
407    }
408
409    pub async fn cancel_device_exclusion(
410        &self,
411        proposal: &coven_protocol::store_commit::StoreDeviceExclusionProposalRef,
412    ) -> Result<(), crate::sync::store::StoreDeviceExclusionError> {
413        self.inner
414            .components
415            .cancel_device_exclusion(proposal)
416            .await
417    }
418
419    pub async fn finalize_device_exclusion(
420        &self,
421        proposal: &coven_protocol::store_commit::StoreDeviceExclusionProposalRef,
422    ) -> Result<(), crate::sync::store::StoreDeviceExclusionError> {
423        self.inner
424            .components
425            .finalize_device_exclusion(proposal)
426            .await
427    }
428
429    pub async fn begin_owner_promotion(
430        &self,
431        device_id: coven_protocol::StoreDeviceId,
432    ) -> Result<
433        coven_protocol::store_commit::OwnerPromotionRequest,
434        crate::sync::store::OwnerPromotionError,
435    > {
436        self.inner.components.begin_owner_promotion(device_id).await
437    }
438
439    pub async fn accept_owner_promotion(
440        &self,
441        request: coven_protocol::store_commit::OwnerPromotionRequest,
442    ) -> Result<
443        coven_protocol::store_commit::OwnerPromotionAcceptance,
444        crate::sync::store::OwnerPromotionError,
445    > {
446        self.inner.components.accept_owner_promotion(request).await
447    }
448
449    pub async fn finalize_owner_promotion(
450        &self,
451        acceptance: coven_protocol::store_commit::OwnerPromotionAcceptance,
452    ) -> Result<(), crate::sync::store::OwnerPromotionError> {
453        self.inner
454            .components
455            .finalize_owner_promotion(acceptance)
456            .await
457    }
458
459    pub async fn begin_device_join_bundle(
460        &self,
461        member_pubkey: &str,
462    ) -> Result<crate::sync::DeviceJoinOfferBundle, crate::sync::store::DeviceJoinTransportError>
463    {
464        self.inner
465            .components
466            .begin_device_join_bundle(member_pubkey)
467            .await
468    }
469
470    pub async fn drive_device_join(
471        &self,
472        bundle: &crate::sync::DeviceJoinOfferBundle,
473        policy: crate::sync::DeviceJoinApprovalPolicy<'_>,
474        access_administrator: Option<&dyn crate::sync::DeviceProviderAccessAdministrator>,
475        on_progress: &(dyn Fn(crate::sync::AdmittingDeviceJoinProgress) + Send + Sync),
476        timing: crate::sync::DeviceJoinTransportTiming,
477    ) -> Result<crate::sync::DeviceJoinDriveOutcome, crate::sync::store::DeviceJoinTransportError>
478    {
479        self.inner
480            .components
481            .drive_device_join(bundle, policy, access_administrator, on_progress, timing)
482            .await
483    }
484
485    pub async fn abandon_device_join_transport(
486        &self,
487        bundle: &crate::sync::DeviceJoinOfferBundle,
488    ) -> Result<crate::sync::DeviceJoinAbandonment, crate::sync::store::DeviceJoinTransportError>
489    {
490        self.inner
491            .components
492            .abandon_device_join_transport(bundle)
493            .await
494    }
495
496    pub async fn abort_device_join_transport(
497        &self,
498        bundle: &crate::sync::DeviceJoinOfferBundle,
499    ) -> Result<(), crate::sync::store::DeviceJoinTransportError> {
500        self.inner
501            .components
502            .abort_device_join_transport(bundle)
503            .await
504    }
505
506    pub async fn begin_device_join(
507        &self,
508        member_pubkey: &str,
509    ) -> Result<crate::sync::DeviceJoinOffer, crate::sync::DeviceJoinError> {
510        self.inner.components.begin_device_join(member_pubkey).await
511    }
512
513    pub async fn abandon_device_join(
514        &self,
515        offer: crate::sync::DeviceJoinOffer,
516    ) -> Result<crate::sync::DeviceJoinAbandonment, crate::sync::DeviceJoinError> {
517        self.inner.components.abandon_device_join(offer).await
518    }
519
520    pub async fn authorize_device_provider_access(
521        &self,
522        request: crate::sync::DeviceProviderAccessRequest,
523        access_administrator: Option<&dyn crate::sync::DeviceProviderAccessAdministrator>,
524    ) -> Result<crate::sync::DeviceProviderAdmissionApproval, crate::sync::DeviceJoinError> {
525        self.inner
526            .components
527            .authorize_device_provider_access(request, access_administrator)
528            .await
529    }
530
531    pub async fn accept_device_registration(
532        &self,
533        request: crate::sync::DeviceRegistrationRequest,
534    ) -> Result<crate::sync::ProvisionalDeviceBootstrap, crate::sync::DeviceJoinError> {
535        self.inner
536            .components
537            .accept_device_registration(request)
538            .await
539    }
540
541    pub async fn publish_device_provider_challenge(
542        &self,
543        bootstrap: crate::sync::ProvisionalDeviceBootstrap,
544    ) -> Result<crate::sync::ProviderReadyDeviceBootstrap, crate::sync::DeviceJoinError> {
545        self.inner
546            .components
547            .publish_device_provider_challenge(bootstrap)
548            .await
549    }
550
551    pub async fn complete_device_provider_admission(
552        &self,
553        readiness: crate::sync::DeviceJoinReadiness,
554    ) -> Result<crate::sync::DeviceProviderAdmissionCompletion, crate::sync::DeviceJoinError> {
555        self.inner
556            .components
557            .complete_device_provider_admission(readiness)
558            .await
559    }
560
561    pub async fn finalize_device_join(
562        &self,
563        completion: crate::sync::DeviceProviderAdmissionCompletion,
564    ) -> Result<crate::sync::DeviceJoinActivation, crate::sync::DeviceJoinError> {
565        self.inner.components.finalize_device_join(completion).await
566    }
567
568    pub fn config(&self) -> &Config {
569        &self.inner.config
570    }
571
572    pub fn blob_path_scheme(&self) -> BlobPathScheme {
573        self.inner.components.blob_path_scheme()
574    }
575
576    pub fn is_encrypted(&self) -> bool {
577        self.inner.components.is_encrypted()
578    }
579
580    pub async fn admit_member(
581        &self,
582        public_key_hex: &str,
583        member_email: Option<&str>,
584        role: coven_protocol::membership::MemberRole,
585        store_name: &str,
586    ) -> Result<crate::sync::store::MemberAdmission, super::store::MembershipOpsError> {
587        self.inner
588            .components
589            .admit_member(public_key_hex, member_email, role, store_name)
590            .await
591    }
592
593    pub async fn remove_member(
594        &self,
595        public_key_hex: &str,
596    ) -> Result<String, super::store::MembershipOpsError> {
597        self.inner.components.remove_member(public_key_hex).await
598    }
599
600    pub async fn resolve_membership_conflict(
601        &self,
602        choice: &coven_protocol::MembershipConflictChoice,
603    ) -> Result<(), super::store::MembershipOpsError> {
604        self.inner
605            .components
606            .resolve_membership_conflict(choice)
607            .await
608    }
609
610    pub async fn drain_uploads(
611        &self,
612    ) -> Result<crate::blob::DrainOutcome, super::store::StoreError> {
613        self.inner
614            .components
615            .drain_uploads(self.inner.clock.as_ref(), self.inner.observer.as_deref())
616            .await
617    }
618
619    pub async fn make_remote(
620        &self,
621        root_table: &str,
622        root_id: &str,
623        root_label: &str,
624        pin: bool,
625        refs: Vec<coven_protocol::blob::RowBlobRef>,
626    ) -> Result<(), crate::blob::transition::MakeRemoteError> {
627        self.inner
628            .components
629            .make_remote(root_table, root_id, root_label, pin, refs)
630            .await
631    }
632
633    pub async fn make_remote_batch(
634        &self,
635        root_table: &str,
636        roots: Vec<crate::blob::MakeRemoteRoot>,
637        pin: bool,
638    ) -> Result<(), crate::blob::transition::MakeRemoteError> {
639        self.inner
640            .components
641            .make_remote_batch(root_table, roots, pin)
642            .await
643    }
644
645    pub async fn cancel_make_remote(
646        &self,
647        root_table: &str,
648        root_id: &str,
649    ) -> Result<(), crate::blob::transition::MakeRemoteError> {
650        self.inner
651            .components
652            .cancel_make_remote(root_table, root_id)
653            .await
654    }
655
656    pub async fn make_local(
657        &self,
658        root_table: &str,
659        root_id: &str,
660        dest: &std::collections::HashMap<String, std::path::PathBuf>,
661        cancel: &tokio::sync::watch::Receiver<bool>,
662    ) -> Result<(), crate::blob::transition::MakeLocalError> {
663        self.inner
664            .components
665            .make_local(root_table, root_id, dest, cancel)
666            .await
667    }
668
669    /// Send a Circle write command to the loop thread and await its reply. Circle
670    /// writes run on the loop thread so they never interleave with a sync cycle.
671    async fn send_circle_command<T>(
672        &self,
673        command: impl FnOnce(CircleReply<T>) -> SyncCommand,
674    ) -> Result<T, crate::sync::store::CircleOperationError> {
675        let (reply, response) = tokio::sync::oneshot::channel();
676        self.command_tx
677            .send(command(reply))
678            .await
679            .map_err(|_| crate::sync::store::CircleOperationError::CommandChannelClosed)?;
680        response
681            .await
682            .map_err(|_| crate::sync::store::CircleOperationError::ReplyChannelClosed)?
683    }
684
685    pub async fn create_circle(
686        &self,
687        name: &str,
688    ) -> Result<coven_protocol::CircleId, crate::sync::store::CircleOperationError> {
689        let name = name.to_string();
690        self.send_circle_command(|reply| SyncCommand::CreateCircle { name, reply })
691            .await
692    }
693
694    pub async fn rename_circle(
695        &self,
696        circle_id: coven_protocol::CircleId,
697        name: &str,
698    ) -> Result<(), crate::sync::store::CircleOperationError> {
699        let name = name.to_string();
700        self.send_circle_command(|reply| SyncCommand::RenameCircle {
701            circle_id,
702            name,
703            reply,
704        })
705        .await
706    }
707
708    pub async fn add_circle_member(
709        &self,
710        circle_id: coven_protocol::CircleId,
711        member_pubkey: String,
712        role: coven_protocol::CircleRole,
713    ) -> Result<(), crate::sync::store::CircleOperationError> {
714        self.send_circle_command(|reply| SyncCommand::AddCircleMember {
715            circle_id,
716            member_pubkey,
717            role,
718            reply,
719        })
720        .await
721    }
722
723    pub async fn remove_circle_member(
724        &self,
725        circle_id: coven_protocol::CircleId,
726        member_pubkey: String,
727    ) -> Result<coven_protocol::CircleOperationId, crate::sync::store::CircleOperationError> {
728        self.send_circle_command(|reply| SyncCommand::RemoveCircleMember {
729            circle_id,
730            member_pubkey,
731            reply,
732        })
733        .await
734    }
735
736    pub async fn resolve_circle_control(
737        &self,
738        circle_id: coven_protocol::CircleId,
739        chosen: coven_protocol::CircleControlCoord,
740    ) -> Result<(), crate::sync::store::CircleOperationError> {
741        self.send_circle_command(|reply| SyncCommand::ResolveCircleControl {
742            circle_id,
743            chosen,
744            reply,
745        })
746        .await
747    }
748
749    pub async fn cancel_circle_epoch_close(
750        &self,
751        circle_id: coven_protocol::CircleId,
752    ) -> Result<coven_protocol::CircleOperationId, crate::sync::store::CircleOperationError> {
753        self.send_circle_command(|reply| SyncCommand::CancelCircleEpochClose { circle_id, reply })
754            .await
755    }
756
757    pub async fn exclude_circle_close_device(
758        &self,
759        circle_id: coven_protocol::CircleId,
760        excluded_device_id: coven_protocol::StoreDeviceId,
761    ) -> Result<(), crate::sync::store::CircleOperationError> {
762        self.send_circle_command(|reply| SyncCommand::ExcludeCircleCloseDevice {
763            circle_id,
764            excluded_device_id,
765            reply,
766        })
767        .await
768    }
769
770    pub async fn delete_circle(
771        &self,
772        circle_id: coven_protocol::CircleId,
773    ) -> Result<(), crate::sync::store::CircleOperationError> {
774        self.send_circle_command(|reply| SyncCommand::DeleteCircle { circle_id, reply })
775            .await
776    }
777
778    pub async fn retry_circle_operation(
779        &self,
780        operation_id: coven_protocol::CircleOperationId,
781    ) -> Result<(), crate::sync::store::CircleOperationError> {
782        self.send_circle_command(|reply| SyncCommand::RetryCircleOperation {
783            operation_id,
784            reply,
785        })
786        .await
787    }
788
789    pub async fn discard_circle_operation(
790        &self,
791        operation_id: coven_protocol::CircleOperationId,
792    ) -> Result<(), crate::sync::store::CircleOperationError> {
793        self.send_circle_command(|reply| SyncCommand::DiscardCircleOperation {
794            operation_id,
795            reply,
796        })
797        .await
798    }
799
800    /// Clear one reclaim operation's stuck mark, so the journal runs it again.
801    ///
802    /// Runs on the loop thread, so it never lands in the middle of a pass that
803    /// has already read the journal; delivering it also ends the loop's wait,
804    /// so the cycle that re-runs the operation begins straight after.
805    pub async fn retry_stuck_reclaim(
806        &self,
807        operation_id: coven_protocol::store_commit::ObjectHash,
808    ) -> Result<(), RetryStuckReclaimError> {
809        let (reply, response) = tokio::sync::oneshot::channel();
810        self.command_tx
811            .send(SyncCommand::RetryStuckReclaim {
812                operation_id,
813                reply,
814            })
815            .await
816            .map_err(|_| RetryStuckReclaimError::CommandChannelClosed)?;
817        response
818            .await
819            .map_err(|_| RetryStuckReclaimError::ReplyChannelClosed)?
820    }
821
822    /// Inspect a Circle's in-flight epoch close. A read, so it runs directly on the
823    /// components rather than serializing behind the write-command channel.
824    pub async fn circle_close_status(
825        &self,
826        circle_id: coven_protocol::CircleId,
827    ) -> Result<coven_protocol::CircleCloseStatus, crate::sync::store::CircleOperationError> {
828        self.inner.components.circle_close_status(circle_id).await
829    }
830
831    #[cfg(any(test, feature = "test-utils"))]
832    pub fn uses_storage_for_test(
833        &self,
834        expected: &Arc<dyn coven_storage::CloudSyncObjectStorage>,
835    ) -> bool {
836        self.inner.components.uses_storage_for_test(expected)
837    }
838
839    #[cfg(any(test, feature = "test-utils"))]
840    pub fn uses_store_dir_for_test(&self, expected: &StoreDir) -> bool {
841        self.inner.components.uses_store_dir_for_test(expected)
842    }
843
844    #[cfg(any(test, feature = "test-utils"))]
845    pub fn encryption_generation_for_test(&self) -> Option<u64> {
846        self.inner.components.encryption_generation_for_test()
847    }
848
849    #[cfg(any(test, feature = "test-utils"))]
850    pub fn open_sealed_blob_for_test(
851        &self,
852        stored: &[u8],
853        aad_context: &[u8],
854    ) -> Result<
855        (coven_keys::encryption::KeyFingerprint, Vec<u8>),
856        coven_keys::encryption::EncryptionError,
857    > {
858        self.inner
859            .components
860            .open_sealed_blob_for_test(stored, aad_context)
861    }
862
863    #[cfg(any(test, feature = "test-utils"))]
864    pub fn adopt_key_rotation_for_test(
865        &self,
866        encryption: coven_keys::encryption::EncryptionService,
867    ) -> Result<String, coven_keys::keys::KeyError> {
868        self.inner.components.adopt_key_rotation(encryption)
869    }
870}
871
872impl SyncLoopHandleInner {
873    async fn execute_command(&self, command: SyncCommand) {
874        match command {
875            SyncCommand::CreateCircle { name, reply } => {
876                reply_circle_command(reply, self.components.create_circle(&name).await);
877            }
878            SyncCommand::RenameCircle {
879                circle_id,
880                name,
881                reply,
882            } => {
883                reply_circle_command(reply, self.components.rename_circle(circle_id, &name).await);
884            }
885            SyncCommand::AddCircleMember {
886                circle_id,
887                member_pubkey,
888                role,
889                reply,
890            } => {
891                reply_circle_command(
892                    reply,
893                    self.components
894                        .add_circle_member(circle_id, member_pubkey, role)
895                        .await,
896                );
897            }
898            SyncCommand::RemoveCircleMember {
899                circle_id,
900                member_pubkey,
901                reply,
902            } => {
903                reply_circle_command(
904                    reply,
905                    self.components
906                        .remove_circle_member(circle_id, member_pubkey)
907                        .await,
908                );
909            }
910            SyncCommand::ResolveCircleControl {
911                circle_id,
912                chosen,
913                reply,
914            } => {
915                reply_circle_command(
916                    reply,
917                    self.components
918                        .resolve_circle_control(circle_id, chosen)
919                        .await,
920                );
921            }
922            SyncCommand::CancelCircleEpochClose { circle_id, reply } => {
923                reply_circle_command(
924                    reply,
925                    self.components.cancel_circle_epoch_close(circle_id).await,
926                );
927            }
928            SyncCommand::ExcludeCircleCloseDevice {
929                circle_id,
930                excluded_device_id,
931                reply,
932            } => {
933                reply_circle_command(
934                    reply,
935                    self.components
936                        .exclude_circle_close_device(circle_id, excluded_device_id)
937                        .await,
938                );
939            }
940            SyncCommand::DeleteCircle { circle_id, reply } => {
941                reply_circle_command(reply, self.components.delete_circle(circle_id).await);
942            }
943            SyncCommand::RetryCircleOperation {
944                operation_id,
945                reply,
946            } => {
947                reply_circle_command(
948                    reply,
949                    self.components.retry_circle_operation(&operation_id).await,
950                );
951            }
952            SyncCommand::DiscardCircleOperation {
953                operation_id,
954                reply,
955            } => {
956                reply_circle_command(
957                    reply,
958                    self.components
959                        .discard_circle_operation(&operation_id)
960                        .await,
961                );
962            }
963            SyncCommand::RetryStuckReclaim {
964                operation_id,
965                reply,
966            } => {
967                let result = self
968                    .components
969                    .retry_stuck_reclaim(operation_id)
970                    .await
971                    .map_err(|error| RetryStuckReclaimError::Database(Box::new(error)));
972                if reply.send(result).is_err() {
973                    debug!("stuck reclaim retry caller dropped its reply receiver");
974                }
975            }
976        }
977    }
978
979    async fn run_single_cycle(
980        &self,
981    ) -> Result<super::cycle::SyncCycleResult, super::cycle::SyncCycleFailure> {
982        self.components
983            .run_cycle(self.clock.as_ref(), self.observer.as_deref())
984            .await
985    }
986}
987
988fn reply_circle_command<T>(
989    reply: CircleReply<T>,
990    result: Result<T, crate::sync::store::CircleOperationError>,
991) {
992    if reply.send(result).is_err() {
993        debug!("Circle command caller dropped its reply receiver");
994    }
995}
996
997#[cfg(test)]
998#[path = "sync_loop_tests.rs"]
999mod tests;
coven_replication/sync/sync_loop.rs

coven_replication/sync/
sync_loop.rs
