Skip to main content

max / makenotwork

synckit: forced clean-change HLC gate via CleanChanges Deep A+ pass, sync axis (D6). detect_conflicts returned a bare Vec of clean changes, and the "gate the clean set against the row's committed HLC" rule was documented but unenforced — so a stale remote edit for an already-committed row could be applied blindly and clobber a newer local value. It now returns an opaque CleanChanges whose only door to the entries is `gated(committed_hlc)`; `row_keys()` lets a caller pre-fetch the clocks first. Applying a clean change without consulting the committed clock is now unrepresentable. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
Author: Max Johnson <me@maxj.phd> · 2026-06-22 22:59 UTC
Commit: adf3718dddef06119d336d43a6d3eac6a03ee066
Parent: eb3b19a
2 files changed, +77 insertions, -4 deletions
M shared/synckit-client/src/conflict.rs +75 -3
@@ -18,7 +18,7 @@ use chrono::{DateTime, Utc};
18 18 use uuid::Uuid;
19 19
20 20 use crate::ids::DeviceId;
21 - use crate::types::{ChangeEntry, PulledChange};
21 + use crate::types::{ChangeEntry, Hlc, PulledChange};
22 22
23 23 /// A remote change that conflicts with a local pending change.
24 24 #[derive(Debug, Clone)]
@@ -29,6 +29,59 @@ pub struct ConflictPair {
29 29 pub local: ChangeEntry,
30 30 }
31 31
32 + /// The non-conflicting changes from a pull, withheld behind the HLC gate.
33 + ///
34 + /// "Clean" means only that no local *pending* edit contests the row — **not**
35 + /// that the remote change is newer than what is already committed locally. A
36 + /// device that applied and pushed a newer edit, then pulls an older remote edit
37 + /// for the same row, sees that older edit here. Applying it blindly clobbers the
38 + /// newer local value.
39 + ///
40 + /// So the applicable entries cannot be read out directly: [`gated`](Self::gated)
41 + /// is the only door, and it requires the row's committed HLC. This makes
42 + /// "apply a clean change without checking the committed clock" unrepresentable
43 + /// rather than a contract the caller has to remember.
44 + #[derive(Debug, Clone)]
45 + pub struct CleanChanges(Vec<PulledChange>);
46 +
47 + impl CleanChanges {
48 + /// HLC-gate the clean changes against committed local clocks.
49 + ///
50 + /// `committed_hlc(table, row_id)` returns the HLC currently committed for a
51 + /// row, or `None` if it has never been applied locally. A change is dropped
52 + /// when its HLC is older than or equal to the committed clock; everything
53 + /// else is returned in pull order, ready to apply.
54 + pub fn gated(self, committed_hlc: impl Fn(&str, &str) -> Option<Hlc>) -> Vec<ChangeEntry> {
55 + self.0
56 + .into_iter()
57 + .filter(|p| {
58 + committed_hlc(&p.entry.table, &p.entry.row_id)
59 + .is_none_or(|committed| p.entry.hlc > committed)
60 + })
61 + .map(|p| p.entry)
62 + .collect()
63 + }
64 +
65 + /// The `(table, row_id)` addresses of the clean changes, so a caller can
66 + /// pre-fetch their committed clocks in one query before [`gated`](Self::gated).
67 + /// Yields addresses only — the change payloads stay sealed until gating.
68 + pub fn row_keys(&self) -> impl Iterator<Item = (&str, &str)> {
69 + self.0
70 + .iter()
71 + .map(|p| (p.entry.table.as_str(), p.entry.row_id.as_str()))
72 + }
73 +
74 + /// Number of clean changes before gating (for metrics/logging).
75 + pub fn len(&self) -> usize {
76 + self.0.len()
77 + }
78 +
79 + /// Whether there are no clean changes at all.
80 + pub fn is_empty(&self) -> bool {
81 + self.0.is_empty()
82 + }
83 + }
84 +
32 85 /// How a conflict should be resolved.
33 86 #[derive(Debug, Clone)]
34 87 pub enum Resolution {
@@ -85,7 +138,7 @@ pub fn detect_conflicts(
85 138 remote: Vec<PulledChange>,
86 139 local_pending: &[ChangeEntry],
87 140 our_device_id: DeviceId,
88 - ) -> (Vec<PulledChange>, Vec<ConflictPair>) {
141 + ) -> (CleanChanges, Vec<ConflictPair>) {
89 142 // Build lookup: (table, row_id) -> last local pending entry.
90 143 // If local_pending has duplicates for the same key, the last entry wins.
91 144 let mut local_map: HashMap<(&str, &str), &ChangeEntry> = HashMap::new();
@@ -114,7 +167,7 @@ pub fn detect_conflicts(
114 167 }
115 168 }
116 169
117 - (clean, conflicts)
170 + (CleanChanges(clean), conflicts)
118 171 }
119 172
120 173 /// Last-write-wins resolution by hybrid logical clock.
@@ -367,6 +420,25 @@ mod tests {
367 420 }
368 421
369 422 #[test]
423 + fn clean_changes_gate_drops_stale_keeps_newer() {
424 + let our_device = Uuid::new_v4();
425 + let other_device = Uuid::new_v4();
426 + let now = Utc::now();
427 + // A clean remote change for tasks/r1; its HLC wall == now_ms.
428 + let remote = vec![make_pulled("tasks", "r1", ChangeOp::Update, now, other_device, 1)];
429 + let (clean, _conflicts) = detect_conflicts(remote, &[], DeviceId::new(our_device));
430 + assert_eq!(clean.len(), 1);
431 +
432 + // No committed clock for the row → kept (first time we've seen it).
433 + assert_eq!(clean.clone().gated(|_, _| None).len(), 1);
434 + // Committed clock older than the remote → kept.
435 + assert_eq!(clean.clone().gated(|_, _| Some(Hlc::zero(other_device))).len(), 1);
436 + // Committed clock newer than the remote → dropped (would clobber newer local).
437 + let newer = Hlc { wall_ms: now.timestamp_millis() + 1, counter: 0, node: other_device };
438 + assert!(clean.gated(move |_, _| Some(newer)).is_empty());
439 + }
440 +
441 + #[test]
370 442 fn different_tables_same_row_id_no_conflict() {
371 443 let our_device = Uuid::new_v4();
372 444 let other_device = Uuid::new_v4();
M shared/synckit-client/src/lib.rs +2 -1
@@ -61,7 +61,8 @@ pub use client::subscription::{
61 61 SubscriptionStatus,
62 62 };
63 63 pub use conflict::{
64 - detect_conflicts, resolve_field_merge, resolve_lww, ConflictPair, ConflictResolver, Resolution,
64 + detect_conflicts, resolve_field_merge, resolve_lww, CleanChanges, ConflictPair,
65 + ConflictResolver, Resolution,
65 66 };
66 67 pub use error::{Result, SyncKitError};
67 68 pub use ids::{AppId, DeviceId, UserId};