Skip to main content

max / synckit

61.2 KB · 1678 lines History Blame Raw
1 //! Client-side conflict detection and resolution for SyncKit.
2 //!
3 //! SyncKit uses E2E encryption, the server never sees row contents and cannot
4 //! merge or compare data. All conflict resolution must happen client-side after
5 //! decryption.
6 //!
7 //! This module provides:
8 //! - [`detect_conflicts`]: pure function that splits pulled changes into clean
9 //! (non-conflicting) and conflicting sets
10 //! - [`resolve_lww`]: last-write-wins resolution by `client_timestamp`
11 //! - [`resolve_field_merge`]: 3-way JSON object merge using a base version
12 //! - [`ConflictResolver`]: trait for custom resolution strategies
13
14 use std::cmp::Ordering;
15 use std::collections::HashMap;
16
17 use chrono::{DateTime, Utc};
18 #[cfg(test)]
19 use uuid::Uuid;
20
21 use crate::ids::DeviceId;
22 use crate::types::{ChangeEntry, Hlc, PulledChange};
23
24 /// A remote change that conflicts with a local pending change.
25 #[derive(Debug, Clone)]
26 pub struct ConflictPair {
27 /// The remote change from pull.
28 pub remote: PulledChange,
29 /// The local pending change that conflicts.
30 pub local: ChangeEntry,
31 }
32
33 /// The non-conflicting changes from a pull, withheld behind the HLC gate.
34 ///
35 /// "Clean" means only that no local *pending* edit contests the row, **not**
36 /// that the remote change is newer than what is already committed locally. A
37 /// device that applied and pushed a newer edit, then pulls an older remote edit
38 /// for the same row, sees that older edit here. Applying it blindly clobbers the
39 /// newer local value.
40 ///
41 /// So the applicable entries cannot be read out directly: [`gated`](Self::gated)
42 /// is the only door, and it requires the row's committed HLC. This makes
43 /// "apply a clean change without checking the committed clock" unrepresentable
44 /// rather than a contract the caller has to remember.
45 #[derive(Debug, Clone)]
46 pub struct CleanChanges(Vec<PulledChange>);
47
48 impl CleanChanges {
49 /// HLC-gate the clean changes against committed local clocks.
50 ///
51 /// `committed_hlc(table, row_id)` returns the HLC currently committed for a
52 /// row, or `None` if it has never been applied locally. A change is dropped
53 /// when its HLC is older than or equal to the committed clock; everything
54 /// else is returned in pull order, ready to apply.
55 ///
56 /// Samples the wall clock once for the poisoning check; use
57 /// [`gated_at`](Self::gated_at) to pin `now` across a whole sync batch (or in
58 /// tests) so every change in the batch is classified against the same instant.
59 pub fn gated(self, committed_hlc: impl Fn(&str, &str) -> Option<Hlc>) -> Vec<ChangeEntry> {
60 self.gated_at(Utc::now(), committed_hlc)
61 }
62
63 /// [`gated`](Self::gated) with an explicit `now`, so the clock-poisoning
64 /// classification is deterministic (testable) and consistent across a batch.
65 ///
66 /// The poisoning threshold is still fundamentally wall-clock-relative: two
67 /// devices whose clocks straddle the drift boundary by less than their skew
68 /// can classify a near-boundary entry differently. Pinning `now` per batch
69 /// removes the *within-device* drift between the batch's first and last change;
70 /// the residual cross-device boundary window is the bounded, documented
71 /// tradeoff of [`MAX_HLC_DRIFT_MS`] (see [`resolve_lww_at`]).
72 pub fn gated_at(
73 self,
74 now: DateTime<Utc>,
75 committed_hlc: impl Fn(&str, &str) -> Option<Hlc>,
76 ) -> Vec<ChangeEntry> {
77 self.0
78 .into_iter()
79 .filter(|p| {
80 // A clean change is applied without passing through `resolve_lww`,
81 // so the clock-poisoning guard must also live here: never apply an
82 // entry whose wall clock is implausibly far in the future.
83 if is_clock_poisoned(&p.entry.hlc, now) {
84 tracing::warn!(
85 table = %p.entry.table,
86 wall_ms = p.entry.hlc.wall_ms,
87 "dropping clean change with implausibly-future HLC"
88 );
89 return false;
90 }
91 let apply = committed_hlc(&p.entry.table, &p.entry.row_id)
92 .is_none_or(|committed| p.entry.hlc > committed);
93 if !apply {
94 tracing::debug!(
95 table = %p.entry.table,
96 row_id = %p.entry.row_id,
97 "clean change gated out: older than the committed HLC"
98 );
99 }
100 apply
101 })
102 .map(|p| p.entry)
103 .collect()
104 }
105
106 /// The `(table, row_id)` addresses of the clean changes, so a caller can
107 /// pre-fetch their committed clocks in one query before [`gated`](Self::gated).
108 /// Yields addresses only, the change payloads stay sealed until gating.
109 pub fn row_keys(&self) -> impl Iterator<Item = (&str, &str)> {
110 self.0
111 .iter()
112 .map(|p| (p.entry.table.as_str(), p.entry.row_id.as_str()))
113 }
114
115 /// Number of clean changes before gating (for metrics/logging).
116 pub fn len(&self) -> usize {
117 self.0.len()
118 }
119
120 /// Whether there are no clean changes at all.
121 pub fn is_empty(&self) -> bool {
122 self.0.is_empty()
123 }
124 }
125
126 /// How a conflict should be resolved.
127 #[derive(Debug, Clone)]
128 pub enum Resolution {
129 /// Keep the local version; skip the remote change.
130 /// The local version will push on the next sync cycle.
131 KeepLocal,
132 /// Apply the remote change, discarding the local version.
133 KeepRemote,
134 /// Apply a merged result that combines both changes.
135 Merged(serde_json::Value),
136 /// Skip both changes (neither apply nor push).
137 Skip,
138 }
139
140 /// Trait for custom conflict resolution strategies.
141 ///
142 /// Implement this to plug in app-specific merge logic. The `base` parameter
143 /// is `Some` only if the app provides a base-store adapter.
144 pub trait ConflictResolver: Send + Sync {
145 fn resolve(
146 &self,
147 local: &ChangeEntry,
148 remote: &PulledChange,
149 base: Option<&serde_json::Value>,
150 ) -> Resolution;
151 }
152
153 /// Split pulled changes into non-conflicting and conflicting sets.
154 ///
155 /// A conflict exists when a remote change and a local pending change both
156 /// modify the same `(table, row_id)`. Classification is by *contest*, not by the
157 /// server-asserted `device_id`: any pulled change that a local pending edit
158 /// contests is resolved, even one labeled as our own echo, trusting that label
159 /// would let a server relabel a hostile row as our echo to route it around
160 /// conflict detection. `our_device_id` is used only to log that spoof signal. A
161 /// pulled change with no contesting local edit is clean (a genuine echo, or a
162 /// remote change we hold no competing edit for); it is still HLC-gated at apply
163 /// time via [`CleanChanges::gated`].
164 ///
165 /// **Precondition:** `local_pending` should contain at most one entry per
166 /// `(table, row_id)`. If duplicates exist, only the last one participates
167 /// in conflict detection (earlier entries are silently ignored). Callers
168 /// should compact local pending changes before calling this function.
169 ///
170 /// If `remote` contains multiple changes for the same `(table, row_id)`,
171 /// each produces a separate `ConflictPair` with a clone of the same local
172 /// entry. The caller must resolve them in order (by `seq`).
173 ///
174 /// Returns `(clean, conflicts)` where `clean` changes have no *un-pushed* local
175 /// edit to conflict with.
176 ///
177 /// **Contract, clean changes still need HLC gating.** "Clean" means only that
178 /// no local *pending* edit contests the row. It does **not** mean the remote
179 /// change is newer than what is already committed locally: a device that has
180 /// applied and pushed a newer edit, then pulls an older remote edit for the same
181 /// row, will see that older edit in `clean`. The caller must compare each clean
182 /// change's [`Hlc`](crate::types::Hlc) against the row's committed HLC and skip
183 /// it if older. Applying clean changes blindly can clobber a newer local value.
184 pub fn detect_conflicts(
185 remote: Vec<PulledChange>,
186 local_pending: &[ChangeEntry],
187 our_device_id: DeviceId,
188 ) -> (CleanChanges, Vec<ConflictPair>) {
189 // Build lookup: (table, row_id) -> last local pending entry.
190 // If local_pending has duplicates for the same key, the last entry wins.
191 let mut local_map: HashMap<(&str, &str), &ChangeEntry> = HashMap::new();
192 for entry in local_pending {
193 local_map.insert((&entry.table, &entry.row_id), entry);
194 }
195
196 let mut clean = Vec::new();
197 let mut conflicts = Vec::new();
198
199 for pulled in remote {
200 let key = (pulled.entry.table.as_str(), pulled.entry.row_id.as_str());
201 if let Some(&local_entry) = local_map.get(&key) {
202 // A pending local edit contests this row. Resolve it regardless of the
203 // asserted device_id: a change claiming to be our echo while contesting
204 // an un-pushed local edit is either a benign re-pull of an in-flight
205 // change or a server relabeling a hostile row to skip detection. Either
206 // way, resolve rather than trust the label, a genuine echo resolves to
207 // an identical value.
208 if pulled.device_id == our_device_id {
209 tracing::debug!(
210 table = %pulled.entry.table,
211 row_id = %pulled.entry.row_id,
212 "pulled change claims our device_id but contests a pending local edit; resolving as a conflict",
213 );
214 }
215 conflicts.push(ConflictPair {
216 remote: pulled,
217 local: local_entry.clone(),
218 });
219 } else {
220 clean.push(pulled);
221 }
222 }
223
224 (CleanChanges(clean), conflicts)
225 }
226
227 /// Last-write-wins resolution by hybrid logical clock.
228 ///
229 /// The higher [`Hlc`](crate::types::Hlc) wins, **regardless of operation**. Because
230 /// the HLC carries the originating device in its `node` component, every value is
231 /// globally unique and the comparison is convergent: device A resolving (local=a,
232 /// remote=b) and device B resolving (local=b, remote=a) compare the same two HLCs
233 /// and keep the same physical change.
234 ///
235 /// This replaces the prior wall-clock + "delete always wins" rule. Delete-wins
236 /// caused permanent loss when a delete on one device beat a strictly-newer edit on
237 /// another (ultra-fuzz Run #28); under HLC ordering a newer edit beats an older
238 /// delete and vice-versa, by clock value alone.
239 ///
240 /// An *exactly* equal HLC normally means a same-device echo, since a distinct
241 /// `node` makes the value unique. It can only collide across devices if two
242 /// installs share a `node` UUID (a cloned config dir or a restored backup). To
243 /// stay convergent even then, an exact tie is broken on the canonical payload
244 /// bytes, a deterministic order both devices compute identically, rather than
245 /// unconditionally keeping local (which would let A keep `a` while B keeps `b`).
246 pub fn resolve_lww(local: &ChangeEntry, remote: &PulledChange) -> Resolution {
247 resolve_lww_at(local, remote, Utc::now())
248 }
249
250 /// Maximum forward wall-clock drift tolerated before an HLC is judged
251 /// clock-poisoned. An entry whose wall component is beyond `now + MAX_HLC_DRIFT_MS`
252 /// carries an implausible future timestamp, from a misconfigured clock or a
253 /// hostile peer holding the shared key, and must not be allowed to win LWW
254 /// against an honest entry. Set generously (5 min) so honest inter-device skew is
255 /// never flagged.
256 pub const MAX_HLC_DRIFT_MS: i64 = 5 * 60 * 1000;
257
258 /// Whether `hlc`'s wall clock is implausibly far in the future relative to `now`,
259 /// the clock-poisoning signal. A bare wall clock, and even an unclamped HLC,
260 /// lets such an entry win *every* conflict until real time catches up (years,
261 /// for a hostile timestamp); this predicate is how [`resolve_lww_at`] refuses to
262 /// let it. Exposed so an app driving its own apply loop can quarantine the same
263 /// entries at ingest.
264 pub fn is_clock_poisoned(hlc: &Hlc, now: DateTime<Utc>) -> bool {
265 hlc.wall_ms > now.timestamp_millis().saturating_add(MAX_HLC_DRIFT_MS)
266 }
267
268 /// [`resolve_lww`] with an explicit `now`, so the poisoning guard is
269 /// deterministic and testable.
270 ///
271 /// A poisoned entry (see [`is_clock_poisoned`]) *loses* to a non-poisoned one
272 /// regardless of raw HLC order, this is what stops a far-future clock from
273 /// winning forever. Merely capping the future value would not: capped at
274 /// `now + drift` it would perpetually track "now" and keep winning by the drift
275 /// margin. When both sides are poisoned, or neither is, the normal HLC order
276 /// applies. Two honest devices agree except in the narrow window where an entry
277 /// straddles the drift threshold by less than their clock skew, vastly better
278 /// than the unbounded, permanent domination an unguarded compare allows.
279 pub fn resolve_lww_at(
280 local: &ChangeEntry,
281 remote: &PulledChange,
282 now: DateTime<Utc>,
283 ) -> Resolution {
284 match (
285 is_clock_poisoned(&local.hlc, now),
286 is_clock_poisoned(&remote.entry.hlc, now),
287 ) {
288 (false, true) => {
289 tracing::warn!(
290 remote_wall_ms = remote.entry.hlc.wall_ms,
291 "remote HLC wall-clock is implausibly far in the future; rejecting it in LWW"
292 );
293 return Resolution::KeepLocal;
294 }
295 (true, false) => {
296 tracing::warn!(
297 local_wall_ms = local.hlc.wall_ms,
298 "local HLC wall-clock is implausibly far in the future; letting remote win LWW"
299 );
300 return Resolution::KeepRemote;
301 }
302 _ => {}
303 }
304 let resolution = match resolve_tie(
305 &local.hlc,
306 &canonical_payload(local.data.as_ref()),
307 &remote.entry.hlc,
308 &canonical_payload(remote.entry.data.as_ref()),
309 ) {
310 Ordering::Less => Resolution::KeepRemote,
311 // Greater or an exact tie both keep local: at a true tie the two changes
312 // are byte-identical, so keeping either side converges.
313 Ordering::Greater | Ordering::Equal => Resolution::KeepLocal,
314 };
315 // Every LWW resolution discards one side; log at debug so a lost edit is
316 // diagnosable after the fact rather than vanishing silently.
317 tracing::debug!(
318 table = %local.table,
319 row_id = %local.row_id,
320 kept = ?resolution,
321 "LWW resolved",
322 );
323 resolution
324 }
325
326 /// The single device-independent tiebreak both resolvers route through.
327 ///
328 /// Orders two conflicting changes by full [`Hlc`] first, globally unique via its
329 /// `node`, so both devices compute the identical result, then breaks an *exact*
330 /// HLC tie on the canonical payload bytes, a value every device derives the same
331 /// way. [`Ordering::Greater`] means the `a` side wins.
332 ///
333 /// One primitive is the point. The old field-merge rule broke ties on "local
334 /// wins" (device-relative → device A keeps `a`, device B keeps `b`, permanent
335 /// silent divergence) and on bare wall-`ms` (a tie window orders of magnitude
336 /// wider than the exact-HLC tie here). With both [`resolve_lww_at`] and
337 /// [`resolve_field_merge`] deferring to this, a device-relative winner rule has
338 /// no code path left.
339 fn resolve_tie(a_hlc: &Hlc, a_payload: &[u8], b_hlc: &Hlc, b_payload: &[u8]) -> Ordering {
340 a_hlc.cmp(b_hlc).then_with(|| a_payload.cmp(b_payload))
341 }
342
343 /// Deterministic byte encoding of a concrete JSON value for the exact-HLC
344 /// tiebreak. `serde_json`'s default `Map` is sorted, so equal values always
345 /// serialize to equal bytes on every device, the property the tiebreak relies on.
346 ///
347 /// INVARIANT: this convergence holds only while `serde_json` keeps insertion
348 /// order OFF (i.e. the `preserve_order` feature is NOT enabled anywhere in the
349 /// dependency tree). If it is ever turned on, map-key order becomes
350 /// insertion-dependent and two devices can compute different tiebreak bytes for
351 /// equal values, divergence. The `canonical_payload_sorts_map_keys` test pins
352 /// this (it fails the moment insertion order leaks in).
353 fn canonical_value(v: &serde_json::Value) -> Vec<u8> {
354 serde_json::to_vec(v).unwrap_or_default()
355 }
356
357 /// [`canonical_value`] for an optional payload; `None` (a delete) canonicalizes
358 /// to empty bytes.
359 fn canonical_payload(data: Option<&serde_json::Value>) -> Vec<u8> {
360 data.map(canonical_value).unwrap_or_default()
361 }
362
363 /// 3-way field-level merge for JSON objects (top-level keys only).
364 ///
365 /// Compares `local` and `remote` against `base` to determine which fields
366 /// each side changed, then merges non-overlapping changes. For overlapping
367 /// fields, the newer timestamp wins.
368 ///
369 /// The overlapping-field winner (and the no-base fallback) is decided by
370 /// [`resolve_tie`] over the two sides' HLCs, not by a device-relative "ties go to
371 /// local" rule, so two devices resolving the mirror image of the same conflict
372 /// converge on the same merged object. Pass each side's [`Hlc`] (`local_hlc` from
373 /// the local [`ChangeEntry`], `remote_hlc` from the pulled entry).
374 ///
375 /// If any input is not a JSON object (including `Value::Null` for a missing base
376 /// snapshot), a field-level merge is impossible, so this falls back to
377 /// last-writer-wins on the HLCs rather than unconditionally keeping remote,
378 /// which would silently discard a strictly-newer local edit whenever no base is
379 /// available (e.g. a first-ever edit).
380 pub fn resolve_field_merge(
381 local: &serde_json::Value,
382 remote: &serde_json::Value,
383 base: &serde_json::Value,
384 local_hlc: &Hlc,
385 remote_hlc: &Hlc,
386 ) -> Resolution {
387 // Device-independent winner of a contested field (and of the whole entry in
388 // the no-base fallback): identical on every device, so the merge converges.
389 // `>= Equal` keeps local; only a strict `Less` hands the field to remote.
390 let local_wins = resolve_tie(
391 local_hlc,
392 &canonical_value(local),
393 remote_hlc,
394 &canonical_value(remote),
395 ) != Ordering::Less;
396
397 let (Some(local_obj), Some(remote_obj), Some(base_obj)) =
398 (local.as_object(), remote.as_object(), base.as_object())
399 else {
400 // No usable base: last-writer-wins on the HLCs, keeping local when it wins
401 // the tie instead of dropping it.
402 tracing::debug!(
403 local_wins,
404 ?local_hlc,
405 ?remote_hlc,
406 "field-merge base unavailable; fell back to HLC LWW"
407 );
408 return if local_wins {
409 Resolution::KeepLocal
410 } else {
411 Resolution::KeepRemote
412 };
413 };
414
415 // Compute diffs: keys where local/remote differ from base
416 let mut local_changed: HashMap<&str, Option<&serde_json::Value>> = HashMap::new();
417 let mut remote_changed: HashMap<&str, Option<&serde_json::Value>> = HashMap::new();
418
419 // Check keys in base for changes or deletions
420 for key in base_obj.keys() {
421 let base_val = &base_obj[key];
422
423 match local_obj.get(key) {
424 Some(local_val) if local_val != base_val => {
425 local_changed.insert(key, Some(local_val));
426 }
427 None => {
428 // Key deleted on local side
429 local_changed.insert(key, None);
430 }
431 _ => {}
432 }
433
434 match remote_obj.get(key) {
435 Some(remote_val) if remote_val != base_val => {
436 remote_changed.insert(key, Some(remote_val));
437 }
438 None => {
439 // Key deleted on remote side
440 remote_changed.insert(key, None);
441 }
442 _ => {}
443 }
444 }
445
446 // Check for new keys added by local (not in base)
447 for (key, val) in local_obj {
448 if !base_obj.contains_key(key) {
449 local_changed.insert(key, Some(val));
450 }
451 }
452
453 // Check for new keys added by remote (not in base)
454 for (key, val) in remote_obj {
455 if !base_obj.contains_key(key) {
456 remote_changed.insert(key, Some(val));
457 }
458 }
459
460 // Build merged result starting from base
461 let mut result = base_obj.clone();
462
463 // Apply local changes
464 for (key, val) in &local_changed {
465 match val {
466 Some(v) => {
467 result.insert((*key).to_string(), (*v).clone());
468 }
469 None => {
470 result.remove(*key);
471 }
472 }
473 }
474
475 // Apply remote changes (non-overlapping only, or remote won the tie)
476 for (key, val) in &remote_changed {
477 if local_changed.contains_key(key) {
478 // Overlapping: the device-independent winner decides every contested
479 // field the same way. When remote wins, apply its value; when local
480 // wins, local was already applied above.
481 if !local_wins {
482 match val {
483 Some(v) => {
484 result.insert((*key).to_string(), (*v).clone());
485 }
486 None => {
487 result.remove(*key);
488 }
489 }
490 }
491 } else {
492 // Non-overlapping: apply remote
493 match val {
494 Some(v) => {
495 result.insert((*key).to_string(), (*v).clone());
496 }
497 None => {
498 result.remove(*key);
499 }
500 }
501 }
502 }
503
504 Resolution::Merged(serde_json::Value::Object(result))
505 }
506
507 #[cfg(test)]
508 mod tests {
509 use super::*;
510 use crate::types::{ChangeOp, Hlc};
511 use serde_json::json;
512
513 /// Fixed node for locally-minted test entries, distinct from any random
514 /// `other_device`, so HLC tiebreaks are deterministic.
515 fn local_node() -> DeviceId {
516 DeviceId::new(Uuid::from_u128(0x1111_1111_1111_1111_1111_1111_1111_1111))
517 }
518
519 /// A second fixed device node, distinct from [`local_node`], for the
520 /// field-merge tests that need to name the remote side's clock.
521 fn remote_node() -> DeviceId {
522 DeviceId::new(Uuid::from_u128(0x2222_2222_2222_2222_2222_2222_2222_2222))
523 }
524
525 /// Map a wall-clock timestamp onto an HLC at `node`, so the timestamp-ordered
526 /// field-merge tests express the same intent against the HLC-based API. A
527 /// strictly later `ts` yields a strictly greater HLC (higher `wall_ms`); equal
528 /// `ts` on distinct nodes ties on the node, which is exactly the convergent
529 /// behavior the F1 fix guarantees.
530 fn ts_hlc(ts: DateTime<Utc>, node: DeviceId) -> Hlc {
531 Hlc::from_legacy(ts.timestamp_millis(), node)
532 }
533
534 fn make_entry(table: &str, row_id: &str, op: ChangeOp, ts: DateTime<Utc>) -> ChangeEntry {
535 // Derive the HLC wall component from the timestamp so the time-ordered
536 // tests below still express the intended ordering.
537 ChangeEntry {
538 table: table.to_string(),
539 op,
540 row_id: row_id.to_string(),
541 timestamp: ts,
542 hlc: Hlc::from_legacy(ts.timestamp_millis(), local_node()),
543 data: Some(json!({"value": "test"})),
544 extra: serde_json::Map::default(),
545 }
546 }
547
548 fn make_pulled(
549 table: &str,
550 row_id: &str,
551 op: ChangeOp,
552 ts: DateTime<Utc>,
553 device_id: Uuid,
554 seq: i64,
555 ) -> PulledChange {
556 let mut entry = make_entry(table, row_id, op, ts);
557 entry.hlc = Hlc::from_legacy(ts.timestamp_millis(), DeviceId::new(device_id));
558 PulledChange {
559 entry,
560 device_id: DeviceId::new(device_id),
561 seq,
562 }
563 }
564
565 /// Build a pulled change with an explicit HLC, for resolution tests that need
566 /// to control the clock independently of the wall timestamp.
567 fn pulled_with_hlc(row_id: &str, op: ChangeOp, hlc: Hlc, device_id: Uuid) -> PulledChange {
568 let mut p = make_pulled("tasks", row_id, op, Utc::now(), device_id, 1);
569 p.entry.hlc = hlc;
570 p
571 }
572
573 // ── detect_conflicts ──
574
575 #[test]
576 fn no_conflicts_when_different_rows() {
577 let our_device = Uuid::new_v4();
578 let other_device = Uuid::new_v4();
579 let now = Utc::now();
580
581 let remote = vec![make_pulled(
582 "tasks",
583 "r1",
584 ChangeOp::Update,
585 now,
586 other_device,
587 1,
588 )];
589 let local = vec![make_entry("tasks", "r2", ChangeOp::Update, now)];
590
591 let (clean, conflicts) = detect_conflicts(remote, &local, DeviceId::new(our_device));
592 assert_eq!(clean.len(), 1);
593 assert!(conflicts.is_empty());
594 }
595
596 #[test]
597 fn conflict_detected_same_row_different_device() {
598 let our_device = Uuid::new_v4();
599 let other_device = Uuid::new_v4();
600 let now = Utc::now();
601
602 let remote = vec![make_pulled(
603 "tasks",
604 "r1",
605 ChangeOp::Update,
606 now,
607 other_device,
608 1,
609 )];
610 let local = vec![make_entry("tasks", "r1", ChangeOp::Update, now)];
611
612 let (clean, conflicts) = detect_conflicts(remote, &local, DeviceId::new(our_device));
613 assert!(clean.is_empty());
614 assert_eq!(conflicts.len(), 1);
615 assert_eq!(conflicts[0].remote.entry.row_id, "r1");
616 assert_eq!(conflicts[0].local.row_id, "r1");
617 }
618
619 #[test]
620 fn own_echo_without_pending_edit_is_clean() {
621 // An echo of our own device with no contesting local pending edit is
622 // clean (it still passes the HLC gate at apply time).
623 let our_device = Uuid::new_v4();
624 let now = Utc::now();
625
626 let remote = vec![make_pulled(
627 "tasks",
628 "r1",
629 ChangeOp::Update,
630 now,
631 our_device,
632 1,
633 )];
634 let (clean, conflicts) = detect_conflicts(remote, &[], DeviceId::new(our_device));
635 assert_eq!(clean.len(), 1);
636 assert!(conflicts.is_empty());
637 }
638
639 #[test]
640 fn echo_contesting_a_pending_edit_is_resolved_not_trusted() {
641 // Hardening: a pulled change labeled as our own echo that contests an
642 // un-pushed local edit is resolved as a conflict, not waved through as
643 // clean. Trusting the device_id label would let a server relabel a hostile
644 // row as our echo to skip conflict detection entirely.
645 let our_device = Uuid::new_v4();
646 let now = Utc::now();
647
648 let remote = vec![make_pulled(
649 "tasks",
650 "r1",
651 ChangeOp::Update,
652 now,
653 our_device,
654 1,
655 )];
656 let local = vec![make_entry("tasks", "r1", ChangeOp::Update, now)];
657
658 let (clean, conflicts) = detect_conflicts(remote, &local, DeviceId::new(our_device));
659 assert!(clean.is_empty());
660 assert_eq!(
661 conflicts.len(),
662 1,
663 "echo contesting a pending edit is resolved"
664 );
665 }
666
667 #[test]
668 fn clean_changes_gate_drops_stale_keeps_newer() {
669 let our_device = Uuid::new_v4();
670 let other_device = Uuid::new_v4();
671 let now = Utc::now();
672 // A clean remote change for tasks/r1; its HLC wall == now_ms.
673 let remote = vec![make_pulled(
674 "tasks",
675 "r1",
676 ChangeOp::Update,
677 now,
678 other_device,
679 1,
680 )];
681 let (clean, _conflicts) = detect_conflicts(remote, &[], DeviceId::new(our_device));
682 assert_eq!(clean.len(), 1);
683
684 // No committed clock for the row → kept (first time we've seen it).
685 assert_eq!(clean.clone().gated(|_, _| None).len(), 1);
686 // Committed clock older than the remote → kept.
687 assert_eq!(
688 clean
689 .clone()
690 .gated(|_, _| Some(Hlc::zero(DeviceId::new(other_device))))
691 .len(),
692 1
693 );
694 // Committed clock newer than the remote → dropped (would clobber newer local).
695 let newer = Hlc {
696 wall_ms: now.timestamp_millis() + 1,
697 counter: 0,
698 node: DeviceId::new(other_device),
699 };
700 assert!(clean.gated(move |_, _| Some(newer)).is_empty());
701 }
702
703 #[test]
704 fn different_tables_same_row_id_no_conflict() {
705 let our_device = Uuid::new_v4();
706 let other_device = Uuid::new_v4();
707 let now = Utc::now();
708
709 let remote = vec![make_pulled(
710 "tasks",
711 "r1",
712 ChangeOp::Update,
713 now,
714 other_device,
715 1,
716 )];
717 let local = vec![make_entry("events", "r1", ChangeOp::Update, now)];
718
719 let (clean, conflicts) = detect_conflicts(remote, &local, DeviceId::new(our_device));
720 assert_eq!(clean.len(), 1);
721 assert!(conflicts.is_empty());
722 }
723
724 #[test]
725 fn detect_conflicts_correct_split() {
726 let our_device = Uuid::new_v4();
727 let other_device = Uuid::new_v4();
728 let now = Utc::now();
729
730 let remote = vec![
731 make_pulled("tasks", "r1", ChangeOp::Update, now, other_device, 1),
732 make_pulled("tasks", "r2", ChangeOp::Insert, now, other_device, 2),
733 make_pulled("events", "r3", ChangeOp::Delete, now, other_device, 3),
734 ];
735 let local = vec![
736 make_entry("tasks", "r1", ChangeOp::Update, now),
737 // r2 not in local → clean
738 // r3 not in local → clean
739 ];
740
741 let (clean, conflicts) = detect_conflicts(remote, &local, DeviceId::new(our_device));
742 assert_eq!(clean.len(), 2);
743 assert_eq!(conflicts.len(), 1);
744 assert_eq!(conflicts[0].remote.entry.row_id, "r1");
745 }
746
747 #[test]
748 fn empty_remote_produces_no_conflicts() {
749 let our_device = Uuid::new_v4();
750 let now = Utc::now();
751
752 let remote = vec![];
753 let local = vec![make_entry("tasks", "r1", ChangeOp::Update, now)];
754
755 let (clean, conflicts) = detect_conflicts(remote, &local, DeviceId::new(our_device));
756 assert!(clean.is_empty());
757 assert!(conflicts.is_empty());
758 }
759
760 #[test]
761 fn empty_local_produces_no_conflicts() {
762 let our_device = Uuid::new_v4();
763 let other_device = Uuid::new_v4();
764 let now = Utc::now();
765
766 let remote = vec![make_pulled(
767 "tasks",
768 "r1",
769 ChangeOp::Update,
770 now,
771 other_device,
772 1,
773 )];
774 let local: Vec<ChangeEntry> = vec![];
775
776 let (clean, conflicts) = detect_conflicts(remote, &local, DeviceId::new(our_device));
777 assert_eq!(clean.len(), 1);
778 assert!(conflicts.is_empty());
779 }
780
781 // ── resolve_lww (HLC ordering) ──
782
783 #[test]
784 fn lww_picks_newer_timestamp() {
785 let other_device = Uuid::new_v4();
786 let old = Utc::now() - chrono::Duration::seconds(60);
787 let new = Utc::now();
788
789 let local = make_entry("tasks", "r1", ChangeOp::Update, old);
790 let remote = make_pulled("tasks", "r1", ChangeOp::Update, new, other_device, 1);
791
792 assert!(matches!(
793 resolve_lww(&local, &remote),
794 Resolution::KeepRemote
795 ));
796 }
797
798 #[test]
799 fn lww_local_wins_when_newer() {
800 let other_device = Uuid::new_v4();
801 let old = Utc::now() - chrono::Duration::seconds(60);
802 let new = Utc::now();
803
804 let local = make_entry("tasks", "r1", ChangeOp::Update, new);
805 let remote = make_pulled("tasks", "r1", ChangeOp::Update, old, other_device, 1);
806
807 assert!(matches!(
808 resolve_lww(&local, &remote),
809 Resolution::KeepLocal
810 ));
811 }
812
813 #[test]
814 fn lww_counter_breaks_same_wall_tie() {
815 // Same wall_ms, higher counter wins regardless of node.
816 let other = Uuid::new_v4();
817 let mut local = make_entry("tasks", "r1", ChangeOp::Update, Utc::now());
818 local.hlc = Hlc {
819 wall_ms: 1000,
820 counter: 2,
821 node: local_node(),
822 };
823 let remote = pulled_with_hlc(
824 "r1",
825 ChangeOp::Update,
826 Hlc {
827 wall_ms: 1000,
828 counter: 5,
829 node: DeviceId::new(other),
830 },
831 other,
832 );
833 assert!(matches!(
834 resolve_lww(&local, &remote),
835 Resolution::KeepRemote
836 ));
837 }
838
839 #[test]
840 fn lww_node_breaks_exact_tie_convergently() {
841 // Identical (wall, counter): the node decides, and BOTH devices must pick
842 // the same physical change. Use nodes with a known order (a < b).
843 let a = Uuid::from_u128(1);
844 let b = Uuid::from_u128(2);
845 let hlc_a = Hlc {
846 wall_ms: 1000,
847 counter: 0,
848 node: DeviceId::new(a),
849 };
850 let hlc_b = Hlc {
851 wall_ms: 1000,
852 counter: 0,
853 node: DeviceId::new(b),
854 };
855
856 // Device A: local is a's change, remote is b's change.
857 let mut local_a = make_entry("tasks", "r1", ChangeOp::Update, Utc::now());
858 local_a.hlc = hlc_a;
859 let remote_b = pulled_with_hlc("r1", ChangeOp::Update, hlc_b, b);
860 // b > a, so A drops its local and keeps remote (b's change).
861 assert!(matches!(
862 resolve_lww(&local_a, &remote_b),
863 Resolution::KeepRemote
864 ));
865
866 // Device B: local is b's change, remote is a's change.
867 let mut local_b = make_entry("tasks", "r1", ChangeOp::Update, Utc::now());
868 local_b.hlc = hlc_b;
869 let remote_a = pulled_with_hlc("r1", ChangeOp::Update, hlc_a, a);
870 // b > a, so B keeps its local (b's change). Both devices converge on b.
871 assert!(matches!(
872 resolve_lww(&local_b, &remote_a),
873 Resolution::KeepLocal
874 ));
875 }
876
877 #[test]
878 fn lww_identical_hlc_node_collision_converges_on_payload() {
879 // Pathological case: two installs share a node UUID (cloned config), so
880 // two genuinely different edits produce a byte-identical HLC. The payload
881 // tiebreak must still make both devices land on the same value.
882 let shared = Uuid::from_u128(7);
883 let hlc = Hlc {
884 wall_ms: 1000,
885 counter: 3,
886 node: DeviceId::new(shared),
887 };
888 let val_a = json!({"v": "aaa"});
889 let val_b = json!({"v": "bbb"}); // canonically greater than val_a
890
891 // Device A: local = a, remote = b.
892 let mut local_a = make_entry("tasks", "r1", ChangeOp::Update, Utc::now());
893 local_a.hlc = hlc;
894 local_a.data = Some(val_a.clone());
895 let mut remote_b = pulled_with_hlc("r1", ChangeOp::Update, hlc, shared);
896 remote_b.entry.data = Some(val_b.clone());
897 // b's payload sorts higher, so A drops local and keeps remote (b).
898 assert!(matches!(
899 resolve_lww(&local_a, &remote_b),
900 Resolution::KeepRemote
901 ));
902
903 // Device B: local = b, remote = a, must keep local (b). Both converge on b.
904 let mut local_b = make_entry("tasks", "r1", ChangeOp::Update, Utc::now());
905 local_b.hlc = hlc;
906 local_b.data = Some(val_b);
907 let mut remote_a = pulled_with_hlc("r1", ChangeOp::Update, hlc, shared);
908 remote_a.entry.data = Some(val_a);
909 assert!(matches!(
910 resolve_lww(&local_b, &remote_a),
911 Resolution::KeepLocal
912 ));
913 }
914
915 #[test]
916 fn lww_newer_update_beats_older_delete() {
917 // The data-loss fix: a strictly-newer UPDATE must beat an older DELETE.
918 // Under the old "delete always wins" rule this edit was silently lost.
919 let other = Uuid::new_v4();
920 let mut local = make_entry("tasks", "r1", ChangeOp::Update, Utc::now());
921 local.hlc = Hlc {
922 wall_ms: 2000,
923 counter: 0,
924 node: local_node(),
925 };
926 let remote = pulled_with_hlc(
927 "r1",
928 ChangeOp::Delete,
929 Hlc {
930 wall_ms: 1000,
931 counter: 0,
932 node: DeviceId::new(other),
933 },
934 other,
935 );
936 assert!(matches!(
937 resolve_lww(&local, &remote),
938 Resolution::KeepLocal
939 ));
940 }
941
942 #[test]
943 fn lww_newer_delete_beats_older_update() {
944 // Symmetric: a strictly-newer DELETE beats an older UPDATE.
945 let other = Uuid::new_v4();
946 let mut local = make_entry("tasks", "r1", ChangeOp::Update, Utc::now());
947 local.hlc = Hlc {
948 wall_ms: 1000,
949 counter: 0,
950 node: local_node(),
951 };
952 let remote = pulled_with_hlc(
953 "r1",
954 ChangeOp::Delete,
955 Hlc {
956 wall_ms: 2000,
957 counter: 0,
958 node: DeviceId::new(other),
959 },
960 other,
961 );
962 assert!(matches!(
963 resolve_lww(&local, &remote),
964 Resolution::KeepRemote
965 ));
966 }
967
968 // ── resolve_field_merge ──
969
970 #[test]
971 fn field_merge_non_overlapping_changes() {
972 let base = json!({"title": "old", "status": "pending", "priority": 1});
973 let local = json!({"title": "new title", "status": "pending", "priority": 1});
974 let remote = json!({"title": "old", "status": "done", "priority": 1});
975 let now = Utc::now();
976
977 let result = resolve_field_merge(
978 &local,
979 &remote,
980 &base,
981 &ts_hlc(now, local_node()),
982 &ts_hlc(now, remote_node()),
983 );
984 match result {
985 Resolution::Merged(v) => {
986 assert_eq!(v["title"], "new title");
987 assert_eq!(v["status"], "done");
988 assert_eq!(v["priority"], 1);
989 }
990 _ => panic!("Expected Merged"),
991 }
992 }
993
994 #[test]
995 fn field_merge_overlapping_newer_wins() {
996 let base = json!({"title": "old"});
997 let local = json!({"title": "local title"});
998 let remote = json!({"title": "remote title"});
999 let old = Utc::now() - chrono::Duration::seconds(60);
1000 let new = Utc::now();
1001
1002 // Remote is newer → remote wins the overlapping field
1003 let result = resolve_field_merge(
1004 &local,
1005 &remote,
1006 &base,
1007 &ts_hlc(old, local_node()),
1008 &ts_hlc(new, remote_node()),
1009 );
1010 match result {
1011 Resolution::Merged(v) => {
1012 assert_eq!(v["title"], "remote title");
1013 }
1014 _ => panic!("Expected Merged"),
1015 }
1016
1017 // Local is newer → local wins the overlapping field
1018 let result = resolve_field_merge(
1019 &local,
1020 &remote,
1021 &base,
1022 &ts_hlc(new, local_node()),
1023 &ts_hlc(old, remote_node()),
1024 );
1025 match result {
1026 Resolution::Merged(v) => {
1027 assert_eq!(v["title"], "local title");
1028 }
1029 _ => panic!("Expected Merged"),
1030 }
1031 }
1032
1033 #[test]
1034 fn field_merge_key_deleted_on_one_side() {
1035 let base = json!({"title": "old", "notes": "some notes"});
1036 let local = json!({"title": "old"}); // notes deleted
1037 let remote = json!({"title": "old", "notes": "some notes"});
1038 let now = Utc::now();
1039
1040 let result = resolve_field_merge(
1041 &local,
1042 &remote,
1043 &base,
1044 &ts_hlc(now, local_node()),
1045 &ts_hlc(now, remote_node()),
1046 );
1047 match result {
1048 Resolution::Merged(v) => {
1049 assert_eq!(v["title"], "old");
1050 assert!(v.get("notes").is_none(), "notes should be deleted");
1051 }
1052 _ => panic!("Expected Merged"),
1053 }
1054 }
1055
1056 #[test]
1057 fn field_merge_non_object_falls_back() {
1058 let base = json!("string value");
1059 let local = json!("local string");
1060 let remote = json!("remote string");
1061 let now = Utc::now();
1062
1063 assert!(matches!(
1064 resolve_field_merge(
1065 &local,
1066 &remote,
1067 &base,
1068 &ts_hlc(now, local_node()),
1069 &ts_hlc(now, remote_node())
1070 ),
1071 Resolution::KeepRemote
1072 ));
1073 }
1074
1075 #[test]
1076 fn field_merge_empty_base_treats_all_as_changed() {
1077 let base = json!({});
1078 let local = json!({"title": "from local"});
1079 let remote = json!({"status": "from remote"});
1080 let now = Utc::now();
1081
1082 let result = resolve_field_merge(
1083 &local,
1084 &remote,
1085 &base,
1086 &ts_hlc(now, local_node()),
1087 &ts_hlc(now, remote_node()),
1088 );
1089 match result {
1090 Resolution::Merged(v) => {
1091 assert_eq!(v["title"], "from local");
1092 assert_eq!(v["status"], "from remote");
1093 }
1094 _ => panic!("Expected Merged"),
1095 }
1096 }
1097
1098 #[test]
1099 fn field_merge_new_keys_from_both_sides() {
1100 let base = json!({"existing": 1});
1101 let local = json!({"existing": 1, "local_new": "a"});
1102 let remote = json!({"existing": 1, "remote_new": "b"});
1103 let now = Utc::now();
1104
1105 let result = resolve_field_merge(
1106 &local,
1107 &remote,
1108 &base,
1109 &ts_hlc(now, local_node()),
1110 &ts_hlc(now, remote_node()),
1111 );
1112 match result {
1113 Resolution::Merged(v) => {
1114 assert_eq!(v["existing"], 1);
1115 assert_eq!(v["local_new"], "a");
1116 assert_eq!(v["remote_new"], "b");
1117 }
1118 _ => panic!("Expected Merged"),
1119 }
1120 }
1121
1122 // ── PulledChange preserves metadata ──
1123
1124 #[test]
1125 fn pulled_change_preserves_device_id_and_seq() {
1126 let device_id = Uuid::new_v4();
1127 let now = Utc::now();
1128 let pulled = make_pulled("tasks", "r1", ChangeOp::Insert, now, device_id, 42);
1129
1130 assert_eq!(pulled.device_id.as_uuid(), device_id);
1131 assert_eq!(pulled.seq, 42);
1132 assert_eq!(pulled.entry.table, "tasks");
1133 assert_eq!(pulled.entry.row_id, "r1");
1134 }
1135
1136 #[test]
1137 fn pulled_change_clone_works() {
1138 let device_id = Uuid::new_v4();
1139 let now = Utc::now();
1140 let pulled = make_pulled("tasks", "r1", ChangeOp::Insert, now, device_id, 1);
1141 let cloned = pulled.clone();
1142
1143 assert_eq!(cloned.device_id, pulled.device_id);
1144 assert_eq!(cloned.seq, pulled.seq);
1145 assert_eq!(cloned.entry.table, pulled.entry.table);
1146 }
1147
1148 // ── Resolution variants ──
1149
1150 #[test]
1151 fn resolution_debug_format() {
1152 let keep_local = Resolution::KeepLocal;
1153 let keep_remote = Resolution::KeepRemote;
1154 let merged = Resolution::Merged(json!({"a": 1}));
1155 let skip = Resolution::Skip;
1156
1157 assert!(format!("{keep_local:?}").contains("KeepLocal"));
1158 assert!(format!("{keep_remote:?}").contains("KeepRemote"));
1159 assert!(format!("{merged:?}").contains("Merged"));
1160 assert!(format!("{skip:?}").contains("Skip"));
1161 }
1162
1163 // ── ConflictResolver trait ──
1164
1165 #[test]
1166 fn custom_resolver_works() {
1167 struct AlwaysRemote;
1168 impl ConflictResolver for AlwaysRemote {
1169 fn resolve(
1170 &self,
1171 _local: &ChangeEntry,
1172 _remote: &PulledChange,
1173 _base: Option<&serde_json::Value>,
1174 ) -> Resolution {
1175 Resolution::KeepRemote
1176 }
1177 }
1178
1179 let resolver = AlwaysRemote;
1180 let now = Utc::now();
1181 let local = make_entry("tasks", "r1", ChangeOp::Update, now);
1182 let remote = make_pulled("tasks", "r1", ChangeOp::Update, now, Uuid::new_v4(), 1);
1183
1184 assert!(matches!(
1185 resolver.resolve(&local, &remote, None),
1186 Resolution::KeepRemote
1187 ));
1188 }
1189
1190 // ── Fuzz: edge cases and attack vectors ──
1191
1192 // Attack vector 1: duplicate local entries for same (table, row_id).
1193 // HashMap insert means last entry wins. Verify the conflict pair uses
1194 // the LATER local entry, not the earlier one.
1195 #[test]
1196 fn detect_conflicts_duplicate_local_uses_last_entry() {
1197 let our_device = Uuid::new_v4();
1198 let other_device = Uuid::new_v4();
1199 let t1 = Utc::now() - chrono::Duration::seconds(60);
1200 let t2 = Utc::now();
1201
1202 let remote = vec![make_pulled(
1203 "tasks",
1204 "r1",
1205 ChangeOp::Update,
1206 t2,
1207 other_device,
1208 1,
1209 )];
1210 // Two local entries for same (table, row_id): Insert then Update.
1211 // The Update (last) should participate in conflict detection.
1212 let local = vec![
1213 make_entry("tasks", "r1", ChangeOp::Insert, t1),
1214 make_entry("tasks", "r1", ChangeOp::Update, t2),
1215 ];
1216
1217 let (_clean, conflicts) = detect_conflicts(remote, &local, DeviceId::new(our_device));
1218 assert_eq!(conflicts.len(), 1);
1219 // The conflict should use the Update (last entry), not the Insert
1220 assert_eq!(conflicts[0].local.op, ChangeOp::Update);
1221 assert_eq!(conflicts[0].local.timestamp, t2);
1222 }
1223
1224 // DELETE vs DELETE under HLC: the higher clock wins like any other pair.
1225 // (Previously local always won and the timestamp was ignored, now a strictly
1226 // newer remote delete wins.)
1227 #[test]
1228 fn lww_both_delete_newer_wins() {
1229 let other_device = Uuid::new_v4();
1230 let old = Utc::now() - chrono::Duration::seconds(60);
1231 let new = Utc::now();
1232
1233 let local = make_entry("tasks", "r1", ChangeOp::Delete, old);
1234 let remote = make_pulled("tasks", "r1", ChangeOp::Delete, new, other_device, 1);
1235
1236 assert!(matches!(
1237 resolve_lww(&local, &remote),
1238 Resolution::KeepRemote
1239 ));
1240 }
1241
1242 // F1 (regression): overlapping fields whose two sides carry the *same*
1243 // wall-ms must resolve convergently, not "ties go to local". The old rule
1244 // broke ties on local, so device A (local=A) kept A while device B (local=B)
1245 // kept B, permanent silent divergence. Now the exact tie breaks on the full
1246 // HLC (distinct node), so both devices land on the same physical value.
1247 #[test]
1248 fn field_merge_overlapping_equal_wall_converges() {
1249 let base = json!({"title": "base"});
1250 let val_a = json!({"title": "device A version"});
1251 let val_b = json!({"title": "device B version"});
1252 let now = Utc::now();
1253 let hlc_a = ts_hlc(now, local_node()); // node 0x111...
1254 let hlc_b = ts_hlc(now, remote_node()); // node 0x222... (> a), same wall
1255
1256 // Device A resolves (local = A's edit, remote = B's edit); device B
1257 // resolves the mirror image (local = B's edit, remote = A's edit).
1258 let on_a = resolve_field_merge(&val_a, &val_b, &base, &hlc_a, &hlc_b);
1259 let on_b = resolve_field_merge(&val_b, &val_a, &base, &hlc_b, &hlc_a);
1260
1261 let (title_a, title_b) = match (on_a, on_b) {
1262 (Resolution::Merged(a), Resolution::Merged(b)) => {
1263 (a["title"].clone(), b["title"].clone())
1264 }
1265 _ => panic!("Expected Merged on both devices"),
1266 };
1267 assert_eq!(
1268 title_a, title_b,
1269 "both devices must converge on the same value at an equal-wall tie"
1270 );
1271 // Specifically on B's edit, since hlc_b > hlc_a on the node tiebreak.
1272 assert_eq!(title_a, "device B version");
1273 }
1274
1275 // Attack vector 4: HashMap iteration order determinism.
1276 // Non-overlapping keys from both sides should merge deterministically
1277 // regardless of HashMap iteration order.
1278 #[test]
1279 fn field_merge_deterministic_with_many_keys() {
1280 let base = json!({});
1281 let local = json!({"a": 1, "c": 3, "e": 5, "g": 7, "i": 9});
1282 let remote = json!({"b": 2, "d": 4, "f": 6, "h": 8, "j": 10});
1283 let now = Utc::now();
1284
1285 // Run multiple times to catch iteration-order bugs
1286 for _ in 0..10 {
1287 let result = resolve_field_merge(
1288 &local,
1289 &remote,
1290 &base,
1291 &ts_hlc(now, local_node()),
1292 &ts_hlc(now, remote_node()),
1293 );
1294 match result {
1295 Resolution::Merged(v) => {
1296 assert_eq!(v["a"], 1);
1297 assert_eq!(v["b"], 2);
1298 assert_eq!(v["c"], 3);
1299 assert_eq!(v["d"], 4);
1300 assert_eq!(v["e"], 5);
1301 assert_eq!(v["f"], 6);
1302 assert_eq!(v["g"], 7);
1303 assert_eq!(v["h"], 8);
1304 assert_eq!(v["i"], 9);
1305 assert_eq!(v["j"], 10);
1306 }
1307 _ => panic!("Expected Merged"),
1308 }
1309 }
1310 }
1311
1312 // Attack vector 5: null base with both local and remote as objects.
1313 // Falls back to KeepRemote, silently discarding local changes.
1314 #[test]
1315 fn field_merge_null_base_discards_local() {
1316 let base = json!(null);
1317 let local = json!({"title": "important local edit"});
1318 let remote = json!({"status": "remote only"});
1319 let now = Utc::now();
1320
1321 // BUG: Both sides are valid objects but null base causes KeepRemote,
1322 // which silently drops "title": "important local edit".
1323 // A better fallback might be to merge both against an empty base,
1324 // or fall back to LWW.
1325 let result = resolve_field_merge(
1326 &local,
1327 &remote,
1328 &base,
1329 &ts_hlc(now, local_node()),
1330 &ts_hlc(now, remote_node()),
1331 );
1332 assert!(
1333 matches!(result, Resolution::KeepRemote),
1334 "Null base should fall back to KeepRemote (current behavior)"
1335 );
1336 }
1337
1338 // Attack vector 6: empty object vs changed value in field_merge.
1339 // Both `{}` and a new value differ from the base value, so both are
1340 // "changed". This is an overlapping-field conflict.
1341 #[test]
1342 fn field_merge_empty_object_counts_as_change() {
1343 let base = json!({"meta": {"nested": "data"}});
1344 let local = json!({"meta": {}}); // Changed to empty object
1345 let remote = json!({"meta": "flat string"}); // Changed to string
1346 let old = Utc::now() - chrono::Duration::seconds(60);
1347 let new = Utc::now();
1348
1349 // Remote is newer, so remote wins the overlapping field
1350 let result = resolve_field_merge(
1351 &local,
1352 &remote,
1353 &base,
1354 &ts_hlc(old, local_node()),
1355 &ts_hlc(new, remote_node()),
1356 );
1357 match result {
1358 Resolution::Merged(v) => {
1359 assert_eq!(v["meta"], "flat string");
1360 }
1361 _ => panic!("Expected Merged"),
1362 }
1363
1364 // Local is newer, so local wins, meta becomes empty object
1365 let result = resolve_field_merge(
1366 &local,
1367 &remote,
1368 &base,
1369 &ts_hlc(new, local_node()),
1370 &ts_hlc(old, remote_node()),
1371 );
1372 match result {
1373 Resolution::Merged(v) => {
1374 assert_eq!(v["meta"], json!({}));
1375 }
1376 _ => panic!("Expected Merged"),
1377 }
1378 }
1379
1380 // Attack vector 7: multiple remote changes for same (table, row_id).
1381 // Each produces a separate ConflictPair with a CLONE of the same local entry.
1382 #[test]
1383 fn detect_conflicts_multiple_remote_same_row() {
1384 let our_device = Uuid::new_v4();
1385 let other_device = Uuid::new_v4();
1386 let now = Utc::now();
1387
1388 let remote = vec![
1389 make_pulled("tasks", "r1", ChangeOp::Update, now, other_device, 1),
1390 make_pulled("tasks", "r1", ChangeOp::Update, now, other_device, 2),
1391 make_pulled("tasks", "r1", ChangeOp::Delete, now, other_device, 3),
1392 ];
1393 let local = vec![make_entry("tasks", "r1", ChangeOp::Update, now)];
1394
1395 let (_clean, conflicts) = detect_conflicts(remote, &local, DeviceId::new(our_device));
1396 // All 3 remote changes conflict with the same local entry
1397 assert_eq!(conflicts.len(), 3);
1398 // Each gets a clone of the same local entry
1399 assert_eq!(conflicts[0].local.row_id, "r1");
1400 assert_eq!(conflicts[1].local.row_id, "r1");
1401 assert_eq!(conflicts[2].local.row_id, "r1");
1402 // Verify seq ordering is preserved
1403 assert_eq!(conflicts[0].remote.seq, 1);
1404 assert_eq!(conflicts[1].remote.seq, 2);
1405 assert_eq!(conflicts[2].remote.seq, 3);
1406 }
1407
1408 // Attack vector 8: numeric type coercion in serde_json PartialEq.
1409 // JSON `1` (u64) and `1.0` (f64) are different Value variants.
1410 // serde_json::Value PartialEq does NOT treat them as equal.
1411 #[test]
1412 fn field_merge_integer_vs_float_treated_as_different() {
1413 let base = json!({"count": 1}); // serde_json: Number(PosInt(1))
1414 let local = json!({"count": 1.0}); // serde_json: Number(Float(1.0))
1415 let remote = json!({"count": 1}); // unchanged from base
1416 let now = Utc::now();
1417
1418 let result = resolve_field_merge(
1419 &local,
1420 &remote,
1421 &base,
1422 &ts_hlc(now, local_node()),
1423 &ts_hlc(now, remote_node()),
1424 );
1425 match result {
1426 Resolution::Merged(v) => {
1427 // BUG CANDIDATE: `1` != `1.0` in serde_json, so local sees
1428 // "count" as changed (1 -> 1.0) even though semantically
1429 // identical. Remote sees no change. Merge applies local's 1.0.
1430 assert_eq!(
1431 v["count"], 1.0,
1432 "serde_json treats 1 and 1.0 as different values"
1433 );
1434 }
1435 _ => panic!("Expected Merged"),
1436 }
1437 }
1438
1439 // INSERT vs DELETE under HLC: operation is irrelevant, the higher clock wins.
1440 // A strictly-newer remote delete beats an older local insert.
1441 #[test]
1442 fn lww_newer_remote_delete_beats_older_local_insert() {
1443 let other_device = Uuid::new_v4();
1444 let old = Utc::now() - chrono::Duration::seconds(60);
1445 let new = Utc::now();
1446
1447 let local = make_entry("tasks", "r1", ChangeOp::Insert, old);
1448 let remote = make_pulled("tasks", "r1", ChangeOp::Delete, new, other_device, 1);
1449
1450 assert!(matches!(
1451 resolve_lww(&local, &remote),
1452 Resolution::KeepRemote
1453 ));
1454 }
1455
1456 // Reverse: a strictly-newer local delete beats an older remote insert.
1457 #[test]
1458 fn lww_newer_local_delete_beats_older_remote_insert() {
1459 let other_device = Uuid::new_v4();
1460 let old = Utc::now() - chrono::Duration::seconds(60);
1461 let new = Utc::now();
1462
1463 let local = make_entry("tasks", "r1", ChangeOp::Delete, new);
1464 let remote = make_pulled("tasks", "r1", ChangeOp::Insert, old, other_device, 1);
1465
1466 assert!(matches!(
1467 resolve_lww(&local, &remote),
1468 Resolution::KeepLocal
1469 ));
1470 }
1471
1472 // Bonus: field_merge where both sides delete the same key.
1473 // Both detect the key as deleted. Local applies deletion first.
1474 // Remote sees it as overlapping but equal-ts, so local's deletion stands.
1475 #[test]
1476 fn field_merge_both_delete_same_key() {
1477 let base = json!({"title": "old", "notes": "old notes"});
1478 let local = json!({"title": "old"}); // deleted "notes"
1479 let remote = json!({"title": "old"}); // also deleted "notes"
1480 let now = Utc::now();
1481
1482 let result = resolve_field_merge(
1483 &local,
1484 &remote,
1485 &base,
1486 &ts_hlc(now, local_node()),
1487 &ts_hlc(now, remote_node()),
1488 );
1489 match result {
1490 Resolution::Merged(v) => {
1491 assert_eq!(v["title"], "old");
1492 assert!(
1493 v.get("notes").is_none(),
1494 "Both sides deleted notes, should stay deleted"
1495 );
1496 }
1497 _ => panic!("Expected Merged"),
1498 }
1499 }
1500
1501 // field_merge where local deletes a key and remote modifies it. The
1502 // higher-HLC side wins the contested field: a strictly-newer local delete
1503 // beats an older remote modify, so the key stays deleted.
1504 #[test]
1505 fn field_merge_newer_local_delete_beats_remote_modify() {
1506 let base = json!({"title": "old", "notes": "original"});
1507 let local = json!({"title": "old"}); // deleted "notes"
1508 let remote = json!({"title": "old", "notes": "updated notes"});
1509 let old = Utc::now() - chrono::Duration::seconds(60);
1510 let new = Utc::now();
1511
1512 // Local strictly newer → its delete wins the overlapping "notes" field.
1513 let result = resolve_field_merge(
1514 &local,
1515 &remote,
1516 &base,
1517 &ts_hlc(new, local_node()),
1518 &ts_hlc(old, remote_node()),
1519 );
1520 match result {
1521 Resolution::Merged(v) => {
1522 assert!(
1523 v.get("notes").is_none(),
1524 "newer local delete should win over older remote modify"
1525 );
1526 }
1527 _ => panic!("Expected Merged"),
1528 }
1529 }
1530
1531 // Bonus: field_merge where both sides add the SAME new key with
1532 // different values. Both are in local_changed and remote_changed.
1533 #[test]
1534 fn field_merge_both_add_same_new_key_different_values() {
1535 let base = json!({"existing": 1});
1536 let local = json!({"existing": 1, "new_key": "local value"});
1537 let remote = json!({"existing": 1, "new_key": "remote value"});
1538 let old = Utc::now() - chrono::Duration::seconds(60);
1539 let new = Utc::now();
1540
1541 // Remote newer: remote wins the overlapping new key
1542 let result = resolve_field_merge(
1543 &local,
1544 &remote,
1545 &base,
1546 &ts_hlc(old, local_node()),
1547 &ts_hlc(new, remote_node()),
1548 );
1549 match result {
1550 Resolution::Merged(v) => {
1551 assert_eq!(v["new_key"], "remote value");
1552 }
1553 _ => panic!("Expected Merged"),
1554 }
1555
1556 // Local newer: local wins the overlapping new key
1557 let result = resolve_field_merge(
1558 &local,
1559 &remote,
1560 &base,
1561 &ts_hlc(new, local_node()),
1562 &ts_hlc(old, remote_node()),
1563 );
1564 match result {
1565 Resolution::Merged(v) => {
1566 assert_eq!(v["new_key"], "local value");
1567 }
1568 _ => panic!("Expected Merged"),
1569 }
1570
1571 // Equal wall: the winner is decided by the HLC node tiebreak, not "local
1572 // wins", and both devices resolving the mirror image agree on it.
1573 let ha = ts_hlc(new, local_node());
1574 let hb = ts_hlc(new, remote_node());
1575 let on_a = resolve_field_merge(&local, &remote, &base, &ha, &hb);
1576 let on_b = resolve_field_merge(&remote, &local, &base, &hb, &ha);
1577 match (on_a, on_b) {
1578 (Resolution::Merged(a), Resolution::Merged(b)) => {
1579 assert_eq!(
1580 a["new_key"], b["new_key"],
1581 "both devices converge on the same new_key"
1582 );
1583 }
1584 _ => panic!("Expected Merged on both devices"),
1585 }
1586 }
1587
1588 // 50 years in milliseconds, far beyond MAX_HLC_DRIFT_MS.
1589 const FIFTY_YEARS_MS: i64 = 50i64 * 365 * 24 * 3600 * 1000;
1590
1591 #[test]
1592 fn lww_rejects_poisoned_remote_for_honest_local() {
1593 let now = Utc::now();
1594 let now_ms = now.timestamp_millis();
1595 let mut local = make_entry("tasks", "r1", ChangeOp::Update, now);
1596 local.hlc = Hlc {
1597 wall_ms: now_ms,
1598 counter: 0,
1599 node: local_node(),
1600 };
1601 let mut pulled = make_pulled("tasks", "r1", ChangeOp::Update, now, Uuid::new_v4(), 1);
1602 // Unguarded, a wall clock 50 years ahead would win every conflict for decades.
1603 pulled.entry.hlc = Hlc {
1604 wall_ms: now_ms + FIFTY_YEARS_MS,
1605 counter: 0,
1606 node: DeviceId::new(Uuid::new_v4()),
1607 };
1608 assert!(is_clock_poisoned(&pulled.entry.hlc, now));
1609 assert!(matches!(
1610 resolve_lww_at(&local, &pulled, now),
1611 Resolution::KeepLocal
1612 ));
1613 }
1614
1615 #[test]
1616 fn lww_rejects_poisoned_local_for_honest_remote() {
1617 let now = Utc::now();
1618 let now_ms = now.timestamp_millis();
1619 let mut local = make_entry("tasks", "r1", ChangeOp::Update, now);
1620 local.hlc = Hlc {
1621 wall_ms: now_ms + FIFTY_YEARS_MS,
1622 counter: 0,
1623 node: local_node(),
1624 };
1625 let mut pulled = make_pulled("tasks", "r1", ChangeOp::Update, now, Uuid::new_v4(), 1);
1626 pulled.entry.hlc = Hlc {
1627 wall_ms: now_ms,
1628 counter: 0,
1629 node: DeviceId::new(Uuid::new_v4()),
1630 };
1631 assert!(matches!(
1632 resolve_lww_at(&local, &pulled, now),
1633 Resolution::KeepRemote
1634 ));
1635 }
1636
1637 #[test]
1638 fn field_merge_null_base_falls_back_to_lww_not_keep_remote() {
1639 let local = json!({"a": 1});
1640 let remote = json!({"a": 2});
1641 let base = serde_json::Value::Null;
1642 let local_ts = Utc::now();
1643 // Local strictly newer: must be kept, not silently discarded (the old bug).
1644 let remote_older = local_ts - chrono::Duration::seconds(10);
1645 assert!(matches!(
1646 resolve_field_merge(
1647 &local,
1648 &remote,
1649 &base,
1650 &ts_hlc(local_ts, local_node()),
1651 &ts_hlc(remote_older, remote_node())
1652 ),
1653 Resolution::KeepLocal
1654 ));
1655 // Remote newer: remote wins.
1656 let remote_newer = local_ts + chrono::Duration::seconds(10);
1657 assert!(matches!(
1658 resolve_field_merge(
1659 &local,
1660 &remote,
1661 &base,
1662 &ts_hlc(local_ts, local_node()),
1663 &ts_hlc(remote_newer, remote_node())
1664 ),
1665 Resolution::KeepRemote
1666 ));
1667 }
1668
1669 #[test]
1670 fn canonical_payload_sorts_map_keys() {
1671 // Pins the exact-HLC tiebreak's convergence invariant: serde_json must
1672 // serialize object keys in sorted order. Fails loudly if `preserve_order`
1673 // is ever enabled anywhere in the dependency tree.
1674 let bytes = canonical_payload(Some(&json!({"b": 1, "a": 2})));
1675 assert_eq!(bytes, br#"{"a":2,"b":1}"#.to_vec());
1676 }
1677 }
1678