Skip to main content

max / synckit

50.2 KB · 1545 lines History Blame Raw
1 //! Tests for [`super`].
2
3 use super::*;
4 use crate::types::{ChangeOp, Hlc};
5 use serde_json::json;
6
7 /// Properties of the resolver.
8 ///
9 /// The contract here is convergence, which is a statement about every pair
10 /// of changes rather than about the pairs someone wrote down. The 43 tests
11 /// below are examples; these state the rule. See wiki `testing-posture`,
12 /// Phase 2.
13 mod properties {
14 use super::*;
15 use proptest::prelude::*;
16
17 /// Small device pool: node is the final tiebreak, so collisions are the
18 /// interesting case and random UUIDs would never produce them.
19 fn device_id() -> impl Strategy<Value = DeviceId> {
20 (0u8..3).prop_map(|n| {
21 let mut bytes = [0u8; 16];
22 bytes[15] = n;
23 DeviceId::new(Uuid::from_bytes(bytes))
24 })
25 }
26
27 /// Walls clustered tightly so ties and near-ties are common, plus a
28 /// far-future band that trips the clock-poisoning guard.
29 fn any_hlc() -> impl Strategy<Value = Hlc> {
30 let wall = prop_oneof![
31 6 => 1_700_000_000_000i64..1_700_000_000_010,
32 2 => 0i64..2_000_000_000_000,
33 2 => 4_000_000_000_000i64..8_000_000_000_000,
34 ];
35 (wall, 0u32..4, device_id()).prop_map(|(wall_ms, counter, node)| Hlc {
36 wall_ms,
37 counter,
38 node,
39 })
40 }
41
42 /// Pairs of clocks, weighted so exact ties are common.
43 ///
44 /// Two independent draws almost never collide, and the tie is exactly
45 /// where convergence is hardest: it is the case the payload tiebreak in
46 /// `resolve_tie` exists for. Generating the pair rather than two
47 /// independent clocks is what gives this property teeth, verified by
48 /// removing that tiebreak and watching the convergence test fail.
49 fn hlc_pair() -> impl Strategy<Value = (Hlc, Hlc)> {
50 prop_oneof![
51 3 => (any_hlc(), any_hlc()),
52 3 => any_hlc().prop_map(|h| (h, h)),
53 2 => (any_hlc(), 0u32..4).prop_map(|(h, counter)| (h, Hlc { counter, ..h })),
54 ]
55 }
56
57 fn entry_with(hlc: Hlc, payload: u8) -> ChangeEntry {
58 let mut e = make_entry("tasks", "row-1", ChangeOp::Update, Utc::now());
59 e.hlc = hlc;
60 e.data = Some(json!({ "v": payload }));
61 e
62 }
63
64 fn pulled_with(hlc: Hlc, payload: u8) -> PulledChange {
65 let mut p = make_pulled(
66 "tasks",
67 "row-1",
68 ChangeOp::Update,
69 Utc::now(),
70 hlc.node.as_uuid(),
71 1,
72 );
73 p.entry.hlc = hlc;
74 p.entry.data = Some(json!({ "v": payload }));
75 p
76 }
77
78 proptest! {
79 /// **Convergence.** Two devices hold the same pair with the roles
80 /// reversed: what is local on A is remote on B. If the answer
81 /// depended on which side the resolver was handed, the two devices
82 /// would keep different rows and never reconcile. No example test
83 /// notices unless it happens to pick that pair.
84 ///
85 /// Stated over the surviving payload rather than the `Resolution`
86 /// variant: at an exact tie both sides keep local, which converges
87 /// precisely because the two changes are then byte-identical.
88 #[test]
89 fn lww_picks_the_same_winner_from_either_side(
90 (a_hlc, b_hlc) in hlc_pair(),
91 a_payload in any::<u8>(),
92 b_payload in any::<u8>(),
93 ) {
94 let now = Utc::now();
95 let on_a = match resolve_lww_at(
96 &entry_with(a_hlc, a_payload),
97 &pulled_with(b_hlc, b_payload),
98 now,
99 ) {
100 Resolution::KeepLocal => a_payload,
101 Resolution::KeepRemote => b_payload,
102 other => return Err(TestCaseError::fail(format!("unexpected {other:?}"))),
103 };
104 let on_b = match resolve_lww_at(
105 &entry_with(b_hlc, b_payload),
106 &pulled_with(a_hlc, a_payload),
107 now,
108 ) {
109 Resolution::KeepLocal => b_payload,
110 Resolution::KeepRemote => a_payload,
111 other => return Err(TestCaseError::fail(format!("unexpected {other:?}"))),
112 };
113
114 prop_assert_eq!(
115 on_a, on_b,
116 "the two devices kept different payloads and will never converge: \
117 A kept {}, B kept {} (a={:?}, b={:?})",
118 on_a, on_b, a_hlc, b_hlc
119 );
120 }
121
122 /// Resolution is a function of its inputs. Cheap to state, and it is
123 /// what lets the resolver be re-run from a retry without
124 /// re-deriving the world.
125 #[test]
126 fn lww_is_deterministic(
127 (a_hlc, b_hlc) in hlc_pair(),
128 a_payload in any::<u8>(),
129 b_payload in any::<u8>(),
130 ) {
131 let now = Utc::now();
132 let local = entry_with(a_hlc, a_payload);
133 let first = resolve_lww_at(&local, &pulled_with(b_hlc, b_payload), now);
134 let second = resolve_lww_at(&local, &pulled_with(b_hlc, b_payload), now);
135 prop_assert_eq!(format!("{first:?}"), format!("{second:?}"));
136 }
137
138 /// A poisoned clock must never beat an honest one. This is the
139 /// guard's whole purpose: an unbounded future timestamp would
140 /// otherwise win every conflict for years.
141 #[test]
142 fn an_honest_clock_beats_a_poisoned_one(
143 honest_wall in 1_700_000_000_000i64..1_700_000_100_000,
144 poison_offset in (MAX_HLC_DRIFT_MS + 1)..10_000_000_000i64,
145 node_a in device_id(),
146 node_b in device_id(),
147 ) {
148 let now = Utc::now();
149 let honest = Hlc { wall_ms: honest_wall, counter: 0, node: node_a };
150 let poisoned = Hlc {
151 wall_ms: now.timestamp_millis().saturating_add(poison_offset),
152 counter: 0,
153 node: node_b,
154 };
155 prop_assume!(!is_clock_poisoned(&honest, now));
156
157 prop_assert!(
158 matches!(
159 resolve_lww_at(&entry_with(honest, 1), &pulled_with(poisoned, 2), now),
160 Resolution::KeepLocal
161 ),
162 "a poisoned remote won against an honest local"
163 );
164 prop_assert!(
165 matches!(
166 resolve_lww_at(&entry_with(poisoned, 2), &pulled_with(honest, 1), now),
167 Resolution::KeepRemote
168 ),
169 "a poisoned local won against an honest remote"
170 );
171 }
172
173 /// **A field merge converges, dependent groups included.**
174 ///
175 /// The two devices see mirror images of one conflict: what is local
176 /// on A is remote on B. They must compute the same merged object, or
177 /// they hold different bytes forever with nothing to detect it.
178 ///
179 /// This is aimed at the group rule specifically. Everything else in
180 /// the merge decides a field from values both devices have, but the
181 /// group rule picks a *side*, and "side" is the one concept that is
182 /// device-relative. It converges because the winner comes from
183 /// `resolve_tie` over the two entries rather than from which one the
184 /// caller happened to label local, and this is what would fail if
185 /// that ever regressed to a "ties go to local" rule.
186 #[test]
187 fn field_merge_converges_on_mirrored_inputs(
188 (a_hlc, b_hlc) in hlc_pair(),
189 a_state in 0u8..3,
190 b_state in 0u8..3,
191 a_at in 0u8..3,
192 b_at in 0u8..3,
193 a_note in 0u8..3,
194 b_note in 0u8..3,
195 ) {
196 const GROUPS: &[&[&str]] = &[&["state", "state_at"]];
197 let base = json!({"state": "s0", "state_at": "t0", "note": "n0"});
198 let a = json!({
199 "state": format!("s{a_state}"),
200 "state_at": format!("t{a_at}"),
201 "note": format!("n{a_note}"),
202 });
203 let b = json!({
204 "state": format!("s{b_state}"),
205 "state_at": format!("t{b_at}"),
206 "note": format!("n{b_note}"),
207 });
208
209 // On device A the local side is `a`; on device B it is `b`.
210 let on_a = resolve_field_merge_with(&a, &b, &base, &a_hlc, &b_hlc, GROUPS);
211 let on_b = resolve_field_merge_with(&b, &a, &base, &b_hlc, &a_hlc, GROUPS);
212
213 prop_assert_eq!(
214 format!("{on_a:?}"),
215 format!("{on_b:?}"),
216 "two devices merged the same conflict differently and will \
217 never converge (a={:?}, b={:?})",
218 a_hlc,
219 b_hlc
220 );
221 }
222
223 /// **A declared group never lands split across the two sides.**
224 ///
225 /// The property the declaration exists to buy. However the merge
226 /// resolves, every member of a contested group has to come from one
227 /// side, so the pair describes a state some device actually held. A
228 /// merge that decided `state` and `state_at` independently fails this
229 /// on the inputs where the two sides disagree about only one of them,
230 /// which is exactly the GoingsOn start()-versus-complete() case.
231 #[test]
232 fn a_contested_group_never_lands_split(
233 (a_hlc, b_hlc) in hlc_pair(),
234 a_state in 0u8..3,
235 b_state in 0u8..3,
236 a_at in 0u8..3,
237 b_at in 0u8..3,
238 ) {
239 const GROUPS: &[&[&str]] = &[&["state", "state_at"]];
240 let base = json!({"state": "s0", "state_at": "t0"});
241 let a = json!({"state": format!("s{a_state}"), "state_at": format!("t{a_at}")});
242 let b = json!({"state": format!("s{b_state}"), "state_at": format!("t{b_at}")});
243
244 let Resolution::Merged(merged) =
245 resolve_field_merge_with(&a, &b, &base, &a_hlc, &b_hlc, GROUPS)
246 else {
247 return Err(TestCaseError::fail("an object base must merge"));
248 };
249
250 // The result's group is allowed to be A's, B's, or the base's
251 // (untouched). What it must never be is one column from one side
252 // and the other from a different one.
253 let pair = (&merged["state"], &merged["state_at"]);
254 let candidates = [
255 (&a["state"], &a["state_at"]),
256 (&b["state"], &b["state_at"]),
257 (&base["state"], &base["state_at"]),
258 ];
259 prop_assert!(
260 candidates.contains(&pair),
261 "the group landed split: got {:?}, which is no device's version \
262 of it (a={a}, b={b})",
263 merged
264 );
265 }
266 }
267 }
268
269 /// Fixed node for locally-minted test entries, distinct from any random
270 /// `other_device`, so HLC tiebreaks are deterministic.
271 fn local_node() -> DeviceId {
272 DeviceId::new(Uuid::from_u128(0x1111_1111_1111_1111_1111_1111_1111_1111))
273 }
274
275 /// A second fixed device node, distinct from [`local_node`], for the
276 /// field-merge tests that need to name the remote side's clock.
277 fn remote_node() -> DeviceId {
278 DeviceId::new(Uuid::from_u128(0x2222_2222_2222_2222_2222_2222_2222_2222))
279 }
280
281 /// Map a wall-clock timestamp onto an HLC at `node`, so the timestamp-ordered
282 /// field-merge tests express the same intent against the HLC-based API. A
283 /// strictly later `ts` yields a strictly greater HLC (higher `wall_ms`); equal
284 /// `ts` on distinct nodes ties on the node, which is exactly the convergent
285 /// behavior the F1 fix guarantees.
286 fn ts_hlc(ts: DateTime<Utc>, node: DeviceId) -> Hlc {
287 Hlc::from_legacy(ts.timestamp_millis(), node)
288 }
289
290 fn make_entry(table: &str, row_id: &str, op: ChangeOp, ts: DateTime<Utc>) -> ChangeEntry {
291 // Derive the HLC wall component from the timestamp so the time-ordered
292 // tests below still express the intended ordering.
293 ChangeEntry {
294 table: table.to_string(),
295 op,
296 row_id: row_id.to_string(),
297 timestamp: ts,
298 hlc: Hlc::from_legacy(ts.timestamp_millis(), local_node()),
299 data: Some(json!({"value": "test"})),
300 extra: serde_json::Map::default(),
301 }
302 }
303
304 fn make_pulled(
305 table: &str,
306 row_id: &str,
307 op: ChangeOp,
308 ts: DateTime<Utc>,
309 device_id: Uuid,
310 seq: i64,
311 ) -> PulledChange {
312 let mut entry = make_entry(table, row_id, op, ts);
313 entry.hlc = Hlc::from_legacy(ts.timestamp_millis(), DeviceId::new(device_id));
314 PulledChange {
315 storage_version: None,
316 entry,
317 device_id: DeviceId::new(device_id),
318 seq,
319 }
320 }
321
322 /// Build a pulled change with an explicit HLC, for resolution tests that need
323 /// to control the clock independently of the wall timestamp.
324 fn pulled_with_hlc(row_id: &str, op: ChangeOp, hlc: Hlc, device_id: Uuid) -> PulledChange {
325 let mut p = make_pulled("tasks", row_id, op, Utc::now(), device_id, 1);
326 p.entry.hlc = hlc;
327 p
328 }
329
330 // ── detect_conflicts ──
331
332 #[test]
333 fn no_conflicts_when_different_rows() {
334 let our_device = Uuid::new_v4();
335 let other_device = Uuid::new_v4();
336 let now = Utc::now();
337
338 let remote = vec![make_pulled(
339 "tasks",
340 "r1",
341 ChangeOp::Update,
342 now,
343 other_device,
344 1,
345 )];
346 let local = vec![make_entry("tasks", "r2", ChangeOp::Update, now)];
347
348 let (clean, conflicts) = detect_conflicts(remote, &local, DeviceId::new(our_device));
349 assert_eq!(clean.len(), 1);
350 assert!(conflicts.is_empty());
351 }
352
353 #[test]
354 fn conflict_detected_same_row_different_device() {
355 let our_device = Uuid::new_v4();
356 let other_device = Uuid::new_v4();
357 let now = Utc::now();
358
359 let remote = vec![make_pulled(
360 "tasks",
361 "r1",
362 ChangeOp::Update,
363 now,
364 other_device,
365 1,
366 )];
367 let local = vec![make_entry("tasks", "r1", ChangeOp::Update, now)];
368
369 let (clean, conflicts) = detect_conflicts(remote, &local, DeviceId::new(our_device));
370 assert!(clean.is_empty());
371 assert_eq!(conflicts.len(), 1);
372 assert_eq!(conflicts[0].remote.entry.row_id, "r1");
373 assert_eq!(conflicts[0].local.row_id, "r1");
374 }
375
376 #[test]
377 fn own_echo_without_pending_edit_is_clean() {
378 // An echo of our own device with no contesting local pending edit is
379 // clean (it still passes the HLC gate at apply time).
380 let our_device = Uuid::new_v4();
381 let now = Utc::now();
382
383 let remote = vec![make_pulled(
384 "tasks",
385 "r1",
386 ChangeOp::Update,
387 now,
388 our_device,
389 1,
390 )];
391 let (clean, conflicts) = detect_conflicts(remote, &[], DeviceId::new(our_device));
392 assert_eq!(clean.len(), 1);
393 assert!(conflicts.is_empty());
394 }
395
396 #[test]
397 fn echo_contesting_a_pending_edit_is_resolved_not_trusted() {
398 // Hardening: a pulled change labeled as our own echo that contests an
399 // un-pushed local edit is resolved as a conflict, not waved through as
400 // clean. Trusting the device_id label would let a server relabel a hostile
401 // row as our echo to skip conflict detection entirely.
402 let our_device = Uuid::new_v4();
403 let now = Utc::now();
404
405 let remote = vec![make_pulled(
406 "tasks",
407 "r1",
408 ChangeOp::Update,
409 now,
410 our_device,
411 1,
412 )];
413 let local = vec![make_entry("tasks", "r1", ChangeOp::Update, now)];
414
415 let (clean, conflicts) = detect_conflicts(remote, &local, DeviceId::new(our_device));
416 assert!(clean.is_empty());
417 assert_eq!(
418 conflicts.len(),
419 1,
420 "echo contesting a pending edit is resolved"
421 );
422 }
423
424 #[test]
425 fn clean_changes_gate_drops_stale_keeps_newer() {
426 let our_device = Uuid::new_v4();
427 let other_device = Uuid::new_v4();
428 let now = Utc::now();
429 // A clean remote change for tasks/r1; its HLC wall == now_ms.
430 let remote = vec![make_pulled(
431 "tasks",
432 "r1",
433 ChangeOp::Update,
434 now,
435 other_device,
436 1,
437 )];
438 let (clean, _conflicts) = detect_conflicts(remote, &[], DeviceId::new(our_device));
439 assert_eq!(clean.len(), 1);
440
441 // No committed clock for the row → kept (first time we've seen it).
442 assert_eq!(clean.clone().gated(|_, _| None).len(), 1);
443 // Committed clock older than the remote → kept.
444 assert_eq!(
445 clean
446 .clone()
447 .gated(|_, _| Some(Hlc::zero(DeviceId::new(other_device))))
448 .len(),
449 1
450 );
451 // Committed clock newer than the remote → dropped (would clobber newer local).
452 let newer = Hlc {
453 wall_ms: now.timestamp_millis() + 1,
454 counter: 0,
455 node: DeviceId::new(other_device),
456 };
457 assert!(clean.gated(move |_, _| Some(newer)).is_empty());
458 }
459
460 #[test]
461 fn different_tables_same_row_id_no_conflict() {
462 let our_device = Uuid::new_v4();
463 let other_device = Uuid::new_v4();
464 let now = Utc::now();
465
466 let remote = vec![make_pulled(
467 "tasks",
468 "r1",
469 ChangeOp::Update,
470 now,
471 other_device,
472 1,
473 )];
474 let local = vec![make_entry("events", "r1", ChangeOp::Update, now)];
475
476 let (clean, conflicts) = detect_conflicts(remote, &local, DeviceId::new(our_device));
477 assert_eq!(clean.len(), 1);
478 assert!(conflicts.is_empty());
479 }
480
481 #[test]
482 fn detect_conflicts_correct_split() {
483 let our_device = Uuid::new_v4();
484 let other_device = Uuid::new_v4();
485 let now = Utc::now();
486
487 let remote = vec![
488 make_pulled("tasks", "r1", ChangeOp::Update, now, other_device, 1),
489 make_pulled("tasks", "r2", ChangeOp::Insert, now, other_device, 2),
490 make_pulled("events", "r3", ChangeOp::Delete, now, other_device, 3),
491 ];
492 let local = vec![
493 make_entry("tasks", "r1", ChangeOp::Update, now),
494 // r2 not in local → clean
495 // r3 not in local → clean
496 ];
497
498 let (clean, conflicts) = detect_conflicts(remote, &local, DeviceId::new(our_device));
499 // The negative side of `is_empty`: every other assertion in this file is
500 // `assert!(clean.is_empty())`, which a constant-`true` `is_empty` also
501 // satisfies. Two clean changes must report non-empty.
502 assert!(
503 !clean.is_empty(),
504 "two uncontested rows must not report an empty CleanChanges"
505 );
506 assert_eq!(clean.len(), 2);
507 assert_eq!(conflicts.len(), 1);
508 assert_eq!(conflicts[0].remote.entry.row_id, "r1");
509 }
510
511 /// `row_keys` is what a caller pre-fetches committed clocks with, so it has
512 /// to name every clean row and nothing else, in the order the changes were
513 /// pulled.
514 #[test]
515 fn row_keys_names_every_clean_row_in_order() {
516 let our_device = Uuid::new_v4();
517 let other_device = Uuid::new_v4();
518 let now = Utc::now();
519
520 let remote = vec![
521 make_pulled("tasks", "r1", ChangeOp::Update, now, other_device, 1),
522 make_pulled("events", "r2", ChangeOp::Insert, now, other_device, 2),
523 ];
524
525 let (clean, conflicts) = detect_conflicts(remote, &[], DeviceId::new(our_device));
526 assert!(conflicts.is_empty());
527 assert_eq!(
528 clean.row_keys().collect::<Vec<_>>(),
529 vec![("tasks", "r1"), ("events", "r2")]
530 );
531 }
532
533 #[test]
534 fn empty_remote_produces_no_conflicts() {
535 let our_device = Uuid::new_v4();
536 let now = Utc::now();
537
538 let remote = vec![];
539 let local = vec![make_entry("tasks", "r1", ChangeOp::Update, now)];
540
541 let (clean, conflicts) = detect_conflicts(remote, &local, DeviceId::new(our_device));
542 assert!(clean.is_empty());
543 assert!(conflicts.is_empty());
544 }
545
546 #[test]
547 fn empty_local_produces_no_conflicts() {
548 let our_device = Uuid::new_v4();
549 let other_device = Uuid::new_v4();
550 let now = Utc::now();
551
552 let remote = vec![make_pulled(
553 "tasks",
554 "r1",
555 ChangeOp::Update,
556 now,
557 other_device,
558 1,
559 )];
560 let local: Vec<ChangeEntry> = vec![];
561
562 let (clean, conflicts) = detect_conflicts(remote, &local, DeviceId::new(our_device));
563 assert_eq!(clean.len(), 1);
564 assert!(conflicts.is_empty());
565 }
566
567 // ── resolve_lww (HLC ordering) ──
568
569 #[test]
570 fn lww_picks_newer_timestamp() {
571 let other_device = Uuid::new_v4();
572 let old = Utc::now() - chrono::Duration::seconds(60);
573 let new = Utc::now();
574
575 let local = make_entry("tasks", "r1", ChangeOp::Update, old);
576 let remote = make_pulled("tasks", "r1", ChangeOp::Update, new, other_device, 1);
577
578 assert!(matches!(
579 resolve_lww(&local, &remote),
580 Resolution::KeepRemote
581 ));
582 }
583
584 #[test]
585 fn lww_local_wins_when_newer() {
586 let other_device = Uuid::new_v4();
587 let old = Utc::now() - chrono::Duration::seconds(60);
588 let new = Utc::now();
589
590 let local = make_entry("tasks", "r1", ChangeOp::Update, new);
591 let remote = make_pulled("tasks", "r1", ChangeOp::Update, old, other_device, 1);
592
593 assert!(matches!(
594 resolve_lww(&local, &remote),
595 Resolution::KeepLocal
596 ));
597 }
598
599 #[test]
600 fn lww_counter_breaks_same_wall_tie() {
601 // Same wall_ms, higher counter wins regardless of node.
602 let other = Uuid::new_v4();
603 let mut local = make_entry("tasks", "r1", ChangeOp::Update, Utc::now());
604 local.hlc = Hlc {
605 wall_ms: 1000,
606 counter: 2,
607 node: local_node(),
608 };
609 let remote = pulled_with_hlc(
610 "r1",
611 ChangeOp::Update,
612 Hlc {
613 wall_ms: 1000,
614 counter: 5,
615 node: DeviceId::new(other),
616 },
617 other,
618 );
619 assert!(matches!(
620 resolve_lww(&local, &remote),
621 Resolution::KeepRemote
622 ));
623 }
624
625 #[test]
626 fn lww_node_breaks_exact_tie_convergently() {
627 // Identical (wall, counter): the node decides, and BOTH devices must pick
628 // the same physical change. Use nodes with a known order (a < b).
629 let a = Uuid::from_u128(1);
630 let b = Uuid::from_u128(2);
631 let hlc_a = Hlc {
632 wall_ms: 1000,
633 counter: 0,
634 node: DeviceId::new(a),
635 };
636 let hlc_b = Hlc {
637 wall_ms: 1000,
638 counter: 0,
639 node: DeviceId::new(b),
640 };
641
642 // Device A: local is a's change, remote is b's change.
643 let mut local_a = make_entry("tasks", "r1", ChangeOp::Update, Utc::now());
644 local_a.hlc = hlc_a;
645 let remote_b = pulled_with_hlc("r1", ChangeOp::Update, hlc_b, b);
646 // b > a, so A drops its local and keeps remote (b's change).
647 assert!(matches!(
648 resolve_lww(&local_a, &remote_b),
649 Resolution::KeepRemote
650 ));
651
652 // Device B: local is b's change, remote is a's change.
653 let mut local_b = make_entry("tasks", "r1", ChangeOp::Update, Utc::now());
654 local_b.hlc = hlc_b;
655 let remote_a = pulled_with_hlc("r1", ChangeOp::Update, hlc_a, a);
656 // b > a, so B keeps its local (b's change). Both devices converge on b.
657 assert!(matches!(
658 resolve_lww(&local_b, &remote_a),
659 Resolution::KeepLocal
660 ));
661 }
662
663 #[test]
664 fn lww_identical_hlc_node_collision_converges_on_payload() {
665 // Pathological case: two installs share a node UUID (cloned config), so
666 // two genuinely different edits produce a byte-identical HLC. The payload
667 // tiebreak must still make both devices land on the same value.
668 let shared = Uuid::from_u128(7);
669 let hlc = Hlc {
670 wall_ms: 1000,
671 counter: 3,
672 node: DeviceId::new(shared),
673 };
674 let val_a = json!({"v": "aaa"});
675 let val_b = json!({"v": "bbb"}); // canonically greater than val_a
676
677 // Device A: local = a, remote = b.
678 let mut local_a = make_entry("tasks", "r1", ChangeOp::Update, Utc::now());
679 local_a.hlc = hlc;
680 local_a.data = Some(val_a.clone());
681 let mut remote_b = pulled_with_hlc("r1", ChangeOp::Update, hlc, shared);
682 remote_b.entry.data = Some(val_b.clone());
683 // b's payload sorts higher, so A drops local and keeps remote (b).
684 assert!(matches!(
685 resolve_lww(&local_a, &remote_b),
686 Resolution::KeepRemote
687 ));
688
689 // Device B: local = b, remote = a, must keep local (b). Both converge on b.
690 let mut local_b = make_entry("tasks", "r1", ChangeOp::Update, Utc::now());
691 local_b.hlc = hlc;
692 local_b.data = Some(val_b);
693 let mut remote_a = pulled_with_hlc("r1", ChangeOp::Update, hlc, shared);
694 remote_a.entry.data = Some(val_a);
695 assert!(matches!(
696 resolve_lww(&local_b, &remote_a),
697 Resolution::KeepLocal
698 ));
699 }
700
701 #[test]
702 fn lww_newer_update_beats_older_delete() {
703 // A strictly-newer UPDATE must beat an older DELETE, or the edit is
704 // silently lost.
705 let other = Uuid::new_v4();
706 let mut local = make_entry("tasks", "r1", ChangeOp::Update, Utc::now());
707 local.hlc = Hlc {
708 wall_ms: 2000,
709 counter: 0,
710 node: local_node(),
711 };
712 let remote = pulled_with_hlc(
713 "r1",
714 ChangeOp::Delete,
715 Hlc {
716 wall_ms: 1000,
717 counter: 0,
718 node: DeviceId::new(other),
719 },
720 other,
721 );
722 assert!(matches!(
723 resolve_lww(&local, &remote),
724 Resolution::KeepLocal
725 ));
726 }
727
728 #[test]
729 fn lww_newer_delete_beats_older_update() {
730 // Symmetric: a strictly-newer DELETE beats an older UPDATE.
731 let other = Uuid::new_v4();
732 let mut local = make_entry("tasks", "r1", ChangeOp::Update, Utc::now());
733 local.hlc = Hlc {
734 wall_ms: 1000,
735 counter: 0,
736 node: local_node(),
737 };
738 let remote = pulled_with_hlc(
739 "r1",
740 ChangeOp::Delete,
741 Hlc {
742 wall_ms: 2000,
743 counter: 0,
744 node: DeviceId::new(other),
745 },
746 other,
747 );
748 assert!(matches!(
749 resolve_lww(&local, &remote),
750 Resolution::KeepRemote
751 ));
752 }
753
754 // ── resolve_field_merge ──
755
756 #[test]
757 fn field_merge_non_overlapping_changes() {
758 let base = json!({"title": "old", "status": "pending", "priority": 1});
759 let local = json!({"title": "new title", "status": "pending", "priority": 1});
760 let remote = json!({"title": "old", "status": "done", "priority": 1});
761 let now = Utc::now();
762
763 let result = resolve_field_merge(
764 &local,
765 &remote,
766 &base,
767 &ts_hlc(now, local_node()),
768 &ts_hlc(now, remote_node()),
769 );
770 match result {
771 Resolution::Merged(v) => {
772 assert_eq!(v["title"], "new title");
773 assert_eq!(v["status"], "done");
774 assert_eq!(v["priority"], 1);
775 }
776 _ => panic!("Expected Merged"),
777 }
778 }
779
780 #[test]
781 fn field_merge_overlapping_newer_wins() {
782 let base = json!({"title": "old"});
783 let local = json!({"title": "local title"});
784 let remote = json!({"title": "remote title"});
785 let old = Utc::now() - chrono::Duration::seconds(60);
786 let new = Utc::now();
787
788 // Remote is newer → remote wins the overlapping field
789 let result = resolve_field_merge(
790 &local,
791 &remote,
792 &base,
793 &ts_hlc(old, local_node()),
794 &ts_hlc(new, remote_node()),
795 );
796 match result {
797 Resolution::Merged(v) => {
798 assert_eq!(v["title"], "remote title");
799 }
800 _ => panic!("Expected Merged"),
801 }
802
803 // Local is newer → local wins the overlapping field
804 let result = resolve_field_merge(
805 &local,
806 &remote,
807 &base,
808 &ts_hlc(new, local_node()),
809 &ts_hlc(old, remote_node()),
810 );
811 match result {
812 Resolution::Merged(v) => {
813 assert_eq!(v["title"], "local title");
814 }
815 _ => panic!("Expected Merged"),
816 }
817 }
818
819 #[test]
820 fn field_merge_key_deleted_on_one_side() {
821 let base = json!({"title": "old", "notes": "some notes"});
822 let local = json!({"title": "old"}); // notes deleted
823 let remote = json!({"title": "old", "notes": "some notes"});
824 let now = Utc::now();
825
826 let result = resolve_field_merge(
827 &local,
828 &remote,
829 &base,
830 &ts_hlc(now, local_node()),
831 &ts_hlc(now, remote_node()),
832 );
833 match result {
834 Resolution::Merged(v) => {
835 assert_eq!(v["title"], "old");
836 assert!(v.get("notes").is_none(), "notes should be deleted");
837 }
838 _ => panic!("Expected Merged"),
839 }
840 }
841
842 #[test]
843 fn field_merge_non_object_falls_back() {
844 let base = json!("string value");
845 let local = json!("local string");
846 let remote = json!("remote string");
847 let now = Utc::now();
848
849 assert!(matches!(
850 resolve_field_merge(
851 &local,
852 &remote,
853 &base,
854 &ts_hlc(now, local_node()),
855 &ts_hlc(now, remote_node())
856 ),
857 Resolution::KeepRemote
858 ));
859 }
860
861 #[test]
862 fn field_merge_empty_base_treats_all_as_changed() {
863 let base = json!({});
864 let local = json!({"title": "from local"});
865 let remote = json!({"status": "from remote"});
866 let now = Utc::now();
867
868 let result = resolve_field_merge(
869 &local,
870 &remote,
871 &base,
872 &ts_hlc(now, local_node()),
873 &ts_hlc(now, remote_node()),
874 );
875 match result {
876 Resolution::Merged(v) => {
877 assert_eq!(v["title"], "from local");
878 assert_eq!(v["status"], "from remote");
879 }
880 _ => panic!("Expected Merged"),
881 }
882 }
883
884 #[test]
885 fn field_merge_new_keys_from_both_sides() {
886 let base = json!({"existing": 1});
887 let local = json!({"existing": 1, "local_new": "a"});
888 let remote = json!({"existing": 1, "remote_new": "b"});
889 let now = Utc::now();
890
891 let result = resolve_field_merge(
892 &local,
893 &remote,
894 &base,
895 &ts_hlc(now, local_node()),
896 &ts_hlc(now, remote_node()),
897 );
898 match result {
899 Resolution::Merged(v) => {
900 assert_eq!(v["existing"], 1);
901 assert_eq!(v["local_new"], "a");
902 assert_eq!(v["remote_new"], "b");
903 }
904 _ => panic!("Expected Merged"),
905 }
906 }
907
908 // ── PulledChange preserves metadata ──
909
910 #[test]
911 fn pulled_change_preserves_device_id_and_seq() {
912 let device_id = Uuid::new_v4();
913 let now = Utc::now();
914 let pulled = make_pulled("tasks", "r1", ChangeOp::Insert, now, device_id, 42);
915
916 assert_eq!(pulled.device_id.as_uuid(), device_id);
917 assert_eq!(pulled.seq, 42);
918 assert_eq!(pulled.entry.table, "tasks");
919 assert_eq!(pulled.entry.row_id, "r1");
920 }
921
922 #[test]
923 fn pulled_change_clone_works() {
924 let device_id = Uuid::new_v4();
925 let now = Utc::now();
926 let pulled = make_pulled("tasks", "r1", ChangeOp::Insert, now, device_id, 1);
927 let cloned = pulled.clone();
928
929 assert_eq!(cloned.device_id, pulled.device_id);
930 assert_eq!(cloned.seq, pulled.seq);
931 assert_eq!(cloned.entry.table, pulled.entry.table);
932 }
933
934 // ── Resolution variants ──
935
936 #[test]
937 fn resolution_debug_format() {
938 let keep_local = Resolution::KeepLocal;
939 let keep_remote = Resolution::KeepRemote;
940 let merged = Resolution::Merged(json!({"a": 1}));
941 let skip = Resolution::Skip;
942
943 assert!(format!("{keep_local:?}").contains("KeepLocal"));
944 assert!(format!("{keep_remote:?}").contains("KeepRemote"));
945 assert!(format!("{merged:?}").contains("Merged"));
946 assert!(format!("{skip:?}").contains("Skip"));
947 }
948
949 // ── ConflictResolver trait ──
950
951 #[test]
952 fn custom_resolver_works() {
953 struct AlwaysRemote;
954 impl ConflictResolver for AlwaysRemote {
955 fn resolve(
956 &self,
957 _local: &ChangeEntry,
958 _remote: &PulledChange,
959 _base: Option<&serde_json::Value>,
960 ) -> Resolution {
961 Resolution::KeepRemote
962 }
963 }
964
965 let resolver = AlwaysRemote;
966 let now = Utc::now();
967 let local = make_entry("tasks", "r1", ChangeOp::Update, now);
968 let remote = make_pulled("tasks", "r1", ChangeOp::Update, now, Uuid::new_v4(), 1);
969
970 assert!(matches!(
971 resolver.resolve(&local, &remote, None),
972 Resolution::KeepRemote
973 ));
974 }
975
976 // ── Fuzz: edge cases and attack vectors ──
977
978 // Attack vector 1: duplicate local entries for same (table, row_id).
979 // HashMap insert means last entry wins. Verify the conflict pair uses
980 // the LATER local entry, not the earlier one.
981 #[test]
982 fn detect_conflicts_duplicate_local_uses_last_entry() {
983 let our_device = Uuid::new_v4();
984 let other_device = Uuid::new_v4();
985 let t1 = Utc::now() - chrono::Duration::seconds(60);
986 let t2 = Utc::now();
987
988 let remote = vec![make_pulled(
989 "tasks",
990 "r1",
991 ChangeOp::Update,
992 t2,
993 other_device,
994 1,
995 )];
996 // Two local entries for same (table, row_id): Insert then Update.
997 // The Update (last) should participate in conflict detection.
998 let local = vec![
999 make_entry("tasks", "r1", ChangeOp::Insert, t1),
1000 make_entry("tasks", "r1", ChangeOp::Update, t2),
1001 ];
1002
1003 let (_clean, conflicts) = detect_conflicts(remote, &local, DeviceId::new(our_device));
1004 assert_eq!(conflicts.len(), 1);
1005 // The conflict should use the Update (last entry), not the Insert
1006 assert_eq!(conflicts[0].local.op, ChangeOp::Update);
1007 assert_eq!(conflicts[0].local.timestamp, t2);
1008 }
1009
1010 // DELETE vs DELETE under HLC: the higher clock wins like any other pair, so
1011 // a strictly newer remote delete wins.
1012 #[test]
1013 fn lww_both_delete_newer_wins() {
1014 let other_device = Uuid::new_v4();
1015 let old = Utc::now() - chrono::Duration::seconds(60);
1016 let new = Utc::now();
1017
1018 let local = make_entry("tasks", "r1", ChangeOp::Delete, old);
1019 let remote = make_pulled("tasks", "r1", ChangeOp::Delete, new, other_device, 1);
1020
1021 assert!(matches!(
1022 resolve_lww(&local, &remote),
1023 Resolution::KeepRemote
1024 ));
1025 }
1026
1027 // F1: overlapping fields whose two sides carry the *same* wall-ms must
1028 // resolve convergently, not "ties go to local". Breaking a tie on local
1029 // would leave device A holding A and device B holding B, a permanent silent
1030 // divergence, so the exact tie breaks on the full HLC (distinct node) and
1031 // both devices land on the same physical value.
1032 #[test]
1033 fn field_merge_overlapping_equal_wall_converges() {
1034 let base = json!({"title": "base"});
1035 let val_a = json!({"title": "device A version"});
1036 let val_b = json!({"title": "device B version"});
1037 let now = Utc::now();
1038 let hlc_a = ts_hlc(now, local_node()); // node 0x111...
1039 let hlc_b = ts_hlc(now, remote_node()); // node 0x222... (> a), same wall
1040
1041 // Device A resolves (local = A's edit, remote = B's edit); device B
1042 // resolves the mirror image (local = B's edit, remote = A's edit).
1043 let on_a = resolve_field_merge(&val_a, &val_b, &base, &hlc_a, &hlc_b);
1044 let on_b = resolve_field_merge(&val_b, &val_a, &base, &hlc_b, &hlc_a);
1045
1046 let (title_a, title_b) = match (on_a, on_b) {
1047 (Resolution::Merged(a), Resolution::Merged(b)) => (a["title"].clone(), b["title"].clone()),
1048 _ => panic!("Expected Merged on both devices"),
1049 };
1050 assert_eq!(
1051 title_a, title_b,
1052 "both devices must converge on the same value at an equal-wall tie"
1053 );
1054 // Specifically on B's edit, since hlc_b > hlc_a on the node tiebreak.
1055 assert_eq!(title_a, "device B version");
1056 }
1057
1058 // Attack vector 4: HashMap iteration order determinism.
1059 // Non-overlapping keys from both sides should merge deterministically
1060 // regardless of HashMap iteration order.
1061 #[test]
1062 fn field_merge_deterministic_with_many_keys() {
1063 let base = json!({});
1064 let local = json!({"a": 1, "c": 3, "e": 5, "g": 7, "i": 9});
1065 let remote = json!({"b": 2, "d": 4, "f": 6, "h": 8, "j": 10});
1066 let now = Utc::now();
1067
1068 // Run multiple times to catch iteration-order bugs
1069 for _ in 0..10 {
1070 let result = resolve_field_merge(
1071 &local,
1072 &remote,
1073 &base,
1074 &ts_hlc(now, local_node()),
1075 &ts_hlc(now, remote_node()),
1076 );
1077 match result {
1078 Resolution::Merged(v) => {
1079 assert_eq!(v["a"], 1);
1080 assert_eq!(v["b"], 2);
1081 assert_eq!(v["c"], 3);
1082 assert_eq!(v["d"], 4);
1083 assert_eq!(v["e"], 5);
1084 assert_eq!(v["f"], 6);
1085 assert_eq!(v["g"], 7);
1086 assert_eq!(v["h"], 8);
1087 assert_eq!(v["i"], 9);
1088 assert_eq!(v["j"], 10);
1089 }
1090 _ => panic!("Expected Merged"),
1091 }
1092 }
1093 }
1094
1095 // Attack vector 5: null base with both local and remote as objects.
1096 // Falls back to KeepRemote, silently discarding local changes.
1097 #[test]
1098 fn field_merge_null_base_discards_local() {
1099 let base = json!(null);
1100 let local = json!({"title": "important local edit"});
1101 let remote = json!({"status": "remote only"});
1102 let now = Utc::now();
1103
1104 // BUG: Both sides are valid objects but null base causes KeepRemote,
1105 // which silently drops "title": "important local edit".
1106 // A better fallback might be to merge both against an empty base,
1107 // or fall back to LWW.
1108 let result = resolve_field_merge(
1109 &local,
1110 &remote,
1111 &base,
1112 &ts_hlc(now, local_node()),
1113 &ts_hlc(now, remote_node()),
1114 );
1115 assert!(
1116 matches!(result, Resolution::KeepRemote),
1117 "Null base should fall back to KeepRemote (current behavior)"
1118 );
1119 }
1120
1121 // Attack vector 6: empty object vs changed value in field_merge.
1122 // Both `{}` and a new value differ from the base value, so both are
1123 // "changed". This is an overlapping-field conflict.
1124 #[test]
1125 fn field_merge_empty_object_counts_as_change() {
1126 let base = json!({"meta": {"nested": "data"}});
1127 let local = json!({"meta": {}}); // Changed to empty object
1128 let remote = json!({"meta": "flat string"}); // Changed to string
1129 let old = Utc::now() - chrono::Duration::seconds(60);
1130 let new = Utc::now();
1131
1132 // Remote is newer, so remote wins the overlapping field
1133 let result = resolve_field_merge(
1134 &local,
1135 &remote,
1136 &base,
1137 &ts_hlc(old, local_node()),
1138 &ts_hlc(new, remote_node()),
1139 );
1140 match result {
1141 Resolution::Merged(v) => {
1142 assert_eq!(v["meta"], "flat string");
1143 }
1144 _ => panic!("Expected Merged"),
1145 }
1146
1147 // Local is newer, so local wins, meta becomes empty object
1148 let result = resolve_field_merge(
1149 &local,
1150 &remote,
1151 &base,
1152 &ts_hlc(new, local_node()),
1153 &ts_hlc(old, remote_node()),
1154 );
1155 match result {
1156 Resolution::Merged(v) => {
1157 assert_eq!(v["meta"], json!({}));
1158 }
1159 _ => panic!("Expected Merged"),
1160 }
1161 }
1162
1163 // Attack vector 7: multiple remote changes for same (table, row_id).
1164 // Each produces a separate ConflictPair with a CLONE of the same local entry.
1165 #[test]
1166 fn detect_conflicts_multiple_remote_same_row() {
1167 let our_device = Uuid::new_v4();
1168 let other_device = Uuid::new_v4();
1169 let now = Utc::now();
1170
1171 let remote = vec![
1172 make_pulled("tasks", "r1", ChangeOp::Update, now, other_device, 1),
1173 make_pulled("tasks", "r1", ChangeOp::Update, now, other_device, 2),
1174 make_pulled("tasks", "r1", ChangeOp::Delete, now, other_device, 3),
1175 ];
1176 let local = vec![make_entry("tasks", "r1", ChangeOp::Update, now)];
1177
1178 let (_clean, conflicts) = detect_conflicts(remote, &local, DeviceId::new(our_device));
1179 // All 3 remote changes conflict with the same local entry
1180 assert_eq!(conflicts.len(), 3);
1181 // Each gets a clone of the same local entry
1182 assert_eq!(conflicts[0].local.row_id, "r1");
1183 assert_eq!(conflicts[1].local.row_id, "r1");
1184 assert_eq!(conflicts[2].local.row_id, "r1");
1185 // Verify seq ordering is preserved
1186 assert_eq!(conflicts[0].remote.seq, 1);
1187 assert_eq!(conflicts[1].remote.seq, 2);
1188 assert_eq!(conflicts[2].remote.seq, 3);
1189 }
1190
1191 // Attack vector 8: numeric type coercion in serde_json PartialEq.
1192 // JSON `1` (u64) and `1.0` (f64) are different Value variants.
1193 // serde_json::Value PartialEq does NOT treat them as equal.
1194 #[test]
1195 fn field_merge_integer_vs_float_treated_as_different() {
1196 let base = json!({"count": 1}); // serde_json: Number(PosInt(1))
1197 let local = json!({"count": 1.0}); // serde_json: Number(Float(1.0))
1198 let remote = json!({"count": 1}); // unchanged from base
1199 let now = Utc::now();
1200
1201 let result = resolve_field_merge(
1202 &local,
1203 &remote,
1204 &base,
1205 &ts_hlc(now, local_node()),
1206 &ts_hlc(now, remote_node()),
1207 );
1208 match result {
1209 Resolution::Merged(v) => {
1210 // BUG CANDIDATE: `1` != `1.0` in serde_json, so local sees
1211 // "count" as changed (1 -> 1.0) even though semantically
1212 // identical. Remote sees no change. Merge applies local's 1.0.
1213 assert_eq!(
1214 v["count"], 1.0,
1215 "serde_json treats 1 and 1.0 as different values"
1216 );
1217 }
1218 _ => panic!("Expected Merged"),
1219 }
1220 }
1221
1222 // INSERT vs DELETE under HLC: operation is irrelevant, the higher clock wins.
1223 // A strictly-newer remote delete beats an older local insert.
1224 #[test]
1225 fn lww_newer_remote_delete_beats_older_local_insert() {
1226 let other_device = Uuid::new_v4();
1227 let old = Utc::now() - chrono::Duration::seconds(60);
1228 let new = Utc::now();
1229
1230 let local = make_entry("tasks", "r1", ChangeOp::Insert, old);
1231 let remote = make_pulled("tasks", "r1", ChangeOp::Delete, new, other_device, 1);
1232
1233 assert!(matches!(
1234 resolve_lww(&local, &remote),
1235 Resolution::KeepRemote
1236 ));
1237 }
1238
1239 // Reverse: a strictly-newer local delete beats an older remote insert.
1240 #[test]
1241 fn lww_newer_local_delete_beats_older_remote_insert() {
1242 let other_device = Uuid::new_v4();
1243 let old = Utc::now() - chrono::Duration::seconds(60);
1244 let new = Utc::now();
1245
1246 let local = make_entry("tasks", "r1", ChangeOp::Delete, new);
1247 let remote = make_pulled("tasks", "r1", ChangeOp::Insert, old, other_device, 1);
1248
1249 assert!(matches!(
1250 resolve_lww(&local, &remote),
1251 Resolution::KeepLocal
1252 ));
1253 }
1254
1255 // Bonus: field_merge where both sides delete the same key.
1256 // Both detect the key as deleted. Local applies deletion first.
1257 // Remote sees it as overlapping but equal-ts, so local's deletion stands.
1258 #[test]
1259 fn field_merge_both_delete_same_key() {
1260 let base = json!({"title": "old", "notes": "old notes"});
1261 let local = json!({"title": "old"}); // deleted "notes"
1262 let remote = json!({"title": "old"}); // also deleted "notes"
1263 let now = Utc::now();
1264
1265 let result = resolve_field_merge(
1266 &local,
1267 &remote,
1268 &base,
1269 &ts_hlc(now, local_node()),
1270 &ts_hlc(now, remote_node()),
1271 );
1272 match result {
1273 Resolution::Merged(v) => {
1274 assert_eq!(v["title"], "old");
1275 assert!(
1276 v.get("notes").is_none(),
1277 "Both sides deleted notes, should stay deleted"
1278 );
1279 }
1280 _ => panic!("Expected Merged"),
1281 }
1282 }
1283
1284 // field_merge where local deletes a key and remote modifies it. The
1285 // higher-HLC side wins the contested field: a strictly-newer local delete
1286 // beats an older remote modify, so the key stays deleted.
1287 #[test]
1288 fn field_merge_newer_local_delete_beats_remote_modify() {
1289 let base = json!({"title": "old", "notes": "original"});
1290 let local = json!({"title": "old"}); // deleted "notes"
1291 let remote = json!({"title": "old", "notes": "updated notes"});
1292 let old = Utc::now() - chrono::Duration::seconds(60);
1293 let new = Utc::now();
1294
1295 // Local strictly newer → its delete wins the overlapping "notes" field.
1296 let result = resolve_field_merge(
1297 &local,
1298 &remote,
1299 &base,
1300 &ts_hlc(new, local_node()),
1301 &ts_hlc(old, remote_node()),
1302 );
1303 match result {
1304 Resolution::Merged(v) => {
1305 assert!(
1306 v.get("notes").is_none(),
1307 "newer local delete should win over older remote modify"
1308 );
1309 }
1310 _ => panic!("Expected Merged"),
1311 }
1312 }
1313
1314 // Bonus: field_merge where both sides add the SAME new key with
1315 // different values. Both are in local_changed and remote_changed.
1316 #[test]
1317 fn field_merge_both_add_same_new_key_different_values() {
1318 let base = json!({"existing": 1});
1319 let local = json!({"existing": 1, "new_key": "local value"});
1320 let remote = json!({"existing": 1, "new_key": "remote value"});
1321 let old = Utc::now() - chrono::Duration::seconds(60);
1322 let new = Utc::now();
1323
1324 // Remote newer: remote wins the overlapping new key
1325 let result = resolve_field_merge(
1326 &local,
1327 &remote,
1328 &base,
1329 &ts_hlc(old, local_node()),
1330 &ts_hlc(new, remote_node()),
1331 );
1332 match result {
1333 Resolution::Merged(v) => {
1334 assert_eq!(v["new_key"], "remote value");
1335 }
1336 _ => panic!("Expected Merged"),
1337 }
1338
1339 // Local newer: local wins the overlapping new key
1340 let result = resolve_field_merge(
1341 &local,
1342 &remote,
1343 &base,
1344 &ts_hlc(new, local_node()),
1345 &ts_hlc(old, remote_node()),
1346 );
1347 match result {
1348 Resolution::Merged(v) => {
1349 assert_eq!(v["new_key"], "local value");
1350 }
1351 _ => panic!("Expected Merged"),
1352 }
1353
1354 // Equal wall: the winner is decided by the HLC node tiebreak, not "local
1355 // wins", and both devices resolving the mirror image agree on it.
1356 let ha = ts_hlc(new, local_node());
1357 let hb = ts_hlc(new, remote_node());
1358 let on_a = resolve_field_merge(&local, &remote, &base, &ha, &hb);
1359 let on_b = resolve_field_merge(&remote, &local, &base, &hb, &ha);
1360 match (on_a, on_b) {
1361 (Resolution::Merged(a), Resolution::Merged(b)) => {
1362 assert_eq!(
1363 a["new_key"], b["new_key"],
1364 "both devices converge on the same new_key"
1365 );
1366 }
1367 _ => panic!("Expected Merged on both devices"),
1368 }
1369 }
1370
1371 // 50 years in milliseconds, far beyond MAX_HLC_DRIFT_MS.
1372 const FIFTY_YEARS_MS: i64 = 50i64 * 365 * 24 * 3600 * 1000;
1373
1374 #[test]
1375 fn lww_rejects_poisoned_remote_for_honest_local() {
1376 let now = Utc::now();
1377 let now_ms = now.timestamp_millis();
1378 let mut local = make_entry("tasks", "r1", ChangeOp::Update, now);
1379 local.hlc = Hlc {
1380 wall_ms: now_ms,
1381 counter: 0,
1382 node: local_node(),
1383 };
1384 let mut pulled = make_pulled("tasks", "r1", ChangeOp::Update, now, Uuid::new_v4(), 1);
1385 // Unguarded, a wall clock 50 years ahead would win every conflict for decades.
1386 pulled.entry.hlc = Hlc {
1387 wall_ms: now_ms + FIFTY_YEARS_MS,
1388 counter: 0,
1389 node: DeviceId::new(Uuid::new_v4()),
1390 };
1391 assert!(is_clock_poisoned(&pulled.entry.hlc, now));
1392 assert!(matches!(
1393 resolve_lww_at(&local, &pulled, now),
1394 Resolution::KeepLocal
1395 ));
1396 }
1397
1398 #[test]
1399 fn lww_rejects_poisoned_local_for_honest_remote() {
1400 let now = Utc::now();
1401 let now_ms = now.timestamp_millis();
1402 let mut local = make_entry("tasks", "r1", ChangeOp::Update, now);
1403 local.hlc = Hlc {
1404 wall_ms: now_ms + FIFTY_YEARS_MS,
1405 counter: 0,
1406 node: local_node(),
1407 };
1408 let mut pulled = make_pulled("tasks", "r1", ChangeOp::Update, now, Uuid::new_v4(), 1);
1409 pulled.entry.hlc = Hlc {
1410 wall_ms: now_ms,
1411 counter: 0,
1412 node: DeviceId::new(Uuid::new_v4()),
1413 };
1414 assert!(matches!(
1415 resolve_lww_at(&local, &pulled, now),
1416 Resolution::KeepRemote
1417 ));
1418 }
1419
1420 #[test]
1421 fn field_merge_null_base_falls_back_to_lww_not_keep_remote() {
1422 let local = json!({"a": 1});
1423 let remote = json!({"a": 2});
1424 let base = serde_json::Value::Null;
1425 let local_ts = Utc::now();
1426 // Local strictly newer: must be kept, not silently discarded.
1427 let remote_older = local_ts - chrono::Duration::seconds(10);
1428 assert!(matches!(
1429 resolve_field_merge(
1430 &local,
1431 &remote,
1432 &base,
1433 &ts_hlc(local_ts, local_node()),
1434 &ts_hlc(remote_older, remote_node())
1435 ),
1436 Resolution::KeepLocal
1437 ));
1438 // Remote newer: remote wins.
1439 let remote_newer = local_ts + chrono::Duration::seconds(10);
1440 assert!(matches!(
1441 resolve_field_merge(
1442 &local,
1443 &remote,
1444 &base,
1445 &ts_hlc(local_ts, local_node()),
1446 &ts_hlc(remote_newer, remote_node())
1447 ),
1448 Resolution::KeepRemote
1449 ));
1450 }
1451
1452 /// The drift window is a stated contract (5 minutes), not an arbitrary
1453 /// number: it is the honest inter-device skew SyncKit promises to tolerate,
1454 /// and every other test in the tree refers to it symbolically, so an
1455 /// arithmetic slip in `5 * 60 * 1000` would change no other outcome.
1456 #[test]
1457 fn max_hlc_drift_is_five_minutes() {
1458 assert_eq!(MAX_HLC_DRIFT_MS, 300_000);
1459 }
1460
1461 /// The poisoning threshold is exclusive: an entry sitting exactly on
1462 /// `now + MAX_HLC_DRIFT_MS` is still honest, and only the next millisecond
1463 /// is poisoned. Both sides of the boundary, because `>` and `>=` differ
1464 /// only at equality.
1465 #[test]
1466 fn poisoning_threshold_is_exclusive_at_the_drift_boundary() {
1467 let now = Utc::now();
1468 let now_ms = now.timestamp_millis();
1469
1470 let at_boundary = Hlc {
1471 wall_ms: now_ms + MAX_HLC_DRIFT_MS,
1472 counter: 0,
1473 node: remote_node(),
1474 };
1475 assert!(
1476 !is_clock_poisoned(&at_boundary, now),
1477 "an entry exactly at the drift limit is within tolerated skew"
1478 );
1479
1480 let one_past = Hlc {
1481 wall_ms: now_ms + MAX_HLC_DRIFT_MS + 1,
1482 counter: 0,
1483 node: remote_node(),
1484 };
1485 assert!(
1486 is_clock_poisoned(&one_past, now),
1487 "one millisecond past the drift limit is poisoned"
1488 );
1489
1490 let one_before = Hlc {
1491 wall_ms: now_ms + MAX_HLC_DRIFT_MS - 1,
1492 counter: 0,
1493 node: remote_node(),
1494 };
1495 assert!(!is_clock_poisoned(&one_before, now));
1496 }
1497
1498 /// The boundary as `resolve_lww_at` sees it: a remote sitting exactly on the
1499 /// limit is honest, so it wins on raw HLC order; one millisecond further out
1500 /// loses to the older local. Same two inputs, opposite resolutions, so the
1501 /// threshold cannot be shifted by a millisecond without this failing.
1502 #[test]
1503 fn lww_keeps_a_remote_at_the_drift_boundary_and_rejects_the_next_ms() {
1504 let now = Utc::now();
1505 let now_ms = now.timestamp_millis();
1506
1507 let mut local = make_entry("tasks", "r1", ChangeOp::Update, now);
1508 local.hlc = Hlc {
1509 wall_ms: now_ms,
1510 counter: 0,
1511 node: local_node(),
1512 };
1513
1514 let mut at_boundary = make_pulled("tasks", "r1", ChangeOp::Update, now, Uuid::new_v4(), 1);
1515 at_boundary.entry.hlc = Hlc {
1516 wall_ms: now_ms + MAX_HLC_DRIFT_MS,
1517 counter: 0,
1518 node: remote_node(),
1519 };
1520 assert!(matches!(
1521 resolve_lww_at(&local, &at_boundary, now),
1522 Resolution::KeepRemote
1523 ));
1524
1525 let mut one_past = make_pulled("tasks", "r1", ChangeOp::Update, now, Uuid::new_v4(), 1);
1526 one_past.entry.hlc = Hlc {
1527 wall_ms: now_ms + MAX_HLC_DRIFT_MS + 1,
1528 counter: 0,
1529 node: remote_node(),
1530 };
1531 assert!(matches!(
1532 resolve_lww_at(&local, &one_past, now),
1533 Resolution::KeepLocal
1534 ));
1535 }
1536
1537 #[test]
1538 fn canonical_payload_sorts_map_keys() {
1539 // Pins the exact-HLC tiebreak's convergence invariant: serde_json must
1540 // serialize object keys in sorted order. Fails loudly if `preserve_order`
1541 // is ever enabled anywhere in the dependency tree.
1542 let bytes = canonical_payload(Some(&json!({"b": 1, "a": 2})));
1543 assert_eq!(bytes, br#"{"a":2,"b":1}"#.to_vec());
1544 }
1545