Skip to main content

max / makenotwork

synckit: converge conflict tie-break, type the HLC node (risk Phase 1) Drive the Conflict/Data-Integrity axis to A+ (crate 0.5.0 -> 0.6.0, breaking). - F1: extract resolve_tie(hlc, payload) and route both resolve_lww_at and resolve_field_merge through it; resolve_field_merge now takes &Hlc,&Hlc instead of bare DateTime, so an equal-ts tie breaks device-independently (full HLC then canonical payload) instead of "ties go to local", which made two devices diverge permanently. Rewrite the :954 bug-test as a mirror-image convergence proof. - Hlc.node: Uuid -> DeviceId; drop Default from Hlc so a silent nil-node Hlc::default() no longer compiles; add DeviceId::nil()/hlc_legacy_floor() as the only sanctioned nil-node clock (serde default for ChangeEntry::hlc). - bump_counter clamps at the {i64::MAX,u32::MAX} ceiling instead of wrapping the counter backwards. - CleanChanges::gated_at(now,..) for batch-consistent, testable poisoning. - detect_conflicts resolves an echo that contests a pending local edit rather than trusting the server-asserted device_id label. - ChangeEntry::extra doc: round-trip-preservation only (resolution reads data). - lib.rs: loud consumer caveat that serde_json preserve_order must stay OFF. Gate: 307 lib + 100 integration + 1 doc tests, clippy --all-targets clean. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Author: Max Johnson <me@maxj.phd> · 2026-07-07 18:35 UTC
Commit: d5d4f2d57cd61d0ed342b5c3faf34088f7af7e38
Parent: aabc70a
8 files changed, +319 insertions, -145 deletions
M shared/synckit-client/Cargo.toml +1 -1
@@ -1,6 +1,6 @@
1 1 [package]
2 2 name = "synckit-client"
3 - version = "0.5.0"
3 + version = "0.6.0"
4 4 edition = "2024"
5 5 description = "SyncKit client SDK with end-to-end encryption"
6 6 license-file = "LICENSE"
M shared/synckit-client/src/client/helpers.rs +19 -17
@@ -198,7 +198,7 @@ impl SyncKitClient {
198 198 table: entry.table,
199 199 op: entry.op,
200 200 row_id: entry.row_id,
201 - hlc: Hlc::from_legacy(entry.timestamp.timestamp_millis(), entry.device_id.as_uuid()),
201 + hlc: Hlc::from_legacy(entry.timestamp.timestamp_millis(), entry.device_id),
202 202 timestamp: entry.timestamp,
203 203 data: None,
204 204 extra: Default::default(),
@@ -230,7 +230,7 @@ impl SyncKitClient {
230 230 /// from `node` + `timestamp_ms`.
231 231 fn split_hlc_envelope(
232 232 decrypted: serde_json::Value,
233 - node: uuid::Uuid,
233 + node: crate::ids::DeviceId,
234 234 timestamp_ms: i64,
235 235 ) -> Result<(Hlc, Option<serde_json::Value>)> {
236 236 if let Some(obj) = decrypted.as_object() {
@@ -346,10 +346,10 @@ impl SyncKitClient {
346 346 let (hlc, data) = match entry.data {
347 347 Some(ref value) => {
348 348 let decrypted = crypto::decrypt_json_aad(value, master_key, &ctx)?;
349 - Self::split_hlc_envelope(decrypted, entry.device_id.as_uuid(), entry.timestamp.timestamp_millis())?
349 + Self::split_hlc_envelope(decrypted, entry.device_id, entry.timestamp.timestamp_millis())?
350 350 }
351 351 None => (
352 - Hlc::from_legacy(entry.timestamp.timestamp_millis(), entry.device_id.as_uuid()),
352 + Hlc::from_legacy(entry.timestamp.timestamp_millis(), entry.device_id),
353 353 None,
354 354 ),
355 355 };
@@ -476,6 +476,8 @@ pub(super) fn is_transient(err: &SyncKitError) -> bool {
476 476 #[cfg(test)]
477 477 mod tests {
478 478 use super::*;
479 + use crate::ids::DeviceId;
480 + use uuid::Uuid;
479 481 use base64::Engine;
480 482 use chrono::Utc;
481 483 use std::time::Duration;
@@ -510,7 +512,7 @@ mod tests {
510 512
511 513 #[test]
512 514 fn split_envelope_dispatches_on_explicit_version() {
513 - let node = uuid::Uuid::from_u128(1);
515 + let node = DeviceId::new(Uuid::from_u128(1));
514 516 let hlc = Hlc { wall_ms: 5, counter: 2, node };
515 517
516 518 // v2 envelope: explicit __skver, parsed by version.
@@ -536,7 +538,7 @@ mod tests {
536 538 fn split_envelope_rejects_unknown_version_loudly() {
537 539 // The X2 hazard: a future envelope version must error, not silently
538 540 // fall back to a bare-row read (which would corrupt the clock).
539 - let node = uuid::Uuid::from_u128(1);
541 + let node = DeviceId::new(Uuid::from_u128(1));
540 542 let hlc = Hlc { wall_ms: 5, counter: 0, node };
541 543 let future = serde_json::json!({ "__skver": 3, "__skhlc": hlc, "data": null });
542 544 let err = SyncKitClient::split_hlc_envelope(future, node, 0).unwrap_err();
@@ -557,7 +559,7 @@ mod tests {
557 559 let key = crypto::generate_master_key();
558 560 *client.master_key.write() = Some(crypto::ZeroizeOnDrop(key));
559 561
560 - let device = uuid::Uuid::new_v4();
562 + let device = DeviceId::new(Uuid::new_v4());
561 563 let hlc = Hlc { wall_ms: 12_345, counter: 7, node: device };
562 564 let entry = ChangeEntry {
563 565 table: "tasks".to_string(),
@@ -575,7 +577,7 @@ mod tests {
575 577
576 578 let pull_entry = PullChangeEntry {
577 579 seq: 1,
578 - device_id: crate::ids::DeviceId::new(device),
580 + device_id: device,
579 581 table: wire.table,
580 582 op: wire.op,
581 583 row_id: wire.row_id,
@@ -598,7 +600,7 @@ mod tests {
598 600 op: ChangeOp::Insert,
599 601 row_id: "row-1".to_string(),
600 602 timestamp: Utc::now(),
601 - hlc: Default::default(),
603 + hlc: Hlc::zero(DeviceId::nil()),
602 604 data: Some(serde_json::json!({"title": "test"})),
603 605 extra: Default::default(),
604 606 };
@@ -619,7 +621,7 @@ mod tests {
619 621 op: ChangeOp::Insert,
620 622 row_id: "row-1".to_string(),
621 623 timestamp: Utc::now(),
622 - hlc: Default::default(),
624 + hlc: Hlc::zero(DeviceId::nil()),
623 625 data: Some(original_data.clone()),
624 626 extra: Default::default(),
625 627 };
@@ -648,7 +650,7 @@ mod tests {
648 650 op: ChangeOp::Update,
649 651 row_id: "row-abc".to_string(),
650 652 timestamp: ts,
651 - hlc: Default::default(),
653 + hlc: Hlc::zero(DeviceId::nil()),
652 654 data: Some(original_data.clone()),
653 655 extra: Default::default(),
654 656 };
@@ -893,7 +895,7 @@ mod tests {
893 895 op: ChangeOp::Update,
894 896 row_id: "unique-row-id".to_string(),
895 897 timestamp: ts,
896 - hlc: Default::default(),
898 + hlc: Hlc::zero(DeviceId::nil()),
897 899 data: Some(serde_json::json!({"name": "Alice"})),
898 900 extra: Default::default(),
899 901 };
@@ -919,7 +921,7 @@ mod tests {
919 921 op: ChangeOp::Insert,
920 922 row_id: "r1".to_string(),
921 923 timestamp: Utc::now(),
922 - hlc: Default::default(),
924 + hlc: Hlc::zero(DeviceId::nil()),
923 925 data: Some(serde_json::json!({"title": "Task 1"})),
924 926 extra: Default::default(),
925 927 },
@@ -928,7 +930,7 @@ mod tests {
928 930 op: ChangeOp::Update,
929 931 row_id: "r2".to_string(),
930 932 timestamp: Utc::now(),
931 - hlc: Default::default(),
933 + hlc: Hlc::zero(DeviceId::nil()),
932 934 data: Some(serde_json::json!({"title": "Task 2", "done": true})),
933 935 extra: Default::default(),
934 936 },
@@ -937,7 +939,7 @@ mod tests {
937 939 op: ChangeOp::Delete,
938 940 row_id: "r3".to_string(),
939 941 timestamp: Utc::now(),
940 - hlc: Default::default(),
942 + hlc: Hlc::zero(DeviceId::nil()),
941 943 data: None,
942 944 extra: Default::default(),
943 945 },
@@ -986,7 +988,7 @@ mod tests {
986 988 op: ChangeOp::Insert,
987 989 row_id: "row-1".into(),
988 990 timestamp: Utc::now(),
989 - hlc: Default::default(),
991 + hlc: Hlc::zero(DeviceId::nil()),
990 992 data: Some(serde_json::json!({"name": "\u{30C6}\u{30B9}\u{30C8}"})),
991 993 extra: Default::default(),
992 994 };
@@ -1020,7 +1022,7 @@ mod tests {
1020 1022 op: ChangeOp::Insert,
1021 1023 row_id: "".into(),
1022 1024 timestamp: Utc::now(),
1023 - hlc: Default::default(),
1025 + hlc: Hlc::zero(DeviceId::nil()),
1024 1026 data: Some(serde_json::json!(42)),
1025 1027 extra: Default::default(),
1026 1028 };
M shared/synckit-client/src/client/sync.rs +2 -2
@@ -281,7 +281,7 @@ mod tests {
281 281 op: ChangeOp::Insert,
282 282 row_id: Uuid::new_v4().to_string(),
283 283 timestamp: Utc::now(),
284 - hlc: Default::default(),
284 + hlc: Hlc::zero(DeviceId::nil()),
285 285 data: Some(serde_json::json!({"title": "Test task", "done": false})),
286 286 extra: Default::default(),
287 287 };
@@ -302,7 +302,7 @@ mod tests {
302 302 op: ChangeOp::Delete,
303 303 row_id: "abc-123".to_string(),
304 304 timestamp: Utc::now(),
305 - hlc: Default::default(),
305 + hlc: Hlc::zero(DeviceId::nil()),
306 306 data: None,
307 307 extra: Default::default(),
308 308 };
M shared/synckit-client/src/conflict.rs +198 -86
@@ -11,6 +11,7 @@
11 11 //! - [`resolve_field_merge`]: 3-way JSON object merge using a base version
12 12 //! - [`ConflictResolver`]: trait for custom resolution strategies
13 13
14 + use std::cmp::Ordering;
14 15 use std::collections::HashMap;
15 16
16 17 use chrono::{DateTime, Utc};
@@ -51,8 +52,28 @@ impl CleanChanges {
51 52 /// row, or `None` if it has never been applied locally. A change is dropped
52 53 /// when its HLC is older than or equal to the committed clock; everything
53 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.
54 59 pub fn gated(self, committed_hlc: impl Fn(&str, &str) -> Option<Hlc>) -> Vec<ChangeEntry> {
55 - let now = Utc::now();
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> {
56 77 self.0
57 78 .into_iter()
58 79 .filter(|p| {
@@ -132,8 +153,14 @@ pub trait ConflictResolver: Send + Sync {
132 153 /// Split pulled changes into non-conflicting and conflicting sets.
133 154 ///
134 155 /// A conflict exists when a remote change and a local pending change both
135 - /// modify the same `(table, row_id)` from different devices. Changes from
136 - /// our own device (echoes) are never treated as conflicts.
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`].
137 164 ///
138 165 /// **Precondition:** `local_pending` should contain at most one entry per
139 166 /// `(table, row_id)`. If duplicates exist, only the last one participates
@@ -170,14 +197,21 @@ pub fn detect_conflicts(
170 197 let mut conflicts = Vec::new();
171 198
172 199 for pulled in remote {
173 - // Our own echo — never a conflict
174 - if pulled.device_id == our_device_id {
175 - clean.push(pulled);
176 - continue;
177 - }
178 -
179 200 let key = (pulled.entry.table.as_str(), pulled.entry.row_id.as_str());
180 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 + }
181 215 conflicts.push(ConflictPair {
182 216 remote: pulled,
183 217 local: local_entry.clone(),
@@ -263,16 +297,16 @@ pub fn resolve_lww_at(local: &ChangeEntry, remote: &PulledChange, now: DateTime<
263 297 }
264 298 _ => {}
265 299 }
266 - let resolution = match local.hlc.cmp(&remote.entry.hlc) {
267 - std::cmp::Ordering::Greater => Resolution::KeepLocal,
268 - std::cmp::Ordering::Less => Resolution::KeepRemote,
269 - std::cmp::Ordering::Equal => {
270 - if canonical_payload(&local.data) >= canonical_payload(&remote.entry.data) {
271 - Resolution::KeepLocal
272 - } else {
273 - Resolution::KeepRemote
274 - }
275 - }
300 + let resolution = match resolve_tie(
301 + &local.hlc,
302 + &canonical_payload(&local.data),
303 + &remote.entry.hlc,
304 + &canonical_payload(&remote.entry.data),
305 + ) {
306 + Ordering::Less => Resolution::KeepRemote,
307 + // Greater or an exact tie both keep local: at a true tie the two changes
308 + // are byte-identical, so keeping either side converges.
309 + Ordering::Greater | Ordering::Equal => Resolution::KeepLocal,
276 310 };
277 311 // Every LWW resolution discards one side; log at debug so a lost edit is
278 312 // diagnosable after the fact rather than vanishing silently.
@@ -285,9 +319,26 @@ pub fn resolve_lww_at(local: &ChangeEntry, remote: &PulledChange, now: DateTime<
285 319 resolution
286 320 }
287 321
288 - /// Deterministic byte encoding of a row payload for the exact-HLC tiebreak.
289 - /// `serde_json`'s default `Map` is sorted, so equal values always serialize to
290 - /// equal bytes on every device — the property the tiebreak relies on.
322 + /// The single device-independent tiebreak both resolvers route through.
323 + ///
324 + /// Orders two conflicting changes by full [`Hlc`] first — globally unique via its
325 + /// `node`, so both devices compute the identical result — then breaks an *exact*
326 + /// HLC tie on the canonical payload bytes, a value every device derives the same
327 + /// way. [`Ordering::Greater`] means the `a` side wins.
328 + ///
329 + /// One primitive is the point. The old field-merge rule broke ties on "local
330 + /// wins" (device-relative → device A keeps `a`, device B keeps `b`, permanent
331 + /// silent divergence) and on bare wall-`ms` (a tie window orders of magnitude
332 + /// wider than the exact-HLC tie here). With both [`resolve_lww_at`] and
333 + /// [`resolve_field_merge`] deferring to this, a device-relative winner rule has
334 + /// no code path left.
335 + fn resolve_tie(a_hlc: &Hlc, a_payload: &[u8], b_hlc: &Hlc, b_payload: &[u8]) -> Ordering {
336 + a_hlc.cmp(b_hlc).then_with(|| a_payload.cmp(b_payload))
337 + }
338 +
339 + /// Deterministic byte encoding of a concrete JSON value for the exact-HLC
340 + /// tiebreak. `serde_json`'s default `Map` is sorted, so equal values always
341 + /// serialize to equal bytes on every device — the property the tiebreak relies on.
291 342 ///
292 343 /// INVARIANT: this convergence holds only while `serde_json` keeps insertion
293 344 /// order OFF (i.e. the `preserve_order` feature is NOT enabled anywhere in the
@@ -295,11 +346,14 @@ pub fn resolve_lww_at(local: &ChangeEntry, remote: &PulledChange, now: DateTime<
295 346 /// insertion-dependent and two devices can compute different tiebreak bytes for
296 347 /// equal values — divergence. The `canonical_payload_sorts_map_keys` test pins
297 348 /// this (it fails the moment insertion order leaks in).
349 + fn canonical_value(v: &serde_json::Value) -> Vec<u8> {
350 + serde_json::to_vec(v).unwrap_or_default()
351 + }
352 +
353 + /// [`canonical_value`] for an optional payload; `None` (a delete) canonicalizes
354 + /// to empty bytes.
298 355 fn canonical_payload(data: &Option<serde_json::Value>) -> Vec<u8> {
299 - match data {
300 - Some(v) => serde_json::to_vec(v).unwrap_or_default(),
301 - None => Vec::new(),
302 - }
356 + data.as_ref().map(canonical_value).unwrap_or_default()
303 357 }
304 358
305 359 /// 3-way field-level merge for JSON objects (top-level keys only).
@@ -308,33 +362,48 @@ fn canonical_payload(data: &Option<serde_json::Value>) -> Vec<u8> {
308 362 /// each side changed, then merges non-overlapping changes. For overlapping
309 363 /// fields, the newer timestamp wins.
310 364 ///
365 + /// The overlapping-field winner (and the no-base fallback) is decided by
366 + /// [`resolve_tie`] over the two sides' HLCs, not by a device-relative "ties go to
367 + /// local" rule — so two devices resolving the mirror image of the same conflict
368 + /// converge on the same merged object. Pass each side's [`Hlc`] (`local_hlc` from
369 + /// the local [`ChangeEntry`], `remote_hlc` from the pulled entry).
370 + ///
311 371 /// If any input is not a JSON object (including `Value::Null` for a missing base
312 372 /// snapshot), a field-level merge is impossible, so this falls back to
313 - /// last-writer-wins on the timestamps rather than unconditionally keeping remote
314 - /// — which would silently discard a strictly-newer local edit whenever no base
315 - /// is available (e.g. a first-ever edit).
373 + /// last-writer-wins on the HLCs rather than unconditionally keeping remote —
374 + /// which would silently discard a strictly-newer local edit whenever no base is
375 + /// available (e.g. a first-ever edit).
316 376 pub fn resolve_field_merge(
317 377 local: &serde_json::Value,
318 378 remote: &serde_json::Value,
319 379 base: &serde_json::Value,
320 - local_ts: DateTime<Utc>,
321 - remote_ts: DateTime<Utc>,
380 + local_hlc: &Hlc,
381 + remote_hlc: &Hlc,
322 382 ) -> Resolution {
383 + // Device-independent winner of a contested field (and of the whole entry in
384 + // the no-base fallback): identical on every device, so the merge converges.
385 + // `>= Equal` keeps local; only a strict `Less` hands the field to remote.
386 + let local_wins = resolve_tie(
387 + local_hlc,
388 + &canonical_value(local),
389 + remote_hlc,
390 + &canonical_value(remote),
391 + ) != Ordering::Less;
392 +
323 393 let (Some(local_obj), Some(remote_obj), Some(base_obj)) = (
324 394 local.as_object(),
325 395 remote.as_object(),
326 396 base.as_object(),
327 397 ) else {
328 - // No usable base: last-writer-wins on timestamp, keeping local when it is
329 - // strictly newer instead of dropping it.
330 - let keep_local = local_ts > remote_ts;
398 + // No usable base: last-writer-wins on the HLCs, keeping local when it wins
399 + // the tie instead of dropping it.
331 400 tracing::debug!(
332 - keep_local,
333 - %local_ts,
334 - %remote_ts,
335 - "field-merge base unavailable; fell back to timestamp LWW"
401 + local_wins,
402 + ?local_hlc,
403 + ?remote_hlc,
404 + "field-merge base unavailable; fell back to HLC LWW"
336 405 );
337 - return if keep_local {
406 + return if local_wins {
338 407 Resolution::KeepLocal
339 408 } else {
340 409 Resolution::KeepRemote
@@ -397,17 +466,18 @@ pub fn resolve_field_merge(
397 466 }
398 467 }
399 468
400 - // Apply remote changes (non-overlapping only, or newer timestamp wins)
469 + // Apply remote changes (non-overlapping only, or remote won the tie)
401 470 for (key, val) in &remote_changed {
402 471 if local_changed.contains_key(key) {
403 - // Overlapping: newer timestamp wins, ties go to local
404 - if remote_ts > local_ts {
472 + // Overlapping: the device-independent winner decides every contested
473 + // field the same way. When remote wins, apply its value; when local
474 + // wins, local was already applied above.
475 + if !local_wins {
405 476 match val {
406 477 Some(v) => { result.insert((*key).to_string(), (*v).clone()); }
407 478 None => { result.remove(*key); }
408 479 }
409 480 }
410 - // else: local already applied above
411 481 } else {
412 482 // Non-overlapping: apply remote
413 483 match val {
@@ -428,8 +498,23 @@ mod tests {
428 498
429 499 /// Fixed node for locally-minted test entries, distinct from any random
430 500 /// `other_device`, so HLC tiebreaks are deterministic.
431 - fn local_node() -> Uuid {
432 - Uuid::from_u128(0x1111_1111_1111_1111_1111_1111_1111_1111)
501 + fn local_node() -> DeviceId {
502 + DeviceId::new(Uuid::from_u128(0x1111_1111_1111_1111_1111_1111_1111_1111))
503 + }
504 +
505 + /// A second fixed device node, distinct from [`local_node`], for the
506 + /// field-merge tests that need to name the remote side's clock.
507 + fn remote_node() -> DeviceId {
508 + DeviceId::new(Uuid::from_u128(0x2222_2222_2222_2222_2222_2222_2222_2222))
509 + }
510 +
511 + /// Map a wall-clock timestamp onto an HLC at `node`, so the timestamp-ordered
512 + /// field-merge tests express the same intent against the HLC-based API. A
513 + /// strictly later `ts` yields a strictly greater HLC (higher `wall_ms`); equal
514 + /// `ts` on distinct nodes ties on the node — which is exactly the convergent
515 + /// behavior the F1 fix guarantees.
516 + fn ts_hlc(ts: DateTime<Utc>, node: DeviceId) -> Hlc {
517 + Hlc::from_legacy(ts.timestamp_millis(), node)
433 518 }
434 519
435 520 fn make_entry(table: &str, row_id: &str, op: ChangeOp, ts: DateTime<Utc>) -> ChangeEntry {
@@ -455,7 +540,7 @@ mod tests {
455 540 seq: i64,
456 541 ) -> PulledChange {
457 542 let mut entry = make_entry(table, row_id, op, ts);
458 - entry.hlc = Hlc::from_legacy(ts.timestamp_millis(), device_id);
543 + entry.hlc = Hlc::from_legacy(ts.timestamp_millis(), DeviceId::new(device_id));
459 544 PulledChange { entry, device_id: DeviceId::new(device_id), seq }
460 545 }
461 546
@@ -508,23 +593,36 @@ mod tests {
508 593 }
509 594
510 595 #[test]
511 - fn own_echo_not_treated_as_conflict() {
596 + fn own_echo_without_pending_edit_is_clean() {
597 + // An echo of our own device with no contesting local pending edit is
598 + // clean (it still passes the HLC gate at apply time).
512 599 let our_device = Uuid::new_v4();
513 600 let now = Utc::now();
514 601
515 - let remote = vec![
516 - make_pulled("tasks", "r1", ChangeOp::Update, now, our_device, 1),
517 - ];
518 - let local = vec![
519 - make_entry("tasks", "r1", ChangeOp::Update, now),
520 - ];
521 -
522 - let (clean, conflicts) = detect_conflicts(remote, &local, DeviceId::new(our_device));
602 + let remote = vec![make_pulled("tasks", "r1", ChangeOp::Update, now, our_device, 1)];
603 + let (clean, conflicts) = detect_conflicts(remote, &[], DeviceId::new(our_device));
523 604 assert_eq!(clean.len(), 1);
524 605 assert!(conflicts.is_empty());
525 606 }
526 607
527 608 #[test]
609 + fn echo_contesting_a_pending_edit_is_resolved_not_trusted() {
610 + // Hardening: a pulled change labeled as our own echo that contests an
611 + // un-pushed local edit is resolved as a conflict, not waved through as
612 + // clean. Trusting the device_id label would let a server relabel a hostile
613 + // row as our echo to skip conflict detection entirely.
614 + let our_device = Uuid::new_v4();
615 + let now = Utc::now();
616 +
617 + let remote = vec![make_pulled("tasks", "r1", ChangeOp::Update, now, our_device, 1)];
618 + let local = vec![make_entry("tasks", "r1", ChangeOp::Update, now)];
619 +
620 + let (clean, conflicts) = detect_conflicts(remote, &local, DeviceId::new(our_device));
621 + assert!(clean.is_empty());
622 + assert_eq!(conflicts.len(), 1, "echo contesting a pending edit is resolved");
623 + }
624 +
625 + #[test]
528 626 fn clean_changes_gate_drops_stale_keeps_newer() {
529 627 let our_device = Uuid::new_v4();
530 628 let other_device = Uuid::new_v4();
@@ -537,9 +635,9 @@ mod tests {
537 635 // No committed clock for the row → kept (first time we've seen it).
538 636 assert_eq!(clean.clone().gated(|_, _| None).len(), 1);
539 637 // Committed clock older than the remote → kept.
540 - assert_eq!(clean.clone().gated(|_, _| Some(Hlc::zero(other_device))).len(), 1);
638 + assert_eq!(clean.clone().gated(|_, _| Some(Hlc::zero(DeviceId::new(other_device)))).len(), 1);
541 639 // Committed clock newer than the remote → dropped (would clobber newer local).
542 - let newer = Hlc { wall_ms: now.timestamp_millis() + 1, counter: 0, node: other_device };
640 + let newer = Hlc { wall_ms: now.timestamp_millis() + 1, counter: 0, node: DeviceId::new(other_device) };
543 641 assert!(clean.gated(move |_, _| Some(newer)).is_empty());
544 642 }
545 643
@@ -647,7 +745,7 @@ mod tests {
647 745 let other = Uuid::new_v4();
648 746 let mut local = make_entry("tasks", "r1", ChangeOp::Update, Utc::now());
649 747 local.hlc = Hlc { wall_ms: 1000, counter: 2, node: local_node() };
650 - let remote = pulled_with_hlc("r1", ChangeOp::Update, Hlc { wall_ms: 1000, counter: 5, node: other }, other);
748 + let remote = pulled_with_hlc("r1", ChangeOp::Update, Hlc { wall_ms: 1000, counter: 5, node: DeviceId::new(other) }, other);
651 749 assert!(matches!(resolve_lww(&local, &remote), Resolution::KeepRemote));
652 750 }
653 751
@@ -657,8 +755,8 @@ mod tests {
657 755 // the same physical change. Use nodes with a known order (a < b).
658 756 let a = Uuid::from_u128(1);
659 757 let b = Uuid::from_u128(2);
660 - let hlc_a = Hlc { wall_ms: 1000, counter: 0, node: a };
661 - let hlc_b = Hlc { wall_ms: 1000, counter: 0, node: b };
758 + let hlc_a = Hlc { wall_ms: 1000, counter: 0, node: DeviceId::new(a) };
759 + let hlc_b = Hlc { wall_ms: 1000, counter: 0, node: DeviceId::new(b) };
662 760
663 761 // Device A: local is a's change, remote is b's change.
664 762 let mut local_a = make_entry("tasks", "r1", ChangeOp::Update, Utc::now());
@@ -681,7 +779,7 @@ mod tests {
681 779 // two genuinely different edits produce a byte-identical HLC. The payload
682 780 // tiebreak must still make both devices land on the same value.
683 781 let shared = Uuid::from_u128(7);
684 - let hlc = Hlc { wall_ms: 1000, counter: 3, node: shared };
782 + let hlc = Hlc { wall_ms: 1000, counter: 3, node: DeviceId::new(shared) };
685 783 let val_a = json!({"v": "aaa"});
686 784 let val_b = json!({"v": "bbb"}); // canonically greater than val_a
687 785
@@ -710,7 +808,7 @@ mod tests {
710 808 let other = Uuid::new_v4();
711 809 let mut local = make_entry("tasks", "r1", ChangeOp::Update, Utc::now());
712 810 local.hlc = Hlc { wall_ms: 2000, counter: 0, node: local_node() };
713 - let remote = pulled_with_hlc("r1", ChangeOp::Delete, Hlc { wall_ms: 1000, counter: 0, node: other }, other);
811 + let remote = pulled_with_hlc("r1", ChangeOp::Delete, Hlc { wall_ms: 1000, counter: 0, node: DeviceId::new(other) }, other);
714 812 assert!(matches!(resolve_lww(&local, &remote), Resolution::KeepLocal));
715 813 }
716 814
@@ -720,7 +818,7 @@ mod tests {
720 818 let other = Uuid::new_v4();
721 819 let mut local = make_entry("tasks", "r1", ChangeOp::Update, Utc::now());
722 820 local.hlc = Hlc { wall_ms: 1000, counter: 0, node: local_node() };
723 - let remote = pulled_with_hlc("r1", ChangeOp::Delete, Hlc { wall_ms: 2000, counter: 0, node: other }, other);
821 + let remote = pulled_with_hlc("r1", ChangeOp::Delete, Hlc { wall_ms: 2000, counter: 0, node: DeviceId::new(other) }, other);
724 822 assert!(matches!(resolve_lww(&local, &remote), Resolution::KeepRemote));
725 823 }
726 824
@@ -733,7 +831,7 @@ mod tests {
733 831 let remote = json!({"title": "old", "status": "done", "priority": 1});
734 832 let now = Utc::now();
735 833
736 - let result = resolve_field_merge(&local, &remote, &base, now, now);
834 + let result = resolve_field_merge(&local, &remote, &base, &ts_hlc(now, local_node()), &ts_hlc(now, remote_node()));
737 835 match result {
738 836 Resolution::Merged(v) => {
739 837 assert_eq!(v["title"], "new title");
@@ -753,7 +851,7 @@ mod tests {
753 851 let new = Utc::now();
754 852
755 853 // Remote is newer → remote wins the overlapping field
756 - let result = resolve_field_merge(&local, &remote, &base, old, new);
854 + let result = resolve_field_merge(&local, &remote, &base, &ts_hlc(old, local_node()), &ts_hlc(new, remote_node()));
757 855 match result {
758 856 Resolution::Merged(v) => {
759 857 assert_eq!(v["title"], "remote title");
@@ -762,7 +860,7 @@ mod tests {
762 860 }
763 861
764 862 // Local is newer → local wins the overlapping field
765 - let result = resolve_field_merge(&local, &remote, &base, new, old);
863 + let result = resolve_field_merge(&local, &remote, &base, &ts_hlc(new, local_node()), &ts_hlc(old, remote_node()));
766 864 match result {
767 865 Resolution::Merged(v) => {
768 866 assert_eq!(v["title"], "local title");
@@ -778,7 +876,7 @@ mod tests {
778 876 let remote = json!({"title": "old", "notes": "some notes"});
779 877 let now = Utc::now();
780 878
781 - let result = resolve_field_merge(&local, &remote, &base, now, now);
879 + let result = resolve_field_merge(&local, &remote, &base, &ts_hlc(now, local_node()), &ts_hlc(now, remote_node()));
782 880 match result {
783 881 Resolution::Merged(v) => {
784 882 assert_eq!(v["title"], "old");
@@ -796,7 +894,7 @@ mod tests {
796 894 let now = Utc::now();
797 895
798 896 assert!(matches!(
799 - resolve_field_merge(&local, &remote, &base, now, now),
897 + resolve_field_merge(&local, &remote, &base, &ts_hlc(now, local_node()), &ts_hlc(now, remote_node())),
800 898 Resolution::KeepRemote
801 899 ));
802 900 }
@@ -808,7 +906,7 @@ mod tests {
808 906 let remote = json!({"status": "from remote"});
809 907 let now = Utc::now();
810 908
811 - let result = resolve_field_merge(&local, &remote, &base, now, now);
909 + let result = resolve_field_merge(&local, &remote, &base, &ts_hlc(now, local_node()), &ts_hlc(now, remote_node()));
812 910 match result {
813 911 Resolution::Merged(v) => {
814 912 assert_eq!(v["title"], "from local");
@@ -825,7 +923,7 @@ mod tests {
825 923 let remote = json!({"existing": 1, "remote_new": "b"});
826 924 let now = Utc::now();
827 925
828 - let result = resolve_field_merge(&local, &remote, &base, now, now);
926 + let result = resolve_field_merge(&local, &remote, &base, &ts_hlc(now, local_node()), &ts_hlc(now, remote_node()));
829 927 match result {
830 928 Resolution::Merged(v) => {
831 929 assert_eq!(v["existing"], 1);
@@ -948,23 +1046,37 @@ mod tests {
948 1046 assert!(matches!(resolve_lww(&local, &remote), Resolution::KeepRemote));
949 1047 }
950 1048
951 - // Attack vector 3: field_merge overlapping fields with equal timestamps.
952 - // Ties go to local (remote_ts > local_ts is false). Verify.
1049 + // F1 (regression): overlapping fields whose two sides carry the *same*
1050 + // wall-ms must resolve convergently, not "ties go to local". The old rule
1051 + // broke ties on local, so device A (local=A) kept A while device B (local=B)
1052 + // kept B — permanent silent divergence. Now the exact tie breaks on the full
1053 + // HLC (distinct node), so both devices land on the same physical value.
953 1054 #[test]
954 - fn field_merge_overlapping_equal_timestamps_local_wins() {
1055 + fn field_merge_overlapping_equal_wall_converges() {
955 1056 let base = json!({"title": "base"});
956 - let local = json!({"title": "local version"});
957 - let remote = json!({"title": "remote version"});
1057 + let val_a = json!({"title": "device A version"});
1058 + let val_b = json!({"title": "device B version"});
958 1059 let now = Utc::now();
1060 + let hlc_a = ts_hlc(now, local_node()); // node 0x111…
1061 + let hlc_b = ts_hlc(now, remote_node()); // node 0x222… (> a), same wall
959 1062
960 - let result = resolve_field_merge(&local, &remote, &base, now, now);
961 - match result {
962 - Resolution::Merged(v) => {
963 - assert_eq!(v["title"], "local version",
964 - "Equal timestamps: local should win for overlapping fields");
1063 + // Device A resolves (local = A's edit, remote = B's edit); device B
1064 + // resolves the mirror image (local = B's edit, remote = A's edit).
1065 + let on_a = resolve_field_merge(&val_a, &val_b, &base, &hlc_a, &hlc_b);
1066 + let on_b = resolve_field_merge(&val_b, &val_a, &base, &hlc_b, &hlc_a);
1067 +
1068 + let (title_a, title_b) = match (on_a, on_b) {
1069 + (Resolution::Merged(a), Resolution::Merged(b)) => {
1070 + (a["title"].clone(), b["title"].clone())
965 1071 }
966 - _ => panic!("Expected Merged"),
967 - }
1072 + _ => panic!("Expected Merged on both devices"),
1073 + };
1074 + assert_eq!(
1075 + title_a, title_b,
1076 + "both devices must converge on the same value at an equal-wall tie"
1077 + );
1078 + // Specifically on B's edit, since hlc_b > hlc_a on the node tiebreak.
1079 + assert_eq!(title_a, "device B version");
968 1080 }
969 1081
970 1082 // Attack vector 4: HashMap iteration order determinism.
@@ -979,7 +1091,7 @@ mod tests {
979 1091
980 1092 // Run multiple times to catch iteration-order bugs
981 1093 for _ in 0..10 {
982 - let result = resolve_field_merge(&local, &remote, &base, now, now);
1094 + let result = resolve_field_merge(&local, &remote, &base, &ts_hlc(now, local_node()), &ts_hlc(now, remote_node()));
983 1095 match result {
984 1096 Resolution::Merged(v) => {
985 1097 assert_eq!(v["a"], 1);
@@ -1011,7 +1123,7 @@ mod tests {
1011 1123 // which silently drops "title": "important local edit".
1012 1124 // A better fallback might be to merge both against an empty base,
1013 1125 // or fall back to LWW.
1014 - let result = resolve_field_merge(&local, &remote, &base, now, now);
1126 + let result = resolve_field_merge(&local, &remote, &base, &ts_hlc(now, local_node()), &ts_hlc(now, remote_node()));
1015 1127 assert!(matches!(result, Resolution::KeepRemote),
1016 1128 "Null base should fall back to KeepRemote (current behavior)");
1017 1129 }
@@ -1028,7 +1140,7 @@ mod tests {
1028 1140 let new = Utc::now();
1029 1141
1030 1142 // Remote is newer, so remote wins the overlapping field
1031 - let result = resolve_field_merge(&local, &remote, &base, old, new);
1143 + let result = resolve_field_merge(&local, &remote, &base, &ts_hlc(old, local_node()), &ts_hlc(new, remote_node()));
1032 1144 match result {
1033 1145 Resolution::Merged(v) => {
1034 1146 assert_eq!(v["meta"], "flat string");
@@ -1037,7 +1149,7 @@ mod tests {
1037 1149 }
1038 1150
1039 1151 // Local is newer, so local wins — meta becomes empty object
1040 - let result = resolve_field_merge(&local, &remote, &base, new, old);
1152 + let result = resolve_field_merge(&local, &remote, &base, &ts_hlc(new, local_node()), &ts_hlc(old, remote_node()));
Lines truncated
M shared/synckit-client/src/ids.rs +7
@@ -36,6 +36,13 @@ macro_rules! id_newtype {
36 36 pub const fn as_uuid(self) -> Uuid {
37 37 self.0
38 38 }
39 +
40 + /// The nil identifier (all-zero UUID). For a device this is the
41 + /// legacy-floor node — an HLC on the nil device always loses, so it
42 + /// only ever clocks a pre-HLC entry, never a live device.
43 + pub const fn nil() -> Self {
44 + Self(Uuid::nil())
45 + }
39 46 }
40 47
41 48 impl From<Uuid> for $name {
M shared/synckit-client/src/lib.rs +17 -1
@@ -3,6 +3,20 @@
3 3 //! All row data is encrypted client-side before leaving the device.
4 4 //! The server only ever sees ciphertext.
5 5 //!
6 + //! # Consumer requirement: `serde_json` insertion order must stay OFF
7 + //!
8 + //! Conflict resolution breaks an exact-HLC tie on the canonical bytes of the row
9 + //! payload (see [`conflict::resolve_field_merge`]). Those bytes are identical on
10 + //! every device **only** while `serde_json` serializes map keys in sorted order —
11 + //! i.e. while its `preserve_order` feature is NOT enabled anywhere in your final
12 + //! dependency tree. Cargo features are additive and global, so a *different* crate
13 + //! in your build turning on `serde_json/preserve_order` would silently make two
14 + //! devices compute different tiebreak bytes for the same value and diverge, with
15 + //! no error. There is no compile-time guard the SDK can enforce for you: if you
16 + //! depend on `serde_json` with `preserve_order`, do not use SyncKit conflict
17 + //! resolution. The `canonical_payload_sorts_map_keys` unit test pins this
18 + //! invariant for this crate's own build.
19 + //!
6 20 //! # Quick start
7 21 //!
8 22 //! ```no_run
@@ -32,7 +46,9 @@
32 46 //! op: ChangeOp::Insert,
33 47 //! row_id: uuid::Uuid::new_v4().to_string(),
34 48 //! timestamp: Utc::now(),
35 - //! hlc: Default::default(),
49 + //! // Minted from the app's persistent clock in real code (see `Hlc::tick`);
50 + //! // the nil-node floor stands in here so the example stays self-contained.
51 + //! hlc: synckit_client::Hlc::zero(device.id),
36 52 //! data: Some(serde_json::json!({"title": "Buy milk"})),
37 53 //! extra: Default::default(),
38 54 //! },
M shared/synckit-client/src/types.rs +65 -28
@@ -69,27 +69,35 @@ impl ChangeOp {
69 69 ///
70 70 /// The value travels inside the E2E-encrypted change payload (the SyncKit server
71 71 /// never sees or orders by it), so adding it required no server-side change.
72 - #[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Default, Serialize, Deserialize)]
72 + ///
73 + /// `Hlc` deliberately does **not** implement `Default`. A defaulted HLC would
74 + /// carry the nil [`DeviceId`] — a non-unique, always-losing clock — and minting
75 + /// one by accident (`Hlc::default()`) silently produced a change that loses every
76 + /// conflict. Mint a real clock with [`Hlc::tick`]/[`Hlc::observe`]; the only
77 + /// sanctioned nil-node clock is the explicit legacy floor ([`hlc_legacy_floor`]),
78 + /// used solely for pre-HLC entries that arrive with no embedded clock.
79 + #[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Serialize, Deserialize)]
73 80 pub struct Hlc {
74 81 /// Wall-clock component, milliseconds since the Unix epoch.
75 82 pub wall_ms: i64,
76 83 /// Monotonic tiebreak within a single `wall_ms`, reset to 0 when `wall_ms` advances.
77 84 pub counter: u32,
78 - /// The device that minted this HLC. Final, deterministic tiebreak.
79 - pub node: Uuid,
85 + /// The device that minted this HLC. Final, deterministic tiebreak, and the
86 + /// component that makes the value globally unique across devices.
87 + pub node: DeviceId,
80 88 }
81 89
82 90 impl Hlc {
83 91 /// The zero HLC for `node`. Used as the initial clock state and as the floor
84 92 /// for legacy entries that predate HLC support.
85 - pub fn zero(node: Uuid) -> Self {
93 + pub fn zero(node: DeviceId) -> Self {
86 94 Hlc { wall_ms: 0, counter: 0, node }
87 95 }
88 96
89 97 /// Synthesize an HLC for a legacy change that carried no embedded clock,
90 98 /// deriving the wall component from its `client_timestamp`. This keeps
91 99 /// pre-HLC entries orderable against new ones during the transition.
92 - pub fn from_legacy(timestamp_ms: i64, node: Uuid) -> Self {
100 + pub fn from_legacy(timestamp_ms: i64, node: DeviceId) -> Self {
93 101 Hlc { wall_ms: timestamp_ms, counter: 0, node }
94 102 }
95 103
@@ -98,7 +106,7 @@ impl Hlc {
98 106 /// Standard HLC send rule: if physical time has moved past the last reading,
99 107 /// adopt it and reset the counter; otherwise hold the wall component and bump
100 108 /// the counter so the new event still sorts strictly after the previous one.
101 - pub fn tick(prev: Hlc, now_ms: i64, node: Uuid) -> Hlc {
109 + pub fn tick(prev: Hlc, now_ms: i64, node: DeviceId) -> Hlc {
102 110 if now_ms > prev.wall_ms {
103 111 Hlc { wall_ms: now_ms, counter: 0, node }
104 112 } else {
@@ -114,7 +122,7 @@ impl Hlc {
114 122 /// strictly greater than any counter seen at that wall component. This keeps
115 123 /// the local clock ahead of everything observed, so subsequent local writes
116 124 /// causally follow the remote change.
117 - pub fn observe(prev: Hlc, remote: Hlc, now_ms: i64, node: Uuid) -> Hlc {
125 + pub fn observe(prev: Hlc, remote: Hlc, now_ms: i64, node: DeviceId) -> Hlc {
118 126 let wall = prev.wall_ms.max(remote.wall_ms).max(now_ms);
119 127 let base = if wall == prev.wall_ms && wall == remote.wall_ms {
120 128 prev.counter.max(remote.counter)
@@ -130,14 +138,32 @@ impl Hlc {
130 138 }
131 139 }
132 140
141 + /// The legacy HLC floor: a zero clock on the nil [`DeviceId`]. This is the only
142 + /// sanctioned way to obtain a nil-node clock — used for pre-HLC entries that
143 + /// deserialize without an embedded clock (they must lose to any real HLC). It is
144 + /// wired as the `serde` default for [`ChangeEntry::hlc`]; everywhere else a clock
145 + /// is minted via [`Hlc::tick`]/[`Hlc::observe`], so the old silent `Hlc::default()`
146 + /// nil-node foot-gun no longer compiles.
147 + pub(crate) fn hlc_legacy_floor() -> Hlc {
148 + Hlc::zero(DeviceId::nil())
149 + }
150 +
133 151 /// Increment a counter at `wall_ms`, rolling into the wall component on the
134 152 /// (astronomically unlikely, but attacker-reachable via a hostile peer's
135 153 /// `counter: u32::MAX`) overflow instead of wrapping to 0 — which would send the
136 154 /// clock *backwards* and break the monotonicity HLC exists to guarantee.
137 - fn bump_counter(wall_ms: i64, counter: u32, node: Uuid) -> Hlc {
155 + fn bump_counter(wall_ms: i64, counter: u32, node: DeviceId) -> Hlc {
138 156 match counter.checked_add(1) {
139 157 Some(next) => Hlc { wall_ms, counter: next, node },
140 - None => Hlc { wall_ms: wall_ms.saturating_add(1), counter: 0, node },
158 + None => match wall_ms.checked_add(1) {
159 + Some(wall_ms) => Hlc { wall_ms, counter: 0, node },
160 + // Absolute ceiling {i64::MAX, u32::MAX}: no representable greater HLC
161 + // exists. Clamp (stay put) rather than reset the counter to 0, which
162 + // would move the clock backwards and break monotonicity. Unreachable
163 + // in practice (year ~292 million); a hostile maximal HLC is poisoned
164 + // and loses in LWW regardless.
165 + None => Hlc { wall_ms, counter: u32::MAX, node },
166 + },
141 167 }
142 168 }
143 169
@@ -207,8 +233,11 @@ pub struct ChangeEntry {
207 233 /// Hybrid logical clock for this change — the value conflict resolution
208 234 /// orders by. The app mints it from its persistent clock at the time the
209 235 /// local change is recorded (see [`Hlc::tick`]). It travels inside the
210 - /// E2E-encrypted payload, so the server never sees it.
211 - #[serde(default)]
236 + /// E2E-encrypted payload, so the server never sees it. A legacy entry that
237 + /// arrives without an embedded clock defaults to the explicit
238 + /// [`hlc_legacy_floor`] (nil node, always loses) rather than a silent
239 + /// `Hlc::default()`.
240 + #[serde(default = "hlc_legacy_floor")]
212 241 pub hlc: Hlc,
213 242 /// The row payload as JSON. `None` for Delete operations.
214 243 /// Serialized as absent (not null) when `None`.
@@ -217,9 +246,17 @@ pub struct ChangeEntry {
217 246 /// Forward-compat capture of any fields a newer client added to this
218 247 /// (decrypted, E2E) payload that this build does not know about. Preserved
219 248 /// verbatim through deserialize -> re-serialize so an older client cannot
220 - /// silently drop a resolution-relevant field (e.g. a future tombstone marker)
221 - /// and thereby change the conflict winner. Empty on current-format entries,
222 - /// so it serializes to nothing and the wire format is unchanged.
249 + /// silently *drop* a field a newer client relies on (e.g. a future tombstone
250 + /// marker). Empty on current-format entries, so it serializes to nothing and
251 + /// the wire format is unchanged.
252 + ///
253 + /// The guarantee is **round-trip preservation only**. Conflict resolution
254 + /// ([`resolve_lww`](crate::conflict::resolve_lww) /
255 + /// [`resolve_field_merge`](crate::conflict::resolve_field_merge)) orders and
256 + /// merges on `data` alone — it does not read `extra`. A newer client that
257 + /// wants a field in `extra` to influence the winner must also promote it into
258 + /// `data`; parking it only in `extra` preserves it across the merge but does
259 + /// not change which side wins.
223 260 #[serde(flatten)]
224 261 pub extra: serde_json::Map<String, serde_json::Value>,
225 262 }
@@ -561,7 +598,7 @@ mod tests {
561 598 op: ChangeOp::Delete,
562 599 row_id: "r".into(),
563 600 timestamp: chrono::Utc::now(),
564 - hlc: Default::default(),
601 + hlc: Hlc::zero(DeviceId::nil()),
565 602 data: None,
566 603 extra: Default::default(),
567 604 };
@@ -576,7 +613,7 @@ mod tests {
576 613 op: ChangeOp::Insert,
577 614 row_id: "r".into(),
578 615 timestamp: chrono::Utc::now(),
579 - hlc: Default::default(),
616 + hlc: Hlc::zero(DeviceId::nil()),
580 617 data: Some(json!({"k": "v"})),
581 618 extra: Default::default(),
582 619 };
@@ -706,8 +743,8 @@ mod tests {
706 743
707 744 #[test]
708 745 fn hlc_orders_by_wall_then_counter_then_node() {
709 - let a = Uuid::from_u128(1);
710 - let b = Uuid::from_u128(2);
746 + let a = DeviceId::new(Uuid::from_u128(1));
747 + let b = DeviceId::new(Uuid::from_u128(2));
711 748 // wall dominates
712 749 assert!(Hlc { wall_ms: 1, counter: 9, node: b } < Hlc { wall_ms: 2, counter: 0, node: a });
713 750 // then counter
@@ -718,7 +755,7 @@ mod tests {
718 755
719 756 #[test]
720 757 fn hlc_tick_adopts_physical_time_and_resets_counter() {
721 - let node = Uuid::from_u128(1);
758 + let node = DeviceId::new(Uuid::from_u128(1));
722 759 let prev = Hlc { wall_ms: 1000, counter: 4, node };
723 760 let next = Hlc::tick(prev, 2000, node);
724 761 assert_eq!(next, Hlc { wall_ms: 2000, counter: 0, node });
@@ -726,7 +763,7 @@ mod tests {
726 763
727 764 #[test]
728 765 fn hlc_tick_bumps_counter_when_clock_has_not_advanced() {
729 - let node = Uuid::from_u128(1);
766 + let node = DeviceId::new(Uuid::from_u128(1));
730 767 // Physical time equal to or behind the last reading: hold wall, bump counter,
731 768 // so the new local event still sorts strictly after the previous one.
732 769 let prev = Hlc { wall_ms: 1000, counter: 4, node };
@@ -744,8 +781,8 @@ mod tests {
744 781
745 782 #[test]
746 783 fn hlc_observe_stays_ahead_of_a_skewed_fast_remote() {
747 - let me = Uuid::from_u128(1);
748 - let them = Uuid::from_u128(2);
784 + let me = DeviceId::new(Uuid::from_u128(1));
785 + let them = DeviceId::new(Uuid::from_u128(2));
749 786 // Our physical clock is at 1000; a remote with a fast clock sends 9000.
750 787 let local = Hlc { wall_ms: 1000, counter: 0, node: me };
751 788 let remote = Hlc { wall_ms: 9000, counter: 3, node: them };
@@ -761,8 +798,8 @@ mod tests {
761 798
762 799 #[test]
763 800 fn hlc_observe_advances_to_physical_time_when_it_leads() {
764 - let me = Uuid::from_u128(1);
765 - let them = Uuid::from_u128(2);
801 + let me = DeviceId::new(Uuid::from_u128(1));
802 + let them = DeviceId::new(Uuid::from_u128(2));
766 803 let local = Hlc { wall_ms: 1000, counter: 2, node: me };
767 804 let remote = Hlc { wall_ms: 1500, counter: 0, node: them };
768 805 // Physical time (3000) leads both: adopt it, counter resets.
@@ -772,7 +809,7 @@ mod tests {
772 809
773 810 #[test]
774 811 fn hlc_counter_overflow_rolls_into_wall_not_backwards() {
775 - let me = Uuid::from_u128(1);
812 + let me = DeviceId::new(Uuid::from_u128(1));
776 813 // A local tick at a saturated counter must not wrap to 0 (which would
777 814 // move the clock backwards); it rolls the wall component forward instead.
778 815 let prev = Hlc { wall_ms: 1000, counter: u32::MAX, node: me };
@@ -783,8 +820,8 @@ mod tests {
783 820
784 821 #[test]
785 822 fn hlc_observe_hostile_max_counter_stays_monotonic() {
786 - let me = Uuid::from_u128(1);
787 - let them = Uuid::from_u128(2);
823 + let me = DeviceId::new(Uuid::from_u128(1));
824 + let them = DeviceId::new(Uuid::from_u128(2));
788 825 // A hostile/buggy peer sends counter: u32::MAX at our exact wall_ms.
789 826 let local = Hlc { wall_ms: 5000, counter: 3, node: me };
790 827 let remote = Hlc { wall_ms: 5000, counter: u32::MAX, node: them };
@@ -796,7 +833,7 @@ mod tests {
796 833
797 834 #[test]
798 835 fn hlc_serde_roundtrip() {
799 - let h = Hlc { wall_ms: 123, counter: 4, node: Uuid::from_u128(7) };
836 + let h = Hlc { wall_ms: 123, counter: 4, node: DeviceId::new(Uuid::from_u128(7)) };
800 837 let j = serde_json::to_value(h).unwrap();
801 838 assert_eq!(serde_json::from_value::<Hlc>(j).unwrap(), h);
802 839 }
M shared/synckit-client/tests/integration.rs +10 -10
@@ -15,7 +15,7 @@ use wiremock::{Mock, MockServer, ResponseTemplate};
15 15
16 16 use std::time::Duration;
17 17 use synckit_client::{
18 - AppId, ChangeEntry, ChangeOp, DeviceId, SyncKitClient, SyncKitConfig, SyncKitError, UserId,
18 + AppId, ChangeEntry, ChangeOp, DeviceId, Hlc, SyncKitClient, SyncKitConfig, SyncKitError, UserId,
19 19 };
20 20
21 21 // ── Helpers ──
@@ -260,7 +260,7 @@ async fn push_encrypts_data() {
260 260 op: ChangeOp::Insert,
261 261 row_id: "row-1".into(),
262 262 timestamp: Utc::now(),
263 - hlc: Default::default(),
263 + hlc: Hlc::zero(DeviceId::nil()),
264 264 data: Some(json!({"title": "Secret task"})),
265 265 extra: Default::default(),
266 266 }],
@@ -1129,7 +1129,7 @@ async fn concurrent_push_operations_no_data_corruption() {
1129 1129 op: ChangeOp::Insert,
1130 1130 row_id: format!("row_{i}"),
1131 1131 timestamp: Utc::now(),
1132 - hlc: Default::default(),
1132 + hlc: Hlc::zero(DeviceId::nil()),
1133 1133 data: Some(json!({"index": i})),
1134 1134 extra: Default::default(),
1135 1135 };
@@ -1214,7 +1214,7 @@ async fn push_many_changes_succeeds() {
1214 1214 op: ChangeOp::Insert,
1215 1215 row_id: format!("row-{i}"),
1216 1216 timestamp: Utc::now(),
1217 - hlc: Default::default(),
1217 + hlc: Hlc::zero(DeviceId::nil()),
1218 1218 data: Some(json!({"index": i, "value": format!("data-{i}")})),
1219 1219 extra: Default::default(),
1220 1220 })
@@ -1391,7 +1391,7 @@ async fn push_with_data_fails_without_master_key() {
1391 1391 op: ChangeOp::Insert,
1392 1392 row_id: "r1".into(),
1393 1393 timestamp: Utc::now(),
1394 - hlc: Default::default(),
1394 + hlc: Hlc::zero(DeviceId::nil()),
1395 1395 data: Some(json!({"title": "test"})),
1396 1396 extra: Default::default(),
1397 1397 }];
@@ -1427,7 +1427,7 @@ async fn push_delete_requires_master_key() {
1427 1427 op: ChangeOp::Delete,
1428 1428 row_id: "r1".into(),
1429 1429 timestamp: Utc::now(),
1430 - hlc: Default::default(),
1430 + hlc: Hlc::zero(DeviceId::nil()),
1431 1431 data: None,
1432 1432 extra: Default::default(),
1433 1433 }];
@@ -1504,7 +1504,7 @@ async fn double_push_same_data_both_succeed() {
1504 1504 op: ChangeOp::Insert,
1505 1505 row_id: "same-row".into(),
1506 1506 timestamp: Utc::now(),
1507 - hlc: Default::default(),
1507 + hlc: Hlc::zero(DeviceId::nil()),
1508 1508 data: Some(json!({"title": "duplicate push test"})),
1509 1509 extra: Default::default(),
1510 1510 };
@@ -1734,7 +1734,7 @@ async fn encryption_setup_cross_device_roundtrip() {
1734 1734 op: ChangeOp::Insert,
1735 1735 row_id: "cross-r1".into(),
1736 1736 timestamp: Utc::now(),
1737 - hlc: Default::default(),
1737 + hlc: Hlc::zero(DeviceId::nil()),
1738 1738 data: Some(original_data.clone()),
1739 1739 extra: Default::default(),
1740 1740 }],
@@ -1992,7 +1992,7 @@ async fn end_to_end_push_pull_encryption_roundtrip() {
1992 1992 op: ChangeOp::Insert,
1993 1993 row_id: "e2e-row".into(),
1994 1994 timestamp: Utc::now(),
1995 - hlc: Default::default(),
1995 + hlc: Hlc::zero(DeviceId::nil()),
1996 1996 data: Some(original_data.clone()),
1997 1997 extra: Default::default(),
1998 1998 }],
@@ -2738,7 +2738,7 @@ async fn concurrent_push_100_entries_each() {
2738 2738 op: ChangeOp::Insert,
2739 2739 row_id: format!("row_{i}"),
2740 2740 timestamp: Utc::now(),
2741 - hlc: Default::default(),
2741 + hlc: Hlc::zero(DeviceId::nil()),
2742 2742 data: Some(json!({"index": i})),
2743 2743 extra: Default::default(),
2744 2744 })