Skip to main content

max / synckit

26.7 KB · 770 lines History Blame Raw
1 //! The `SyncStore` facade: the public handle that assembles the schema, the
2 //! client, and the database into a one-call sync (`sync_now`) and an optional
3 //! background scheduler.
4
5 use std::collections::HashSet;
6 use std::sync::Arc;
7 use std::time::{Duration, Instant};
8
9 use chrono::{DateTime, Utc};
10
11 use super::blob::{BlobOutcome, BlobPolicy, BlobTransport, sync_blobs};
12 use super::db::{
13 DbSource, clear_applying_remote, count_pending_changes, get_sync_state, get_sync_state_or,
14 set_sync_state,
15 };
16 use super::scheduler::{
17 SyncObserver, SyncState, backoff_delay, interval_elapsed, is_auth_lost,
18 is_subscription_required,
19 };
20 use super::schema::SyncSchema;
21 use super::sync::{
22 SyncScope, SyncTransport, cleanup_changelog, create_initial_snapshot,
23 enforce_changelog_retention, ensure_device_registered, pull_changes, pull_scope, push_changes,
24 push_scope,
25 };
26 use crate::client::SyncKitClient;
27 use crate::error::{Result, SyncKitError};
28
29 fn join_err(e: &tokio::task::JoinError) -> SyncKitError {
30 SyncKitError::Internal(format!("blocking task failed: {e}"))
31 }
32
33 /// Tunables for a [`SyncStore`].
34 #[derive(Debug, Clone)]
35 pub struct SyncConfig {
36 /// How often the scheduler wakes to consider syncing (the timer granularity;
37 /// the actual sync cadence is the `sync_interval_minutes` sync-state value).
38 pub check_interval: Duration,
39 /// Changelog size cap enforced after each cycle (`None` = no cap).
40 pub retention_cap: Option<i64>,
41 }
42
43 impl Default for SyncConfig {
44 fn default() -> Self {
45 Self {
46 check_interval: Duration::from_mins(1),
47 retention_cap: None,
48 }
49 }
50 }
51
52 /// Outcome of one [`SyncStore::sync_now`] cycle.
53 #[derive(Debug, Default, PartialEq, Eq)]
54 pub struct SyncOutcome {
55 pub pushed: u64,
56 pub pulled: u64,
57 pub changed_tables: HashSet<String>,
58 pub blobs: BlobOutcome,
59 }
60
61 /// The engine handle. Generic over the transport so it can be driven by the real
62 /// [`SyncKitClient`] or a test double; the scheduler loop is available only for
63 /// the concrete client (it needs SSE and auth state).
64 pub struct SyncStore<C> {
65 db: DbSource,
66 client: Arc<C>,
67 schema: SyncSchema,
68 device_name: String,
69 platform: String,
70 config: SyncConfig,
71 blob_policy: Option<Arc<dyn BlobPolicy>>,
72 }
73
74 /// Builder for [`SyncStore`].
75 pub struct SyncStoreBuilder<C> {
76 db: DbSource,
77 client: C,
78 schema: SyncSchema,
79 device_name: String,
80 platform: String,
81 config: SyncConfig,
82 blob_policy: Option<Arc<dyn BlobPolicy>>,
83 }
84
85 fn default_device_name() -> String {
86 std::env::var("HOSTNAME")
87 .or_else(|_| std::env::var("COMPUTERNAME"))
88 .unwrap_or_else(|_| "SyncKit Device".to_string())
89 }
90
91 impl<C> SyncStore<C> {
92 /// Start building a store from its database, client, and schema.
93 pub fn builder(db: DbSource, client: C, schema: SyncSchema) -> SyncStoreBuilder<C> {
94 SyncStoreBuilder {
95 db,
96 client,
97 schema,
98 device_name: default_device_name(),
99 platform: std::env::consts::OS.to_string(),
100 config: SyncConfig::default(),
101 blob_policy: None,
102 }
103 }
104
105 /// The engine's device name (for display / registration).
106 pub fn device_name(&self) -> &str {
107 &self.device_name
108 }
109
110 /// The underlying transport client. A consumer that also drives auth,
111 /// subscription, or tier flows against the same client (GoingsOn's sync
112 /// commands) reaches it here instead of holding a second handle.
113 pub fn client(&self) -> &Arc<C> {
114 &self.client
115 }
116 }
117
118 impl<C> SyncStoreBuilder<C> {
119 /// Override the device name and platform used at registration.
120 #[must_use]
121 pub fn device(mut self, name: impl Into<String>, platform: impl Into<String>) -> Self {
122 self.device_name = name.into();
123 self.platform = platform.into();
124 self
125 }
126
127 /// Override the config.
128 #[must_use]
129 pub fn config(mut self, config: SyncConfig) -> Self {
130 self.config = config;
131 self
132 }
133
134 /// Attach a blob policy (enables the blob pass in each cycle).
135 #[must_use]
136 pub fn blob_policy(mut self, policy: Arc<dyn BlobPolicy>) -> Self {
137 self.blob_policy = Some(policy);
138 self
139 }
140
141 /// Finish building.
142 pub fn build(self) -> SyncStore<C> {
143 SyncStore {
144 db: self.db,
145 client: Arc::new(self.client),
146 schema: self.schema,
147 device_name: self.device_name,
148 platform: self.platform,
149 config: self.config,
150 blob_policy: self.blob_policy,
151 }
152 }
153 }
154
155 impl<C: SyncTransport + BlobTransport> SyncStore<C> {
156 /// Run one full sync cycle: ensure the device is registered, take the initial
157 /// snapshot if needed, push local changes, pull and apply remote changes, sync
158 /// blobs (if a policy is set), then clean up and stamp the last-sync time.
159 pub async fn sync_now(&self) -> Result<SyncOutcome> {
160 // Defensive: a database written by an older client might carry a stale flag.
161 self.blocking(clear_applying_remote).await?;
162
163 let device_id =
164 ensure_device_registered(&self.db, &*self.client, &self.device_name, &self.platform)
165 .await?;
166 self.maybe_snapshot().await?;
167
168 let mut pushed = push_changes(&self.db, &*self.client, device_id).await?;
169 let pull = pull_changes(&self.db, &*self.client, &self.schema, device_id).await?;
170 let mut pulled = pull.applied;
171 let mut changed_tables = pull.changed_tables;
172
173 // Group scopes: sync each group the user belongs to, under its own key and
174 // cursor. A transport without groups reports none (the default), so this
175 // loop is a no-op for personal-only apps.
176 for (id, gck_version) in self.client.list_group_scopes().await? {
177 let scope = SyncScope::Group { id, gck_version };
178 pushed += push_scope(&self.db, &*self.client, device_id, scope).await?;
179 let gp = pull_scope(&self.db, &*self.client, &self.schema, device_id, scope).await?;
180 pulled += gp.applied;
181 changed_tables.extend(gp.changed_tables);
182 }
183
184 let blobs = match &self.blob_policy {
185 Some(policy) => sync_blobs(&self.db, &*self.client, policy).await?,
186 None => BlobOutcome::default(),
187 };
188
189 self.finalize().await?;
190 Ok(SyncOutcome {
191 pushed,
192 pulled,
193 changed_tables,
194 blobs,
195 })
196 }
197
198 /// Count unpushed local changes.
199 pub async fn pending_changes(&self) -> Result<i64> {
200 self.blocking(count_pending_changes).await
201 }
202
203 async fn maybe_snapshot(&self) -> Result<()> {
204 let db = self.db.clone();
205 let schema = self.schema.clone();
206 tokio::task::spawn_blocking(move || -> Result<()> {
207 let conn = db.open()?;
208 if get_sync_state_or(&conn, "initial_snapshot_done", "0")? != "1" {
209 create_initial_snapshot(&conn, &schema)?;
210 }
211 Ok(())
212 })
213 .await
214 .map_err(|e| join_err(&e))?
215 }
216
217 async fn finalize(&self) -> Result<()> {
218 let db = self.db.clone();
219 let cap = self.config.retention_cap;
220 tokio::task::spawn_blocking(move || -> Result<()> {
221 let conn = db.open()?;
222 cleanup_changelog(&conn)?;
223 if let Some(c) = cap {
224 enforce_changelog_retention(&conn, c)?;
225 }
226 set_sync_state(&conn, "last_sync_at", &Utc::now().to_rfc3339())?;
227 Ok(())
228 })
229 .await
230 .map_err(|e| join_err(&e))?
231 }
232
233 /// Run a synchronous DB closure on a blocking thread with a fresh connection.
234 async fn blocking<F, R>(&self, f: F) -> Result<R>
235 where
236 F: FnOnce(&rusqlite::Connection) -> Result<R> + Send + 'static,
237 R: Send + 'static,
238 {
239 let db = self.db.clone();
240 tokio::task::spawn_blocking(move || f(&db.open()?))
241 .await
242 .map_err(|e| join_err(&e))?
243 }
244 }
245
246 /// Handle to a running background scheduler. Dropping it does not stop the loop;
247 /// call [`stop`](Self::stop).
248 pub struct SchedulerHandle {
249 handle: tokio::task::JoinHandle<()>,
250 }
251
252 impl SchedulerHandle {
253 /// Stop the scheduler.
254 pub fn stop(self) {
255 self.handle.abort();
256 }
257 }
258
259 // The scheduler and session teardown need the concrete client's SSE stream and
260 // auth state, so they are not part of the generic impl.
261 impl SyncStore<SyncKitClient> {
262 /// Spawn the background sync loop, reporting through `observer`. The loop
263 /// races a timer against the server's change notifications, applies the gate
264 /// chain (authenticated, key loaded, auto-sync enabled, not backed off), and
265 /// backs off exponentially on failure (an hour on a subscription block).
266 pub fn spawn_scheduler(self: Arc<Self>, observer: Arc<dyn SyncObserver>) -> SchedulerHandle {
267 let handle = tokio::spawn(async move { self.run_scheduler(observer).await });
268 SchedulerHandle { handle }
269 }
270
271 async fn run_scheduler(self: Arc<Self>, observer: Arc<dyn SyncObserver>) {
272 let mut failures: u32 = 0;
273 let mut backoff_until: Option<Instant> = None;
274 let mut ticker = tokio::time::interval(self.config.check_interval);
275 ticker.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay);
276 // The SDK's stream auto-reconnects internally; None means SSE is
277 // unavailable and we fall back to the timer alone.
278 let mut sse = self.client.subscribe().await.ok();
279
280 loop {
281 let mut sse_triggered = false;
282 match sse.as_mut() {
283 Some(stream) => {
284 tokio::select! {
285 _ = ticker.tick() => {}
286 change = stream.next_change() => match change {
287 Some(()) => sse_triggered = true,
288 None => sse = None, // stream ended fatally → timer-only
289 }
290 }
291 }
292 None => {
293 ticker.tick().await;
294 }
295 }
296
297 // Gate chain.
298 if self.client.session_info().is_none() || self.client.is_token_expired() {
299 observer.on_status(SyncState::LoggedOut);
300 continue;
301 }
302 if !self.client.has_master_key() {
303 continue;
304 }
305 if !self.auto_sync_enabled().await.unwrap_or(false) {
306 continue;
307 }
308 if backoff_until.is_some_and(|u| Instant::now() < u) {
309 continue;
310 }
311 if !sse_triggered && !self.timer_due().await.unwrap_or(true) {
312 continue;
313 }
314
315 observer.on_status(SyncState::Syncing);
316 match self.sync_now().await {
317 Ok(outcome) => {
318 failures = 0;
319 backoff_until = None;
320 observer.on_status(SyncState::Idle);
321 if outcome.pulled > 0 {
322 observer.on_changes_applied(&outcome.changed_tables);
323 }
324 }
325 Err(e) if is_subscription_required(&e) => {
326 observer.on_subscription_required();
327 backoff_until = Some(Instant::now() + Duration::from_hours(1));
328 }
329 Err(e) if is_auth_lost(&e) => {
330 observer.on_status(SyncState::LoggedOut);
331 backoff_until = Some(Instant::now() + backoff_delay(failures));
332 }
333 Err(e) => {
334 failures = failures.saturating_add(1);
335 observer.on_status(SyncState::Error(e.to_string()));
336 backoff_until = Some(Instant::now() + backoff_delay(failures));
337 }
338 }
339 }
340 }
341
342 /// Clear the in-memory session and master key. Does not touch the OS keychain
343 /// or the local database (a later re-auth reuses the stored `device_id`).
344 pub fn disconnect(&self) {
345 self.client.clear_session();
346 }
347
348 async fn auto_sync_enabled(&self) -> Result<bool> {
349 self.blocking(|c| Ok(get_sync_state_or(c, "auto_sync_enabled", "1")? == "1"))
350 .await
351 }
352
353 async fn timer_due(&self) -> Result<bool> {
354 self.blocking(|c| {
355 let interval: i64 = get_sync_state_or(c, "sync_interval_minutes", "15")?
356 .parse()
357 .unwrap_or(15);
358 let last = get_sync_state(c, "last_sync_at")?
359 .and_then(|s| DateTime::parse_from_rfc3339(&s).ok())
360 .map(|d| d.with_timezone(&Utc));
361 Ok(interval_elapsed(last, interval, Utc::now()))
362 })
363 .await
364 }
365 }
366
367 #[cfg(test)]
368 mod tests {
369 use super::super::blob::BlobTransport;
370 use super::super::schema::SyncTable;
371 use super::*;
372 use crate::client::BlobUploadOutcome;
373 use crate::ids::{AppId, DeviceId, UserId};
374 use crate::types::{ChangeEntry, Device, PulledChange};
375 use std::collections::HashMap;
376 use std::future::Future;
377 use std::path::PathBuf;
378 use std::sync::Mutex;
379
380 /// A combined fake implementing both transports over a shared changelog log
381 /// and blob store, tagged with this device's registered id.
382 type GroupLog = Arc<Mutex<Vec<(crate::ids::GroupId, DeviceId, ChangeEntry)>>>;
383
384 #[derive(Clone)]
385 struct Fake {
386 device_id: DeviceId,
387 log: Arc<Mutex<Vec<(DeviceId, ChangeEntry)>>>,
388 blobs: Arc<Mutex<HashMap<String, Vec<u8>>>>,
389 /// Groups this member belongs to, reported by `list_group_scopes`.
390 groups: Vec<(crate::ids::GroupId, i32)>,
391 /// Shared group changelog (plaintext, the fake bypasses encryption, which
392 /// is covered by the `client::groups` tests).
393 group_log: GroupLog,
394 }
395 impl Fake {
396 fn new(device_id: u128, log: Arc<Mutex<Vec<(DeviceId, ChangeEntry)>>>) -> Self {
397 Self {
398 device_id: DeviceId::new(uuid::Uuid::from_u128(device_id)),
399 log,
400 blobs: Arc::new(Mutex::new(HashMap::new())),
401 groups: Vec::new(),
402 group_log: Arc::new(Mutex::new(Vec::new())),
403 }
404 }
405
406 /// Enroll this member in a group sharing `group_log`.
407 fn in_group(mut self, id: crate::ids::GroupId, version: i32, group_log: GroupLog) -> Self {
408 self.groups.push((id, version));
409 self.group_log = group_log;
410 self
411 }
412 }
413
414 impl SyncTransport for Fake {
415 fn register_device(
416 &self,
417 _name: &str,
418 _platform: &str,
419 ) -> impl Future<Output = Result<Device>> + Send {
420 let id = self.device_id;
421 async move {
422 Ok(Device {
423 id,
424 app_id: AppId::nil(),
425 user_id: UserId::nil(),
426 device_name: "fake".into(),
427 platform: "test".into(),
428 last_seen_at: Utc::now(),
429 created_at: Utc::now(),
430 })
431 }
432 }
433 fn push(
434 &self,
435 device_id: DeviceId,
436 changes: Vec<ChangeEntry>,
437 ) -> impl Future<Output = Result<i64>> + Send {
438 let log = self.log.clone();
439 async move {
440 let mut l = log.lock().unwrap();
441 for c in changes {
442 l.push((device_id, c));
443 }
444 Ok(l.len() as i64)
445 }
446 }
447 fn pull_rich(
448 &self,
449 _device_id: DeviceId,
450 cursor: i64,
451 ) -> impl Future<Output = Result<(Vec<PulledChange>, i64, bool)>> + Send {
452 let log = self.log.clone();
453 async move {
454 let l = log.lock().unwrap();
455 let out: Vec<PulledChange> = l
456 .iter()
457 .enumerate()
458 .filter(|(i, _)| (*i as i64 + 1) > cursor)
459 .map(|(i, (d, e))| PulledChange {
460 entry: e.clone(),
461 device_id: *d,
462 seq: i as i64 + 1,
463 })
464 .collect();
465 Ok((out, l.len() as i64, false))
466 }
467 }
468
469 fn list_group_scopes(
470 &self,
471 ) -> impl Future<Output = Result<Vec<(crate::ids::GroupId, i32)>>> + Send {
472 let groups = self.groups.clone();
473 async move { Ok(groups) }
474 }
475
476 fn group_scope_push(
477 &self,
478 group_id: crate::ids::GroupId,
479 _gck_version: i32,
480 device_id: DeviceId,
481 changes: Vec<ChangeEntry>,
482 ) -> impl Future<Output = Result<i64>> + Send {
483 let group_log = self.group_log.clone();
484 async move {
485 let mut l = group_log.lock().unwrap();
486 for c in changes {
487 l.push((group_id, device_id, c));
488 }
489 Ok(l.len() as i64)
490 }
491 }
492
493 fn group_scope_pull(
494 &self,
495 group_id: crate::ids::GroupId,
496 _gck_version: i32,
497 _device_id: DeviceId,
498 cursor: i64,
499 ) -> impl Future<Output = Result<(Vec<PulledChange>, i64, bool)>> + Send {
500 let group_log = self.group_log.clone();
501 async move {
502 let l = group_log.lock().unwrap();
503 // Filter to this group, then page by global position (mirrors the
504 // personal fake's seq scheme).
505 let out: Vec<PulledChange> = l
506 .iter()
507 .enumerate()
508 .filter(|(i, (gid, _, _))| *gid == group_id && (*i as i64 + 1) > cursor)
509 .map(|(i, (_, dev, entry))| PulledChange {
510 entry: entry.clone(),
511 device_id: *dev,
512 seq: i as i64 + 1,
513 })
514 .collect();
515 Ok((out, l.len() as i64, false))
516 }
517 }
518 }
519
520 impl BlobTransport for Fake {
521 async fn upload_file(
522 &self,
523 hash: &str,
524 path: &std::path::Path,
525 ) -> Result<BlobUploadOutcome> {
526 if self.blobs.lock().unwrap().contains_key(hash) {
527 return Ok(BlobUploadOutcome::AlreadyPresent);
528 }
529 let data = std::fs::read(path)
530 .map_err(|e| crate::error::SyncKitError::Internal(e.to_string()))?;
531 self.blobs.lock().unwrap().insert(hash.to_string(), data);
532 Ok(BlobUploadOutcome::Uploaded)
533 }
534 async fn confirm(&self, _hash: &str, _size: u64) -> Result<()> {
535 Ok(())
536 }
537 async fn download_url(&self, hash: &str) -> Result<String> {
538 Ok(format!("fake://{hash}"))
539 }
540 async fn download(&self, hash: &str, _url: &str) -> Result<Vec<u8>> {
541 self.blobs
542 .lock()
543 .unwrap()
544 .get(hash)
545 .cloned()
546 .ok_or_else(|| SyncKitError::Internal("no blob".into()))
547 }
548 }
549
550 fn schema() -> SyncSchema {
551 SyncSchema::new(vec![SyncTable::full("note", &["id", "name"])])
552 }
553
554 fn tempdir() -> PathBuf {
555 use std::sync::atomic::{AtomicU64, Ordering};
556 static N: AtomicU64 = AtomicU64::new(0);
557 let mut p = std::env::temp_dir();
558 p.push(format!(
559 "synckit_b7_{}_{}",
560 std::process::id(),
561 N.fetch_add(1, Ordering::Relaxed)
562 ));
563 std::fs::create_dir_all(&p).unwrap();
564 p
565 }
566
567 fn make_db(path: &std::path::Path) -> DbSource {
568 let db = DbSource::path(path);
569 let conn = db.open().unwrap();
570 conn.execute_batch("CREATE TABLE note (id TEXT PRIMARY KEY, name TEXT);")
571 .unwrap();
572 conn.execute_batch(&schema().migration_sql()).unwrap();
573 db
574 }
575
576 fn store(db: DbSource, fake: Fake) -> SyncStore<Fake> {
577 SyncStore::builder(db, fake, schema())
578 .device("dev", "test")
579 .build()
580 }
581
582 // ── Group-scope sync ──
583
584 fn group_schema() -> SyncSchema {
585 SyncSchema::new(vec![
586 SyncTable::full("task", &["id", "name"]).group_scoped("group_id"),
587 ])
588 }
589
590 fn group_make_db(path: &std::path::Path) -> DbSource {
591 let db = DbSource::path(path);
592 let conn = db.open().unwrap();
593 conn.execute_batch("CREATE TABLE task (id TEXT PRIMARY KEY, name TEXT, group_id TEXT);")
594 .unwrap();
595 conn.execute_batch(&group_schema().migration_sql()).unwrap();
596 db
597 }
598
599 fn group_store(db: DbSource, fake: Fake) -> SyncStore<Fake> {
600 SyncStore::builder(db, fake, group_schema())
601 .device("dev", "test")
602 .build()
603 }
604
605 #[tokio::test]
606 async fn two_members_converge_through_a_group_scope() {
607 let dir = tempdir();
608 let gid = crate::ids::GroupId::new(uuid::Uuid::from_u128(0x9));
609 let group_log = Arc::new(Mutex::new(Vec::new()));
610 let plog = Arc::new(Mutex::new(Vec::new()));
611
612 let da = group_make_db(&dir.join("a.db"));
613 let dbb = group_make_db(&dir.join("b.db"));
614 let sa = group_store(
615 da.clone(),
616 Fake::new(0xA, plog.clone()).in_group(gid, 1, group_log.clone()),
617 );
618 let sb = group_store(
619 dbb.clone(),
620 Fake::new(0xB, plog.clone()).in_group(gid, 1, group_log.clone()),
621 );
622
623 // Member A creates a task in the group.
624 da.open()
625 .unwrap()
626 .execute(
627 "INSERT INTO task (id, name, group_id) VALUES ('t', 'from-A', ?1)",
628 [gid.to_string()],
629 )
630 .unwrap();
631
632 sa.sync_now().await.unwrap(); // pushes the group task to the shared log
633 sb.sync_now().await.unwrap(); // pulls it and applies, stamping group_id
634
635 // Member B now has the task, tagged with the group it came from (stamped
636 // from the channel, not the payload, the payload carried only id + name).
637 let (name, group_id): (String, String) = dbb
638 .open()
639 .unwrap()
640 .query_row("SELECT name, group_id FROM task WHERE id = 't'", [], |r| {
641 Ok((r.get(0)?, r.get(1)?))
642 })
643 .unwrap();
644 assert_eq!(name, "from-A");
645 assert_eq!(group_id, gid.to_string());
646
647 // The group write never touched B's personal changelog scope.
648 let personal_rows: i64 = dbb
649 .open()
650 .unwrap()
651 .query_row(
652 "SELECT COUNT(*) FROM sync_changelog WHERE scope = ''",
653 [],
654 |r| r.get(0),
655 )
656 .unwrap();
657 assert_eq!(personal_rows, 0);
658 }
659
660 /// Pre-stamp a device's pending edit at a controlled time so cross-device HLC
661 /// ordering is deterministic (sync_now's internal stamp is then a no-op).
662 fn stamp_pending(db: &DbSource, node: DeviceId, now_ms: i64) {
663 super::super::hlc::stamp_pending(&db.open().unwrap(), node, now_ms).unwrap();
664 }
665
666 #[tokio::test]
667 async fn sync_now_registers_snapshots_pushes_and_finalizes() {
668 let dir = tempdir();
669 let db = make_db(&dir.join("a.db"));
670 // Pre-existing row (trigger fired), plus clear changelog to test snapshot.
671 {
672 let c = db.open().unwrap();
673 c.execute("INSERT INTO note (id, name) VALUES ('r', 'v')", [])
674 .unwrap();
675 c.execute("DELETE FROM sync_changelog", []).unwrap();
676 }
677 let log = Arc::new(Mutex::new(Vec::new()));
678 let s = store(db.clone(), Fake::new(0xA, log.clone()));
679
680 let out = s.sync_now().await.unwrap();
681 assert_eq!(out.pushed, 1, "the snapshot row is pushed");
682 // device_id registered, last_sync stamped, snapshot marked done.
683 let c = db.open().unwrap();
684 assert!(get_sync_state(&c, "device_id").unwrap().is_some());
685 assert_eq!(
686 get_sync_state_or(&c, "initial_snapshot_done", "0").unwrap(),
687 "1"
688 );
689 assert!(
690 !get_sync_state_or(&c, "last_sync_at", "")
691 .unwrap()
692 .is_empty()
693 );
694 // The server received the snapshot entry.
695 assert_eq!(log.lock().unwrap().len(), 1);
696 }
697
698 #[tokio::test]
699 async fn two_devices_converge_through_sync_now() {
700 let dir = tempdir();
701 let log = Arc::new(Mutex::new(Vec::new()));
702 let da = make_db(&dir.join("a.db"));
703 let db = make_db(&dir.join("b.db"));
704
705 // A edits at t=100, B edits the same row at t=200 (B wins).
706 da.open()
707 .unwrap()
708 .execute("INSERT INTO note (id,name) VALUES ('r','from-A')", [])
709 .unwrap();
710 stamp_pending(&da, DeviceId::new(uuid::Uuid::from_u128(0xA)), 100);
711 db.open()
712 .unwrap()
713 .execute("INSERT INTO note (id,name) VALUES ('r','from-B')", [])
714 .unwrap();
715 stamp_pending(&db, DeviceId::new(uuid::Uuid::from_u128(0xB)), 200);
716
717 let sa = store(da.clone(), Fake::new(0xA, log.clone()));
718 let sb = store(db.clone(), Fake::new(0xB, log.clone()));
719
720 // Push order then pull order.
721 sa.sync_now().await.unwrap();
722 sb.sync_now().await.unwrap();
723 sa.sync_now().await.unwrap();
724 sb.sync_now().await.unwrap();
725
726 let name_a: String = da
727 .open()
728 .unwrap()
729 .query_row("SELECT name FROM note WHERE id='r'", [], |r| r.get(0))
730 .unwrap();
731 let name_b: String = db
732 .open()
733 .unwrap()
734 .query_row("SELECT name FROM note WHERE id='r'", [], |r| r.get(0))
735 .unwrap();
736 assert_eq!(name_a, "from-B");
737 assert_eq!(name_b, "from-B");
738 }
739
740 #[tokio::test]
741 async fn builder_defaults_and_overrides() {
742 let dir = tempdir();
743 let db = make_db(&dir.join("a.db"));
744 let log = Arc::new(Mutex::new(Vec::new()));
745 let s = SyncStore::builder(db, Fake::new(0xA, log), schema())
746 .device("My Mac", "macos")
747 .config(SyncConfig {
748 check_interval: Duration::from_secs(30),
749 retention_cap: Some(1000),
750 })
751 .build();
752 assert_eq!(s.device_name(), "My Mac");
753 assert_eq!(s.platform, "macos");
754 assert_eq!(s.config.retention_cap, Some(1000));
755 }
756
757 #[tokio::test]
758 async fn pending_changes_counts_unpushed() {
759 let dir = tempdir();
760 let db = make_db(&dir.join("a.db"));
761 db.open()
762 .unwrap()
763 .execute("INSERT INTO note (id,name) VALUES ('r','v')", [])
764 .unwrap();
765 let log = Arc::new(Mutex::new(Vec::new()));
766 let s = store(db, Fake::new(0xA, log));
767 assert_eq!(s.pending_changes().await.unwrap(), 1);
768 }
769 }
770