Skip to main content

max / synckit

17.8 KB · 511 lines History Blame Raw
1 //! Hybrid logical clock ledger and conflict-strategy orchestration.
2 //!
3 //! This is the wiring between the changelog and the crate's audited conflict
4 //! primitives ([`detect_conflicts`], [`CleanChanges::gated`](crate::conflict),
5 //! [`resolve_lww_at`]). It owns three pieces of persistent HLC state:
6 //!
7 //! - the **local clock** (this device's monotonic HLC), kept in `sync_state`;
8 //! - the **per-row stamp** on each `sync_changelog` row (`hlc_wall`/`hlc_counter`);
9 //! - the **committed ledger** (`sync_committed_hlc`), the HLC last applied for a
10 //! row, which the gate uses to drop a re-pulled older change.
11 //!
12 //! [`resolve_pull`] is the entry point: it turns a pulled batch into the resolved
13 //! `Vec<ChangeEntry>` the apply engine writes, dispatching on the schema's
14 //! [`ConflictStrategy`].
15
16 use std::collections::HashMap;
17
18 use chrono::{DateTime, Utc};
19 use rusqlite::{Connection, OptionalExtension};
20 use uuid::Uuid;
21
22 use super::db::{get_sync_state, set_sync_state};
23 use super::schema::{ConflictStrategy, SyncSchema};
24 use crate::conflict::{Resolution, detect_conflicts, resolve_lww_at};
25 use crate::error::Result;
26 use crate::ids::DeviceId;
27 use crate::types::{ChangeEntry, ChangeOp, Hlc, PulledChange};
28
29 // local clock (persisted in sync_state)
30
31 /// Load this device's persisted clock, or `Hlc::zero(node)` if never set.
32 pub fn load_clock(conn: &Connection, node: DeviceId) -> Result<Hlc> {
33 let wall_ms = get_sync_state(conn, "hlc_wall")?
34 .and_then(|s| s.parse::<i64>().ok())
35 .unwrap_or(0);
36 let counter = get_sync_state(conn, "hlc_counter")?
37 .and_then(|s| s.parse::<u32>().ok())
38 .unwrap_or(0);
39 Ok(Hlc {
40 wall_ms,
41 counter,
42 node,
43 })
44 }
45
46 fn save_clock(conn: &Connection, clock: &Hlc) -> Result<()> {
47 set_sync_state(conn, "hlc_wall", &clock.wall_ms.to_string())?;
48 set_sync_state(conn, "hlc_counter", &clock.counter.to_string())?;
49 Ok(())
50 }
51
52 /// Mint an HLC for every unpushed changelog row that lacks one, advancing the
53 /// local clock once per row. Returns how many rows were stamped.
54 pub fn stamp_pending(conn: &Connection, node: DeviceId, now_ms: i64) -> Result<usize> {
55 let ids: Vec<i64> = {
56 let mut stmt = conn.prepare(
57 "SELECT id FROM sync_changelog WHERE pushed = 0 AND hlc_wall IS NULL ORDER BY id",
58 )?;
59 stmt.query_map([], |r| r.get(0))?
60 .collect::<rusqlite::Result<_>>()?
61 };
62 if ids.is_empty() {
63 return Ok(0);
64 }
65 let mut clock = load_clock(conn, node)?;
66 for id in &ids {
67 clock = Hlc::tick(clock, now_ms, node);
68 conn.execute(
69 "UPDATE sync_changelog SET hlc_wall = ?1, hlc_counter = ?2 WHERE id = ?3",
70 (clock.wall_ms, clock.counter, id),
71 )?;
72 }
73 save_clock(conn, &clock)?;
74 Ok(ids.len())
75 }
76
77 /// Advance the local clock past every observed remote HLC (monotonic merge).
78 pub fn observe(
79 conn: &Connection,
80 node: DeviceId,
81 remotes: impl IntoIterator<Item = Hlc>,
82 now_ms: i64,
83 ) -> Result<()> {
84 let mut clock = load_clock(conn, node)?;
85 let mut any = false;
86 for remote in remotes {
87 clock = Hlc::observe(clock, remote, now_ms, node);
88 any = true;
89 }
90 if any {
91 save_clock(conn, &clock)?;
92 }
93 Ok(())
94 }
95
96 // committed ledger
97
98 /// The HLC currently committed for a row, or `None` if never applied locally.
99 pub fn committed_hlc(conn: &Connection, table: &str, row_id: &str) -> Result<Option<Hlc>> {
100 conn.query_row(
101 "SELECT hlc_wall, hlc_counter, hlc_node FROM sync_committed_hlc \
102 WHERE table_name = ?1 AND row_id = ?2",
103 (table, row_id),
104 |r| {
105 let wall_ms: i64 = r.get(0)?;
106 let counter: i64 = r.get(1)?;
107 let node: String = r.get(2)?;
108 Ok((wall_ms, counter as u32, node))
109 },
110 )
111 .optional()?
112 .map(|(wall_ms, counter, node)| {
113 let node = Uuid::parse_str(&node)
114 .map(DeviceId::new)
115 .map_err(|e| crate::error::SyncKitError::Database(format!("bad hlc_node: {e}")))?;
116 Ok(Hlc {
117 wall_ms,
118 counter,
119 node,
120 })
121 })
122 .transpose()
123 }
124
125 /// Record a row's committed HLC, advancing only (never regress the ledger).
126 pub fn set_committed(conn: &Connection, table: &str, row_id: &str, hlc: &Hlc) -> Result<()> {
127 if let Some(existing) = committed_hlc(conn, table, row_id)?
128 && cmp_hlc(&existing, hlc).is_ge()
129 {
130 return Ok(());
131 }
132 conn.execute(
133 "INSERT INTO sync_committed_hlc (table_name, row_id, hlc_wall, hlc_counter, hlc_node) \
134 VALUES (?1, ?2, ?3, ?4, ?5) \
135 ON CONFLICT(table_name, row_id) DO UPDATE SET \
136 hlc_wall = excluded.hlc_wall, hlc_counter = excluded.hlc_counter, hlc_node = excluded.hlc_node",
137 (
138 table,
139 row_id,
140 hlc.wall_ms,
141 hlc.counter,
142 hlc.node.as_uuid().to_string(),
143 ),
144 )?;
145 Ok(())
146 }
147
148 /// Advance the committed ledger for a batch of applied entries.
149 pub fn record_committed(conn: &Connection, entries: &[ChangeEntry]) -> Result<()> {
150 for e in entries {
151 set_committed(conn, &e.table, &e.row_id, &e.hlc)?;
152 }
153 Ok(())
154 }
155
156 // pull resolution
157
158 /// Turn a pulled batch into the resolved changes to apply, per the schema's
159 /// [`ConflictStrategy`].
160 ///
161 /// `ServerOrder` trusts the server's sequence order and returns the entries
162 /// verbatim (last delivered wins). `HybridLogicalClock` runs the full pipeline:
163 /// advance the clock past the remote HLCs, stamp local pending edits, split into
164 /// clean vs conflicting against local pending, HLC-gate the clean set against the
165 /// committed ledger, resolve conflicts by `resolve_lww`, and collapse to one
166 /// entry per row (highest HLC wins, operation-agnostic).
167 pub fn resolve_pull(
168 conn: &Connection,
169 schema: &SyncSchema,
170 node: DeviceId,
171 pulled: Vec<PulledChange>,
172 now: DateTime<Utc>,
173 ) -> Result<Vec<ChangeEntry>> {
174 match schema.conflict {
175 ConflictStrategy::ServerOrder => Ok(pulled.into_iter().map(|p| p.entry).collect()),
176 ConflictStrategy::HybridLogicalClock => {
177 let now_ms = now.timestamp_millis();
178 observe(conn, node, pulled.iter().map(|p| p.entry.hlc), now_ms)?;
179 stamp_pending(conn, node, now_ms)?;
180
181 let local_pending = load_local_pending(conn, node)?;
182 let (clean, conflicts) = detect_conflicts(pulled, &local_pending, node);
183
184 let mut resolved: Vec<ChangeEntry> =
185 clean.gated_at(now, |t, r| lookup_committed(conn, t, r));
186
187 for pair in conflicts {
188 match resolve_lww_at(&pair.local, &pair.remote, now) {
189 Resolution::KeepRemote => resolved.push(pair.remote.entry),
190 Resolution::KeepLocal | Resolution::Skip => {}
191 Resolution::Merged(data) => {
192 let hlc = max_hlc(pair.local.hlc, pair.remote.entry.hlc);
193 resolved.push(ChangeEntry {
194 table: pair.remote.entry.table,
195 op: ChangeOp::Update,
196 row_id: pair.remote.entry.row_id,
197 timestamp: pair.remote.entry.timestamp,
198 hlc,
199 data: Some(data),
200 extra: serde_json::Map::default(),
201 });
202 }
203 }
204 }
205
206 Ok(collapse_max_hlc(resolved))
207 }
208 }
209 }
210
211 /// Committed-HLC lookup for the gate. A read error is logged and treated as
212 /// "never applied", the gate then keeps the change, which the idempotent apply
213 /// path can safely re-run.
214 fn lookup_committed(conn: &Connection, table: &str, row_id: &str) -> Option<Hlc> {
215 match committed_hlc(conn, table, row_id) {
216 Ok(v) => v,
217 Err(e) => {
218 tracing::warn!(
219 table,
220 row_id,
221 "committed_hlc lookup failed, treating as unset: {e}"
222 );
223 None
224 }
225 }
226 }
227
228 /// Load the unpushed, HLC-stamped local changelog rows as `ChangeEntry` values
229 /// for conflict detection. `node` is this device's id, the node of every local
230 /// stamp.
231 fn load_local_pending(conn: &Connection, node: DeviceId) -> Result<Vec<ChangeEntry>> {
232 let mut stmt = conn.prepare(
233 "SELECT table_name, op, row_id, data, hlc_wall, hlc_counter \
234 FROM sync_changelog WHERE pushed = 0 AND hlc_wall IS NOT NULL ORDER BY id",
235 )?;
236 let rows = stmt.query_map([], |r| {
237 let table: String = r.get(0)?;
238 let op: String = r.get(1)?;
239 let row_id: String = r.get(2)?;
240 let data: Option<String> = r.get(3)?;
241 let wall_ms: i64 = r.get(4)?;
242 let counter: i64 = r.get(5)?;
243 Ok((table, op, row_id, data, wall_ms, counter as u32))
244 })?;
245
246 let mut out = Vec::new();
247 for row in rows {
248 let (table, op, row_id, data, wall_ms, counter) = row?;
249 let Some(op) = ChangeOp::from_str_opt(&op) else {
250 continue; // unknown op, not a resolvable local edit
251 };
252 let data = match data {
253 Some(s) => Some(serde_json::from_str(&s)?),
254 None => None,
255 };
256 out.push(ChangeEntry {
257 table,
258 op,
259 row_id,
260 timestamp: Utc::now(),
261 hlc: Hlc {
262 wall_ms,
263 counter,
264 node,
265 },
266 data,
267 extra: serde_json::Map::default(),
268 });
269 }
270 Ok(out)
271 }
272
273 /// Collapse to one entry per `(table, row_id)`, keeping the highest HLC,
274 /// operation-agnostic, so a newer delete beats an older edit and vice versa, and
275 /// every device converges on the same winner. First-seen order is preserved
276 /// (apply re-orders by schema anyway).
277 fn collapse_max_hlc(entries: Vec<ChangeEntry>) -> Vec<ChangeEntry> {
278 let mut best: HashMap<(String, String), usize> = HashMap::new();
279 let mut kept: Vec<ChangeEntry> = Vec::new();
280 for e in entries {
281 let key = (e.table.clone(), e.row_id.clone());
282 match best.get(&key) {
283 Some(&i) if cmp_hlc(&kept[i].hlc, &e.hlc).is_ge() => {}
284 Some(&i) => kept[i] = e,
285 None => {
286 best.insert(key, kept.len());
287 kept.push(e);
288 }
289 }
290 }
291 kept
292 }
293
294 fn cmp_hlc(a: &Hlc, b: &Hlc) -> std::cmp::Ordering {
295 (a.wall_ms, a.counter, a.node.as_uuid()).cmp(&(b.wall_ms, b.counter, b.node.as_uuid()))
296 }
297
298 fn max_hlc(a: Hlc, b: Hlc) -> Hlc {
299 if cmp_hlc(&a, &b).is_ge() { a } else { b }
300 }
301
302 #[cfg(test)]
303 mod tests {
304 use super::super::apply::apply_remote_changes;
305 use super::super::db::configure_connection;
306 use super::super::schema::{SyncSchema, SyncTable};
307 use super::*;
308 use rusqlite::Connection;
309
310 fn node(n: u128) -> DeviceId {
311 DeviceId::new(Uuid::from_u128(n))
312 }
313
314 fn schema() -> SyncSchema {
315 SyncSchema::new(vec![SyncTable::full("note", &["id", "name"])])
316 }
317
318 fn server_order_schema() -> SyncSchema {
319 schema().conflict_strategy(ConflictStrategy::ServerOrder)
320 }
321
322 /// A device: in-memory DB with the note table + migration, and a node id.
323 fn device(n: u128) -> (Connection, DeviceId) {
324 let conn = Connection::open_in_memory().unwrap();
325 configure_connection(&conn).unwrap();
326 conn.execute_batch("CREATE TABLE note (id TEXT PRIMARY KEY, name TEXT);")
327 .unwrap();
328 conn.execute_batch(&schema().migration_sql()).unwrap();
329 (conn, node(n))
330 }
331
332 /// Make a local edit (domain write → trigger captures it), stamp it at
333 /// `now_ms`, and return it as a pulled change from `node` (as a peer would
334 /// receive it via the server).
335 fn local_edit_as_pulled(
336 conn: &Connection,
337 node: DeviceId,
338 id: &str,
339 name: &str,
340 now_ms: i64,
341 seq: i64,
342 ) -> PulledChange {
343 conn.execute(
344 "INSERT INTO note (id, name) VALUES (?1, ?2) \
345 ON CONFLICT(id) DO UPDATE SET name = excluded.name",
346 (id, name),
347 )
348 .unwrap();
349 stamp_pending(conn, node, now_ms).unwrap();
350 let entry = load_local_pending(conn, node)
351 .unwrap()
352 .into_iter()
353 .find(|e| e.row_id == id)
354 .unwrap();
355 PulledChange {
356 entry,
357 device_id: node,
358 seq,
359 }
360 }
361
362 fn note_name(conn: &Connection, id: &str) -> Option<String> {
363 conn.query_row("SELECT name FROM note WHERE id = ?1", [id], |r| r.get(0))
364 .optional()
365 .unwrap()
366 }
367
368 fn pull_apply(
369 conn: &mut Connection,
370 s: &SyncSchema,
371 node: DeviceId,
372 pulled: Vec<PulledChange>,
373 ) {
374 let now = Utc::now();
375 let resolved = resolve_pull(conn, s, node, pulled, now).unwrap();
376 apply_remote_changes(conn, s, &resolved, "").unwrap();
377 record_committed(conn, &resolved).unwrap();
378 }
379
380 #[test]
381 fn stamp_pending_assigns_monotonic_hlcs() {
382 let (conn, n) = device(1);
383 conn.execute("INSERT INTO note (id, name) VALUES ('a', '1')", [])
384 .unwrap();
385 conn.execute("INSERT INTO note (id, name) VALUES ('b', '2')", [])
386 .unwrap();
387 assert_eq!(stamp_pending(&conn, n, 1000).unwrap(), 2);
388 // Re-stamping is a no-op (both already stamped).
389 assert_eq!(stamp_pending(&conn, n, 2000).unwrap(), 0);
390 let stamps: Vec<(i64, i64)> = {
391 let mut s = conn
392 .prepare("SELECT hlc_wall, hlc_counter FROM sync_changelog ORDER BY id")
393 .unwrap();
394 s.query_map([], |r| Ok((r.get(0)?, r.get(1)?)))
395 .unwrap()
396 .map(|r| r.unwrap())
397 .collect()
398 };
399 assert!(stamps.iter().all(|(w, _)| *w == 1000));
400 assert_eq!(
401 stamps[0].1 + 1,
402 stamps[1].1,
403 "counter increments within a wall_ms"
404 );
405 }
406
407 #[test]
408 fn committed_ledger_advances_only() {
409 let (conn, _) = device(1);
410 let older = Hlc {
411 wall_ms: 100,
412 counter: 0,
413 node: node(9),
414 };
415 let newer = Hlc {
416 wall_ms: 200,
417 counter: 0,
418 node: node(9),
419 };
420 set_committed(&conn, "note", "r", &newer).unwrap();
421 set_committed(&conn, "note", "r", &older).unwrap(); // must not regress
422 assert_eq!(
423 committed_hlc(&conn, "note", "r").unwrap().unwrap().wall_ms,
424 200
425 );
426 }
427
428 #[test]
429 fn conflicting_edits_converge_to_higher_hlc_both_directions() {
430 // A edits at t=100, B edits at t=200 → B wins everywhere.
431 let (mut a, an) = device(1);
432 let (mut b, bn) = device(2);
433 let a_change = local_edit_as_pulled(&a, an, "r", "from-A", 100, 1);
434 let b_change = local_edit_as_pulled(&b, bn, "r", "from-B", 200, 1);
435
436 pull_apply(&mut a, &schema(), an, vec![b_change]); // A pulls B (newer) → adopts B
437 pull_apply(&mut b, &schema(), bn, vec![a_change]); // B pulls A (older) → keeps B
438
439 assert_eq!(note_name(&a, "r").as_deref(), Some("from-B"));
440 assert_eq!(note_name(&b, "r").as_deref(), Some("from-B"));
441 }
442
443 #[test]
444 fn gate_drops_repulled_older_change() {
445 let (mut a, an) = device(1);
446 let (b, bn) = device(2);
447 // B's change is applied on A.
448 let b_change = local_edit_as_pulled(&b, bn, "r", "v-old", 100, 1);
449 pull_apply(&mut a, &schema(), an, vec![b_change.clone()]);
450 // A then makes a NEWER local edit and commits it.
451 let _a_new = local_edit_as_pulled(&a, an, "r", "v-new", 300, 2);
452 // Mark A's edit committed (as a push would) so the gate has a committed clock.
453 let a_pending = load_local_pending(&a, an);
454 record_committed(&a, &a_pending.unwrap()).unwrap();
455 // Re-pulling B's OLD change must be gated out (older than committed).
456 let resolved = resolve_pull(&a, &schema(), an, vec![b_change], Utc::now()).unwrap();
457 assert!(
458 resolved.iter().all(|e| e.row_id != "r"),
459 "stale re-pull must be gated"
460 );
461 }
462
463 #[test]
464 fn newer_delete_beats_older_edit() {
465 let (mut a, an) = device(1);
466 let (b, bn) = device(2);
467 // A has an older edit locally.
468 local_edit_as_pulled(&a, an, "r", "edit", 100, 1);
469 // B deletes the same row, newer.
470 b.execute("INSERT INTO note (id, name) VALUES ('r', 'x')", [])
471 .unwrap();
472 stamp_pending(&b, bn, 50).unwrap();
473 b.execute("DELETE FROM note WHERE id = 'r'", []).unwrap();
474 stamp_pending(&b, bn, 200).unwrap();
475 let del = load_local_pending(&b, bn)
476 .unwrap()
477 .into_iter()
478 .find(|e| e.op == ChangeOp::Delete)
479 .unwrap();
480 let pulled = PulledChange {
481 entry: del,
482 device_id: bn,
483 seq: 2,
484 };
485 pull_apply(&mut a, &schema(), an, vec![pulled]);
486 assert_eq!(
487 note_name(&a, "r"),
488 None,
489 "newer delete wins over older edit"
490 );
491 }
492
493 #[test]
494 fn server_order_applies_last_delivered_no_hlc() {
495 let (mut a, an) = device(1);
496 let s = server_order_schema();
497 // Two pulled changes for the same row; server order = last wins, HLC ignored.
498 let (src, sn) = device(2);
499 let first = local_edit_as_pulled(&src, sn, "r", "first", 999, 1); // higher wall
500 let (src2, sn2) = device(3);
501 let second = local_edit_as_pulled(&src2, sn2, "r", "second", 1, 2); // lower wall, later seq
502 let resolved = resolve_pull(&a, &s, an, vec![first, second], Utc::now()).unwrap();
503 apply_remote_changes(&mut a, &s, &resolved, "").unwrap();
504 assert_eq!(
505 note_name(&a, "r").as_deref(),
506 Some("second"),
507 "server order: last delivered wins"
508 );
509 }
510 }
511