Skip to main content

max / synckit

30.2 KB · 879 lines History Blame Raw
1 //! Push/pull loops and changelog lifecycle.
2 //!
3 //! This ties B2–B4 together against a transport: the drain-loop push (stamp ->
4 //! batch -> send -> mark-pushed + record committed), the paginated pull (fetch ->
5 //! resolve -> apply -> record committed -> persist cursor), plus the initial
6 //! snapshot, changelog cleanup, and retention cap.
7 //!
8 //! The engine's SQLite connection is `!Send`, so every database step runs on a
9 //! blocking thread (`spawn_blocking`) with a fresh connection from the
10 //! [`DbSource`](super::db::DbSource); the transport calls are the only `.await`
11 //! points, and nothing `!Send` is held across them. The transport is abstracted
12 //! behind [`SyncTransport`] so the loops test end-to-end against an in-memory
13 //! fake, and `SyncKitClient` satisfies it in production.
14
15 use std::collections::HashSet;
16 use std::future::Future;
17
18 use chrono::Utc;
19 use rusqlite::Connection;
20
21 use super::apply::{ApplyOutcome, apply_remote_changes};
22 use super::db::{
23 DbSource, device_id, get_scope_cursor, set_device_id, set_scope_cursor, set_sync_state,
24 };
25 use super::hlc::{record_committed, resolve_pull, set_committed, stamp_pending};
26 use super::migrate::{json_object, row_id_expr};
27 use super::schema::{SyncMode, SyncSchema};
28 use crate::client::SyncKitClient;
29 use crate::error::{Result, SyncKitError};
30 use crate::ids::{DeviceId, GroupId};
31 use crate::types::{ChangeEntry, ChangeOp, Device, Hlc, PulledChange};
32
33 /// A sync scope: the personal changelog, or one group's shared changelog. Each
34 /// scope has its own changelog partition (`sync_changelog.scope`), pull cursor
35 /// (`sync_scope_cursor`), key, and endpoint. See the wiki design note
36 /// `synckit-multiscope-design`.
37 #[derive(Debug, Clone, Copy, PartialEq, Eq)]
38 pub enum SyncScope {
39 /// The user's personal data, encrypted under the master key.
40 Personal,
41 /// A group's shared data, encrypted under the group's GCK at `gck_version`.
42 Group { id: GroupId, gck_version: i32 },
43 }
44
45 impl SyncScope {
46 /// The `sync_changelog.scope` / `sync_scope_cursor.scope` string for this
47 /// scope: `""` for personal, the group id otherwise.
48 pub fn key(&self) -> String {
49 match self {
50 SyncScope::Personal => String::new(),
51 SyncScope::Group { id, .. } => id.to_string(),
52 }
53 }
54 }
55
56 /// Maximum changes sent in one push batch.
57 pub const PUSH_BATCH_LIMIT: usize = 500;
58
59 /// The two operations the sync loops need from a server.
60 ///
61 /// A seam for testing (an in-memory fake) and the reference point a future
62 /// non-Rust SDK mirrors. [`SyncKitClient`] implements it by forwarding to its
63 /// inherent methods.
64 pub trait SyncTransport: Sync {
65 /// Register (or look up) this device with the server.
66 fn register_device(
67 &self,
68 device_name: &str,
69 platform: &str,
70 ) -> impl Future<Output = Result<Device>> + Send;
71
72 /// Push encrypted changes; returns the server's new cursor.
73 fn push(
74 &self,
75 device_id: DeviceId,
76 changes: Vec<ChangeEntry>,
77 ) -> impl Future<Output = Result<i64>> + Send;
78
79 /// Pull decrypted changes since `cursor`; returns `(changes, new_cursor,
80 /// has_more)` with each change's originating device and sequence.
81 fn pull_rich(
82 &self,
83 device_id: DeviceId,
84 cursor: i64,
85 ) -> impl Future<Output = Result<(Vec<PulledChange>, i64, bool)>> + Send;
86
87 /// The groups this user belongs to, as `(group_id, current_gck_version)`,
88 /// the scopes to sync beyond personal. Defaults to none, so a transport that
89 /// does not support groups (a test double, a future minimal SDK) never syncs
90 /// any group scope.
91 // `-> impl Future + Send` (not `async fn`) keeps the explicit Send bound the
92 // store's spawned tasks need; that is the whole trait's convention.
93 #[allow(clippy::manual_async_fn)]
94 fn list_group_scopes(&self) -> impl Future<Output = Result<Vec<(GroupId, i32)>>> + Send {
95 async move { Ok(Vec::new()) }
96 }
97
98 /// Push encrypted changes to a group's shared changelog. The implementation
99 /// resolves the group's GCK internally; the caller never handles keys. The
100 /// default errors, it is only reached if [`list_group_scopes`] returns a
101 /// group, which the default never does.
102 #[allow(clippy::manual_async_fn)]
103 fn group_scope_push(
104 &self,
105 _group_id: GroupId,
106 _gck_version: i32,
107 _device_id: DeviceId,
108 _changes: Vec<ChangeEntry>,
109 ) -> impl Future<Output = Result<i64>> + Send {
110 async move {
111 Err(SyncKitError::Internal(
112 "group sync not supported by this transport".into(),
113 ))
114 }
115 }
116
117 /// Pull a group's changes since `cursor`, decrypted under its GCK. Default
118 /// errors; see [`group_scope_push`](Self::group_scope_push).
119 #[allow(clippy::manual_async_fn)]
120 fn group_scope_pull(
121 &self,
122 _group_id: GroupId,
123 _gck_version: i32,
124 _device_id: DeviceId,
125 _cursor: i64,
126 ) -> impl Future<Output = Result<(Vec<PulledChange>, i64, bool)>> + Send {
127 async move {
128 Err(SyncKitError::Internal(
129 "group sync not supported by this transport".into(),
130 ))
131 }
132 }
133 }
134
135 // The group methods below return `-> impl Future + Send` with an `async move`
136 // body (not `async fn`) to keep the explicit Send bound the trait requires.
137 #[allow(clippy::manual_async_fn)]
138 impl SyncTransport for SyncKitClient {
139 fn register_device(
140 &self,
141 device_name: &str,
142 platform: &str,
143 ) -> impl Future<Output = Result<Device>> + Send {
144 SyncKitClient::register_device(self, device_name, platform)
145 }
146
147 fn push(
148 &self,
149 device_id: DeviceId,
150 changes: Vec<ChangeEntry>,
151 ) -> impl Future<Output = Result<i64>> + Send {
152 // Inherent method (preferred over the trait method of the same name).
153 SyncKitClient::push(self, device_id, changes)
154 }
155
156 fn pull_rich(
157 &self,
158 device_id: DeviceId,
159 cursor: i64,
160 ) -> impl Future<Output = Result<(Vec<PulledChange>, i64, bool)>> + Send {
161 SyncKitClient::pull_rich(self, device_id, cursor)
162 }
163
164 fn list_group_scopes(&self) -> impl Future<Output = Result<Vec<(GroupId, i32)>>> + Send {
165 async move {
166 Ok(self
167 .list_groups()
168 .await?
169 .into_iter()
170 .map(|g| (g.id, g.gck_version))
171 .collect())
172 }
173 }
174
175 fn group_scope_push(
176 &self,
177 group_id: GroupId,
178 gck_version: i32,
179 device_id: DeviceId,
180 changes: Vec<ChangeEntry>,
181 ) -> impl Future<Output = Result<i64>> + Send {
182 async move {
183 let gck = self.group_content_key(group_id, gck_version).await?;
184 self.group_push(group_id, &gck, device_id, changes).await
185 }
186 }
187
188 fn group_scope_pull(
189 &self,
190 group_id: GroupId,
191 gck_version: i32,
192 device_id: DeviceId,
193 cursor: i64,
194 ) -> impl Future<Output = Result<(Vec<PulledChange>, i64, bool)>> + Send {
195 async move {
196 let gck = self.group_content_key(group_id, gck_version).await?;
197 match self
198 .group_pull_rich(group_id, &gck, device_id, cursor)
199 .await
200 {
201 // A decrypt failure means this client is holding a stale GCK (it
202 // missed a rotation): drop it, re-fetch the current grant, retry
203 // once. A second failure surfaces.
204 Err(SyncKitError::DecryptionFailed) => {
205 self.invalidate_gck(group_id);
206 let gck = self.group_content_key(group_id, gck_version).await?;
207 self.group_pull_rich(group_id, &gck, device_id, cursor)
208 .await
209 }
210 other => other,
211 }
212 }
213 }
214 }
215
216 /// Ensure this install has a registered device, returning its id (which doubles
217 /// as the HLC node). Fast path: a stored `device_id` returns without a network
218 /// call. Otherwise register with the server and persist the assigned id.
219 pub async fn ensure_device_registered<T: SyncTransport>(
220 db: &DbSource,
221 client: &T,
222 device_name: &str,
223 platform: &str,
224 ) -> Result<DeviceId> {
225 let db_read = db.clone();
226 let existing = tokio::task::spawn_blocking(move || device_id(&db_read.open()?))
227 .await
228 .map_err(|e| join_err(&e))??;
229 if let Some(id) = existing {
230 return Ok(id);
231 }
232
233 let device = client.register_device(device_name, platform).await?;
234 let id = device.id;
235
236 let db_write = db.clone();
237 tokio::task::spawn_blocking(move || set_device_id(&db_write.open()?, id))
238 .await
239 .map_err(|e| join_err(&e))??;
240 Ok(id)
241 }
242
243 /// Outcome of a pull pass.
244 #[derive(Debug, Default, PartialEq, Eq)]
245 pub struct PullOutcome {
246 pub applied: u64,
247 pub changed_tables: HashSet<String>,
248 }
249
250 fn join_err(e: &tokio::task::JoinError) -> SyncKitError {
251 SyncKitError::Internal(format!("blocking task failed: {e}"))
252 }
253
254 // push
255
256 struct PushBatch {
257 wire: Vec<ChangeEntry>,
258 wire_ids: Vec<i64>,
259 /// Rows marked pushed without sending (corrupt/unknown-op) so they can't wedge.
260 skip_ids: Vec<i64>,
261 total_read: usize,
262 }
263
264 /// Drain all unpushed local changes for the **personal** scope. Thin wrapper over
265 /// [`push_scope`] preserved for the personal call sites.
266 pub async fn push_changes<T: SyncTransport>(
267 db: &DbSource,
268 client: &T,
269 device_id: DeviceId,
270 ) -> Result<u64> {
271 push_scope(db, client, device_id, SyncScope::Personal).await
272 }
273
274 /// Drain all unpushed local changes for one `scope` to the server, batch by batch.
275 ///
276 /// Each batch is stamped and read on a blocking thread, sent to the scope's
277 /// endpoint (personal push, or a group push under its GCK), then marked pushed
278 /// with its HLC recorded as the committed clock (so a peer's later stale re-pull
279 /// of this row is gated out). `device_id` doubles as the HLC node. The drain
280 /// reads only rows whose `sync_changelog.scope` matches, so a group's changes
281 /// never leak into the personal push and vice versa.
282 pub async fn push_scope<T: SyncTransport>(
283 db: &DbSource,
284 client: &T,
285 device_id: DeviceId,
286 scope: SyncScope,
287 ) -> Result<u64> {
288 let scope_key = scope.key();
289 let mut pushed_total = 0u64;
290 loop {
291 let db_read = db.clone();
292 let sk = scope_key.clone();
293 let batch =
294 tokio::task::spawn_blocking(move || read_push_batch(&db_read.open()?, device_id, &sk))
295 .await
296 .map_err(|e| join_err(&e))??;
297
298 if batch.wire.is_empty() && batch.skip_ids.is_empty() {
299 break;
300 }
301
302 if !batch.wire.is_empty() {
303 match scope {
304 SyncScope::Personal => {
305 client.push(device_id, batch.wire.clone()).await?;
306 }
307 SyncScope::Group { id, gck_version } => {
308 client
309 .group_scope_push(id, gck_version, device_id, batch.wire.clone())
310 .await?;
311 }
312 }
313 }
314
315 let db_write = db.clone();
316 let PushBatch {
317 wire,
318 wire_ids,
319 skip_ids,
320 total_read,
321 } = batch;
322 let sent = wire.len() as u64;
323 tokio::task::spawn_blocking(move || {
324 mark_pushed_committed(&mut db_write.open()?, &wire_ids, &skip_ids, &wire)
325 })
326 .await
327 .map_err(|e| join_err(&e))??;
328
329 pushed_total += sent;
330 if total_read < PUSH_BATCH_LIMIT {
331 break;
332 }
333 }
334 Ok(pushed_total)
335 }
336
337 fn read_push_batch(conn: &Connection, device_id: DeviceId, scope: &str) -> Result<PushBatch> {
338 stamp_pending(conn, device_id, Utc::now().timestamp_millis())?;
339 let mut stmt = conn.prepare(
340 "SELECT id, table_name, op, row_id, data, hlc_wall, hlc_counter \
341 FROM sync_changelog WHERE pushed = 0 AND scope = ?2 ORDER BY id LIMIT ?1",
342 )?;
343 let rows = stmt.query_map(rusqlite::params![PUSH_BATCH_LIMIT as i64, scope], |r| {
344 Ok((
345 r.get::<_, i64>(0)?,
346 r.get::<_, String>(1)?,
347 r.get::<_, String>(2)?,
348 r.get::<_, String>(3)?,
349 r.get::<_, Option<String>>(4)?,
350 r.get::<_, i64>(5)?,
351 r.get::<_, i64>(6)?,
352 ))
353 })?;
354
355 let mut batch = PushBatch {
356 wire: Vec::new(),
357 wire_ids: Vec::new(),
358 skip_ids: Vec::new(),
359 total_read: 0,
360 };
361 for row in rows {
362 batch.total_read += 1;
363 let (id, table, op, row_id, data, wall_ms, counter) = row?;
364 // Corrupt/unknown rows are marked pushed (not sent) so they can't wedge
365 // the drain by being re-read forever.
366 let Some(op) = ChangeOp::from_str_opt(&op) else {
367 batch.skip_ids.push(id);
368 continue;
369 };
370 let Ok(data) = data.as_deref().map(serde_json::from_str).transpose() else {
371 batch.skip_ids.push(id);
372 continue;
373 };
374 batch.wire.push(ChangeEntry {
375 table,
376 op,
377 row_id,
378 timestamp: Utc::now(),
379 hlc: Hlc {
380 wall_ms,
381 counter: counter as u32,
382 node: device_id,
383 },
384 data,
385 extra: serde_json::Map::default(),
386 });
387 batch.wire_ids.push(id);
388 }
389 Ok(batch)
390 }
391
392 fn mark_pushed_committed(
393 conn: &mut Connection,
394 wire_ids: &[i64],
395 skip_ids: &[i64],
396 wire: &[ChangeEntry],
397 ) -> Result<()> {
398 let tx = conn.transaction()?;
399 for id in wire_ids.iter().chain(skip_ids) {
400 tx.execute("UPDATE sync_changelog SET pushed = 1 WHERE id = ?1", [id])?;
401 }
402 // Push-side committed-ledger update: our own edit's HLC becomes the row's
403 // committed clock, so a peer's later stale re-pull of it is gated out.
404 for e in wire {
405 set_committed(&tx, &e.table, &e.row_id, &e.hlc)?;
406 }
407 tx.commit()?;
408 Ok(())
409 }
410
411 // pull
412
413 /// Pull, resolve, and apply all remote changes for the **personal** scope. Thin
414 /// wrapper over [`pull_scope`] preserved for the personal call sites.
415 pub async fn pull_changes<T: SyncTransport>(
416 db: &DbSource,
417 client: &T,
418 schema: &SyncSchema,
419 device_id: DeviceId,
420 ) -> Result<PullOutcome> {
421 pull_scope(db, client, schema, device_id, SyncScope::Personal).await
422 }
423
424 /// Pull, resolve, and apply all remote changes for one `scope`, page by page.
425 ///
426 /// Reads from the scope's own cursor (`sync_scope_cursor`), pulls from the
427 /// scope's endpoint (personal pull, or a group pull under its GCK), and applies
428 /// with the scope stamped so group rows are tagged with their group. Follows the
429 /// SDK contract: the cursor advances only *after* the batch is applied and
430 /// committed, in the same blocking step, so a crash re-pulls a batch the
431 /// idempotent apply/gate absorbs.
432 pub async fn pull_scope<T: SyncTransport>(
433 db: &DbSource,
434 client: &T,
435 schema: &SyncSchema,
436 device_id: DeviceId,
437 scope: SyncScope,
438 ) -> Result<PullOutcome> {
439 let scope_key = scope.key();
440 let mut out = PullOutcome::default();
441 loop {
442 let db_cursor = db.clone();
443 let sk = scope_key.clone();
444 let cursor = tokio::task::spawn_blocking(move || get_scope_cursor(&db_cursor.open()?, &sk))
445 .await
446 .map_err(|e| join_err(&e))??;
447
448 let (pulled, new_cursor, has_more) = match scope {
449 SyncScope::Personal => client.pull_rich(device_id, cursor).await?,
450 SyncScope::Group { id, gck_version } => {
451 client
452 .group_scope_pull(id, gck_version, device_id, cursor)
453 .await?
454 }
455 };
456
457 if pulled.is_empty() {
458 let db_c = db.clone();
459 let sk = scope_key.clone();
460 tokio::task::spawn_blocking(move || set_scope_cursor(&db_c.open()?, &sk, new_cursor))
461 .await
462 .map_err(|e| join_err(&e))??;
463 break;
464 }
465
466 let db_apply = db.clone();
467 let schema = schema.clone();
468 let sk = scope_key.clone();
469 let outcome = tokio::task::spawn_blocking(move || {
470 apply_pull(
471 &mut db_apply.open()?,
472 &schema,
473 device_id,
474 pulled,
475 new_cursor,
476 &sk,
477 )
478 })
479 .await
480 .map_err(|e| join_err(&e))??;
481
482 out.applied += outcome.applied as u64;
483 out.changed_tables.extend(outcome.changed_tables);
484 if !has_more {
485 break;
486 }
487 }
488 Ok(out)
489 }
490
491 fn apply_pull(
492 conn: &mut Connection,
493 schema: &SyncSchema,
494 device_id: DeviceId,
495 pulled: Vec<PulledChange>,
496 new_cursor: i64,
497 scope_key: &str,
498 ) -> Result<ApplyOutcome> {
499 let resolved = resolve_pull(conn, schema, device_id, pulled, Utc::now())?;
500 let outcome = apply_remote_changes(conn, schema, &resolved, scope_key)?;
501 record_committed(conn, &resolved)?;
502 set_scope_cursor(conn, scope_key, new_cursor)?;
503 Ok(outcome)
504 }
505
506 // lifecycle (synchronous)
507
508 /// One-time backfill: snapshot every existing row into the changelog so a fresh
509 /// device's pre-existing data pushes on the first sync. Idempotent, rows already
510 /// represented in the changelog are skipped. Reuses the trigger's row-id and
511 /// payload expressions, so snapshot entries are byte-identical to trigger entries.
512 pub fn create_initial_snapshot(conn: &Connection, schema: &SyncSchema) -> Result<u64> {
513 let mut total = 0u64;
514 for table in &schema.tables {
515 let name = table.name;
516 let cols = table.emitted_columns();
517 let row_id = row_id_expr(table, name);
518 let data = json_object(name, &cols);
519 // PartialUpdate tables reflect state changes; snapshot them as UPDATEs so
520 // a peer applies them through the same partial path.
521 let op = if matches!(table.mode, SyncMode::PartialUpdate { .. }) {
522 "UPDATE"
523 } else {
524 "INSERT"
525 };
526 let exclude = table
527 .exclude_where
528 .map(|p| format!(" AND ({})", p.replace("{row}", name)))
529 .unwrap_or_default();
530
531 let sql = format!(
532 "INSERT INTO sync_changelog (table_name, op, row_id, data) \
533 SELECT '{name}', '{op}', {row_id}, {data} FROM {name} \
534 WHERE NOT EXISTS ( \
535 SELECT 1 FROM sync_changelog sc \
536 WHERE sc.table_name = '{name}' AND sc.row_id = {row_id} \
537 ){exclude}"
538 );
539 total += conn.execute(&sql, [])? as u64;
540 }
541 set_sync_state(conn, "initial_snapshot_done", "1")?;
542 Ok(total)
543 }
544
545 /// Prune pushed changelog entries older than seven days.
546 pub fn cleanup_changelog(conn: &Connection) -> Result<u64> {
547 let n = conn.execute(
548 "DELETE FROM sync_changelog WHERE pushed = 1 AND timestamp < datetime('now', '-7 days')",
549 [],
550 )?;
551 Ok(n as u64)
552 }
553
554 /// Cap changelog size: drop the oldest *pushed* entries beyond the most recent
555 /// `cap`. Unpushed entries are never dropped (that would lose un-synced data), so
556 /// a backlog larger than `cap` is preserved until it syncs.
557 pub fn enforce_changelog_retention(conn: &Connection, cap: i64) -> Result<u64> {
558 let n = conn.execute(
559 "DELETE FROM sync_changelog WHERE pushed = 1 AND id NOT IN ( \
560 SELECT id FROM sync_changelog ORDER BY id DESC LIMIT ?1 \
561 )",
562 [cap],
563 )?;
564 Ok(n as u64)
565 }
566
567 #[cfg(test)]
568 mod tests {
569 use super::super::db::get_sync_state_or;
570 use super::super::schema::SyncTable;
571 use super::*;
572 use std::sync::{Arc, Mutex};
573
574 /// A shared in-memory "server": an append-only personal log of (origin
575 /// device, entry), plus a group log of (group, origin device, entry).
576 #[derive(Clone, Default)]
577 struct FakeServer {
578 log: Arc<Mutex<Vec<(DeviceId, ChangeEntry)>>>,
579 group_log: Arc<Mutex<Vec<(GroupId, DeviceId, ChangeEntry)>>>,
580 }
581
582 impl SyncTransport for FakeServer {
583 fn group_scope_push(
584 &self,
585 group_id: GroupId,
586 _gck_version: i32,
587 device_id: DeviceId,
588 changes: Vec<ChangeEntry>,
589 ) -> impl Future<Output = Result<i64>> + Send {
590 let group_log = self.group_log.clone();
591 async move {
592 let mut l = group_log.lock().unwrap();
593 for c in changes {
594 l.push((group_id, device_id, c));
595 }
596 Ok(l.len() as i64)
597 }
598 }
599
600 async fn register_device(&self, _name: &str, _platform: &str) -> Result<Device> {
601 Ok(Device {
602 id: DeviceId::new(uuid::Uuid::from_u128(0xDE)),
603 app_id: crate::ids::AppId::nil(),
604 user_id: crate::ids::UserId::nil(),
605 device_name: "fake".into(),
606 platform: "test".into(),
607 last_seen_at: Utc::now(),
608 created_at: Utc::now(),
609 })
610 }
611
612 fn push(
613 &self,
614 device_id: DeviceId,
615 changes: Vec<ChangeEntry>,
616 ) -> impl Future<Output = Result<i64>> + Send {
617 let log = self.log.clone();
618 async move {
619 let mut l = log.lock().unwrap();
620 for c in changes {
621 l.push((device_id, c));
622 }
623 Ok(l.len() as i64)
624 }
625 }
626
627 fn pull_rich(
628 &self,
629 _device_id: DeviceId,
630 cursor: i64,
631 ) -> impl Future<Output = Result<(Vec<PulledChange>, i64, bool)>> + Send {
632 let log = self.log.clone();
633 async move {
634 let l = log.lock().unwrap();
635 let out: Vec<PulledChange> = l
636 .iter()
637 .enumerate()
638 .filter(|(i, _)| (*i as i64 + 1) > cursor)
639 .map(|(i, (dev, entry))| PulledChange {
640 entry: entry.clone(),
641 device_id: *dev,
642 seq: i as i64 + 1,
643 })
644 .collect();
645 Ok((out, l.len() as i64, false))
646 }
647 }
648 }
649
650 fn schema() -> SyncSchema {
651 SyncSchema::new(vec![SyncTable::full("note", &["id", "name"])])
652 }
653
654 /// A schema with a group-scoped table: `task` carries a local `group_id`
655 /// provenance column (not a synced column), declared via `group_scoped`.
656 fn group_schema() -> SyncSchema {
657 SyncSchema::new(vec![
658 SyncTable::full("task", &["id", "name"]).group_scoped("group_id"),
659 ])
660 }
661
662 fn group_device(path: &std::path::Path, n: u128) -> (DbSource, DeviceId) {
663 let db = DbSource::path(path);
664 let conn = db.open().unwrap();
665 conn.execute_batch("CREATE TABLE task (id TEXT PRIMARY KEY, name TEXT, group_id TEXT);")
666 .unwrap();
667 conn.execute_batch(&group_schema().migration_sql()).unwrap();
668 (db, DeviceId::new(uuid::Uuid::from_u128(n)))
669 }
670
671 #[tokio::test]
672 async fn push_scope_drains_only_its_own_scope() {
673 let dir = tempdir();
674 let (db, node) = group_device(&dir.join("g.db"), 7);
675 let server = FakeServer::default();
676 let gid = GroupId::new(uuid::Uuid::from_u128(0x6971));
677
678 // A personal task (group_id NULL) and a group task (group_id = gid).
679 {
680 let c = db.open().unwrap();
681 c.execute(
682 "INSERT INTO task (id, name, group_id) VALUES ('p', 'personal', NULL)",
683 [],
684 )
685 .unwrap();
686 c.execute(
687 "INSERT INTO task (id, name, group_id) VALUES ('g', 'grouped', ?1)",
688 [gid.to_string()],
689 )
690 .unwrap();
691 }
692
693 // Personal push drains only the personal row.
694 let pushed = push_scope(&db, &server, node, SyncScope::Personal)
695 .await
696 .unwrap();
697 assert_eq!(pushed, 1);
698 assert_eq!(server.log.lock().unwrap().len(), 1);
699 assert_eq!(server.log.lock().unwrap()[0].1.row_id, "p");
700 assert!(server.group_log.lock().unwrap().is_empty());
701
702 // Group push drains only the group row, to that group.
703 let pushed = push_scope(
704 &db,
705 &server,
706 node,
707 SyncScope::Group {
708 id: gid,
709 gck_version: 1,
710 },
711 )
712 .await
713 .unwrap();
714 assert_eq!(pushed, 1);
715 let gl = server.group_log.lock().unwrap();
716 assert_eq!(gl.len(), 1);
717 assert_eq!(gl[0].0, gid);
718 assert_eq!(gl[0].2.row_id, "g");
719 // The personal log did not grow.
720 assert_eq!(server.log.lock().unwrap().len(), 1);
721 }
722
723 fn device(path: &std::path::Path, n: u128) -> (DbSource, DeviceId) {
724 let db = DbSource::path(path);
725 let conn = db.open().unwrap();
726 conn.execute_batch("CREATE TABLE note (id TEXT PRIMARY KEY, name TEXT);")
727 .unwrap();
728 conn.execute_batch(&schema().migration_sql()).unwrap();
729 (db, DeviceId::new(uuid::Uuid::from_u128(n)))
730 }
731
732 fn edit(db: &DbSource, id: &str, name: &str) {
733 let conn = db.open().unwrap();
734 conn.execute(
735 "INSERT INTO note (id, name) VALUES (?1, ?2) ON CONFLICT(id) DO UPDATE SET name = excluded.name",
736 (id, name),
737 )
738 .unwrap();
739 }
740
741 fn stamp_at(db: &DbSource, node: DeviceId, now_ms: i64) {
742 stamp_pending(&db.open().unwrap(), node, now_ms).unwrap();
743 }
744
745 fn note_name(db: &DbSource, id: &str) -> Option<String> {
746 db.open()
747 .unwrap()
748 .query_row("SELECT name FROM note WHERE id = ?1", [id], |r| r.get(0))
749 .ok()
750 }
751
752 #[tokio::test]
753 async fn two_device_push_pull_converges_to_higher_hlc() {
754 let dir = tempdir();
755 let (da, na) = device(&dir.join("a.db"), 1);
756 let (db_, nb) = device(&dir.join("b.db"), 2);
757 let server = FakeServer::default();
758
759 // A edits first (t=100), B edits the same row later (t=200) → B wins.
760 edit(&da, "r", "from-A");
761 stamp_at(&da, na, 100);
762 edit(&db_, "r", "from-B");
763 stamp_at(&db_, nb, 200);
764
765 push_changes(&da, &server, na).await.unwrap();
766 push_changes(&db_, &server, nb).await.unwrap();
767
768 pull_changes(&da, &server, &schema(), na).await.unwrap();
769 pull_changes(&db_, &server, &schema(), nb).await.unwrap();
770
771 assert_eq!(note_name(&da, "r").as_deref(), Some("from-B"));
772 assert_eq!(note_name(&db_, "r").as_deref(), Some("from-B"));
773 }
774
775 #[tokio::test]
776 async fn push_marks_rows_and_advances_cursor() {
777 let dir = tempdir();
778 let (da, na) = device(&dir.join("a.db"), 1);
779 let server = FakeServer::default();
780 edit(&da, "r1", "x");
781 edit(&da, "r2", "y");
782
783 let pushed = push_changes(&da, &server, na).await.unwrap();
784 assert_eq!(pushed, 2);
785 // All local rows are marked pushed.
786 let unpushed: i64 = da
787 .open()
788 .unwrap()
789 .query_row(
790 "SELECT COUNT(*) FROM sync_changelog WHERE pushed = 0",
791 [],
792 |r| r.get(0),
793 )
794 .unwrap();
795 assert_eq!(unpushed, 0);
796
797 // A fresh device pulls both and advances its cursor to 2.
798 let (db_, nb) = device(&dir.join("b.db"), 2);
799 let out = pull_changes(&db_, &server, &schema(), nb).await.unwrap();
800 assert_eq!(out.applied, 2);
801 assert_eq!(note_name(&db_, "r1").as_deref(), Some("x"));
802 // Personal pull advances the personal ('') scope cursor.
803 let cursor = get_scope_cursor(&db_.open().unwrap(), "").unwrap();
804 assert_eq!(cursor, 2);
805 }
806
807 #[test]
808 fn initial_snapshot_captures_existing_rows_once() {
809 let dir = tempdir();
810 let (da, _) = device(&dir.join("a.db"), 1);
811 let conn = da.open().unwrap();
812 // Pre-existing rows inserted with triggers suppressed (as if before sync).
813 conn.execute("INSERT INTO note (id, name) VALUES ('r1', 'a')", [])
814 .unwrap();
815 conn.execute("INSERT INTO note (id, name) VALUES ('r2', 'b')", [])
816 .unwrap();
817 // (the inserts above DID fire triggers; clear the changelog to simulate a
818 // pre-sync backfill scenario cleanly)
819 conn.execute("DELETE FROM sync_changelog", []).unwrap();
820
821 let n = create_initial_snapshot(&conn, &schema()).unwrap();
822 assert_eq!(n, 2);
823 // Idempotent: a second snapshot adds nothing.
824 assert_eq!(create_initial_snapshot(&conn, &schema()).unwrap(), 0);
825 let done = get_sync_state_or(&conn, "initial_snapshot_done", "0").unwrap();
826 assert_eq!(done, "1");
827 }
828
829 #[test]
830 fn retention_caps_pushed_but_keeps_unpushed() {
831 let dir = tempdir();
832 let (da, _) = device(&dir.join("a.db"), 1);
833 let conn = da.open().unwrap();
834 conn.execute("DELETE FROM sync_changelog", []).unwrap();
835 // 5 pushed + 2 unpushed entries.
836 for i in 0..5 {
837 conn.execute(
838 "INSERT INTO sync_changelog (table_name, op, row_id, data, pushed) VALUES ('note','INSERT',?1,'{}',1)",
839 [format!("p{i}")],
840 )
841 .unwrap();
842 }
843 for i in 0..2 {
844 conn.execute(
845 "INSERT INTO sync_changelog (table_name, op, row_id, data, pushed) VALUES ('note','INSERT',?1,'{}',0)",
846 [format!("u{i}")],
847 )
848 .unwrap();
849 }
850 // Cap to 3: keep the 3 most recent rows, drop older PUSHED ones only.
851 let dropped = enforce_changelog_retention(&conn, 3).unwrap();
852 let unpushed: i64 = conn
853 .query_row(
854 "SELECT COUNT(*) FROM sync_changelog WHERE pushed = 0",
855 [],
856 |r| r.get(0),
857 )
858 .unwrap();
859 assert_eq!(unpushed, 2, "unpushed entries are never dropped");
860 assert!(dropped >= 1);
861 }
862
863 // minimal temp-dir helper (no external dep)
864 fn tempdir() -> std::path::PathBuf {
865 let mut p = std::env::temp_dir();
866 // Unique-ish per test via a monotonic counter; tests run in the same
867 // process so a static AtomicU64 keeps paths distinct.
868 use std::sync::atomic::{AtomicU64, Ordering};
869 static N: AtomicU64 = AtomicU64::new(0);
870 p.push(format!(
871 "synckit_b5_{}_{}",
872 std::process::id(),
873 N.fetch_add(1, Ordering::Relaxed)
874 ));
875 std::fs::create_dir_all(&p).unwrap();
876 p
877 }
878 }
879