Skip to main content

max / makenotwork

21.1 KB · 687 lines History Blame Raw
1 //! DB-layer contract tests for `db::synckit::log`: the pull side, where a
2 //! dropped entry is invisible.
3 //!
4 //! The change log is the sync engine's source of truth, so the failure this
5 //! file exists to catch is the silent one: a pull that drops an entry at a page
6 //! boundary, a cursor that is inclusive at one end and exclusive at the other,
7 //! or a filter that turns a full drain into a partial one. Every pagination
8 //! test here asserts the same thing from two directions, that draining the log
9 //! a page at a time yields exactly what one unpaginated pull yields, because
10 //! that equality is what a client's convergence depends on and neither half
11 //! alone would notice a missing row.
12 //!
13 //! Pinned here: cursor exclusivity at both ends, the empty-list table filter
14 //! meaning "nothing" rather than "everything", `since` being inclusive of its
15 //! own boundary, pull ordering being arrival order and not the client clock,
16 //! and the two filters composing as AND rather than OR.
17 //!
18 //! The write side of the same module is `db_synckit_log_push_layer`, and the
19 //! audit trail is `db_synckit_security_layer`. Neighbouring ground is
20 //! deliberately not repeated: `db_synckit_layer` pins append order, per-user
21 //! scoping, blobs and keys; `db_synckit_groups` pins group membership and the
22 //! structural isolation of `sync_group_log` from personal pulls.
23 //!
24 //! Delete this file and the pagination contract of the sync protocol is
25 //! unasserted at this layer.
26
27 use crate::harness::db::TestDb;
28 use crate::harness::seed_user;
29
30 use makenotwork::db::synckit;
31 use makenotwork::db::{SyncAppId, SyncDeviceId, UserId};
32 use uuid::Uuid;
33
34 /// One change tuple in the shape `push_sync_changes` and `push_group_changes`
35 /// both take.
36 type Change = (
37 String,
38 String,
39 String,
40 chrono::DateTime<chrono::Utc>,
41 Option<serde_json::Value>,
42 );
43
44 /// Seed a sync app owned by `user`.
45 async fn seed_app(pool: &sqlx::PgPool, user: UserId, name: &str) -> SyncAppId {
46 sqlx::query_scalar::<_, SyncAppId>(
47 "INSERT INTO sync_apps (creator_id, name, api_key_hash, api_key_prefix)
48 VALUES ($1, $2, $3, $4) RETURNING id",
49 )
50 .bind(user)
51 .bind(name)
52 .bind(format!("hash_{name}"))
53 .bind(&name[..name.len().min(8)])
54 .fetch_one(pool)
55 .await
56 .expect("seed sync app")
57 }
58
59 /// Seed a device row for a user within an app.
60 async fn seed_device(
61 pool: &sqlx::PgPool,
62 app: SyncAppId,
63 user: UserId,
64 name: &str,
65 ) -> SyncDeviceId {
66 sqlx::query_scalar::<_, SyncDeviceId>(
67 "INSERT INTO sync_devices (app_id, user_id, device_name, platform)
68 VALUES ($1, $2, $3, 'macos') RETURNING id",
69 )
70 .bind(app)
71 .bind(user)
72 .bind(name)
73 .fetch_one(pool)
74 .await
75 .expect("seed device")
76 }
77
78 /// A fixed instant offset by whole hours. Fixed rather than `Utc::now()` so a
79 /// `since` boundary can be asserted for equality without depending on the
80 /// wall clock.
81 fn ts(offset_hours: i64) -> chrono::DateTime<chrono::Utc> {
82 "2024-03-01T12:00:00Z"
83 .parse::<chrono::DateTime<chrono::Utc>>()
84 .expect("fixed base timestamp parses")
85 + chrono::Duration::hours(offset_hours)
86 }
87
88 /// An INSERT change on `table` for `row`, client-stamped at `ts(offset_hours)`.
89 fn change_at(table: &str, row: &str, offset_hours: i64) -> Change {
90 (
91 table.to_string(),
92 "INSERT".to_string(),
93 row.to_string(),
94 ts(offset_hours),
95 Some(serde_json::json!({ "row": row })),
96 )
97 }
98
99 /// An INSERT change on `table` for `row` at the base instant.
100 fn change(table: &str, row: &str) -> Change {
101 change_at(table, row, 0)
102 }
103
104 /// Push `changes` as one fresh batch, returning the cursor the push reports.
105 async fn push(
106 pool: &sqlx::PgPool,
107 app: SyncAppId,
108 user: UserId,
109 device: SyncDeviceId,
110 changes: &[Change],
111 ) -> i64 {
112 synckit::push_sync_changes(pool, app, user, device, Uuid::new_v4(), changes)
113 .await
114 .expect("push sync changes")
115 }
116
117 /// Walk the personal log `limit` entries at a time the way a client does,
118 /// carrying the last returned seq forward as the next cursor. Returns the
119 /// row ids in the order they were handed over.
120 ///
121 /// The iteration guard is what makes a non-terminating cursor fail as a test
122 /// rather than hang the suite.
123 async fn drain(
124 pool: &sqlx::PgPool,
125 app: SyncAppId,
126 user: UserId,
127 limit: i64,
128 tables: Option<&[String]>,
129 since: Option<chrono::DateTime<chrono::Utc>>,
130 ) -> Vec<String> {
131 let mut seen: Vec<String> = Vec::new();
132 let mut cursor = 0i64;
133 let mut pages = 0;
134 loop {
135 let page =
136 synckit::pull_sync_changes_filtered(pool, app, user, cursor, limit, tables, since)
137 .await
138 .expect("paginated pull");
139 if page.is_empty() {
140 break;
141 }
142 assert!(
143 i64::try_from(page.len()).expect("page length fits i64") <= limit,
144 "a page must never exceed the limit it was asked for: got {} for limit {limit}",
145 page.len()
146 );
147 cursor = page.last().expect("non-empty page has a last entry").seq;
148 seen.extend(page.into_iter().map(|e| e.row_id));
149 pages += 1;
150 assert!(
151 pages <= 64,
152 "the cursor did not terminate after 64 pages of limit {limit}; seen so far: {seen:?}"
153 );
154 }
155 seen
156 }
157
158 fn names(v: &[String]) -> Vec<&str> {
159 v.iter().map(String::as_str).collect()
160 }
161 // ── log: pagination and cursor termination ──────────────────────────────────
162
163 #[tokio::test]
164 async fn a_paginated_drain_yields_exactly_what_one_unpaginated_pull_yields() {
165 let db = TestDb::new().await;
166 let user = seed_user(&db.pool, "sklogp_drain").await;
167 let app = seed_app(&db.pool, user, "logdrain").await;
168 let device = seed_device(&db.pool, app, user, "laptop").await;
169
170 // Seven entries across three batches of unequal size, so the batch
171 // boundaries do not line up with any of the page sizes below.
172 push(
173 &db.pool,
174 app,
175 user,
176 device,
177 &[change("tasks", "r0"), change("tasks", "r1")],
178 )
179 .await;
180 push(
181 &db.pool,
182 app,
183 user,
184 device,
185 &[
186 change("tasks", "r2"),
187 change("tasks", "r3"),
188 change("tasks", "r4"),
189 ],
190 )
191 .await;
192 push(
193 &db.pool,
194 app,
195 user,
196 device,
197 &[change("tasks", "r5"), change("tasks", "r6")],
198 )
199 .await;
200
201 let whole = synckit::pull_sync_changes_filtered(&db.pool, app, user, 0, 1000, None, None)
202 .await
203 .unwrap();
204 let whole_rows: Vec<String> = whole.iter().map(|e| e.row_id.clone()).collect();
205 assert_eq!(
206 names(&whole_rows),
207 ["r0", "r1", "r2", "r3", "r4", "r5", "r6"],
208 "one unpaginated pull is the reference: {whole_rows:?}"
209 );
210
211 // 2 and 3 leave a short final page; 7 is an exact multiple, which is the
212 // case where a client only stops because the pull after the last full page
213 // comes back empty.
214 for limit in [2i64, 3, 7] {
215 let paged = drain(&db.pool, app, user, limit, None, None).await;
216 assert_eq!(
217 paged, whole_rows,
218 "draining {limit} at a time must lose and duplicate nothing: {paged:?}"
219 );
220 }
221 }
222
223 #[tokio::test]
224 async fn the_pull_cursor_is_exclusive_at_both_ends() {
225 let db = TestDb::new().await;
226 let user = seed_user(&db.pool, "sklogp_cursor").await;
227 let app = seed_app(&db.pool, user, "logcursor").await;
228 let device = seed_device(&db.pool, app, user, "laptop").await;
229
230 push(
231 &db.pool,
232 app,
233 user,
234 device,
235 &[
236 change("tasks", "r0"),
237 change("tasks", "r1"),
238 change("tasks", "r2"),
239 change("tasks", "r3"),
240 change("tasks", "r4"),
241 ],
242 )
243 .await;
244
245 let all = synckit::pull_sync_changes_filtered(&db.pool, app, user, 0, 100, None, None)
246 .await
247 .unwrap();
248 assert_eq!(all.len(), 5, "five entries were pushed: {all:?}");
249
250 // Resuming at the seq of r1 must hand back r2 onward. An inclusive `>=`
251 // would re-deliver r1 and return four entries, which is why the count and
252 // the first row id are both asserted.
253 let after_second =
254 synckit::pull_sync_changes_filtered(&db.pool, app, user, all[1].seq, 100, None, None)
255 .await
256 .unwrap();
257 let rows: Vec<&str> = after_second.iter().map(|e| e.row_id.as_str()).collect();
258 assert_eq!(
259 rows,
260 ["r2", "r3", "r4"],
261 "a cursor is the last entry already seen, not the next one to send: {rows:?}"
262 );
263
264 // The cursor a client holds after a full drain returns nothing, which is
265 // the only reason the drain loop above terminates.
266 let at_end = synckit::pull_sync_changes_filtered(
267 &db.pool,
268 app,
269 user,
270 all.last().unwrap().seq,
271 100,
272 None,
273 None,
274 )
275 .await
276 .unwrap();
277 assert!(
278 at_end.is_empty(),
279 "the final cursor must drain empty, got {at_end:?}"
280 );
281
282 // A cursor past the end (a log truncated behind a client, or a client that
283 // held a cursor from another app) is empty rather than an error or a rewind.
284 let past_end = synckit::pull_sync_changes_filtered(
285 &db.pool,
286 app,
287 user,
288 all.last().unwrap().seq + 5_000,
289 100,
290 None,
291 None,
292 )
293 .await
294 .unwrap();
295 assert!(
296 past_end.is_empty(),
297 "a cursor beyond the highest seq returns nothing, got {past_end:?}"
298 );
299 }
300
301 #[tokio::test]
302 async fn a_limit_bounds_one_page_and_a_zero_limit_returns_no_page_at_all() {
303 let db = TestDb::new().await;
304 let user = seed_user(&db.pool, "sklogp_limit").await;
305 let app = seed_app(&db.pool, user, "loglimit").await;
306 let device = seed_device(&db.pool, app, user, "laptop").await;
307
308 push(
309 &db.pool,
310 app,
311 user,
312 device,
313 &[
314 change("tasks", "r0"),
315 change("tasks", "r1"),
316 change("tasks", "r2"),
317 change("tasks", "r3"),
318 change("tasks", "r4"),
319 ],
320 )
321 .await;
322
323 let page = synckit::pull_sync_changes_filtered(&db.pool, app, user, 0, 3, None, None)
324 .await
325 .unwrap();
326 let rows: Vec<&str> = page.iter().map(|e| e.row_id.as_str()).collect();
327 assert_eq!(
328 rows,
329 ["r0", "r1", "r2"],
330 "a limit takes the lowest seqs, not an arbitrary three: {rows:?}"
331 );
332
333 // A limit larger than the log is not an error and does not pad.
334 let over = synckit::pull_sync_changes_filtered(&db.pool, app, user, 0, 500, None, None)
335 .await
336 .unwrap();
337 assert_eq!(over.len(), 5, "a generous limit returns all five: {over:?}");
338
339 // Pinned because it is a live hazard for a caller that computes a page
340 // size: a zero limit is an empty page, indistinguishable from "drained",
341 // so a client that ever asks for zero silently stops syncing.
342 let none = synckit::pull_sync_changes_filtered(&db.pool, app, user, 0, 0, None, None)
343 .await
344 .unwrap();
345 assert!(
346 none.is_empty(),
347 "a zero limit reads as a finished drain, got {none:?}"
348 );
349 }
350
351 #[tokio::test]
352 async fn the_unfiltered_pull_matches_the_filtered_pull_with_no_filters() {
353 let db = TestDb::new().await;
354 let user = seed_user(&db.pool, "sklogp_parity").await;
355 let app = seed_app(&db.pool, user, "logparity").await;
356 let device = seed_device(&db.pool, app, user, "laptop").await;
357
358 push(
359 &db.pool,
360 app,
361 user,
362 device,
363 &[
364 change("tasks", "r0"),
365 change("notes", "r1"),
366 change("tasks", "r2"),
367 change("tags", "r3"),
368 ],
369 )
370 .await;
371
372 let legacy = synckit::pull_sync_changes(&db.pool, app, user, 0, 100)
373 .await
374 .unwrap();
375 let filtered = synckit::pull_sync_changes_filtered(&db.pool, app, user, 0, 100, None, None)
376 .await
377 .unwrap();
378
379 let legacy_rows: Vec<&str> = legacy.iter().map(|e| e.row_id.as_str()).collect();
380 let filtered_rows: Vec<&str> = filtered.iter().map(|e| e.row_id.as_str()).collect();
381 assert_eq!(
382 legacy_rows, filtered_rows,
383 "passing no filters is documented as identical to the older pull: {legacy_rows:?} vs {filtered_rows:?}"
384 );
385 assert_eq!(
386 legacy.iter().map(|e| e.seq).collect::<Vec<_>>(),
387 filtered.iter().map(|e| e.seq).collect::<Vec<_>>(),
388 "and hands back the same cursors"
389 );
390 assert_eq!(legacy_rows, ["r0", "r1", "r2", "r3"]);
391 }
392
393 #[tokio::test]
394 async fn pull_order_is_arrival_order_and_not_the_client_clock() {
395 let db = TestDb::new().await;
396 let user = seed_user(&db.pool, "sklogp_order").await;
397 let app = seed_app(&db.pool, user, "logorder").await;
398 let device = seed_device(&db.pool, app, user, "laptop").await;
399
400 // Client timestamps descend as the entries arrive, so ordering by
401 // client_timestamp would reverse this list. A device with a skewed clock
402 // is the real case: its entries must still replay in the order the server
403 // accepted them.
404 push(
405 &db.pool,
406 app,
407 user,
408 device,
409 &[
410 change_at("tasks", "arrived-first", 9),
411 change_at("tasks", "arrived-second", 5),
412 change_at("tasks", "arrived-third", 1),
413 ],
414 )
415 .await;
416
417 let entries = synckit::pull_sync_changes_filtered(&db.pool, app, user, 0, 100, None, None)
418 .await
419 .unwrap();
420 let rows: Vec<&str> = entries.iter().map(|e| e.row_id.as_str()).collect();
421 assert_eq!(
422 rows,
423 ["arrived-first", "arrived-second", "arrived-third"],
424 "seq order, not client_timestamp order: {rows:?}"
425 );
426 assert!(
427 entries.windows(2).all(|w| w[0].seq < w[1].seq),
428 "and seq is strictly increasing: {:?}",
429 entries.iter().map(|e| e.seq).collect::<Vec<_>>()
430 );
431 }
432 // ── log: filters ────────────────────────────────────────────────────────────
433
434 #[tokio::test]
435 async fn the_table_filter_selects_the_named_tables_and_an_empty_list_selects_none() {
436 let db = TestDb::new().await;
437 let user = seed_user(&db.pool, "sklogp_tables").await;
438 let app = seed_app(&db.pool, user, "logtables").await;
439 let device = seed_device(&db.pool, app, user, "laptop").await;
440
441 push(
442 &db.pool,
443 app,
444 user,
445 device,
446 &[
447 change("tasks", "t0"),
448 change("notes", "n0"),
449 change("tasks", "t1"),
450 change("tags", "g0"),
451 change("notes", "n1"),
452 change("tasks", "t2"),
453 ],
454 )
455 .await;
456
457 let tasks = vec!["tasks".to_string()];
458 let picked = synckit::pull_sync_changes_filtered(
459 &db.pool,
460 app,
461 user,
462 0,
463 100,
464 Some(tasks.as_slice()),
465 None,
466 )
467 .await
468 .unwrap();
469 let rows: Vec<&str> = picked.iter().map(|e| e.row_id.as_str()).collect();
470 assert_eq!(
471 rows,
472 ["t0", "t1", "t2"],
473 "one named table yields three of the six: {rows:?}"
474 );
475
476 // Two names, so a filter that only ever honoured the first element would
477 // return three here instead of four.
478 let two = vec!["tasks".to_string(), "tags".to_string()];
479 let picked_two = synckit::pull_sync_changes_filtered(
480 &db.pool,
481 app,
482 user,
483 0,
484 100,
485 Some(two.as_slice()),
486 None,
487 )
488 .await
489 .unwrap();
490 let rows_two: Vec<&str> = picked_two.iter().map(|e| e.row_id.as_str()).collect();
491 assert_eq!(
492 rows_two,
493 ["t0", "t1", "g0", "t2"],
494 "every named table is honoured, in seq order: {rows_two:?}"
495 );
496
497 // An empty list is not NULL, so it means "no table matches", not "no
498 // filter". A caller that builds the list from a user's selection and lets
499 // it come back empty gets nothing, not everything.
500 let empty: Vec<String> = Vec::new();
501 let none = synckit::pull_sync_changes_filtered(
502 &db.pool,
503 app,
504 user,
505 0,
506 100,
507 Some(empty.as_slice()),
508 None,
509 )
510 .await
511 .unwrap();
512 assert!(
513 none.is_empty(),
514 "an empty table list selects nothing, got {none:?}"
515 );
516
517 // And a name nobody pushed is empty rather than a wildcard.
518 let absent = vec!["ledger".to_string()];
519 let missing = synckit::pull_sync_changes_filtered(
520 &db.pool,
521 app,
522 user,
523 0,
524 100,
525 Some(absent.as_slice()),
526 None,
527 )
528 .await
529 .unwrap();
530 assert!(
531 missing.is_empty(),
532 "an unknown table selects nothing, got {missing:?}"
533 );
534 }
535
536 #[tokio::test]
537 async fn the_since_filter_includes_an_entry_stamped_exactly_at_the_boundary() {
538 let db = TestDb::new().await;
539 let user = seed_user(&db.pool, "sklogp_since").await;
540 let app = seed_app(&db.pool, user, "logsince").await;
541 let device = seed_device(&db.pool, app, user, "laptop").await;
542
543 push(
544 &db.pool,
545 app,
546 user,
547 device,
548 &[
549 change_at("tasks", "before", -2),
550 change_at("tasks", "exactly", 0),
551 change_at("tasks", "after", 2),
552 ],
553 )
554 .await;
555
556 // ts(0) is the stamp of "exactly", so the boundary row separates `>=` from
557 // `>`; "before" separates `>=` from an unfiltered pull.
558 let from_boundary =
559 synckit::pull_sync_changes_filtered(&db.pool, app, user, 0, 100, None, Some(ts(0)))
560 .await
561 .unwrap();
562 let rows: Vec<&str> = from_boundary.iter().map(|e| e.row_id.as_str()).collect();
563 assert_eq!(
564 rows,
565 ["exactly", "after"],
566 "since is inclusive of its own instant and excludes what came before: {rows:?}"
567 );
568
569 let from_later =
570 synckit::pull_sync_changes_filtered(&db.pool, app, user, 0, 100, None, Some(ts(3)))
571 .await
572 .unwrap();
573 assert!(
574 from_later.is_empty(),
575 "a since past every entry returns nothing, got {from_later:?}"
576 );
577
578 let from_earlier =
579 synckit::pull_sync_changes_filtered(&db.pool, app, user, 0, 100, None, Some(ts(-5)))
580 .await
581 .unwrap();
582 assert_eq!(
583 from_earlier.len(),
584 3,
585 "a since before every entry returns all three: {from_earlier:?}"
586 );
587 }
588
589 #[tokio::test]
590 async fn the_table_and_since_filters_compose_as_and_not_or() {
591 let db = TestDb::new().await;
592 let user = seed_user(&db.pool, "sklogp_compose").await;
593 let app = seed_app(&db.pool, user, "logcompose").await;
594 let device = seed_device(&db.pool, app, user, "laptop").await;
595
596 // One row in each quadrant of (right table, right time), so an OR would
597 // return three and an ignored filter would return two or four.
598 push(
599 &db.pool,
600 app,
601 user,
602 device,
603 &[
604 change_at("tasks", "old-task", -4),
605 change_at("tasks", "new-task", 4),
606 change_at("notes", "old-note", -4),
607 change_at("notes", "new-note", 4),
608 ],
609 )
610 .await;
611
612 let tasks = vec!["tasks".to_string()];
613 let both = synckit::pull_sync_changes_filtered(
614 &db.pool,
615 app,
616 user,
617 0,
618 100,
619 Some(tasks.as_slice()),
620 Some(ts(0)),
621 )
622 .await
623 .unwrap();
624 let rows: Vec<&str> = both.iter().map(|e| e.row_id.as_str()).collect();
625 assert_eq!(
626 rows,
627 ["new-task"],
628 "both filters must hold at once: {rows:?}"
629 );
630 }
631
632 #[tokio::test]
633 async fn a_filtered_paginated_drain_matches_the_filtered_unpaginated_pull() {
634 let db = TestDb::new().await;
635 let user = seed_user(&db.pool, "sklogp_fdrain").await;
636 let app = seed_app(&db.pool, user, "logfdrain").await;
637 let device = seed_device(&db.pool, app, user, "laptop").await;
638
639 // The matching entries are deliberately non-contiguous in seq, so a page
640 // boundary always falls on a skipped row. This is where a filtered pull
641 // loses data if the cursor is advanced by anything other than the seq of
642 // the last entry actually returned.
643 push(
644 &db.pool,
645 app,
646 user,
647 device,
648 &[
649 change("tasks", "t0"),
650 change("notes", "n0"),
651 change("tasks", "t1"),
652 change("notes", "n1"),
653 change("tasks", "t2"),
654 change("notes", "n2"),
655 change("tasks", "t3"),
656 ],
657 )
658 .await;
659
660 let tasks = vec!["tasks".to_string()];
661 let whole = synckit::pull_sync_changes_filtered(
662 &db.pool,
663 app,
664 user,
665 0,
666 1000,
667 Some(tasks.as_slice()),
668 None,
669 )
670 .await
671 .unwrap();
672 let whole_rows: Vec<String> = whole.iter().map(|e| e.row_id.clone()).collect();
673 assert_eq!(
674 names(&whole_rows),
675 ["t0", "t1", "t2", "t3"],
676 "the filtered reference is the four task rows: {whole_rows:?}"
677 );
678
679 for limit in [2i64, 3, 4] {
680 let paged = drain(&db.pool, app, user, limit, Some(tasks.as_slice()), None).await;
681 assert_eq!(
682 paged, whole_rows,
683 "a filtered drain at {limit} per page must equal the whole filtered pull: {paged:?}"
684 );
685 }
686 }
687