Skip to main content

max / synckit

30.0 KB · 928 lines History Blame Raw
1 //! Tests for [`super`].
2
3 use super::super::db::get_sync_state_or;
4 use super::super::schema::SyncTable;
5 use super::*;
6 use std::sync::{Arc, Mutex};
7
8 /// A shared in-memory "server": an append-only personal log of (origin
9 /// device, entry), plus a group log of (group, origin device, entry).
10 #[derive(Clone, Default)]
11 struct FakeServer {
12 log: Arc<Mutex<Vec<(DeviceId, ChangeEntry)>>>,
13 group_log: Arc<Mutex<Vec<(GroupId, DeviceId, ChangeEntry)>>>,
14 /// The storage version a pulled page appears to have been sealed under.
15 /// Stands in for the `__sksv` the real transport reads out of the
16 /// envelope; the fake log holds decrypted entries, so there is no
17 /// envelope here to carry it.
18 peer_version: Arc<Mutex<Option<u32>>>,
19 }
20
21 impl SyncTransport for FakeServer {
22 fn group_scope_push(
23 &self,
24 group_id: GroupId,
25 _gck_version: i32,
26 device_id: DeviceId,
27 changes: Vec<ChangeEntry>,
28 ) -> impl Future<Output = Result<i64>> + Send {
29 let group_log = self.group_log.clone();
30 async move {
31 let mut l = group_log.lock().unwrap();
32 for c in changes {
33 l.push((group_id, device_id, c));
34 }
35 Ok(l.len() as i64)
36 }
37 }
38
39 async fn register_device(&self, _name: &str, _platform: &str) -> Result<Device> {
40 Ok(Device {
41 id: DeviceId::new(uuid::Uuid::from_u128(0xDE)),
42 app_id: crate::ids::AppId::nil(),
43 user_id: crate::ids::UserId::nil(),
44 device_name: "fake".into(),
45 platform: "test".into(),
46 last_seen_at: Utc::now(),
47 created_at: Utc::now(),
48 })
49 }
50
51 fn push(
52 &self,
53 device_id: DeviceId,
54 changes: Vec<ChangeEntry>,
55 ) -> impl Future<Output = Result<i64>> + Send {
56 let log = self.log.clone();
57 async move {
58 let mut l = log.lock().unwrap();
59 for c in changes {
60 l.push((device_id, c));
61 }
62 Ok(l.len() as i64)
63 }
64 }
65
66 fn pull_rich(
67 &self,
68 _device_id: DeviceId,
69 cursor: i64,
70 ) -> impl Future<Output = Result<(Vec<PulledChange>, i64, bool)>> + Send {
71 let log = self.log.clone();
72 let peer_version = *self.peer_version.lock().unwrap();
73 async move {
74 let l = log.lock().unwrap();
75 let out: Vec<PulledChange> = l
76 .iter()
77 .enumerate()
78 .filter(|(i, _)| (*i as i64 + 1) > cursor)
79 .map(|(i, (dev, entry))| PulledChange {
80 storage_version: peer_version,
81 entry: entry.clone(),
82 device_id: *dev,
83 seq: i as i64 + 1,
84 })
85 .collect();
86 Ok((out, l.len() as i64, false))
87 }
88 }
89 }
90
91 fn schema() -> SyncSchema {
92 SyncSchema::new(vec![SyncTable::full("note", &["id", "name"])])
93 }
94
95 /// A schema with a group-scoped table: `task` carries a local `group_id`
96 /// provenance column (not a synced column), declared via `group_scoped`.
97 fn group_schema() -> SyncSchema {
98 SyncSchema::new(vec![
99 SyncTable::full("task", &["id", "name"]).group_scoped("group_id"),
100 ])
101 }
102
103 fn group_device(path: &std::path::Path, n: u128) -> (DbSource, DeviceId) {
104 let db = DbSource::path(path);
105 let conn = db.open().unwrap();
106 conn.execute_batch("CREATE TABLE task (id TEXT PRIMARY KEY, name TEXT, group_id TEXT);")
107 .unwrap();
108 conn.execute_batch(&group_schema().migration_sql()).unwrap();
109 (db, DeviceId::new(uuid::Uuid::from_u128(n)))
110 }
111
112 #[tokio::test]
113 async fn push_scope_drains_only_its_own_scope() {
114 let dir = tempdir();
115 let (db, node) = group_device(&dir.join("g.db"), 7);
116 let server = FakeServer::default();
117 let gid = GroupId::new(uuid::Uuid::from_u128(0x6971));
118
119 // A personal task (group_id NULL) and a group task (group_id = gid).
120 {
121 let c = db.open().unwrap();
122 c.execute(
123 "INSERT INTO task (id, name, group_id) VALUES ('p', 'personal', NULL)",
124 [],
125 )
126 .unwrap();
127 c.execute(
128 "INSERT INTO task (id, name, group_id) VALUES ('g', 'grouped', ?1)",
129 [gid.to_string()],
130 )
131 .unwrap();
132 }
133
134 // Personal push drains only the personal row.
135 let pushed = push_scope(&db, &server, &group_schema(), node, SyncScope::Personal)
136 .await
137 .unwrap();
138 assert_eq!(pushed, 1);
139 assert_eq!(server.log.lock().unwrap().len(), 1);
140 assert_eq!(server.log.lock().unwrap()[0].1.row_id, "p");
141 assert!(server.group_log.lock().unwrap().is_empty());
142
143 // Group push drains only the group row, to that group.
144 let pushed = push_scope(
145 &db,
146 &server,
147 &group_schema(),
148 node,
149 SyncScope::Group {
150 id: gid,
151 gck_version: 1,
152 },
153 )
154 .await
155 .unwrap();
156 assert_eq!(pushed, 1);
157 let gl = server.group_log.lock().unwrap();
158 assert_eq!(gl.len(), 1);
159 assert_eq!(gl[0].0, gid);
160 assert_eq!(gl[0].2.row_id, "g");
161 // The personal log did not grow.
162 assert_eq!(server.log.lock().unwrap().len(), 1);
163 }
164
165 fn device(path: &std::path::Path, n: u128) -> (DbSource, DeviceId) {
166 let db = DbSource::path(path);
167 let conn = db.open().unwrap();
168 conn.execute_batch("CREATE TABLE note (id TEXT PRIMARY KEY, name TEXT);")
169 .unwrap();
170 conn.execute_batch(&schema().migration_sql()).unwrap();
171 (db, DeviceId::new(uuid::Uuid::from_u128(n)))
172 }
173
174 /// The gate's end-to-end case: a peer on a different manifest is refused
175 /// before anything is applied, and the cursor does not move, so the page is
176 /// still there once both sides agree.
177 #[tokio::test]
178 async fn a_peer_on_another_storage_version_is_refused_and_nothing_lands() {
179 let dir = tempdir();
180 let (writer, writer_node) = device(&dir.join("a.db"), 1);
181 let (reader, reader_node) = device(&dir.join("b.db"), 2);
182 let server = FakeServer::default();
183 let gated = schema().storage_version(4);
184
185 edit(&writer, "n1", "from the newer device");
186 push_scope(&writer, &server, &gated, writer_node, SyncScope::Personal)
187 .await
188 .unwrap();
189 // The peer is a manifest ahead.
190 *server.peer_version.lock().unwrap() = Some(5);
191
192 let err = pull_scope(&reader, &server, &gated, reader_node, SyncScope::Personal)
193 .await
194 .unwrap_err();
195 let r = match err {
196 SyncKitError::StorageVersion(r) => r,
197 other => panic!("expected a storage-version refusal, got {other:?}"),
198 };
199 assert_eq!((r.mine, r.theirs), (4, 5));
200 assert_eq!(r.message(), "Update this device.");
201
202 // No partial write, no dropped records, and the cursor is where it was.
203 let conn = reader.open().unwrap();
204 let rows: i64 = conn
205 .query_row("SELECT COUNT(*) FROM note", [], |r| r.get(0))
206 .unwrap();
207 assert_eq!(rows, 0, "nothing was applied");
208 assert_eq!(
209 get_scope_cursor(&conn, "").unwrap(),
210 0,
211 "the cursor did not advance past a page that was never applied"
212 );
213
214 // Once the reader catches up, the same page applies.
215 let matched = schema().storage_version(5);
216 pull_scope(&reader, &server, &matched, reader_node, SyncScope::Personal)
217 .await
218 .unwrap();
219 let name: String = conn
220 .query_row("SELECT name FROM note WHERE id = 'n1'", [], |r| r.get(0))
221 .unwrap();
222 assert_eq!(name, "from the newer device");
223 }
224
225 /// An app that has not adopted the gate must be entirely unaffected.
226 #[tokio::test]
227 async fn an_undeclared_manifest_pulls_a_stamped_page_as_before() {
228 let dir = tempdir();
229 let (writer, writer_node) = device(&dir.join("a.db"), 1);
230 let (reader, reader_node) = device(&dir.join("b.db"), 2);
231 let server = FakeServer::default();
232
233 edit(&writer, "n1", "hello");
234 push_scope(
235 &writer,
236 &server,
237 &schema(),
238 writer_node,
239 SyncScope::Personal,
240 )
241 .await
242 .unwrap();
243 *server.peer_version.lock().unwrap() = Some(9);
244
245 pull_scope(
246 &reader,
247 &server,
248 &schema(),
249 reader_node,
250 SyncScope::Personal,
251 )
252 .await
253 .unwrap();
254 let conn = reader.open().unwrap();
255 let name: String = conn
256 .query_row("SELECT name FROM note WHERE id = 'n1'", [], |r| r.get(0))
257 .unwrap();
258 assert_eq!(name, "hello");
259 }
260
261 fn edit(db: &DbSource, id: &str, name: &str) {
262 let conn = db.open().unwrap();
263 conn.execute(
264 "INSERT INTO note (id, name) VALUES (?1, ?2) ON CONFLICT(id) DO UPDATE SET name = excluded.name",
265 (id, name),
266 )
267 .unwrap();
268 }
269
270 /// A schema whose `child` table has a real foreign key to `parent`, so a
271 /// child arriving first violates a constraint instead of quietly landing.
272 fn fk_schema() -> SyncSchema {
273 SyncSchema::new(vec![
274 SyncTable::full("parent", &["id", "name"]),
275 SyncTable::full("child", &["id", "parent_id"]),
276 ])
277 }
278
279 fn fk_device(path: &std::path::Path, n: u128) -> (DbSource, DeviceId) {
280 let db = DbSource::path(path);
281 let conn = db.open().unwrap();
282 conn.execute_batch(
283 "CREATE TABLE parent (id TEXT PRIMARY KEY, name TEXT);
284 CREATE TABLE child (id TEXT PRIMARY KEY, parent_id TEXT NOT NULL REFERENCES parent(id));",
285 )
286 .unwrap();
287 conn.execute_batch(&fk_schema().migration_sql()).unwrap();
288 (db, DeviceId::new(uuid::Uuid::from_u128(n)))
289 }
290
291 /// Put an entry on the fake server as if another device had pushed it.
292 fn serve(server: &FakeServer, table: &str, row_id: &str, data: serde_json::Value) {
293 server.log.lock().unwrap().push((
294 DeviceId::new(uuid::Uuid::from_u128(0xAA)),
295 ChangeEntry {
296 table: table.into(),
297 op: ChangeOp::Insert,
298 row_id: row_id.into(),
299 timestamp: Utc::now(),
300 hlc: crate::types::hlc_legacy_floor(),
301 data: Some(data),
302 extra: serde_json::Map::default(),
303 },
304 ));
305 }
306
307 fn row_count(db: &DbSource, table: &str) -> i64 {
308 db.open()
309 .unwrap()
310 .query_row(&format!("SELECT COUNT(*) FROM {table}"), [], |r| r.get(0))
311 .unwrap()
312 }
313
314 #[tokio::test]
315 async fn a_child_that_arrives_before_its_parent_is_held_and_lands_on_the_next_pull() {
316 let dir = tempdir();
317 let (db, node) = fk_device(&dir.join("fk.db"), 11);
318 let server = FakeServer::default();
319
320 // The child arrives alone. Its parent does not exist yet, so it cannot be
321 // written; before the hold existed this row was gone for good, because the
322 // cursor moved past it and the server never sends an entry twice.
323 serve(
324 &server,
325 "child",
326 "c1",
327 serde_json::json!({"id":"c1","parent_id":"p1"}),
328 );
329 let out = pull_scope(&db, &server, &fk_schema(), node, SyncScope::Personal)
330 .await
331 .unwrap();
332
333 assert_eq!(out.applied, 0);
334 assert_eq!(out.deferred, 1);
335 assert_eq!(row_count(&db, "child"), 0);
336 {
337 let conn = db.open().unwrap();
338 assert_eq!(
339 get_scope_cursor(&conn, "").unwrap(),
340 1,
341 "the cursor still advances; the hold is what makes that safe"
342 );
343 assert_eq!(deferred::counts(&conn, "").unwrap().deferred, 1);
344 }
345
346 // The parent lands on the next pull, and the held child rides in with it.
347 serve(
348 &server,
349 "parent",
350 "p1",
351 serde_json::json!({"id":"p1","name":"p"}),
352 );
353 let out = pull_scope(&db, &server, &fk_schema(), node, SyncScope::Personal)
354 .await
355 .unwrap();
356
357 assert_eq!(out.applied, 2, "the new parent plus the retried child");
358 assert_eq!(out.deferred, 0);
359 assert_eq!(row_count(&db, "child"), 1);
360 assert_eq!(
361 deferred::counts(&db.open().unwrap(), "").unwrap().total(),
362 0,
363 "a held entry that lands is cleared"
364 );
365 }
366
367 #[tokio::test]
368 async fn a_parent_that_never_arrives_stops_being_retried_at_the_cap() {
369 let dir = tempdir();
370 let (db, node) = fk_device(&dir.join("fk_cap.db"), 12);
371 let server = FakeServer::default();
372
373 serve(
374 &server,
375 "child",
376 "c1",
377 serde_json::json!({"id":"c1","parent_id":"nope"}),
378 );
379
380 // Each pull spends one attempt. The first holds it; MAX_ATTEMPTS more
381 // exhaust it. An empty page still runs the retry, which is the point.
382 for _ in 0..=deferred::MAX_ATTEMPTS {
383 pull_scope(&db, &server, &fk_schema(), node, SyncScope::Personal)
384 .await
385 .unwrap();
386 }
387
388 let conn = db.open().unwrap();
389 let counts = deferred::counts(&conn, "").unwrap();
390 assert_eq!(counts.deferred, 0, "no longer retried");
391 assert_eq!(counts.rejected, 1, "but still visible, not discarded");
392 let listed = deferred::list(&conn, "").unwrap();
393 assert_eq!(listed[0].row_id, "c1");
394 assert_eq!(listed[0].attempts, deferred::MAX_ATTEMPTS);
395 }
396
397 #[tokio::test]
398 async fn a_deferred_entry_is_not_recorded_as_committed() {
399 let dir = tempdir();
400 let (db, node) = fk_device(&dir.join("fk_gate.db"), 13);
401 let server = FakeServer::default();
402
403 serve(
404 &server,
405 "child",
406 "c1",
407 serde_json::json!({"id":"c1","parent_id":"p1"}),
408 );
409 pull_scope(&db, &server, &fk_schema(), node, SyncScope::Personal)
410 .await
411 .unwrap();
412
413 // Recording an unapplied row's HLC would gate its own retry out on the
414 // next pull, since the gate drops anything not newer than what is
415 // committed, and the hold would be a queue that never drains.
416 assert!(
417 super::super::hlc::committed_hlc(&db.open().unwrap(), "child", "c1")
418 .unwrap()
419 .is_none()
420 );
421 }
422
423 fn stamp_at(db: &DbSource, node: DeviceId, now_ms: i64) {
424 stamp_pending(&db.open().unwrap(), node, now_ms).unwrap();
425 }
426
427 fn note_name(db: &DbSource, id: &str) -> Option<String> {
428 db.open()
429 .unwrap()
430 .query_row("SELECT name FROM note WHERE id = ?1", [id], |r| r.get(0))
431 .ok()
432 }
433
434 #[tokio::test]
435 async fn two_device_push_pull_converges_to_higher_hlc() {
436 let dir = tempdir();
437 let (da, na) = device(&dir.join("a.db"), 1);
438 let (db_, nb) = device(&dir.join("b.db"), 2);
439 let server = FakeServer::default();
440
441 // A edits first (t=100), B edits the same row later (t=200) → B wins.
442 edit(&da, "r", "from-A");
443 stamp_at(&da, na, 100);
444 edit(&db_, "r", "from-B");
445 stamp_at(&db_, nb, 200);
446
447 push_changes(&da, &server, &schema(), na).await.unwrap();
448 push_changes(&db_, &server, &schema(), nb).await.unwrap();
449
450 pull_changes(&da, &server, &schema(), na).await.unwrap();
451 pull_changes(&db_, &server, &schema(), nb).await.unwrap();
452
453 assert_eq!(note_name(&da, "r").as_deref(), Some("from-B"));
454 assert_eq!(note_name(&db_, "r").as_deref(), Some("from-B"));
455 }
456
457 #[tokio::test]
458 async fn push_marks_rows_and_advances_cursor() {
459 let dir = tempdir();
460 let (da, na) = device(&dir.join("a.db"), 1);
461 let server = FakeServer::default();
462 edit(&da, "r1", "x");
463 edit(&da, "r2", "y");
464
465 let pushed = push_changes(&da, &server, &schema(), na).await.unwrap();
466 assert_eq!(pushed, 2);
467 // All local rows are marked pushed.
468 let unpushed: i64 = da
469 .open()
470 .unwrap()
471 .query_row(
472 "SELECT COUNT(*) FROM sync_changelog WHERE pushed = 0",
473 [],
474 |r| r.get(0),
475 )
476 .unwrap();
477 assert_eq!(unpushed, 0);
478
479 // A fresh device pulls both and advances its cursor to 2.
480 let (db_, nb) = device(&dir.join("b.db"), 2);
481 let out = pull_changes(&db_, &server, &schema(), nb).await.unwrap();
482 assert_eq!(out.applied, 2);
483 assert_eq!(note_name(&db_, "r1").as_deref(), Some("x"));
484 // Personal pull advances the personal ('') scope cursor.
485 let cursor = get_scope_cursor(&db_.open().unwrap(), "").unwrap();
486 assert_eq!(cursor, 2);
487 }
488
489 /// The push half of the field-merge base.
490 ///
491 /// A pull is the obvious moment a row becomes common ground and it is only
492 /// half of them: once the server takes an edit, that edit is what a peer will
493 /// pull, so it is the version the two devices next diverge from. Re-basing
494 /// only on pull would leave the base stuck at whatever this device last
495 /// *received*, and every merge afterwards would report this device's own
496 /// already-shared edits as changes, handing itself fields it never contested.
497 #[tokio::test]
498 async fn an_acknowledged_push_rebases_the_row() {
499 let dir = tempdir();
500 let db = DbSource::path(dir.join("a.db"));
501 {
502 let conn = db.open().unwrap();
503 conn.execute_batch("CREATE TABLE note (id TEXT PRIMARY KEY, name TEXT);")
504 .unwrap();
505 conn.execute_batch(&merge_schema().migration_sql()).unwrap();
506 }
507 let node = DeviceId::new(uuid::Uuid::from_u128(1));
508 let server = FakeServer::default();
509
510 edit(&db, "r1", "mine");
511 assert_eq!(
512 base(&db, "r1"),
513 serde_json::Value::Null,
514 "nothing shared yet"
515 );
516
517 push_changes(&db, &server, &merge_schema(), node)
518 .await
519 .unwrap();
520 assert_eq!(
521 base(&db, "r1")["name"],
522 "mine",
523 "the server took this edit, so it is the version a peer will pull"
524 );
525
526 // And a pushed delete drops the base rather than leaving it describing a
527 // row that no longer exists.
528 db.open()
529 .unwrap()
530 .execute("DELETE FROM note WHERE id = 'r1'", [])
531 .unwrap();
532 push_changes(&db, &server, &merge_schema(), node)
533 .await
534 .unwrap();
535 assert_eq!(base(&db, "r1"), serde_json::Value::Null);
536 }
537
538 /// A table that did not opt in must store no base, so the storage cost lands
539 /// only where someone asked for it.
540 #[tokio::test]
541 async fn a_push_stores_no_base_for_a_table_that_did_not_opt_in() {
542 let dir = tempdir();
543 let (db, node) = device(&dir.join("a.db"), 1);
544 let server = FakeServer::default();
545
546 edit(&db, "r1", "mine");
547 push_changes(&db, &server, &schema(), node).await.unwrap();
548
549 assert_eq!(base(&db, "r1"), serde_json::Value::Null);
550 }
551
552 fn merge_schema() -> SyncSchema {
553 SyncSchema::new(vec![
554 SyncTable::full("note", &["id", "name"]).field_merge(&[]),
555 ])
556 }
557
558 fn base(db: &DbSource, row_id: &str) -> serde_json::Value {
559 super::snapshot::load(&db.open().unwrap(), "note", row_id)
560 }
561
562 #[test]
563 fn initial_snapshot_captures_existing_rows_once() {
564 let dir = tempdir();
565 let (da, _) = device(&dir.join("a.db"), 1);
566 let conn = da.open().unwrap();
567 // Pre-existing rows inserted with triggers suppressed (as if before sync).
568 conn.execute("INSERT INTO note (id, name) VALUES ('r1', 'a')", [])
569 .unwrap();
570 conn.execute("INSERT INTO note (id, name) VALUES ('r2', 'b')", [])
571 .unwrap();
572 // (the inserts above DID fire triggers; clear the changelog to simulate a
573 // pre-sync backfill scenario cleanly)
574 conn.execute("DELETE FROM sync_changelog", []).unwrap();
575
576 let n = create_initial_snapshot(&conn, &schema()).unwrap();
577 assert_eq!(n, 2);
578 // Idempotent: a second snapshot adds nothing.
579 assert_eq!(create_initial_snapshot(&conn, &schema()).unwrap(), 0);
580 let done = get_sync_state_or(&conn, "initial_snapshot_done", "0").unwrap();
581 assert_eq!(done, "1");
582 }
583
584 #[test]
585 fn retention_caps_pushed_but_keeps_unpushed() {
586 let dir = tempdir();
587 let (da, _) = device(&dir.join("a.db"), 1);
588 let conn = da.open().unwrap();
589 conn.execute("DELETE FROM sync_changelog", []).unwrap();
590 // 5 pushed + 2 unpushed entries.
591 for i in 0..5 {
592 conn.execute(
593 "INSERT INTO sync_changelog (table_name, op, row_id, data, pushed) VALUES ('note','INSERT',?1,'{}',1)",
594 [format!("p{i}")],
595 )
596 .unwrap();
597 }
598 for i in 0..2 {
599 conn.execute(
600 "INSERT INTO sync_changelog (table_name, op, row_id, data, pushed) VALUES ('note','INSERT',?1,'{}',0)",
601 [format!("u{i}")],
602 )
603 .unwrap();
604 }
605 let before: i64 = conn
606 .query_row("SELECT COUNT(*) FROM sync_changelog", [], |r| r.get(0))
607 .unwrap();
608 // Cap to 3: keep the 3 most recent rows, drop older PUSHED ones only.
609 let dropped = enforce_changelog_retention(&conn, 3).unwrap();
610 let after: i64 = conn
611 .query_row("SELECT COUNT(*) FROM sync_changelog", [], |r| r.get(0))
612 .unwrap();
613 let unpushed: i64 = conn
614 .query_row(
615 "SELECT COUNT(*) FROM sync_changelog WHERE pushed = 0",
616 [],
617 |r| r.get(0),
618 )
619 .unwrap();
620 assert_eq!(unpushed, 2, "unpushed entries are never dropped");
621 // The 3 newest rows by id are the 2 unpushed plus the newest pushed one,
622 // so exactly the 4 oldest pushed rows go. A count that is not 4 - a
623 // hardcoded 0 or 1, or the row count of some other query - disagrees.
624 assert_eq!(
625 dropped, 4,
626 "the four oldest pushed rows are the only ones dropped"
627 );
628 assert_eq!(
629 dropped,
630 (before - after) as u64,
631 "the returned count is the number of rows actually deleted"
632 );
633 assert_eq!(after, 3, "the cap is the number of surviving rows here");
634 }
635
636 #[test]
637 fn retention_keeps_exactly_cap_rows_and_drops_the_next_one() {
638 let dir = tempdir();
639 let (da, _) = device(&dir.join("retention_edge.db"), 27);
640 let conn = da.open().unwrap();
641 conn.execute("DELETE FROM sync_changelog", []).unwrap();
642 // 6 pushed rows, ids ascending with insertion order.
643 for i in 0..6 {
644 conn.execute(
645 "INSERT INTO sync_changelog (table_name, op, row_id, data, pushed) VALUES ('note','INSERT',?1,'{}',1)",
646 [format!("p{i}")],
647 )
648 .unwrap();
649 }
650 let ids: Vec<i64> = {
651 let mut st = conn
652 .prepare("SELECT id FROM sync_changelog ORDER BY id ASC")
653 .unwrap();
654 let rows: Vec<i64> = st
655 .query_map([], |r| r.get(0))
656 .unwrap()
657 .map(|r| r.unwrap())
658 .collect();
659 rows
660 };
661 assert_eq!(ids.len(), 6);
662
663 // Cap 4 of 6 keeps the newest four and drops two. Both sides of the
664 // boundary are named: ids[1] is the last row dropped, ids[2] the first
665 // row kept, so an off-by-one in the LIMIT changes the answer.
666 let dropped = enforce_changelog_retention(&conn, 4).unwrap();
667 assert_eq!(dropped, 2, "6 rows capped at 4 drops 2");
668 let survivors: Vec<i64> = {
669 let mut st = conn
670 .prepare("SELECT id FROM sync_changelog ORDER BY id ASC")
671 .unwrap();
672 st.query_map([], |r| r.get(0))
673 .unwrap()
674 .map(|r| r.unwrap())
675 .collect()
676 };
677 assert_eq!(survivors, ids[2..].to_vec(), "the newest four rows survive");
678 assert!(
679 !survivors.contains(&ids[1]),
680 "the row just past the cap is gone"
681 );
682
683 // Re-running at the same cap is a no-op: nothing is left to drop, so a
684 // constant return value of 1 or 2 disagrees with 0 here.
685 assert_eq!(
686 enforce_changelog_retention(&conn, 4).unwrap(),
687 0,
688 "a second pass at the same cap drops nothing"
689 );
690 }
691
692 #[tokio::test]
693 async fn push_drains_past_the_batch_limit() {
694 let dir = tempdir();
695 let (db, node) = device(&dir.join("big.db"), 21);
696 let server = FakeServer::default();
697
698 // One more than two full batches, so the drain has to come back for a
699 // third: a loop that stops after the first batch loses 501 rows and
700 // still reports a clean sync.
701 let total = PUSH_BATCH_LIMIT * 2 + 1;
702 {
703 let conn = db.open().unwrap();
704 conn.execute("BEGIN", []).unwrap();
705 for i in 0..total {
706 conn.execute(
707 "INSERT INTO note (id, name) VALUES (?1, 'n')",
708 [format!("r{i:04}")],
709 )
710 .unwrap();
711 }
712 conn.execute("COMMIT", []).unwrap();
713 let pending: i64 = conn
714 .query_row(
715 "SELECT COUNT(*) FROM sync_changelog WHERE pushed = 0",
716 [],
717 |r| r.get(0),
718 )
719 .unwrap();
720 assert_eq!(pending as usize, total, "the triggers logged every insert");
721 }
722
723 let pushed = push_scope(&db, &server, &schema(), node, SyncScope::Personal)
724 .await
725 .unwrap();
726 assert_eq!(pushed as usize, total);
727
728 {
729 let log = server.log.lock().unwrap();
730 assert_eq!(log.len(), total, "every row reached the server");
731 let mut seen: Vec<String> = log.iter().map(|(_, e)| e.row_id.clone()).collect();
732 seen.sort();
733 seen.dedup();
734 assert_eq!(
735 seen.len(),
736 total,
737 "no row was sent twice in place of another"
738 );
739 }
740
741 let left: i64 = db
742 .open()
743 .unwrap()
744 .query_row(
745 "SELECT COUNT(*) FROM sync_changelog WHERE pushed = 0",
746 [],
747 |r| r.get(0),
748 )
749 .unwrap();
750 assert_eq!(left, 0, "every row is marked pushed");
751
752 // A second drain has nothing to do and sends nothing.
753 let again = push_scope(&db, &server, &schema(), node, SyncScope::Personal)
754 .await
755 .unwrap();
756 assert_eq!(again, 0);
757 assert_eq!(server.log.lock().unwrap().len(), total);
758 }
759
760 /// A schema wide enough for one pull to produce all four outcomes at once.
761 fn mix_schema() -> SyncSchema {
762 SyncSchema::new(vec![
763 SyncTable::full("parent", &["id", "name"]),
764 SyncTable::full("child", &["id", "parent_id"]),
765 SyncTable::full("cfg", &["key", "value"])
766 .pk(&["key"])
767 .exclude_where("{row}.key NOT LIKE 'sync\\_%' ESCAPE '\\'"),
768 ])
769 }
770
771 fn mix_device(path: &std::path::Path, n: u128) -> (DbSource, DeviceId) {
772 let db = DbSource::path(path);
773 let conn = db.open().unwrap();
774 conn.execute_batch(
775 "CREATE TABLE parent (id TEXT PRIMARY KEY, name TEXT);
776 CREATE TABLE child (id TEXT PRIMARY KEY, parent_id TEXT NOT NULL REFERENCES parent(id));
777 CREATE TABLE cfg (key TEXT PRIMARY KEY, value TEXT);",
778 )
779 .unwrap();
780 conn.execute_batch(&mix_schema().migration_sql()).unwrap();
781 (db, DeviceId::new(uuid::Uuid::from_u128(n)))
782 }
783
784 /// Put an arbitrary entry on the fake server, including one with no payload.
785 fn serve_entry(server: &FakeServer, entry: ChangeEntry) {
786 server
787 .log
788 .lock()
789 .unwrap()
790 .push((DeviceId::new(uuid::Uuid::from_u128(0xAA)), entry));
791 }
792
793 #[tokio::test]
794 async fn pull_reports_all_four_outcome_counters() {
795 let dir = tempdir();
796 let (db, node) = mix_device(&dir.join("mix.db"), 22);
797 let server = FakeServer::default();
798
799 // applied x4
800 for i in 1..=4 {
801 serve(
802 &server,
803 "parent",
804 &format!("p{i}"),
805 serde_json::json!({"id": format!("p{i}"), "name": "p"}),
806 );
807 }
808 // filtered x3: cfg's include predicate excludes the sync_ prefix.
809 for k in ["sync_a", "sync_b", "sync_c"] {
810 serve(
811 &server,
812 "cfg",
813 k,
814 serde_json::json!({"key": k, "value": "v"}),
815 );
816 }
817 // rejected x1: no payload, so the same bytes always fail.
818 serve_entry(
819 &server,
820 ChangeEntry {
821 table: "parent".into(),
822 op: ChangeOp::Insert,
823 row_id: "p9".into(),
824 timestamp: Utc::now(),
825 hlc: crate::types::hlc_legacy_floor(),
826 data: None,
827 extra: serde_json::Map::default(),
828 },
829 );
830 // deferred x2: a parent that is not in this page and never will be.
831 for i in 1..=2 {
832 serve(
833 &server,
834 "child",
835 &format!("c{i}"),
836 serde_json::json!({"id": format!("c{i}"), "parent_id": "absent"}),
837 );
838 }
839
840 let out = pull_scope(&db, &server, &mix_schema(), node, SyncScope::Personal)
841 .await
842 .unwrap();
843
844 assert_eq!(out.applied, 4);
845 assert_eq!(out.filtered, 3);
846 assert_eq!(out.rejected, 1);
847 assert_eq!(out.deferred, 2);
848 assert!(out.changed_tables.contains("parent"));
849 assert!(
850 !out.changed_tables.contains("child"),
851 "nothing landed in child"
852 );
853 assert_eq!(row_count(&db, "parent"), 4);
854 assert_eq!(row_count(&db, "child"), 0);
855 assert_eq!(row_count(&db, "cfg"), 0);
856 }
857
858 #[test]
859 fn cleanup_changelog_returns_what_it_deleted() {
860 let dir = tempdir();
861 let (da, _) = device(&dir.join("cleanup.db"), 23);
862 let conn = da.open().unwrap();
863 conn.execute("DELETE FROM sync_changelog", []).unwrap();
864
865 // 4 pushed and old (deletable), 2 pushed and recent, 3 unpushed and old.
866 let insert = |row_id: &str, pushed: i64, ts: &str| {
867 conn.execute(
868 "INSERT INTO sync_changelog (table_name, op, row_id, data, pushed, timestamp) \
869 VALUES ('note','INSERT',?1,'{}',?2,?3)",
870 rusqlite::params![row_id, pushed, ts],
871 )
872 .unwrap();
873 };
874 let old = "2000-01-01T00:00:00.000Z";
875 let now = "2999-01-01T00:00:00.000Z";
876 for i in 0..4 {
877 insert(&format!("old{i}"), 1, old);
878 }
879 for i in 0..2 {
880 insert(&format!("new{i}"), 1, now);
881 }
882 for i in 0..3 {
883 insert(&format!("unp{i}"), 0, old);
884 }
885
886 let before: i64 = conn
887 .query_row("SELECT COUNT(*) FROM sync_changelog", [], |r| r.get(0))
888 .unwrap();
889 let removed = cleanup_changelog(&conn).unwrap();
890 let after: i64 = conn
891 .query_row("SELECT COUNT(*) FROM sync_changelog", [], |r| r.get(0))
892 .unwrap();
893
894 assert_eq!(removed, 4, "only the old pushed rows go");
895 assert_eq!(
896 removed,
897 (before - after) as u64,
898 "the returned count is the number of rows actually deleted"
899 );
900 let unpushed: i64 = conn
901 .query_row(
902 "SELECT COUNT(*) FROM sync_changelog WHERE pushed = 0",
903 [],
904 |r| r.get(0),
905 )
906 .unwrap();
907 assert_eq!(unpushed, 3, "an unpushed row is never pruned by age");
908
909 // Nothing left to prune, so a second pass reports zero.
910 assert_eq!(cleanup_changelog(&conn).unwrap(), 0);
911 }
912
913 // minimal temp-dir helper (no external dep)
914 fn tempdir() -> std::path::PathBuf {
915 let mut p = std::env::temp_dir();
916 // Unique-ish per test via a monotonic counter; tests run in the same
917 // process so a static AtomicU64 keeps paths distinct.
918 use std::sync::atomic::{AtomicU64, Ordering};
919 static N: AtomicU64 = AtomicU64::new(0);
920 p.push(format!(
921 "synckit_b5_{}_{}",
922 std::process::id(),
923 N.fetch_add(1, Ordering::Relaxed)
924 ));
925 std::fs::create_dir_all(&p).unwrap();
926 p
927 }
928