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. [`Conflict`](Self::Conflict)
51/// and [`Blocked`](Self::Blocked) are successful storage cycles with unresolved
52/// durable writes; [`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`, `Conflict`, 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 the Serial branch was based on
76    /// an older global head and require explicit discard or replacement.
77    Conflict {
78        success: SyncLoopSuccess,
79        branch: crate::PendingBranch,
80    },
81    /// The cycle reached storage, but one or more writes cannot publish until
82    /// their named prerequisite is supplied or repaired.
83    Blocked {
84        success: SyncLoopSuccess,
85        writes: Vec<crate::PendingWrite>,
86    },
87    /// The cycle failed as a whole — no outcome to report, only the fault.
88    Failed { error: String },
89}
90
91/// Manages the background sync loop and provides access to sync components.
92pub(crate) struct SyncLoopHandle {
93    inner: Arc<SyncLoopInner>,
94    clock: ClockRef,
95    config: Config,
96    store_dir: StoreDir,
97    trigger_tx: tokio::sync::mpsc::Sender<()>,
98    trigger_rx: std::sync::Mutex<Option<tokio::sync::mpsc::Receiver<()>>>,
99    command_tx: tokio::sync::mpsc::Sender<SyncCommand>,
100    command_rx: std::sync::Mutex<Option<tokio::sync::mpsc::Receiver<SyncCommand>>>,
101    stop_tx: tokio::sync::watch::Sender<bool>,
102    stop_rx: std::sync::Mutex<Option<tokio::sync::watch::Receiver<bool>>>,
103    /// The current status value, owned by the [`CovenHandle`] and cloned into each
104    /// loop it starts, so a subscription survives a loop restart (a reconnect
105    /// builds a fresh loop but keeps this same sender). The loop only sends here.
106    status_tx: tokio::sync::watch::Sender<SyncLoopStatus>,
107    thread_handle: std::sync::Mutex<Option<std::thread::JoinHandle<()>>>,
108    running: Arc<AtomicBool>,
109}
110
111struct SyncLoopInner {
112    components: SyncComponents,
113    custody: Arc<dyn MasterKeyCustody>,
114    observer: Option<Arc<dyn BlobTransitionObserver>>,
115
116    /// The store-directory lock, held so it releases only when the loop's
117    /// thread exits. The running thread keeps a clone of this `SyncLoopInner`
118    /// alive across its whole cycle, so the last handle dropping never releases
119    /// `.coven-lock` while a mid-cycle pull or upload is still writing — a second
120    /// `open()` of the same store stays refused until this writer is gone.
121    _open_guard: Arc<StoreOpenGuard>,
122}
123
124enum SyncCommand {
125    CreateCircle {
126        name: String,
127        reply: tokio::sync::oneshot::Sender<
128            Result<crate::CircleId, crate::sync::circle_ops::CircleOperationError>,
129        >,
130    },
131}
132
133impl SyncLoopHandle {
134    pub(crate) fn new(
135        components: SyncComponents,
136        custody: Arc<dyn MasterKeyCustody>,
137        clock: ClockRef,
138        config: Config,
139        observer: Option<Arc<dyn BlobTransitionObserver>>,
140        open_guard: Arc<StoreOpenGuard>,
141        status_tx: tokio::sync::watch::Sender<SyncLoopStatus>,
142    ) -> Self {
143        let (trigger_tx, trigger_rx) = tokio::sync::mpsc::channel(1);
144        let (command_tx, command_rx) = tokio::sync::mpsc::channel(16);
145        let (stop_tx, stop_rx) = tokio::sync::watch::channel(false);
146        let store_dir = config.store_dir.clone();
147
148        Self {
149            inner: Arc::new(SyncLoopInner {
150                components,
151                custody,
152                observer,
153                _open_guard: open_guard,
154            }),
155            clock,
156            config,
157            store_dir,
158            trigger_tx,
159            trigger_rx: std::sync::Mutex::new(Some(trigger_rx)),
160            command_tx,
161            command_rx: std::sync::Mutex::new(Some(command_rx)),
162            stop_tx,
163            stop_rx: std::sync::Mutex::new(Some(stop_rx)),
164            status_tx,
165            thread_handle: std::sync::Mutex::new(None),
166            running: Arc::new(AtomicBool::new(false)),
167        }
168    }
169
170    /// Start the background sync loop on a dedicated OS thread. No-op if already
171    /// running.
172    ///
173    /// The thread runs a current-thread tokio runtime that `block_on`s the loop.
174    /// A dedicated thread (rather than `tokio::spawn` on the host runtime) is for
175    /// the *stack*: aws-sdk-s3's endpoint/auth resolution recurses deeply enough
176    /// to overflow the default secondary-thread stack in debug builds (SIGBUS in
177    /// `resolve_endpoint`), so the thread is given a main-thread-sized stack.
178    /// Everything the loop holds is `Send`; database access goes through async
179    /// calls on the [`Database`] handle, so nothing here is bound to this thread
180    /// except by choice of stack size.
181    pub(crate) fn start(&self) -> Result<(), SyncLoopError> {
182        if self.running.swap(true, Ordering::AcqRel) {
183            return Ok(());
184        }
185
186        let mut trigger_rx = match self.trigger_rx.lock().unwrap().take() {
187            Some(rx) => rx,
188            None => {
189                self.running.store(false, Ordering::Release);
190                return Err(SyncLoopError::NotRestartable);
191            }
192        };
193        let mut stop_rx = match self.stop_rx.lock().unwrap().take() {
194            Some(rx) => rx,
195            None => {
196                self.running.store(false, Ordering::Release);
197                return Err(SyncLoopError::NotRestartable);
198            }
199        };
200        let mut command_rx = match self.command_rx.lock().unwrap().take() {
201            Some(rx) => rx,
202            None => {
203                self.running.store(false, Ordering::Release);
204                return Err(SyncLoopError::NotRestartable);
205            }
206        };
207
208        let inner = Arc::clone(&self.inner);
209        let status_tx = self.status_tx.clone();
210        let clock = self.clock.clone();
211        let store_dir = self.store_dir.clone();
212        let running = Arc::clone(&self.running);
213
214        let handle = std::thread::Builder::new()
215            .name("coven-sync-loop".to_string())
216            // aws-sdk-s3's endpoint/auth resolution recurses deeply enough to blow
217            // the ~2 MiB default secondary-thread stack in debug builds (SIGBUS in
218            // resolve_endpoint). Give this thread a main-thread-sized stack so S3
219            // sync doesn't overflow it.
220            .stack_size(8 * 1024 * 1024)
221            .spawn(move || {
222                let _running_guard = RunningGuard {
223                    running: Arc::clone(&running),
224                };
225                let rt = match tokio::runtime::Builder::new_current_thread()
226                    .enable_all()
227                    .build()
228                {
229                    Ok(rt) => rt,
230                    Err(e) => {
231                        let error = format!("failed to create sync loop runtime: {e}");
232                        error!("{error}");
233                        status_tx.send_replace(SyncLoopStatus::Failed { error });
234                        return;
235                    }
236                };
237
238                rt.block_on(async move {
239                    // Short delay to avoid racing with app startup.
240                    let startup_delay = tokio::time::sleep(Duration::from_secs(3));
241                    tokio::pin!(startup_delay);
242                    loop {
243                        tokio::select! {
244                            _ = &mut startup_delay => break,
245                            changed = stop_rx.changed() => {
246                                if changed.is_err() || *stop_rx.borrow() {
247                                    info!("Sync loop stopped before first cycle");
248                                    return;
249                                }
250                            }
251                            command = command_rx.recv() => {
252                                let Some(command) = command else {
253                                    info!("Sync command channel closed before first cycle");
254                                    return;
255                                };
256                                execute_command(&inner, command).await;
257                            }
258                        }
259                    }
260
261                    let mut consecutive_failures: u32 = 0;
262                    while running.load(Ordering::Acquire) && !*stop_rx.borrow() {
263                        status_tx.send_replace(SyncLoopStatus::CheckingStorage);
264                        let reachable = inner
265                            .components
266                            .storage()
267                            .cloud_home()
268                            .probe()
269                            .await
270                            .map_err(crate::sync::storage::StorageError::from);
271                        let (decision, status) = match reachable {
272                            Err(error) => {
273                                let status = storage_check_failure_status(&error);
274                                let decision = loop_policy::after_failure(
275                                    format!("check sync storage: {error}"),
276                                    consecutive_failures,
277                                    300,
278                                );
279                                (decision, status)
280                            }
281                            Ok(_) => {
282                                status_tx.send_replace(SyncLoopStatus::Publishing);
283                                let (decision, cycle_went_offline) = match run_single_cycle(&inner, clock.as_ref(), &store_dir).await {
284                                    Ok(result) => (loop_policy::after_success(result), false),
285                                    Err(error) => {
286                                        let offline = error.is_offline();
287                                        (
288                                            loop_policy::after_failure(
289                                                error.to_string(),
290                                                consecutive_failures,
291                                                300,
292                                            ),
293                                            offline,
294                                        )
295                                    }
296                                };
297                                let status = match &decision.report {
298                                    SyncLoopReport::Success(success) => {
299                                        match current_success_status(inner.components.database(), success.clone()).await {
300                                            Ok(status) => status,
301                                            Err(error) => SyncLoopStatus::Failed { error },
302                                        }
303                                    }
304                                    SyncLoopReport::Failure(_) if cycle_went_offline => {
305                                        SyncLoopStatus::Offline
306                                    }
307                                    SyncLoopReport::Failure(error) => SyncLoopStatus::Failed {
308                                        error: error.clone(),
309                                    },
310                                };
311                                (decision, status)
312                            }
313                        };
314                        consecutive_failures = decision.consecutive_failures;
315                        status_tx.send_replace(status);
316
317                        let wait = match decision.wait {
318                            LoopWait::Immediate => Duration::ZERO,
319                            LoopWait::Idle => Duration::from_secs(super::backoff::backoff_secs(0, 300)),
320                            LoopWait::BackoffSecs(secs) => Duration::from_secs(secs),
321                        };
322                        if matches!(decision.wait, LoopWait::BackoffSecs(_)) {
323                            debug!(
324                                "Backing off {wait:?} after {consecutive_failures} consecutive failure(s)",
325                            );
326                        }
327                        tokio::select! {
328                            _ = tokio::time::sleep(wait) => {}
329                            changed = stop_rx.changed() => {
330                                if changed.is_err() || *stop_rx.borrow() {
331                                    info!("Sync loop stop requested");
332                                    break;
333                                }
334                            }
335                            msg = trigger_rx.recv() => {
336                                if msg.is_none() {
337                                    info!("Sync trigger channel closed, stopping sync loop");
338                                    break;
339                                }
340                            }
341                            command = command_rx.recv() => {
342                                let Some(command) = command else {
343                                    info!("Sync command channel closed, stopping sync loop");
344                                    break;
345                                };
346                                execute_command(&inner, command).await;
347                            }
348                        }
349                    }
350                });
351            })
352            .map_err(|e| {
353                self.running.store(false, Ordering::Release);
354                SyncLoopError::ThreadSpawn(e)
355            })?;
356
357        *self.thread_handle.lock().unwrap() = Some(handle);
358        Ok(())
359    }
360
361    /// Whether the background sync thread is running.
362    pub(crate) fn is_running(&self) -> bool {
363        self.running.load(Ordering::Acquire)
364    }
365
366    /// Request loop shutdown and join the sync thread.
367    pub(crate) fn stop(&self) -> Result<(), SyncLoopError> {
368        let handle = {
369            let mut guard = self.thread_handle.lock().unwrap();
370            if guard.is_none() && !self.running.load(Ordering::Acquire) {
371                return Ok(());
372            }
373            if self.stop_tx.send(true).is_err() {
374                debug!("sync loop stop requested after stop receiver closed");
375            }
376            self.trigger();
377            guard.take()
378        };
379
380        if let Some(handle) = handle {
381            if handle.join().is_err() {
382                self.running.store(false, Ordering::Release);
383                return Err(SyncLoopError::ThreadPanicked);
384            }
385        }
386        self.running.store(false, Ordering::Release);
387        Ok(())
388    }
389
390    /// Signal the sync loop to run a cycle immediately.
391    ///
392    /// `Full` means a trigger is already pending — our request collapses into the
393    /// existing one, which is exactly what the capacity-1 channel is for.
394    /// `Closed` means the loop is gone, so the trigger is moot.
395    pub(crate) fn trigger(&self) {
396        match self.trigger_tx.try_send(()) {
397            Ok(()) | Err(TrySendError::Full(())) => {}
398            Err(TrySendError::Closed(())) => {
399                debug!("Sync trigger channel closed, loop is not running");
400            }
401        }
402    }
403
404    // -- Accessors for membership operations --
405
406    pub(crate) fn storage(&self) -> &Arc<CloudSyncStorage> {
407        self.inner.components.storage()
408    }
409
410    pub(crate) fn store_dir(&self) -> &StoreDir {
411        &self.store_dir
412    }
413
414    pub(crate) fn config(&self) -> &Config {
415        &self.config
416    }
417
418    pub(crate) fn blob_path_scheme(&self) -> BlobPathScheme {
419        self.inner.components.blob_path_scheme()
420    }
421
422    pub(crate) fn self_uploader(&self) -> String {
423        self.inner.components.self_uploader()
424    }
425
426    pub(crate) fn hlc(&self) -> &Arc<Hlc> {
427        self.inner.components.hlc()
428    }
429
430    pub(crate) fn current_encryption(&self) -> Option<crate::encryption::EncryptionService> {
431        self.inner.components.current_encryption()
432    }
433
434    pub(crate) async fn invite_member(
435        &self,
436        public_key_hex: &str,
437        invitee_email: Option<&str>,
438        role: super::membership::MemberRole,
439        store_name: &str,
440    ) -> Result<crate::join_code::InviteCode, super::membership_ops::MembershipOpsError> {
441        self.inner
442            .components
443            .invite_member(public_key_hex, invitee_email, role, store_name)
444            .await
445    }
446
447    pub(crate) async fn remove_member(
448        &self,
449        public_key_hex: &str,
450    ) -> Result<String, super::membership_ops::MembershipOpsError> {
451        self.inner
452            .components
453            .remove_member(public_key_hex, self.inner.custody.as_ref())
454            .await
455    }
456
457    pub(crate) async fn persist_pending_rotation(&self) -> Result<(), crate::database::DbError> {
458        self.inner.components.persist_pending_rotation().await
459    }
460
461    #[cfg(test)]
462    pub(crate) fn adopt_key_rotation_for_test(
463        &self,
464        encryption: crate::encryption::EncryptionService,
465    ) -> Result<String, crate::keys::KeyError> {
466        self.inner
467            .components
468            .adopt_key_rotation(encryption, self.inner.custody.as_ref())
469    }
470
471    pub(crate) async fn drain_uploads(
472        &self,
473    ) -> Result<crate::blob::upload::DrainOutcome, crate::database::DbError> {
474        self.inner
475            .components
476            .drain_uploads(
477                self.clock.as_ref(),
478                &self.store_dir,
479                self.inner.observer.as_deref(),
480            )
481            .await
482    }
483
484    pub(crate) async fn create_circle(
485        &self,
486        name: &str,
487    ) -> Result<crate::CircleId, crate::sync::circle_ops::CircleOperationError> {
488        let (reply, response) = tokio::sync::oneshot::channel();
489        self.command_tx
490            .send(SyncCommand::CreateCircle {
491                name: name.to_string(),
492                reply,
493            })
494            .await
495            .map_err(|_| crate::sync::circle_ops::CircleOperationError::CommandChannelClosed)?;
496        response
497            .await
498            .map_err(|_| crate::sync::circle_ops::CircleOperationError::ReplyChannelClosed)?
499    }
500}
501
502async fn execute_command(inner: &SyncLoopInner, command: SyncCommand) {
503    match command {
504        SyncCommand::CreateCircle { name, reply } => {
505            let result = inner.components.create_circle(&name).await;
506            if reply.send(result).is_err() {
507                debug!("create_circle caller dropped its reply receiver");
508            }
509        }
510    }
511}
512
513fn storage_check_failure_status(error: &crate::sync::storage::StorageError) -> SyncLoopStatus {
514    if error.is_transport() {
515        SyncLoopStatus::Offline
516    } else {
517        SyncLoopStatus::Failed {
518            error: format!("check sync storage: {error}"),
519        }
520    }
521}
522
523async fn current_success_status(
524    db: &crate::database::Database,
525    success: SyncLoopSuccess,
526) -> Result<SyncLoopStatus, String> {
527    let branches = db
528        .pending_branches()
529        .await
530        .map_err(|error| format!("read pending Serial branches after sync: {error}"))?;
531    if let Some(branch) = branches {
532        return Ok(SyncLoopStatus::Conflict { success, branch });
533    }
534    let writes: Vec<_> = db
535        .pending_writes()
536        .await
537        .map_err(|error| format!("read pending writes after sync: {error}"))?
538        .into_iter()
539        .filter(|write| matches!(write.status, crate::WriteStatus::Blocked(_)))
540        .collect();
541    if !writes.is_empty() {
542        return Ok(SyncLoopStatus::Blocked { success, writes });
543    }
544    Ok(SyncLoopStatus::Synchronized(success))
545}
546
547struct RunningGuard {
548    running: Arc<AtomicBool>,
549}
550
551impl Drop for RunningGuard {
552    fn drop(&mut self) {
553        self.running.store(false, Ordering::Release);
554    }
555}
556
557/// Run a single sync cycle.
558async fn run_single_cycle(
559    inner: &SyncLoopInner,
560    clock: &dyn crate::clock::Clock,
561    store_dir: &StoreDir,
562) -> Result<super::cycle::SyncCycleResult, super::cycle::SyncCycleFailure> {
563    inner
564        .components
565        .run_cycle(
566            clock,
567            Some(inner.custody.as_ref()),
568            store_dir,
569            inner.observer.as_deref(),
570        )
571        .await
572}
573
574#[cfg(test)]
575mod tests {
576    use super::*;
577
578    fn success() -> SyncLoopSuccess {
579        SyncLoopSuccess {
580            last_sync_time: "2026-07-14T00:00:00Z".to_string(),
581            device_count: 1,
582            device_activity: Vec::new(),
583            data_changed: false,
584            row_changes: None,
585            alerts: crate::SyncLoopAlerts {
586                rotation_pending: None,
587                held_positions: Vec::new(),
588                asset_downloads_failed: false,
589                local_blob_cleanup_pending: false,
590            },
591        }
592    }
593
594    #[test]
595    fn storage_configuration_failure_is_terminal() {
596        let status = storage_check_failure_status(
597            &crate::sync::storage::StorageError::Configuration("missing bucket".to_string()),
598        );
599
600        assert!(matches!(status, SyncLoopStatus::Failed { .. }));
601    }
602
603    fn database() -> crate::database::Database {
604        coven_core::database::Database::open(
605            std::path::Path::new(":memory:"),
606            Vec::new(),
607            chrono::Duration::days(30),
608            coven_core::blob::TransferLimits::serial(),
609            crate::WritePolicy::Serial,
610            "status-test".to_string(),
611            &[],
612        )
613        .expect("open status test database")
614        .0
615    }
616
617    async fn insert_write_status(
618        db: &crate::database::Database,
619        write_id: &'static str,
620        branch_id: &'static str,
621        status: crate::WriteStatus,
622    ) {
623        let base = format!(r#"{{"serial":{{"branch_id":"{branch_id}","base":null}}}}"#);
624        let status = serde_json::to_string(&status).expect("serialize durable write status");
625        db.call(move |conn| {
626            conn.execute(
627                r#"INSERT INTO store_writes
628                 (write_id, status, affected_rows, changeset, inverse_changeset, base, blob_facts)
629                 VALUES (?1, ?2, '[]', X'', X'', ?3, '{"blobs":[]}')"#,
630                (write_id, status, base),
631            )
632            .map(|_| ())
633            .map_err(crate::DbError::from)
634        })
635        .await
636        .expect("insert durable write status");
637    }
638
639    #[tokio::test]
640    async fn successful_cycle_projects_durable_blocked_and_conflict_states() {
641        let db = database();
642        insert_write_status(
643            &db,
644            "blocked-write",
645            "blocked-write",
646            crate::WriteStatus::Blocked(crate::WriteBlock::MissingBlob {
647                namespace: "audio".to_string(),
648                id: "missing".to_string(),
649            }),
650        )
651        .await;
652        let blocked = current_success_status(&db, success())
653            .await
654            .expect("project blocked state");
655        assert!(matches!(
656            blocked,
657            SyncLoopStatus::Blocked { writes, .. }
658                if writes.len() == 1 && writes[0].write_id.as_str() == "blocked-write"
659        ));
660        db.call(|conn| {
661            conn.execute(
662                "DELETE FROM store_writes WHERE write_id = 'blocked-write'",
663                [],
664            )
665            .map(|_| ())
666            .map_err(crate::DbError::from)
667        })
668        .await
669        .expect("remove blocked projection fixture");
670
671        let store = crate::sync::test_helpers::TestStore::create(
672            &db,
673            "status-test",
674            crate::keys::UserKeypair::generate(),
675        )
676        .await
677        .expect("create exact Serial status test Store");
678        let founder_registration = db
679            .call(|conn| {
680                let encoded: String = conn
681                    .query_row(
682                        "SELECT registration_object FROM store_device_registration_activations",
683                        [],
684                        |row| row.get(0),
685                    )
686                    .map_err(crate::DbError::from)?;
687                serde_json::from_str(&encoded)
688                    .map_err(|error| crate::DbError::Message(error.to_string()))
689            })
690            .await
691            .expect("read exact founder registration reference");
692        let genesis = crate::StoreSerialPredecessor::Genesis {
693            root: store.root.clone(),
694            founder_registration,
695        };
696        let branch_id =
697            serde_json::from_str(r#""branch-write""#).expect("parse test pending branch identity");
698
699        insert_write_status(
700            &db,
701            "branch-write",
702            "branch-write",
703            crate::WriteStatus::Conflict(Box::new(crate::SerializationConflict {
704                branch_id,
705                base: genesis.clone(),
706                current: genesis,
707            })),
708        )
709        .await;
710        let conflict = current_success_status(&db, success())
711            .await
712            .expect("project conflict state");
713        assert!(matches!(
714            conflict,
715            SyncLoopStatus::Conflict { branch, .. }
716                if branch.branch_id.first_write_id().as_str() == "branch-write"
717        ));
718    }
719}
coven/sync/sync_loop.rs

coven/sync/
sync_loop.rs
