Skip to main content

max / synckit

20.3 KB · 684 lines History Blame Raw
1 //! Push and pull: the changelog round-trip, its encryption, pagination, and the
2 //! guards that stop a push without a master key or a session.
3
4 use crate::common::*;
5
6 const PUSH_PATH: &str = "/api/v1/sync/push";
7 const PULL_PATH: &str = "/api/v1/sync/pull";
8
9 /// One Insert the way a caller builds it, with a zero clock: these tests assert
10 /// on what crosses the wire, never on HLC ordering.
11 fn insert(table: &str, row_id: &str, data: serde_json::Value) -> ChangeEntry {
12 ChangeEntry {
13 table: table.into(),
14 op: ChangeOp::Insert,
15 row_id: row_id.into(),
16 timestamp: Utc::now(),
17 hlc: Hlc::zero(DeviceId::nil()),
18 data: Some(data),
19 extra: serde_json::Map::default(),
20 }
21 }
22
23 // ── Push / Pull with encryption ──
24
25 #[tokio::test]
26 async fn push_encrypts_data() {
27 let kit = MockKit::start().await;
28 kit.post(PUSH_PATH).json(json!({"cursor": 1})).await;
29
30 let (client, _key) = kit.keyed();
31 let device_id = DeviceId::new(Uuid::new_v4());
32 let cursor = client
33 .push(
34 device_id,
35 vec![insert("tasks", "row-1", json!({"title": "Secret task"}))],
36 )
37 .await
38 .unwrap();
39
40 assert_eq!(cursor, 1);
41
42 // Verify the request body was sent with encrypted data (not plaintext)
43 let body = kit.body(PUSH_PATH).await;
44 let wire_data = body["changes"][0]["data"].as_str().unwrap();
45 assert!(
46 !wire_data.contains("Secret task"),
47 "Plaintext should not appear on the wire"
48 );
49 }
50
51 #[tokio::test]
52 async fn pull_decrypts_data() {
53 let kit = MockKit::start().await;
54 let (client, key) = kit.keyed();
55
56 // Encrypt a value to simulate what the server would return
57 let plaintext = json!({"title": "Decrypted task"});
58 let encrypted = synckit_client::crypto::encrypt_json(&plaintext, &key).unwrap();
59
60 let device_id = DeviceId::new(Uuid::new_v4());
61 kit.post(PULL_PATH)
62 .json(json!({
63 "changes": [{
64 "seq": 1,
65 "device_id": device_id,
66 "table": "tasks",
67 "op": "INSERT",
68 "row_id": "row-1",
69 "timestamp": "2025-06-01T12:00:00Z",
70 "data": encrypted,
71 }],
72 "cursor": 1,
73 "has_more": false,
74 }))
75 .await;
76
77 let (changes, cursor, has_more) = client.pull(device_id, 0).await.unwrap();
78 assert_eq!(changes.len(), 1);
79 assert_eq!(cursor, 1);
80 assert!(!has_more);
81 assert_eq!(changes[0].data.as_ref().unwrap(), &plaintext);
82 }
83
84 #[tokio::test]
85 async fn push_retries_on_503() {
86 let kit = MockKit::start().await;
87 kit.post(PUSH_PATH).code(503).once().empty().await;
88 kit.post(PUSH_PATH).json(json!({"cursor": 5})).await;
89
90 let (client, _key) = kit.keyed();
91 let cursor = client
92 .push(DeviceId::new(Uuid::new_v4()), vec![])
93 .await
94 .unwrap();
95 assert_eq!(cursor, 5);
96 }
97
98 #[tokio::test]
99 async fn push_fails_immediately_on_401() {
100 let kit = MockKit::start().await;
101 kit.post(PUSH_PATH)
102 .code(401)
103 .exactly(1)
104 .text("Unauthorized")
105 .await;
106
107 let (client, _key) = kit.keyed();
108 let err = client
109 .push(DeviceId::new(Uuid::new_v4()), vec![])
110 .await
111 .unwrap_err();
112 assert!(matches!(err, SyncKitError::Server { status: 401, .. }));
113 }
114
115 #[tokio::test]
116 async fn pull_with_has_more_pagination() {
117 let kit = MockKit::start().await;
118 let (client, _key) = kit.keyed();
119 let device_id = DeviceId::new(Uuid::new_v4());
120
121 // First pull: has_more = true
122 kit.post(PULL_PATH)
123 .once()
124 .json(json!({
125 "changes": [],
126 "cursor": 50,
127 "has_more": true,
128 }))
129 .await;
130
131 let (changes, cursor, has_more) = client.pull(device_id, 0).await.unwrap();
132 assert!(changes.is_empty());
133 assert_eq!(cursor, 50);
134 assert!(has_more);
135
136 // Second pull from cursor 50: has_more = false
137 kit.post(PULL_PATH)
138 .json(json!({
139 "changes": [],
140 "cursor": 100,
141 "has_more": false,
142 }))
143 .await;
144
145 let (_, cursor2, has_more2) = client.pull(device_id, 50).await.unwrap();
146 assert_eq!(cursor2, 100);
147 assert!(!has_more2);
148 }
149
150 // ── Empty changelog push ──
151
152 #[tokio::test]
153 async fn push_empty_changes_succeeds() {
154 let kit = MockKit::start().await;
155 kit.post(PUSH_PATH).json(json!({"cursor": 0})).await;
156
157 let (client, _key) = kit.keyed();
158 let cursor = client
159 .push(DeviceId::new(Uuid::new_v4()), vec![])
160 .await
161 .unwrap();
162 assert_eq!(cursor, 0);
163 }
164
165 // ── Large payload handling ──
166
167 #[tokio::test]
168 async fn push_many_changes_succeeds() {
169 let kit = MockKit::start().await;
170 kit.post(PUSH_PATH).json(json!({"cursor": 1000})).await;
171
172 let (client, _key) = kit.keyed();
173
174 // Create 1000+ change entries
175 let changes: Vec<ChangeEntry> = (0..1100)
176 .map(|i| {
177 insert(
178 "bulk_table",
179 &format!("row-{i}"),
180 json!({"index": i, "value": format!("data-{i}")}),
181 )
182 })
183 .collect();
184
185 let cursor = client
186 .push(DeviceId::new(Uuid::new_v4()), changes)
187 .await
188 .unwrap();
189 assert_eq!(cursor, 1000);
190 }
191
192 // ── Push without master key ──
193
194 #[tokio::test]
195 async fn push_with_data_fails_without_master_key() {
196 let kit = MockKit::start().await;
197 let client = kit.authed(); // No master key
198
199 let err = client
200 .push(
201 DeviceId::new(Uuid::new_v4()),
202 vec![insert("tasks", "r1", json!({"title": "test"}))],
203 )
204 .await
205 .unwrap_err();
206 assert!(
207 matches!(err, SyncKitError::NoMasterKey),
208 "Push with data should fail without master key: {err:?}"
209 );
210 }
211
212 #[tokio::test]
213 async fn push_delete_requires_master_key() {
214 // Deletes used to push without a key (no payload to encrypt). With HLC, a
215 // Delete now seals its clock into an encrypted envelope, so the master key is
216 // required for every push, Deletes included.
217 let kit = MockKit::start().await;
218 kit.post(PUSH_PATH).json(json!({"cursor": 1})).await;
219
220 let client = kit.authed(); // No master key loaded.
221
222 let changes = vec![ChangeEntry {
223 table: "tasks".into(),
224 op: ChangeOp::Delete,
225 row_id: "r1".into(),
226 timestamp: Utc::now(),
227 hlc: Hlc::zero(DeviceId::nil()),
228 data: None,
229 extra: serde_json::Map::default(),
230 }];
231
232 let err = client
233 .push(DeviceId::new(Uuid::new_v4()), changes)
234 .await
235 .unwrap_err();
236 assert!(
237 matches!(err, SyncKitError::NoMasterKey),
238 "Delete now seals an HLC envelope and needs the key: {err:?}"
239 );
240 }
241
242 // ── Double-push same data ──
243
244 #[tokio::test]
245 async fn double_push_same_data_both_succeed() {
246 let kit = MockKit::start().await;
247
248 // Server returns incrementing cursors
249 kit.post(PUSH_PATH).once().json(json!({"cursor": 1})).await;
250 kit.post(PUSH_PATH).json(json!({"cursor": 2})).await;
251
252 let (client, _key) = kit.keyed();
253 let entry = insert("tasks", "same-row", json!({"title": "duplicate push test"}));
254
255 let cursor1 = client
256 .push(DeviceId::new(Uuid::new_v4()), vec![entry.clone()])
257 .await
258 .unwrap();
259 let cursor2 = client
260 .push(DeviceId::new(Uuid::new_v4()), vec![entry])
261 .await
262 .unwrap();
263
264 assert_eq!(cursor1, 1);
265 assert_eq!(cursor2, 2);
266 }
267
268 // ── Pull without auth ──
269
270 #[tokio::test]
271 async fn pull_without_auth_returns_not_authenticated() {
272 let kit = MockKit::start().await;
273 let err = kit
274 .client()
275 .pull(DeviceId::new(Uuid::new_v4()), 0)
276 .await
277 .unwrap_err();
278 assert!(matches!(err, SyncKitError::NotAuthenticated));
279 }
280
281 // ── Encryption roundtrip through push/pull (end-to-end) ──
282
283 #[tokio::test]
284 async fn end_to_end_push_pull_encryption_roundtrip() {
285 let kit = MockKit::start().await;
286 let (client, _key) = kit.keyed();
287
288 let device_id = DeviceId::new(Uuid::new_v4());
289 let original_data = json!({
290 "title": "End-to-end test",
291 "tags": ["e2e", "encryption"],
292 "nested": {"key": "value"},
293 "count": 42
294 });
295
296 kit.post(PUSH_PATH).json(json!({"cursor": 1})).await;
297
298 client
299 .push(
300 device_id,
301 vec![insert("tasks", "e2e-row", original_data.clone())],
302 )
303 .await
304 .unwrap();
305
306 // Extract the encrypted data that was sent to the server
307 let push_body = kit.body(PUSH_PATH).await;
308 let wire_entry = &push_body["changes"][0];
309
310 // Feed encrypted data back through pull
311 kit.post(PULL_PATH)
312 .json(json!({
313 "changes": [{
314 "seq": 1,
315 "device_id": device_id,
316 "table": wire_entry["table"],
317 "op": wire_entry["op"],
318 "row_id": wire_entry["row_id"],
319 "timestamp": wire_entry["timestamp"],
320 "data": wire_entry["data"],
321 }],
322 "cursor": 1,
323 "has_more": false,
324 }))
325 .await;
326
327 let (changes, _, _) = client.pull(device_id, 0).await.unwrap();
328 assert_eq!(changes.len(), 1);
329 assert_eq!(
330 changes[0].data.as_ref().unwrap(),
331 &original_data,
332 "Data must survive push encryption + pull decryption"
333 );
334 }
335
336 // ── Pagination is a transport detail, not a content one ──
337
338 /// Build `n` encrypted changes with distinguishable payloads, starting at
339 /// `first_seq`. Returns the wire JSON and the plaintexts they should decrypt to.
340 fn encrypted_changes(
341 key: &[u8; 32],
342 device_id: DeviceId,
343 first_seq: i64,
344 n: i64,
345 ) -> (Vec<serde_json::Value>, Vec<serde_json::Value>) {
346 let mut wire = Vec::new();
347 let mut plain = Vec::new();
348 for i in 0..n {
349 let seq = first_seq + i;
350 let payload = json!({"title": format!("task {seq}"), "n": seq});
351 let encrypted = synckit_client::crypto::encrypt_json(&payload, key).unwrap();
352 wire.push(json!({
353 "seq": seq,
354 "device_id": device_id,
355 "table": "tasks",
356 "op": "INSERT",
357 "row_id": format!("row-{seq}"),
358 "timestamp": "2025-06-01T12:00:00Z",
359 "data": encrypted,
360 }));
361 plain.push(payload);
362 }
363 (wire, plain)
364 }
365
366 /// **Metamorphic relation:** the same six changes delivered as one page and as
367 /// three pages must decrypt to the same sequence. Pagination is a property of
368 /// the transport, so anything it changes about the content is a bug.
369 ///
370 /// The existing `pull_with_has_more_pagination` asserts the cursor plumbing
371 /// against empty `changes` arrays, so it cannot see a page boundary that drops,
372 /// duplicates or reorders a row. This relates two runs instead of judging one,
373 /// which needs no expected-value table (Chen et al. 1998).
374 #[tokio::test]
375 async fn a_paginated_pull_yields_what_an_unpaginated_pull_yields() {
376 let key = synckit_client::crypto::generate_master_key();
377 let device_id = DeviceId::new(Uuid::new_v4());
378 const TOTAL: i64 = 6;
379
380 // Run A: one page.
381 let unpaginated = {
382 let kit = MockKit::start().await;
383 let client = kit.authed();
384 client.set_master_key_raw(key);
385 let (wire, _) = encrypted_changes(&key, device_id, 1, TOTAL);
386 kit.post(PULL_PATH)
387 .json(json!({
388 "changes": wire,
389 "cursor": TOTAL,
390 "has_more": false,
391 }))
392 .await;
393
394 let (changes, cursor, has_more) = client.pull(device_id, 0).await.unwrap();
395 assert_eq!(cursor, TOTAL);
396 assert!(!has_more);
397 changes
398 };
399
400 // Run B: the same changes, three pages of two, drained the way a caller
401 // drains them.
402 let paginated = {
403 let kit = MockKit::start().await;
404 let client = kit.authed();
405 client.set_master_key_raw(key);
406 for page in 0..3i64 {
407 let first = page * 2 + 1;
408 let (wire, _) = encrypted_changes(&key, device_id, first, 2);
409 let cursor = first + 1;
410 kit.post(PULL_PATH)
411 .once()
412 .json(json!({
413 "changes": wire,
414 "cursor": cursor,
415 "has_more": page < 2,
416 }))
417 .await;
418 }
419
420 let mut collected = Vec::new();
421 let mut cursor = 0i64;
422 loop {
423 let (changes, next, has_more) = client.pull(device_id, cursor).await.unwrap();
424 collected.extend(changes);
425 cursor = next;
426 if !has_more {
427 break;
428 }
429 }
430 assert_eq!(cursor, TOTAL, "the drain should end on the same cursor");
431 collected
432 };
433
434 assert_eq!(
435 unpaginated.len(),
436 TOTAL as usize,
437 "the single-page run delivered nothing, so the comparison below is vacuous"
438 );
439 assert_eq!(
440 unpaginated.len(),
441 paginated.len(),
442 "pagination changed how many changes arrived: {} vs {}",
443 unpaginated.len(),
444 paginated.len()
445 );
446 for (a, b) in unpaginated.iter().zip(paginated.iter()) {
447 assert_eq!(a.row_id, b.row_id, "pagination reordered or dropped a row");
448 assert_eq!(a.data, b.data, "pagination changed a decrypted payload");
449 assert_eq!(a.table, b.table);
450 }
451 }
452
453 // ── The three pull variants ──
454 //
455 // `pull_inner` is covered through the base `pull` above (cursor advance,
456 // has_more, decryption, pagination). What each wrapper adds on top is the shape
457 // of the request body it posts and the shape of what it hands back, so that is
458 // what these pin.
459
460 /// One encrypted change on the wire, as the server would return it.
461 fn served_change(
462 seq: i64,
463 device_id: DeviceId,
464 row_id: &str,
465 key: &[u8; 32],
466 plaintext: &serde_json::Value,
467 ) -> serde_json::Value {
468 let encrypted = synckit_client::crypto::encrypt_json(plaintext, key).unwrap();
469 json!({
470 "seq": seq,
471 "device_id": device_id,
472 "table": "tasks",
473 "op": "INSERT",
474 "row_id": row_id,
475 "timestamp": "2025-06-01T12:00:00Z",
476 "data": encrypted,
477 })
478 }
479
480 #[tokio::test]
481 async fn pull_filtered_puts_tables_and_since_in_the_body() {
482 let kit = MockKit::start().await;
483 let (client, key) = kit.keyed();
484 let device_id = DeviceId::new(Uuid::new_v4());
485 let plaintext = json!({"title": "filtered"});
486
487 kit.post(PULL_PATH)
488 .json(json!({
489 "changes": [served_change(7, device_id, "row-1", &key, &plaintext)],
490 "cursor": 7,
491 "has_more": true,
492 }))
493 .await;
494
495 let since = "2025-05-01T00:00:00Z"
496 .parse::<chrono::DateTime<Utc>>()
497 .unwrap();
498 let filter = synckit_client::PullFilter {
499 tables: Some(vec!["tasks".to_string(), "notes".to_string()]),
500 since: Some(since),
501 };
502 let (changes, cursor, has_more) = client.pull_filtered(device_id, 3, filter).await.unwrap();
503
504 assert_eq!(cursor, 7);
505 assert!(has_more);
506 assert_eq!(changes.len(), 1);
507 assert_eq!(changes[0].row_id, "row-1");
508 assert_eq!(changes[0].data.as_ref().unwrap(), &plaintext);
509
510 let body = kit.body(PULL_PATH).await;
511 assert_eq!(body["cursor"], 3);
512 assert_eq!(body["device_id"], json!(device_id));
513 assert_eq!(body["tables"], json!(["tasks", "notes"]));
514 assert_eq!(
515 body["since"]
516 .as_str()
517 .unwrap()
518 .parse::<chrono::DateTime<Utc>>()
519 .unwrap(),
520 since
521 );
522 }
523
524 #[tokio::test]
525 async fn an_empty_filter_omits_both_fields_from_the_body() {
526 let kit = MockKit::start().await;
527 let (client, _key) = kit.keyed();
528 let device_id = DeviceId::new(Uuid::new_v4());
529
530 kit.post(PULL_PATH)
531 .json(json!({"changes": [], "cursor": 0, "has_more": false}))
532 .await;
533
534 client
535 .pull_filtered(device_id, 0, synckit_client::PullFilter::default())
536 .await
537 .unwrap();
538
539 let body = kit.body(PULL_PATH).await;
540 assert!(
541 body.get("tables").is_none(),
542 "an empty table list is not sent"
543 );
544 assert!(body.get("since").is_none(), "an absent since is not sent");
545 }
546
547 #[tokio::test]
548 async fn pull_rich_keeps_device_id_and_seq() {
549 let kit = MockKit::start().await;
550 let (client, key) = kit.keyed();
551 let origin = DeviceId::new(Uuid::new_v4());
552 let me = DeviceId::new(Uuid::new_v4());
553 let first = json!({"title": "one"});
554 let second = json!({"title": "two"});
555
556 kit.post(PULL_PATH)
557 .json(json!({
558 "changes": [
559 served_change(41, origin, "row-a", &key, &first),
560 served_change(42, origin, "row-b", &key, &second),
561 ],
562 "cursor": 42,
563 "has_more": false,
564 }))
565 .await;
566
567 let (changes, cursor, has_more) = client.pull_rich(me, 40).await.unwrap();
568
569 assert_eq!(cursor, 42);
570 assert!(!has_more);
571 assert_eq!(changes.len(), 2);
572 assert_eq!(changes[0].seq, 41);
573 assert_eq!(changes[1].seq, 42);
574 assert_eq!(changes[0].device_id, origin);
575 assert_eq!(changes[1].device_id, origin);
576 assert_ne!(
577 changes[0].device_id, me,
578 "the wrapper carries the originating device, not the puller"
579 );
580 assert_eq!(changes[0].entry.row_id, "row-a");
581 assert_eq!(changes[0].entry.data.as_ref().unwrap(), &first);
582 assert_eq!(changes[1].entry.data.as_ref().unwrap(), &second);
583
584 let body = kit.body(PULL_PATH).await;
585 assert_eq!(body["cursor"], 40);
586 assert_eq!(body["device_id"], json!(me));
587 assert!(body.get("tables").is_none(), "pull_rich sends no filter");
588 }
589
590 #[tokio::test]
591 async fn pull_filtered_rich_carries_both_the_filter_and_the_metadata() {
592 let kit = MockKit::start().await;
593 let (client, key) = kit.keyed();
594 let origin = DeviceId::new(Uuid::new_v4());
595 let me = DeviceId::new(Uuid::new_v4());
596 let plaintext = json!({"title": "both"});
597
598 kit.post(PULL_PATH)
599 .json(json!({
600 "changes": [served_change(9, origin, "row-c", &key, &plaintext)],
601 "cursor": 9,
602 "has_more": false,
603 }))
604 .await;
605
606 let since = "2025-04-02T03:04:05Z"
607 .parse::<chrono::DateTime<Utc>>()
608 .unwrap();
609 let filter = synckit_client::PullFilter {
610 tables: Some(vec!["tasks".to_string()]),
611 since: Some(since),
612 };
613 let (changes, cursor, has_more) = client.pull_filtered_rich(me, 8, filter).await.unwrap();
614
615 assert_eq!(cursor, 9);
616 assert!(!has_more);
617 assert_eq!(changes.len(), 1);
618 assert_eq!(changes[0].seq, 9);
619 assert_eq!(changes[0].device_id, origin);
620 assert_eq!(changes[0].entry.row_id, "row-c");
621 assert_eq!(changes[0].entry.data.as_ref().unwrap(), &plaintext);
622
623 let body = kit.body(PULL_PATH).await;
624 assert_eq!(body["cursor"], 8);
625 assert_eq!(body["device_id"], json!(me));
626 assert_eq!(body["tables"], json!(["tasks"]));
627 assert_eq!(
628 body["since"]
629 .as_str()
630 .unwrap()
631 .parse::<chrono::DateTime<Utc>>()
632 .unwrap(),
633 since
634 );
635 }
636
637 #[tokio::test]
638 async fn pull_rich_drains_a_second_page() {
639 let kit = MockKit::start().await;
640 let (client, key) = kit.keyed();
641 let origin = DeviceId::new(Uuid::new_v4());
642 let me = DeviceId::new(Uuid::new_v4());
643
644 kit.post(PULL_PATH)
645 .once()
646 .json(json!({
647 "changes": [served_change(1, origin, "p1", &key, &json!({"n": 1}))],
648 "cursor": 1,
649 "has_more": true,
650 }))
651 .await;
652 kit.post(PULL_PATH)
653 .json(json!({
654 "changes": [served_change(2, origin, "p2", &key, &json!({"n": 2}))],
655 "cursor": 2,
656 "has_more": false,
657 }))
658 .await;
659
660 let mut cursor = 0;
661 let mut collected: Vec<synckit_client::PulledChange> = Vec::new();
662 loop {
663 let (changes, next, has_more) = client.pull_rich(me, cursor).await.unwrap();
664 collected.extend(changes);
665 cursor = next;
666 if !has_more {
667 break;
668 }
669 }
670
671 assert_eq!(cursor, 2);
672 assert_eq!(collected.len(), 2, "the drain needed both round trips");
673 assert_eq!(collected[0].seq, 1);
674 assert_eq!(collected[1].seq, 2);
675 assert_eq!(collected[0].entry.row_id, "p1");
676 assert_eq!(collected[1].entry.row_id, "p2");
677 assert_eq!(kit.hits(PULL_PATH).await, 2);
678
679 // The second request resumes from the cursor the first returned.
680 let bodies = kit.bodies("POST", PULL_PATH).await;
681 assert_eq!(bodies[0]["cursor"], 0);
682 assert_eq!(bodies[1]["cursor"], 1);
683 }
684