Skip to main content

max / goingson

7.2 KB · 178 lines History Blame Raw
1 //! Hybrid logical clock persistence for cross-device sync conflict resolution.
2 //!
3 //! SyncKit orders conflicts by an [`Hlc`] rather than a bare wall-clock timestamp
4 //! (ultra-fuzz Run #28): a skewed-fast device no longer wins every conflict, and a
5 //! strictly-newer edit beats an older delete. This module owns the device's
6 //! persistent clock — `(wall_ms, counter)` in the `hlc_state` table; the HLC's node
7 //! component is always this device's id, so it is not stored.
8 //!
9 //! Local changelog rows are stamped **lazily**: when a row is first read for push or
10 //! conflict detection it gets the next HLC, and the stamp is written back onto the
11 //! row. This keeps the changelog triggers unchanged and guarantees push and
12 //! conflict-detection read the same HLC for a given change.
13
14 use chrono::Utc;
15 use goingson_core::CoreError;
16 use sqlx::SqlitePool;
17 use std::collections::{HashMap, HashSet};
18 use synckit_client::{DeviceId, Hlc};
19 use uuid::Uuid;
20
21 /// Load the persistent clock, reattaching `node` (which is not stored).
22 async fn load_clock(pool: &SqlitePool, node: Uuid) -> Result<Hlc, CoreError> {
23 let (wall_ms, counter): (i64, i64) =
24 sqlx::query_as("SELECT wall_ms, counter FROM hlc_state WHERE id = 1")
25 .fetch_one(pool)
26 .await
27 .map_err(CoreError::database)?;
28 Ok(Hlc { wall_ms, counter: counter as u32, node: DeviceId::new(node) })
29 }
30
31 /// Stamp every unpushed changelog row that lacks an HLC, in id order, advancing the
32 /// persistent clock once per row. Idempotent — already-stamped rows are skipped — so
33 /// push and conflict-detection observe identical stamps. Runs in one transaction so a
34 /// crash mid-stamp leaves the clock and the rows consistent (all-or-nothing).
35 #[tracing::instrument(skip_all)]
36 pub async fn assign_pending_hlcs(pool: &SqlitePool, device_id: Uuid) -> Result<(), CoreError> {
37 let ids: Vec<(i64, String, String)> = sqlx::query_as(
38 "SELECT id, table_name, row_id FROM sync_changelog \
39 WHERE pushed = 0 AND hlc_wall IS NULL ORDER BY id ASC",
40 )
41 .fetch_all(pool)
42 .await
43 .map_err(CoreError::database)?;
44
45 if ids.is_empty() {
46 return Ok(());
47 }
48
49 let mut tx = pool.begin().await.map_err(CoreError::database)?;
50
51 let (wall_ms, counter): (i64, i64) =
52 sqlx::query_as("SELECT wall_ms, counter FROM hlc_state WHERE id = 1")
53 .fetch_one(&mut *tx)
54 .await
55 .map_err(CoreError::database)?;
56 let node = DeviceId::new(device_id);
57 let mut clock = Hlc { wall_ms, counter: counter as u32, node };
58
59 for (id, table_name, row_id) in ids {
60 clock = Hlc::tick(clock, Utc::now().timestamp_millis(), node);
61 sqlx::query("UPDATE sync_changelog SET hlc_wall = ?, hlc_counter = ? WHERE id = ?")
62 .bind(clock.wall_ms)
63 .bind(clock.counter as i64)
64 .bind(id)
65 .execute(&mut *tx)
66 .await
67 .map_err(CoreError::database)?;
68 // The row now holds this local value at `clock`; record it as the row's
69 // committed HLC so a later older remote edit gates out instead of clobbering it.
70 record_committed_hlc(&mut *tx, &table_name, &row_id, clock).await?;
71 }
72
73 sqlx::query("UPDATE hlc_state SET wall_ms = ?, counter = ? WHERE id = 1")
74 .bind(clock.wall_ms)
75 .bind(clock.counter as i64)
76 .execute(&mut *tx)
77 .await
78 .map_err(CoreError::database)?;
79
80 tx.commit().await.map_err(CoreError::database)?;
81 Ok(())
82 }
83
84 /// Advance the persistent clock past a remote HLC observed during pull, so that
85 /// subsequent local changes causally follow it. Monotonic: advancing past the batch
86 /// maximum is enough to cover every change in the batch.
87 #[tracing::instrument(skip_all)]
88 pub async fn observe_remote(pool: &SqlitePool, remote: Hlc, device_id: Uuid) -> Result<(), CoreError> {
89 let clock = load_clock(pool, device_id).await?;
90 let advanced = Hlc::observe(clock, remote, Utc::now().timestamp_millis(), DeviceId::new(device_id));
91 sqlx::query("UPDATE hlc_state SET wall_ms = ?, counter = ? WHERE id = 1")
92 .bind(advanced.wall_ms)
93 .bind(advanced.counter as i64)
94 .execute(pool)
95 .await
96 .map_err(CoreError::database)?;
97 Ok(())
98 }
99
100 /// Record the committed HLC for a row, keeping the maximum.
101 ///
102 /// This is the committed-clock store SyncKit's `CleanChanges` gate reads: a clean
103 /// remote change older-or-equal to a row's committed HLC is dropped rather than
104 /// clobbering a newer local value. Called on every applied change — a local stamp
105 /// ([`assign_pending_hlcs`]) or a remote apply (`pull::apply_changes_inner`) — so
106 /// the stored clock always reflects the value currently in the row. The
107 /// `ON CONFLICT ... WHERE` keeps the larger HLC, comparing `(wall, counter, node)`
108 /// as a row value (node is the canonical lowercase UUID string, whose lexicographic
109 /// order matches the byte order [`Hlc`] compares on).
110 pub(crate) async fn record_committed_hlc<'e, E>(
111 executor: E,
112 table: &str,
113 row_id: &str,
114 hlc: Hlc,
115 ) -> Result<(), CoreError>
116 where
117 E: sqlx::Executor<'e, Database = sqlx::Sqlite>,
118 {
119 sqlx::query(
120 "INSERT INTO sync_committed_hlc (table_name, row_id, hlc_wall, hlc_counter, hlc_node) \
121 VALUES (?, ?, ?, ?, ?) \
122 ON CONFLICT(table_name, row_id) DO UPDATE SET \
123 hlc_wall = excluded.hlc_wall, \
124 hlc_counter = excluded.hlc_counter, \
125 hlc_node = excluded.hlc_node \
126 WHERE (excluded.hlc_wall, excluded.hlc_counter, excluded.hlc_node) > \
127 (sync_committed_hlc.hlc_wall, sync_committed_hlc.hlc_counter, sync_committed_hlc.hlc_node)",
128 )
129 .bind(table)
130 .bind(row_id)
131 .bind(hlc.wall_ms)
132 .bind(hlc.counter as i64)
133 .bind(hlc.node.to_string())
134 .execute(executor)
135 .await
136 .map_err(CoreError::database)?;
137 Ok(())
138 }
139
140 /// Load the committed HLCs for a set of `(table, row_id)` keys, for pre-fetching
141 /// before the `CleanChanges` gate. Queries by `row_id` (one statement) and filters
142 /// the exact `(table, row_id)` pairs in memory, so a row never seen locally is
143 /// simply absent from the map (the gate then keeps that change).
144 pub(crate) async fn load_committed_hlcs(
145 pool: &SqlitePool,
146 keys: &[(String, String)],
147 ) -> Result<HashMap<(String, String), Hlc>, CoreError> {
148 let mut map = HashMap::new();
149 if keys.is_empty() {
150 return Ok(map);
151 }
152
153 let placeholders = vec!["?"; keys.len()].join(",");
154 let sql = format!(
155 "SELECT table_name, row_id, hlc_wall, hlc_counter, hlc_node \
156 FROM sync_committed_hlc WHERE row_id IN ({placeholders})"
157 );
158 let mut query = sqlx::query_as::<_, (String, String, i64, i64, String)>(&sql);
159 for (_, row_id) in keys {
160 query = query.bind(row_id);
161 }
162 let rows = query.fetch_all(pool).await.map_err(CoreError::database)?;
163
164 let want: HashSet<(&str, &str)> =
165 keys.iter().map(|(t, r)| (t.as_str(), r.as_str())).collect();
166 for (table, row_id, wall, counter, node) in rows {
167 if want.contains(&(table.as_str(), row_id.as_str()))
168 && let Ok(node) = Uuid::parse_str(&node)
169 {
170 map.insert(
171 (table, row_id),
172 Hlc { wall_ms: wall, counter: counter as u32, node: DeviceId::new(node) },
173 );
174 }
175 }
176 Ok(map)
177 }
178