Skip to main content

max / makenotwork

synckit-client: order sync conflicts by hybrid logical clock Replace wall-clock LWW + "delete always wins" with an HLC carried inside the E2E-encrypted change payload (server untouched). Fixes two conflict-resolution defects (GoingsOn ultra-fuzz Run #28): a skewed-fast device no longer wins every conflict, and a strictly-newer edit beats an older delete instead of being lost. - Hlc {wall_ms, counter, node} with tick/observe rules; node makes resolution globally convergent without a side channel. - resolve_lww now orders by HLC, op-agnostic; delete-priority removed. - HLC rides in the encrypted envelope (Deletes too), with a legacy fallback that synthesizes an HLC from client_timestamp for pre-HLC entries. - ChangeEntry gains an `hlc` field. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
Author: Max Johnson <me@maxj.phd> · 2026-06-19 19:28 UTC
Commit: 27defcd84369135deedf35b915714104be82f98e
Parent: 2696457
6 files changed, +392 insertions, -127 deletions
M shared/synckit-client/src/client/helpers.rs +103 -40
@@ -108,31 +108,17 @@ impl SyncKitClient {
108 108 Err(last_err.expect("loop ran at least once"))
109 109 }
110 110
111 - /// Encrypt the data field of a change entry for the wire.
111 + /// Encrypt a change entry for the wire. Every change now seals an HLC
112 + /// envelope (Deletes included), so the master key is always required.
112 113 #[cfg(test)]
113 114 pub(super) fn encrypt_change(&self, entry: ChangeEntry) -> Result<WireChangeEntry> {
114 - if entry.data.is_some() {
115 - let master_key = self.require_master_key()?;
116 - Self::encrypt_change_with_key(entry, &master_key)
117 - } else {
118 - Self::encrypt_change_no_data(entry)
119 - }
115 + let master_key = self.require_master_key()?;
116 + Self::encrypt_change_with_key(entry, &master_key)
120 117 }
121 118
122 - /// Build a wire entry for a change with no data (e.g. Delete). No key needed.
123 - #[cfg(test)]
124 - pub(super) fn encrypt_change_no_data(entry: ChangeEntry) -> Result<WireChangeEntry> {
125 - debug_assert!(entry.data.is_none());
126 - Ok(WireChangeEntry {
127 - table: entry.table,
128 - op: entry.op,
129 - row_id: entry.row_id,
130 - timestamp: entry.timestamp,
131 - data: None,
132 - })
133 - }
134 -
135 - /// Decrypt a pulled entry that has no data. No key needed.
119 + /// Decrypt a pulled legacy entry that has no encrypted payload (a pre-HLC
120 + /// Delete). The HLC is synthesized from the entry's `client_timestamp`. No key
121 + /// needed.
136 122 #[cfg(test)]
137 123 pub(super) fn decrypt_change_no_data(entry: PullChangeEntry) -> Result<ChangeEntry> {
138 124 debug_assert!(entry.data.is_none());
@@ -140,17 +126,47 @@ impl SyncKitClient {
140 126 table: entry.table,
141 127 op: entry.op,
142 128 row_id: entry.row_id,
129 + hlc: Hlc::from_legacy(entry.timestamp.timestamp_millis(), entry.device_id),
143 130 timestamp: entry.timestamp,
144 131 data: None,
145 132 })
146 133 }
147 134
148 - /// Encrypt with a pre-loaded key. Used by `push()` to avoid per-entry lock acquisition.
135 + /// Wrap a change's HLC and payload into one JSON envelope. Encrypting the
136 + /// envelope (rather than the bare row payload) is what carries the HLC inside
137 + /// the E2E ciphertext — including for Deletes, which have no row payload. The
138 + /// server stores the ciphertext opaquely and never sees the clock.
139 + fn hlc_envelope(hlc: &Hlc, data: &Option<serde_json::Value>) -> serde_json::Value {
140 + serde_json::json!({ "__skhlc": hlc, "data": data })
141 + }
142 +
143 + /// Split a decrypted payload back into `(hlc, data)`. Handles both the HLC
144 + /// envelope and legacy bare-row payloads (entries pushed before HLC support),
145 + /// for which the HLC is synthesized from `node` + `timestamp_ms`.
146 + fn split_hlc_envelope(
147 + decrypted: serde_json::Value,
148 + node: uuid::Uuid,
149 + timestamp_ms: i64,
150 + ) -> (Hlc, Option<serde_json::Value>) {
151 + if let Some(obj) = decrypted.as_object()
152 + && obj.contains_key("data")
153 + && let Some(hlc) = obj
154 + .get("__skhlc")
155 + .and_then(|v| serde_json::from_value::<Hlc>(v.clone()).ok())
156 + {
157 + let data = obj.get("data").cloned().filter(|v| !v.is_null());
158 + return (hlc, data);
159 + }
160 + // Legacy bare-row payload: the decrypted value is the row data itself.
161 + (Hlc::from_legacy(timestamp_ms, node), Some(decrypted))
162 + }
163 +
164 + /// Encrypt with a pre-loaded key. Used by `push()` to avoid per-entry lock
165 + /// acquisition. Every change (including Deletes) is encrypted, because the HLC
166 + /// envelope always needs sealing.
149 167 pub(super) fn encrypt_change_with_key(entry: ChangeEntry, master_key: &[u8; 32]) -> Result<WireChangeEntry> {
150 - let encrypted_data = match entry.data {
151 - Some(ref value) => Some(crypto::encrypt_json(value, master_key)?),
152 - None => None,
153 - };
168 + let envelope = Self::hlc_envelope(&entry.hlc, &entry.data);
169 + let encrypted_data = Some(crypto::encrypt_json(&envelope, master_key)?);
154 170
155 171 Ok(WireChangeEntry {
156 172 table: entry.table,
@@ -214,33 +230,47 @@ impl SyncKitClient {
214 230 /// Inner decryption that borrows the entry (for fallback logic).
215 231 #[allow(dead_code)]
216 232 fn decrypt_change_with_key_inner(entry: &PullChangeEntry, master_key: &[u8; 32]) -> Result<ChangeEntry> {
217 - let decrypted_data = match entry.data {
218 - Some(ref value) => Some(crypto::decrypt_json(value, master_key)?),
219 - None => None,
233 + let (hlc, data) = match entry.data {
234 + Some(ref value) => {
235 + let decrypted = crypto::decrypt_json(value, master_key)?;
236 + Self::split_hlc_envelope(decrypted, entry.device_id, entry.timestamp.timestamp_millis())
237 + }
238 + None => (
239 + Hlc::from_legacy(entry.timestamp.timestamp_millis(), entry.device_id),
240 + None,
241 + ),
220 242 };
221 243
222 244 Ok(ChangeEntry {
223 245 table: entry.table.clone(),
224 246 op: entry.op,
225 247 row_id: entry.row_id.clone(),
248 + hlc,
226 249 timestamp: entry.timestamp,
227 - data: decrypted_data,
250 + data,
228 251 })
229 252 }
230 253
231 254 /// Decrypt with a pre-loaded key. Used by `pull()` to avoid per-entry lock acquisition.
232 255 pub(super) fn decrypt_change_with_key(entry: PullChangeEntry, master_key: &[u8; 32]) -> Result<ChangeEntry> {
233 - let decrypted_data = match entry.data {
234 - Some(ref value) => Some(crypto::decrypt_json(value, master_key)?),
235 - None => None,
256 + let (hlc, data) = match entry.data {
257 + Some(ref value) => {
258 + let decrypted = crypto::decrypt_json(value, master_key)?;
259 + Self::split_hlc_envelope(decrypted, entry.device_id, entry.timestamp.timestamp_millis())
260 + }
261 + None => (
262 + Hlc::from_legacy(entry.timestamp.timestamp_millis(), entry.device_id),
263 + None,
264 + ),
236 265 };
237 266
238 267 Ok(ChangeEntry {
239 268 table: entry.table,
240 269 op: entry.op,
241 270 row_id: entry.row_id,
271 + hlc,
242 272 timestamp: entry.timestamp,
243 - data: decrypted_data,
273 + data,
244 274 })
245 275 }
246 276 }
@@ -373,21 +403,44 @@ mod tests {
373 403 // ── encrypt_change / decrypt_change ──
374 404
375 405 #[test]
376 - fn encrypt_change_with_no_data() {
406 + fn delete_seals_hlc_envelope_and_roundtrips() {
407 + // A Delete carries no row payload, but its HLC must still travel — so it is
408 + // now encrypted into an envelope (data = Some), and decrypt restores the
409 + // op, a None payload, and the exact HLC.
377 410 let client = SyncKitClient::new(test_config());
411 + let key = crypto::generate_master_key();
412 + *client.master_key.write() = Some(crypto::ZeroizeOnDrop(key));
413 +
414 + let device = uuid::Uuid::new_v4();
415 + let hlc = Hlc { wall_ms: 12_345, counter: 7, node: device };
378 416 let entry = ChangeEntry {
379 417 table: "tasks".to_string(),
380 418 op: ChangeOp::Delete,
381 419 row_id: "row-1".to_string(),
382 420 timestamp: Utc::now(),
421 + hlc,
383 422 data: None,
384 423 };
385 424
386 - let wire = client.encrypt_change(entry.clone()).unwrap();
387 - assert_eq!(wire.table, "tasks");
425 + let wire = client.encrypt_change(entry).unwrap();
388 426 assert_eq!(wire.op, ChangeOp::Delete);
389 - assert_eq!(wire.row_id, "row-1");
390 - assert!(wire.data.is_none());
427 + assert!(wire.data.is_some(), "delete now seals an encrypted HLC envelope");
428 +
429 + let pull_entry = PullChangeEntry {
430 + seq: 1,
431 + device_id: device,
432 + table: wire.table,
433 + op: wire.op,
434 + row_id: wire.row_id,
435 + timestamp: wire.timestamp,
436 + data: wire.data,
437 + key_id: None,
438 + };
439 + let decrypted = client.decrypt_change(pull_entry).unwrap();
440 + assert_eq!(decrypted.op, ChangeOp::Delete);
441 + assert_eq!(decrypted.row_id, "row-1");
442 + assert!(decrypted.data.is_none(), "payload is still None after the envelope unwraps");
443 + assert_eq!(decrypted.hlc, hlc, "HLC survives the round trip");
391 444 }
392 445
393 446 #[test]
@@ -398,6 +451,7 @@ mod tests {
398 451 op: ChangeOp::Insert,
399 452 row_id: "row-1".to_string(),
400 453 timestamp: Utc::now(),
454 + hlc: Default::default(),
401 455 data: Some(serde_json::json!({"title": "test"})),
402 456 };
403 457
@@ -417,6 +471,7 @@ mod tests {
417 471 op: ChangeOp::Insert,
418 472 row_id: "row-1".to_string(),
419 473 timestamp: Utc::now(),
474 + hlc: Default::default(),
420 475 data: Some(original_data.clone()),
421 476 };
422 477
@@ -444,6 +499,7 @@ mod tests {
444 499 op: ChangeOp::Update,
445 500 row_id: "row-abc".to_string(),
446 501 timestamp: ts,
502 + hlc: Default::default(),
447 503 data: Some(original_data.clone()),
448 504 };
449 505
@@ -687,6 +743,7 @@ mod tests {
687 743 op: ChangeOp::Update,
688 744 row_id: "unique-row-id".to_string(),
689 745 timestamp: ts,
746 + hlc: Default::default(),
690 747 data: Some(serde_json::json!({"name": "Alice"})),
691 748 };
692 749
@@ -711,6 +768,7 @@ mod tests {
711 768 op: ChangeOp::Insert,
712 769 row_id: "r1".to_string(),
713 770 timestamp: Utc::now(),
771 + hlc: Default::default(),
714 772 data: Some(serde_json::json!({"title": "Task 1"})),
715 773 },
716 774 ChangeEntry {
@@ -718,6 +776,7 @@ mod tests {
718 776 op: ChangeOp::Update,
719 777 row_id: "r2".to_string(),
720 778 timestamp: Utc::now(),
779 + hlc: Default::default(),
721 780 data: Some(serde_json::json!({"title": "Task 2", "done": true})),
722 781 },
723 782 ChangeEntry {
@@ -725,6 +784,7 @@ mod tests {
725 784 op: ChangeOp::Delete,
726 785 row_id: "r3".to_string(),
727 786 timestamp: Utc::now(),
787 + hlc: Default::default(),
728 788 data: None,
729 789 },
730 790 ];
@@ -738,7 +798,8 @@ mod tests {
738 798 assert_eq!(wire_entries.len(), 3);
739 799 assert!(wire_entries[0].data.is_some());
740 800 assert!(wire_entries[1].data.is_some());
741 - assert!(wire_entries[2].data.is_none());
801 + // The Delete now also seals an encrypted HLC envelope (was None before HLC).
802 + assert!(wire_entries[2].data.is_some());
742 803
743 804 for (i, wire) in wire_entries.into_iter().enumerate() {
744 805 let pull = PullChangeEntry {
@@ -771,6 +832,7 @@ mod tests {
771 832 op: ChangeOp::Insert,
772 833 row_id: "row-1".into(),
773 834 timestamp: Utc::now(),
835 + hlc: Default::default(),
774 836 data: Some(serde_json::json!({"name": "\u{30C6}\u{30B9}\u{30C8}"})),
775 837 };
776 838
@@ -803,6 +865,7 @@ mod tests {
803 865 op: ChangeOp::Insert,
804 866 row_id: "".into(),
805 867 timestamp: Utc::now(),
868 + hlc: Default::default(),
806 869 data: Some(serde_json::json!(42)),
807 870 };
808 871
M shared/synckit-client/src/client/sync.rs +7 -5
@@ -70,12 +70,12 @@ impl SyncKitClient {
70 70 ) -> Result<i64> {
71 71 let token = self.require_token()?;
72 72
73 - // Extract key once for the entire batch (only needed if any entry has data)
74 - let has_data = changes.iter().any(|c| c.data.is_some());
75 - let key_holder = if has_data {
76 - self.require_master_key()?
77 - } else {
73 + // Every change now seals an HLC envelope (Deletes included), so the master
74 + // key is needed whenever there is anything to push.
75 + let key_holder = if changes.is_empty() {
78 76 crypto::ZeroizeOnDrop([0u8; 32])
77 + } else {
78 + self.require_master_key()?
79 79 };
80 80 let master_key: &[u8; 32] = &key_holder;
81 81 let wire_changes = changes
@@ -270,6 +270,7 @@ mod tests {
270 270 op: ChangeOp::Insert,
271 271 row_id: Uuid::new_v4().to_string(),
272 272 timestamp: Utc::now(),
273 + hlc: Default::default(),
273 274 data: Some(serde_json::json!({"title": "Test task", "done": false})),
274 275 };
275 276
@@ -289,6 +290,7 @@ mod tests {
289 290 op: ChangeOp::Delete,
290 291 row_id: "abc-123".to_string(),
291 292 timestamp: Utc::now(),
293 + hlc: Default::default(),
292 294 data: None,
293 295 };
294 296
M shared/synckit-client/src/conflict.rs +96 -76
@@ -16,7 +16,7 @@ use std::collections::HashMap;
16 16 use chrono::{DateTime, Utc};
17 17 use uuid::Uuid;
18 18
19 - use crate::types::{ChangeEntry, ChangeOp, PulledChange};
19 + use crate::types::{ChangeEntry, PulledChange};
20 20
21 21 /// A remote change that conflicts with a local pending change.
22 22 #[derive(Debug, Clone)]
@@ -106,22 +106,22 @@ pub fn detect_conflicts(
106 106 (clean, conflicts)
107 107 }
108 108
109 - /// Last-write-wins resolution by `client_timestamp`.
109 + /// Last-write-wins resolution by hybrid logical clock.
110 110 ///
111 - /// - DELETE vs non-DELETE: delete wins regardless of timestamp.
112 - /// - Both INSERT/UPDATE: newer timestamp wins. Ties go to local.
111 + /// The higher [`Hlc`](crate::types::Hlc) wins, **regardless of operation**. Because
112 + /// the HLC carries the originating device in its `node` component, every value is
113 + /// globally unique and the comparison is convergent: device A resolving (local=a,
114 + /// remote=b) and device B resolving (local=b, remote=a) compare the same two HLCs
115 + /// and keep the same physical change.
116 + ///
117 + /// This replaces the prior wall-clock + "delete always wins" rule. Delete-wins
118 + /// caused permanent loss when a delete on one device beat a strictly-newer edit on
119 + /// another (ultra-fuzz Run #28); under HLC ordering a newer edit beats an older
120 + /// delete and vice-versa, by clock value alone. `>=` only ties to local for an
121 + /// identical HLC, which (distinct `node`) means a same-device echo, not a real
122 + /// cross-device conflict.
113 123 pub fn resolve_lww(local: &ChangeEntry, remote: &PulledChange) -> Resolution {
114 - // DELETE wins over non-DELETE
115 - if local.op == ChangeOp::Delete || remote.entry.op == ChangeOp::Delete {
116 - return if local.op == ChangeOp::Delete {
117 - Resolution::KeepLocal
118 - } else {
119 - Resolution::KeepRemote
120 - };
121 - }
122 -
123 - // Both are INSERT/UPDATE: newer timestamp wins, ties go to local
124 - if local.timestamp >= remote.entry.timestamp {
124 + if local.hlc >= remote.entry.hlc {
125 125 Resolution::KeepLocal
126 126 } else {
127 127 Resolution::KeepRemote
@@ -234,14 +234,24 @@ pub fn resolve_field_merge(
234 234 #[cfg(test)]
235 235 mod tests {
236 236 use super::*;
237 + use crate::types::{ChangeOp, Hlc};
237 238 use serde_json::json;
238 239
240 + /// Fixed node for locally-minted test entries, distinct from any random
241 + /// `other_device`, so HLC tiebreaks are deterministic.
242 + fn local_node() -> Uuid {
243 + Uuid::from_u128(0x1111_1111_1111_1111_1111_1111_1111_1111)
244 + }
245 +
239 246 fn make_entry(table: &str, row_id: &str, op: ChangeOp, ts: DateTime<Utc>) -> ChangeEntry {
247 + // Derive the HLC wall component from the timestamp so the time-ordered
248 + // tests below still express the intended ordering.
240 249 ChangeEntry {
241 250 table: table.to_string(),
242 251 op,
243 252 row_id: row_id.to_string(),
244 253 timestamp: ts,
254 + hlc: Hlc::from_legacy(ts.timestamp_millis(), local_node()),
245 255 data: Some(json!({"value": "test"})),
246 256 }
247 257 }
@@ -254,11 +264,17 @@ mod tests {
254 264 device_id: Uuid,
255 265 seq: i64,
256 266 ) -> PulledChange {
257 - PulledChange {
258 - entry: make_entry(table, row_id, op, ts),
259 - device_id,
260 - seq,
261 - }
267 + let mut entry = make_entry(table, row_id, op, ts);
268 + entry.hlc = Hlc::from_legacy(ts.timestamp_millis(), device_id);
269 + PulledChange { entry, device_id, seq }
270 + }
271 +
272 + /// Build a pulled change with an explicit HLC, for resolution tests that need
273 + /// to control the clock independently of the wall timestamp.
274 + fn pulled_with_hlc(row_id: &str, op: ChangeOp, hlc: Hlc, device_id: Uuid) -> PulledChange {
275 + let mut p = make_pulled("tasks", row_id, op, Utc::now(), device_id, 1);
276 + p.entry.hlc = hlc;
277 + p
262 278 }
263 279
264 280 // ── detect_conflicts ──
@@ -390,7 +406,7 @@ mod tests {
390 406 assert!(conflicts.is_empty());
391 407 }
392 408
393 - // ── resolve_lww ──
409 + // ── resolve_lww (HLC ordering) ──
394 410
395 411 #[test]
396 412 fn lww_picks_newer_timestamp() {
@@ -417,47 +433,58 @@ mod tests {
417 433 }
418 434
419 435 #[test]
420 - fn lww_equal_timestamps_local_wins() {
421 - let other_device = Uuid::new_v4();
422 - let now = Utc::now();
423 -
424 - let local = make_entry("tasks", "r1", ChangeOp::Update, now);
425 - let remote = make_pulled("tasks", "r1", ChangeOp::Update, now, other_device, 1);
426 -
427 - assert!(matches!(resolve_lww(&local, &remote), Resolution::KeepLocal));
436 + fn lww_counter_breaks_same_wall_tie() {
437 + // Same wall_ms, higher counter wins regardless of node.
438 + let other = Uuid::new_v4();
439 + let mut local = make_entry("tasks", "r1", ChangeOp::Update, Utc::now());
440 + local.hlc = Hlc { wall_ms: 1000, counter: 2, node: local_node() };
441 + let remote = pulled_with_hlc("r1", ChangeOp::Update, Hlc { wall_ms: 1000, counter: 5, node: other }, other);
442 + assert!(matches!(resolve_lww(&local, &remote), Resolution::KeepRemote));
428 443 }
429 444
430 445 #[test]
431 - fn lww_local_delete_wins_over_update() {
432 - let other_device = Uuid::new_v4();
433 - let now = Utc::now();
446 + fn lww_node_breaks_exact_tie_convergently() {
447 + // Identical (wall, counter): the node decides, and BOTH devices must pick
448 + // the same physical change. Use nodes with a known order (a < b).
449 + let a = Uuid::from_u128(1);
450 + let b = Uuid::from_u128(2);
451 + let hlc_a = Hlc { wall_ms: 1000, counter: 0, node: a };
452 + let hlc_b = Hlc { wall_ms: 1000, counter: 0, node: b };
434 453
435 - let local = make_entry("tasks", "r1", ChangeOp::Delete, now);
436 - let remote = make_pulled("tasks", "r1", ChangeOp::Update, now, other_device, 1);
454 + // Device A: local is a's change, remote is b's change.
455 + let mut local_a = make_entry("tasks", "r1", ChangeOp::Update, Utc::now());
456 + local_a.hlc = hlc_a;
457 + let remote_b = pulled_with_hlc("r1", ChangeOp::Update, hlc_b, b);
458 + // b > a, so A drops its local and keeps remote (b's change).
459 + assert!(matches!(resolve_lww(&local_a, &remote_b), Resolution::KeepRemote));
437 460
438 - assert!(matches!(resolve_lww(&local, &remote), Resolution::KeepLocal));
461 + // Device B: local is b's change, remote is a's change.
462 + let mut local_b = make_entry("tasks", "r1", ChangeOp::Update, Utc::now());
463 + local_b.hlc = hlc_b;
464 + let remote_a = pulled_with_hlc("r1", ChangeOp::Update, hlc_a, a);
465 + // b > a, so B keeps its local (b's change). Both devices converge on b.
466 + assert!(matches!(resolve_lww(&local_b, &remote_a), Resolution::KeepLocal));
439 467 }
440 468
441 469 #[test]
442 - fn lww_remote_delete_wins_over_update() {
443 - let other_device = Uuid::new_v4();
444 - let now = Utc::now();
445 -
446 - let local = make_entry("tasks", "r1", ChangeOp::Update, now);
447 - let remote = make_pulled("tasks", "r1", ChangeOp::Delete, now, other_device, 1);
448 -
449 - assert!(matches!(resolve_lww(&local, &remote), Resolution::KeepRemote));
470 + fn lww_newer_update_beats_older_delete() {
471 + // The data-loss fix: a strictly-newer UPDATE must beat an older DELETE.
472 + // Under the old "delete always wins" rule this edit was silently lost.
473 + let other = Uuid::new_v4();
474 + let mut local = make_entry("tasks", "r1", ChangeOp::Update, Utc::now());
475 + local.hlc = Hlc { wall_ms: 2000, counter: 0, node: local_node() };
476 + let remote = pulled_with_hlc("r1", ChangeOp::Delete, Hlc { wall_ms: 1000, counter: 0, node: other }, other);
477 + assert!(matches!(resolve_lww(&local, &remote), Resolution::KeepLocal));
450 478 }
451 479
452 480 #[test]
453 - fn lww_both_delete_local_wins() {
454 - let other_device = Uuid::new_v4();
455 - let now = Utc::now();
456 -
457 - let local = make_entry("tasks", "r1", ChangeOp::Delete, now);
458 - let remote = make_pulled("tasks", "r1", ChangeOp::Delete, now, other_device, 1);
459 -
460 - assert!(matches!(resolve_lww(&local, &remote), Resolution::KeepLocal));
481 + fn lww_newer_delete_beats_older_update() {
482 + // Symmetric: a strictly-newer DELETE beats an older UPDATE.
483 + let other = Uuid::new_v4();
484 + let mut local = make_entry("tasks", "r1", ChangeOp::Update, Utc::now());
485 + local.hlc = Hlc { wall_ms: 1000, counter: 0, node: local_node() };
486 + let remote = pulled_with_hlc("r1", ChangeOp::Delete, Hlc { wall_ms: 2000, counter: 0, node: other }, other);
487 + assert!(matches!(resolve_lww(&local, &remote), Resolution::KeepRemote));
461 488 }
462 489
463 490 // ── resolve_field_merge ──
@@ -669,12 +696,11 @@ mod tests {
669 696 assert_eq!(conflicts[0].local.timestamp, t2);
670 697 }
671 698
672 - // Attack vector 2: DELETE vs DELETE in LWW.
673 - // Both are Delete, so it hits `local.op == ChangeOp::Delete` and returns
674 - // KeepLocal — even if remote Delete has a newer timestamp. This is the
675 - // current behavior. Verify it explicitly.
699 + // DELETE vs DELETE under HLC: the higher clock wins like any other pair.
700 + // (Previously local always won and the timestamp was ignored — now a strictly
701 + // newer remote delete wins.)
676 702 #[test]
677 - fn lww_both_delete_local_wins_even_when_remote_newer() {
703 + fn lww_both_delete_newer_wins() {
678 704 let other_device = Uuid::new_v4();
679 705 let old = Utc::now() - chrono::Duration::seconds(60);
680 706 let new = Utc::now();
@@ -682,11 +708,7 @@ mod tests {
682 708 let local = make_entry("tasks", "r1", ChangeOp::Delete, old);
683 709 let remote = make_pulled("tasks", "r1", ChangeOp::Delete, new, other_device, 1);
684 710
685 - // BUG CANDIDATE: remote Delete is newer but local wins because the
686 - // code checks `local.op == Delete` first, returning KeepLocal always.
687 - // For Delete-vs-Delete this is semantically fine (both delete the row),
688 - // but the timestamp is ignored entirely.
689 - assert!(matches!(resolve_lww(&local, &remote), Resolution::KeepLocal));
711 + assert!(matches!(resolve_lww(&local, &remote), Resolution::KeepRemote));
690 712 }
691 713
692 714 // Attack vector 3: field_merge overlapping fields with equal timestamps.
@@ -840,31 +862,29 @@ mod tests {
840 862 }
841 863 }
842 864
843 - // Bonus: INSERT vs DELETE in LWW. Insert is not Delete, so the Delete
844 - // branch fires. If local is Insert and remote is Delete, remote wins.
845 - // This means a remote delete can kill a row that was just created locally.
865 + // INSERT vs DELETE under HLC: operation is irrelevant, the higher clock wins.
866 + // A strictly-newer remote delete beats an older local insert.
846 867 #[test]
847 - fn lww_remote_delete_beats_local_insert() {
868 + fn lww_newer_remote_delete_beats_older_local_insert() {
848 869 let other_device = Uuid::new_v4();
849 - let now = Utc::now();
870 + let old = Utc::now() - chrono::Duration::seconds(60);
871 + let new = Utc::now();
850 872
851 - let local = make_entry("tasks", "r1", ChangeOp::Insert, now);
852 - let remote = make_pulled("tasks", "r1", ChangeOp::Delete, now, other_device, 1);
873 + let local = make_entry("tasks", "r1", ChangeOp::Insert, old);
874 + let remote = make_pulled("tasks", "r1", ChangeOp::Delete, new, other_device, 1);
853 875
854 - // Remote Delete wins over local Insert — the local insert is discarded.
855 - // This could be surprising: user creates a row, but a concurrent
856 - // delete from another device kills it even though the insert is "newer".
857 876 assert!(matches!(resolve_lww(&local, &remote), Resolution::KeepRemote));
858 877 }
859 878
860 - // Bonus: local delete wins over remote insert. Same logic in reverse.
879 + // Reverse: a strictly-newer local delete beats an older remote insert.
861 880 #[test]
862 - fn lww_local_delete_beats_remote_insert() {
881 + fn lww_newer_local_delete_beats_older_remote_insert() {
863 882 let other_device = Uuid::new_v4();
864 - let now = Utc::now();
883 + let old = Utc::now() - chrono::Duration::seconds(60);
884 + let new = Utc::now();
865 885
866 - let local = make_entry("tasks", "r1", ChangeOp::Delete, now);
867 - let remote = make_pulled("tasks", "r1", ChangeOp::Insert, now, other_device, 1);
886 + let local = make_entry("tasks", "r1", ChangeOp::Delete, new);
887 + let remote = make_pulled("tasks", "r1", ChangeOp::Insert, old, other_device, 1);
868 888
869 889 assert!(matches!(resolve_lww(&local, &remote), Resolution::KeepLocal));
870 890 }
M shared/synckit-client/src/lib.rs +2 -1
@@ -32,6 +32,7 @@
32 32 //! op: ChangeOp::Insert,
33 33 //! row_id: uuid::Uuid::new_v4().to_string(),
34 34 //! timestamp: Utc::now(),
35 + //! hlc: Default::default(),
35 36 //! data: Some(serde_json::json!({"title": "Buy milk"})),
36 37 //! },
37 38 //! ]).await?;
@@ -61,4 +62,4 @@ pub use conflict::{
61 62 detect_conflicts, resolve_field_merge, resolve_lww, ConflictPair, ConflictResolver, Resolution,
62 63 };
63 64 pub use error::{Result, SyncKitError};
64 - pub use types::{ChangeEntry, ChangeOp, Device, PullFilter, PulledChange, SyncStatus};
65 + pub use types::{ChangeEntry, ChangeOp, Device, Hlc, PullFilter, PulledChange, SyncStatus};
M shared/synckit-client/src/types.rs +165 -1
@@ -45,6 +45,85 @@ impl ChangeOp {
45 45 }
46 46 }
47 47
48 + // ── Hybrid logical clock ──
49 +
50 + /// A hybrid logical clock (HLC) timestamp: a wall-clock millisecond reading kept
51 + /// monotonic by a tiebreak counter, with the originating device as a final
52 + /// tiebreak. HLCs are the ordering used for conflict resolution.
53 + ///
54 + /// Why not a bare wall-clock timestamp: a device with a fast-running clock would
55 + /// otherwise win every last-write-wins conflict, and two edits in the same
56 + /// millisecond would be unordered. The HLC keeps causality (it never goes
57 + /// backwards and advances past anything it has seen) and breaks exact ties
58 + /// deterministically by `node`, so every device resolves a conflict identically.
59 + ///
60 + /// Ordering is lexicographic: `(wall_ms, counter, node)`. The `node` field makes
61 + /// the value globally unique across devices, so resolution is convergent without
62 + /// any side-channel — comparing two complete HLCs yields the same winner on every
63 + /// device.
64 + ///
65 + /// The value travels inside the E2E-encrypted change payload (the SyncKit server
66 + /// never sees or orders by it), so adding it required no server-side change.
67 + #[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Default, Serialize, Deserialize)]
68 + pub struct Hlc {
69 + /// Wall-clock component, milliseconds since the Unix epoch.
70 + pub wall_ms: i64,
71 + /// Monotonic tiebreak within a single `wall_ms`, reset to 0 when `wall_ms` advances.
72 + pub counter: u32,
73 + /// The device that minted this HLC. Final, deterministic tiebreak.
74 + pub node: Uuid,
75 + }
76 +
77 + impl Hlc {
78 + /// The zero HLC for `node`. Used as the initial clock state and as the floor
79 + /// for legacy entries that predate HLC support.
80 + pub fn zero(node: Uuid) -> Self {
81 + Hlc { wall_ms: 0, counter: 0, node }
82 + }
83 +
84 + /// Synthesize an HLC for a legacy change that carried no embedded clock,
85 + /// deriving the wall component from its `client_timestamp`. This keeps
86 + /// pre-HLC entries orderable against new ones during the transition.
87 + pub fn from_legacy(timestamp_ms: i64, node: Uuid) -> Self {
88 + Hlc { wall_ms: timestamp_ms, counter: 0, node }
89 + }
90 +
91 + /// Advance the clock for a **local** event happening at `now_ms` on `node`.
92 + ///
93 + /// Standard HLC send rule: if physical time has moved past the last reading,
94 + /// adopt it and reset the counter; otherwise hold the wall component and bump
95 + /// the counter so the new event still sorts strictly after the previous one.
96 + pub fn tick(prev: Hlc, now_ms: i64, node: Uuid) -> Hlc {
97 + if now_ms > prev.wall_ms {
98 + Hlc { wall_ms: now_ms, counter: 0, node }
99 + } else {
100 + Hlc { wall_ms: prev.wall_ms, counter: prev.counter + 1, node }
101 + }
102 + }
103 +
104 + /// Advance the clock on **receiving** `remote`, at local physical time
105 + /// `now_ms` on `node`.
106 + ///
107 + /// Standard HLC receive rule: take the max wall component across the local
108 + /// clock, the incoming HLC, and physical time, then derive a counter that is
109 + /// strictly greater than any counter seen at that wall component. This keeps
110 + /// the local clock ahead of everything observed, so subsequent local writes
111 + /// causally follow the remote change.
112 + pub fn observe(prev: Hlc, remote: Hlc, now_ms: i64, node: Uuid) -> Hlc {
113 + let wall = prev.wall_ms.max(remote.wall_ms).max(now_ms);
114 + let counter = if wall == prev.wall_ms && wall == remote.wall_ms {
115 + prev.counter.max(remote.counter) + 1
116 + } else if wall == prev.wall_ms {
117 + prev.counter + 1
118 + } else if wall == remote.wall_ms {
119 + remote.counter + 1
120 + } else {
121 + 0
122 + };
123 + Hlc { wall_ms: wall, counter, node }
124 + }
125 + }
126 +
48 127 // ── Auth ──
49 128
50 129 #[derive(Serialize)]
@@ -103,8 +182,16 @@ pub struct ChangeEntry {
103 182 pub op: ChangeOp,
104 183 /// App-assigned row identifier (typically a UUID string).
105 184 pub row_id: String,
106 - /// When this change was made on the originating device.
185 + /// When this change was made on the originating device. Retained for the
186 + /// server's incremental-pull cursor and for display; conflict ordering uses
187 + /// [`Self::hlc`], not this field.
107 188 pub timestamp: DateTime<Utc>,
189 + /// Hybrid logical clock for this change — the value conflict resolution
190 + /// orders by. The app mints it from its persistent clock at the time the
191 + /// local change is recorded (see [`Hlc::tick`]). It travels inside the
192 + /// E2E-encrypted payload, so the server never sees it.
193 + #[serde(default)]
194 + pub hlc: Hlc,
108 195 /// The row payload as JSON. `None` for Delete operations.
109 196 /// Serialized as absent (not null) when `None`.
110 197 #[serde(skip_serializing_if = "Option::is_none")]
@@ -440,6 +527,7 @@ mod tests {
440 527 op: ChangeOp::Delete,
441 528 row_id: "r".into(),
442 529 timestamp: chrono::Utc::now(),
530 + hlc: Default::default(),
443 531 data: None,
444 532 };
445 533 let json = serde_json::to_string(&entry).unwrap();
@@ -453,6 +541,7 @@ mod tests {
453 541 op: ChangeOp::Insert,
454 542 row_id: "r".into(),
455 543 timestamp: chrono::Utc::now(),
544 + hlc: Default::default(),
456 545 data: Some(json!({"k": "v"})),
457 546 };
458 547 let json = serde_json::to_string(&entry).unwrap();
@@ -568,4 +657,79 @@ mod tests {
568 657 assert_eq!(parsed["tables"].as_array().unwrap().len(), 1);
569 658 assert!(parsed["since"].is_string());
570 659 }
660 +
661 + // ── Hybrid logical clock ──
662 +
663 + #[test]
664 + fn hlc_orders_by_wall_then_counter_then_node() {
665 + let a = Uuid::from_u128(1);
666 + let b = Uuid::from_u128(2);
667 + // wall dominates
668 + assert!(Hlc { wall_ms: 1, counter: 9, node: b } < Hlc { wall_ms: 2, counter: 0, node: a });
669 + // then counter
670 + assert!(Hlc { wall_ms: 5, counter: 0, node: b } < Hlc { wall_ms: 5, counter: 1, node: a });
671 + // then node
672 + assert!(Hlc { wall_ms: 5, counter: 1, node: a } < Hlc { wall_ms: 5, counter: 1, node: b });
673 + }
674 +
675 + #[test]
676 + fn hlc_tick_adopts_physical_time_and_resets_counter() {
677 + let node = Uuid::from_u128(1);
678 + let prev = Hlc { wall_ms: 1000, counter: 4, node };
679 + let next = Hlc::tick(prev, 2000, node);
680 + assert_eq!(next, Hlc { wall_ms: 2000, counter: 0, node });
681 + }
682 +
683 + #[test]
684 + fn hlc_tick_bumps_counter_when_clock_has_not_advanced() {
685 + let node = Uuid::from_u128(1);
686 + // Physical time equal to or behind the last reading: hold wall, bump counter,
687 + // so the new local event still sorts strictly after the previous one.
688 + let prev = Hlc { wall_ms: 1000, counter: 4, node };
689 + assert_eq!(Hlc::tick(prev, 1000, node), Hlc { wall_ms: 1000, counter: 5, node });
690 + assert_eq!(Hlc::tick(prev, 500, node), Hlc { wall_ms: 1000, counter: 5, node });
691 + // Strictly increasing under a stuck clock.
692 + let mut h = Hlc::zero(node);
693 + let mut last = h;
694 + for _ in 0..100 {
695 + h = Hlc::tick(h, 0, node);
696 + assert!(h > last);
697 + last = h;
698 + }
699 + }
700 +
701 + #[test]
702 + fn hlc_observe_stays_ahead_of_a_skewed_fast_remote() {
703 + let me = Uuid::from_u128(1);
704 + let them = Uuid::from_u128(2);
705 + // Our physical clock is at 1000; a remote with a fast clock sends 9000.
706 + let local = Hlc { wall_ms: 1000, counter: 0, node: me };
707 + let remote = Hlc { wall_ms: 9000, counter: 3, node: them };
708 + let merged = Hlc::observe(local, remote, 1000, me);
709 + // We adopt the remote wall and a strictly-greater counter, so our next
710 + // local write causally follows the remote change despite the skew.
711 + assert_eq!(merged.wall_ms, 9000);
712 + assert!(merged.counter > remote.counter);
713 + assert_eq!(merged.node, me);
714 + let next_local = Hlc::tick(merged, 1001, me);
715 + assert!(next_local > remote, "a later local write must outrank the skewed remote");
716 + }
717 +
718 + #[test]
719 + fn hlc_observe_advances_to_physical_time_when_it_leads() {
720 + let me = Uuid::from_u128(1);
721 + let them = Uuid::from_u128(2);
722 + let local = Hlc { wall_ms: 1000, counter: 2, node: me };
723 + let remote = Hlc { wall_ms: 1500, counter: 0, node: them };
724 + // Physical time (3000) leads both: adopt it, counter resets.
725 + let merged = Hlc::observe(local, remote, 3000, me);
726 + assert_eq!(merged, Hlc { wall_ms: 3000, counter: 0, node: me });
727 + }
728 +
729 + #[test]
730 + fn hlc_serde_roundtrip() {
731 + let h = Hlc { wall_ms: 123, counter: 4, node: Uuid::from_u128(7) };
732 + let j = serde_json::to_value(h).unwrap();
733 + assert_eq!(serde_json::from_value::<Hlc>(j).unwrap(), h);
734 + }
571 735 }
M shared/synckit-client/tests/integration.rs +19 -4
@@ -256,6 +256,7 @@ async fn push_encrypts_data() {
256 256 op: ChangeOp::Insert,
257 257 row_id: "row-1".into(),
258 258 timestamp: Utc::now(),
259 + hlc: Default::default(),
259 260 data: Some(json!({"title": "Secret task"})),
260 261 }],
261 262 )
@@ -1131,6 +1132,7 @@ async fn concurrent_push_operations_no_data_corruption() {
1131 1132 op: ChangeOp::Insert,
1132 1133 row_id: format!("row_{i}"),
1133 1134 timestamp: Utc::now(),
1135 + hlc: Default::default(),
1134 1136 data: Some(json!({"index": i})),
1135 1137 };
1136 1138 c.push(device_id, vec![entry]).await
@@ -1211,6 +1213,7 @@ async fn push_many_changes_succeeds() {
1211 1213 op: ChangeOp::Insert,
1212 1214 row_id: format!("row-{i}"),
1213 1215 timestamp: Utc::now(),
1216 + hlc: Default::default(),
1214 1217 data: Some(json!({"index": i, "value": format!("data-{i}")})),
1215 1218 })
1216 1219 .collect();
@@ -1377,6 +1380,7 @@ async fn push_with_data_fails_without_master_key() {
1377 1380 op: ChangeOp::Insert,
1378 1381 row_id: "r1".into(),
1379 1382 timestamp: Utc::now(),
1383 + hlc: Default::default(),
1380 1384 data: Some(json!({"title": "test"})),
1381 1385 }];
1382 1386
@@ -1388,7 +1392,10 @@ async fn push_with_data_fails_without_master_key() {
1388 1392 }
1389 1393
1390 1394 #[tokio::test]
1391 - async fn push_delete_without_master_key_succeeds() {
1395 + async fn push_delete_requires_master_key() {
1396 + // Deletes used to push without a key (no payload to encrypt). With HLC, a
1397 + // Delete now seals its clock into an encrypted envelope, so the master key is
1398 + // required for every push, Deletes included.
1392 1399 let server = MockServer::start().await;
1393 1400
1394 1401 Mock::given(method("POST"))
@@ -1398,18 +1405,22 @@ async fn push_delete_without_master_key_succeeds() {
1398 1405 .await;
1399 1406
1400 1407 let client = authed_client(&server);
1401 - // No master key -- but Delete entries have no data to encrypt
1408 + // No master key loaded.
1402 1409
1403 1410 let changes = vec![ChangeEntry {
1404 1411 table: "tasks".into(),
1405 1412 op: ChangeOp::Delete,
1406 1413 row_id: "r1".into(),
1407 1414 timestamp: Utc::now(),
1415 + hlc: Default::default(),
1408 1416 data: None,
1409 1417 }];
1410 1418
1411 - let cursor = client.push(Uuid::new_v4(), changes).await.unwrap();
1412 - assert_eq!(cursor, 1);
1419 + let err = client.push(Uuid::new_v4(), changes).await.unwrap_err();
1420 + assert!(
1421 + matches!(err, SyncKitError::NoMasterKey),
1422 + "Delete now seals an HLC envelope and needs the key: {err:?}"
1423 + );
1413 1424 }
1414 1425
1415 1426 // ── Blob operations require auth ──
@@ -1474,6 +1485,7 @@ async fn double_push_same_data_both_succeed() {
1474 1485 op: ChangeOp::Insert,
1475 1486 row_id: "same-row".into(),
1476 1487 timestamp: Utc::now(),
1488 + hlc: Default::default(),
1477 1489 data: Some(json!({"title": "duplicate push test"})),
1478 1490 };
1479 1491
@@ -1713,6 +1725,7 @@ async fn encryption_setup_cross_device_roundtrip() {
1713 1725 op: ChangeOp::Insert,
1714 1726 row_id: "cross-r1".into(),
1715 1727 timestamp: Utc::now(),
1728 + hlc: Default::default(),
1716 1729 data: Some(original_data.clone()),
1717 1730 }],
1718 1731 )
@@ -1973,6 +1986,7 @@ async fn end_to_end_push_pull_encryption_roundtrip() {
1973 1986 op: ChangeOp::Insert,
1974 1987 row_id: "e2e-row".into(),
1975 1988 timestamp: Utc::now(),
1989 + hlc: Default::default(),
1976 1990 data: Some(original_data.clone()),
1977 1991 }],
1978 1992 )
@@ -2649,6 +2663,7 @@ async fn concurrent_push_100_entries_each() {
2649 2663 op: ChangeOp::Insert,
2650 2664 row_id: format!("row_{i}"),
2651 2665 timestamp: Utc::now(),
2666 + hlc: Default::default(),
2652 2667 data: Some(json!({"index": i})),
2653 2668 })
2654 2669 .collect();