Skip to main content

max / goingson

sync: per-row committed-HLC store to back SyncKit's clean-change gate SyncKit now withholds clean pulled changes behind CleanChanges::gated, which needs each row's committed HLC. GO had none (only the device clock + per-row stamps for local *pending* changes). Adds migration 056 sync_committed_hlc, a record/load pair (max-keeping), recording on every local stamp and remote apply, and wires the gate into pull. Closes the latent hazard where an older remote edit for an already-committed row could clobber a newer local value. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
Author: Max Johnson <me@maxj.phd> · 2026-06-22 23:00 UTC
Commit: dd878b25935dd2ea71ef9e9afd28852727b1a629
Parent: 66fb0ed
4 files changed, +177 insertions, -16 deletions
A migrations/sqlite/056_committed_hlc.sql +22
@@ -0,0 +1,22 @@
1 + -- Per-row committed HLC, for gating clean remote changes (deep A+ remediation).
2 + --
3 + -- "Clean" pulled changes (those with no contesting local *pending* edit) were
4 + -- applied blindly. That can clobber a newer value already committed locally when
5 + -- an older remote edit arrives for an already-applied row -- a latent multi-device
6 + -- hazard the SyncKit audit flagged. SyncKit's CleanChanges gate now drops any clean
7 + -- change whose HLC is older-or-equal to the row's committed HLC; this table is the
8 + -- committed-clock store the gate reads.
9 + --
10 + -- Updated on every applied change -- a local stamp (hlc.rs::assign_pending_hlcs) or
11 + -- a remote apply (pull.rs::apply_changes_inner) -- keeping the maximum HLC per row.
12 + -- The node component is stored as its canonical lowercase UUID string, whose
13 + -- lexicographic order matches the byte order SyncKit's Hlc compares on.
14 +
15 + CREATE TABLE IF NOT EXISTS sync_committed_hlc (
16 + table_name TEXT NOT NULL,
17 + row_id TEXT NOT NULL,
18 + hlc_wall INTEGER NOT NULL,
19 + hlc_counter INTEGER NOT NULL,
20 + hlc_node TEXT NOT NULL,
21 + PRIMARY KEY (table_name, row_id)
22 + ) WITHOUT ROWID;
M src-tauri/src/sync_service/hlc.rs +87 -3
@@ -14,6 +14,7 @@
14 14 use chrono::Utc;
15 15 use goingson_core::CoreError;
16 16 use sqlx::SqlitePool;
17 + use std::collections::{HashMap, HashSet};
17 18 use synckit_client::Hlc;
18 19 use uuid::Uuid;
19 20
@@ -32,8 +33,9 @@ async fn load_clock(pool: &SqlitePool, node: Uuid) -> Result<Hlc, CoreError> {
32 33 /// push and conflict-detection observe identical stamps. Runs in one transaction so a
33 34 /// crash mid-stamp leaves the clock and the rows consistent (all-or-nothing).
34 35 pub async fn assign_pending_hlcs(pool: &SqlitePool, device_id: Uuid) -> Result<(), CoreError> {
35 - let ids: Vec<(i64,)> = sqlx::query_as(
36 - "SELECT id FROM sync_changelog WHERE pushed = 0 AND hlc_wall IS NULL ORDER BY id ASC",
36 + let ids: Vec<(i64, String, String)> = sqlx::query_as(
37 + "SELECT id, table_name, row_id FROM sync_changelog \
38 + WHERE pushed = 0 AND hlc_wall IS NULL ORDER BY id ASC",
37 39 )
38 40 .fetch_all(pool)
39 41 .await
@@ -52,7 +54,7 @@ pub async fn assign_pending_hlcs(pool: &SqlitePool, device_id: Uuid) -> Result<(
52 54 .map_err(CoreError::database)?;
53 55 let mut clock = Hlc { wall_ms, counter: counter as u32, node: device_id };
54 56
55 - for (id,) in ids {
57 + for (id, table_name, row_id) in ids {
56 58 clock = Hlc::tick(clock, Utc::now().timestamp_millis(), device_id);
57 59 sqlx::query("UPDATE sync_changelog SET hlc_wall = ?, hlc_counter = ? WHERE id = ?")
58 60 .bind(clock.wall_ms)
@@ -61,6 +63,9 @@ pub async fn assign_pending_hlcs(pool: &SqlitePool, device_id: Uuid) -> Result<(
61 63 .execute(&mut *tx)
62 64 .await
63 65 .map_err(CoreError::database)?;
66 + // The row now holds this local value at `clock`; record it as the row's
67 + // committed HLC so a later older remote edit gates out instead of clobbering it.
68 + record_committed_hlc(&mut *tx, &table_name, &row_id, clock).await?;
64 69 }
65 70
66 71 sqlx::query("UPDATE hlc_state SET wall_ms = ?, counter = ? WHERE id = 1")
@@ -88,3 +93,82 @@ pub async fn observe_remote(pool: &SqlitePool, remote: Hlc, device_id: Uuid) ->
88 93 .map_err(CoreError::database)?;
89 94 Ok(())
90 95 }
96 +
97 + /// Record the committed HLC for a row, keeping the maximum.
98 + ///
99 + /// This is the committed-clock store SyncKit's `CleanChanges` gate reads: a clean
100 + /// remote change older-or-equal to a row's committed HLC is dropped rather than
101 + /// clobbering a newer local value. Called on every applied change — a local stamp
102 + /// ([`assign_pending_hlcs`]) or a remote apply (`pull::apply_changes_inner`) — so
103 + /// the stored clock always reflects the value currently in the row. The
104 + /// `ON CONFLICT ... WHERE` keeps the larger HLC, comparing `(wall, counter, node)`
105 + /// as a row value (node is the canonical lowercase UUID string, whose lexicographic
106 + /// order matches the byte order [`Hlc`] compares on).
107 + pub(crate) async fn record_committed_hlc<'e, E>(
108 + executor: E,
109 + table: &str,
110 + row_id: &str,
111 + hlc: Hlc,
112 + ) -> Result<(), CoreError>
113 + where
114 + E: sqlx::Executor<'e, Database = sqlx::Sqlite>,
115 + {
116 + sqlx::query(
117 + "INSERT INTO sync_committed_hlc (table_name, row_id, hlc_wall, hlc_counter, hlc_node) \
118 + VALUES (?, ?, ?, ?, ?) \
119 + ON CONFLICT(table_name, row_id) DO UPDATE SET \
120 + hlc_wall = excluded.hlc_wall, \
121 + hlc_counter = excluded.hlc_counter, \
122 + hlc_node = excluded.hlc_node \
123 + WHERE (excluded.hlc_wall, excluded.hlc_counter, excluded.hlc_node) > \
124 + (sync_committed_hlc.hlc_wall, sync_committed_hlc.hlc_counter, sync_committed_hlc.hlc_node)",
125 + )
126 + .bind(table)
127 + .bind(row_id)
128 + .bind(hlc.wall_ms)
129 + .bind(hlc.counter as i64)
130 + .bind(hlc.node.to_string())
131 + .execute(executor)
132 + .await
133 + .map_err(CoreError::database)?;
134 + Ok(())
135 + }
136 +
137 + /// Load the committed HLCs for a set of `(table, row_id)` keys, for pre-fetching
138 + /// before the `CleanChanges` gate. Queries by `row_id` (one statement) and filters
139 + /// the exact `(table, row_id)` pairs in memory, so a row never seen locally is
140 + /// simply absent from the map (the gate then keeps that change).
141 + pub(crate) async fn load_committed_hlcs(
142 + pool: &SqlitePool,
143 + keys: &[(String, String)],
144 + ) -> Result<HashMap<(String, String), Hlc>, CoreError> {
145 + let mut map = HashMap::new();
146 + if keys.is_empty() {
147 + return Ok(map);
148 + }
149 +
150 + let placeholders = vec!["?"; keys.len()].join(",");
151 + let sql = format!(
152 + "SELECT table_name, row_id, hlc_wall, hlc_counter, hlc_node \
153 + FROM sync_committed_hlc WHERE row_id IN ({placeholders})"
154 + );
155 + let mut query = sqlx::query_as::<_, (String, String, i64, i64, String)>(&sql);
156 + for (_, row_id) in keys {
157 + query = query.bind(row_id);
158 + }
159 + let rows = query.fetch_all(pool).await.map_err(CoreError::database)?;
160 +
161 + let want: HashSet<(&str, &str)> =
162 + keys.iter().map(|(t, r)| (t.as_str(), r.as_str())).collect();
163 + for (table, row_id, wall, counter, node) in rows {
164 + if want.contains(&(table.as_str(), row_id.as_str()))
165 + && let Ok(node) = Uuid::parse_str(&node)
166 + {
167 + map.insert(
168 + (table, row_id),
169 + Hlc { wall_ms: wall, counter: counter as u32, node },
170 + );
171 + }
172 + }
173 + Ok(map)
174 + }
M src-tauri/src/sync_service/pull.rs +34 -13
@@ -11,7 +11,7 @@ use tracing::{debug, warn};
11 11 use uuid::Uuid;
12 12
13 13 use super::state::{get_sync_state, set_sync_state};
14 - use super::hlc::{assign_pending_hlcs, observe_remote};
14 + use super::hlc::{assign_pending_hlcs, load_committed_hlcs, observe_remote, record_committed_hlc};
15 15 use super::{UPSERT_ORDER, DELETE_ORDER};
16 16 use super::apply::{apply_upsert, apply_delete};
17 17
@@ -49,13 +49,24 @@ pub async fn pull_changes(
49 49 // Detect conflicts between remote changes and local pending
50 50 let (clean, conflicts) = detect_conflicts(changes, &local_pending, DeviceId::new(device_id));
51 51
52 - // Apply non-conflicting changes directly
52 + // Apply non-conflicting changes — but HLC-gate them first. "Clean" only
53 + // means no local *pending* edit contests the row; an older remote edit for
54 + // an already-committed row also lands here, and applying it blindly would
55 + // clobber the newer local value. Pre-fetch the rows' committed clocks, then
56 + // let SyncKit's CleanChanges gate drop anything older-or-equal.
53 57 let clean_count = clean.len() as i64;
54 58 if !clean.is_empty() {
59 + let keys: Vec<(String, String)> = clean
60 + .row_keys()
61 + .map(|(t, r)| (t.to_string(), r.to_string()))
62 + .collect();
63 + let committed = load_committed_hlcs(pool, &keys).await?;
55 64 let clean_entries: Vec<ChangeEntry> =
56 - clean.into_iter().map(|p| p.entry).collect();
57 - changed_tables.extend(clean_entries.iter().map(|e| e.table.clone()));
58 - apply_remote_changes(pool, clean_entries).await?;
65 + clean.gated(|t, r| committed.get(&(t.to_string(), r.to_string())).copied());
66 + if !clean_entries.is_empty() {
67 + changed_tables.extend(clean_entries.iter().map(|e| e.table.clone()));
68 + apply_remote_changes(pool, clean_entries).await?;
69 + }
59 70 }
60 71
61 72 // Resolve conflicts with LWW
@@ -285,10 +296,15 @@ pub(crate) async fn apply_changes_inner(
285 296 let Some(ref data) = change.data else {
286 297 continue;
287 298 };
288 - if let Err(e) = apply_upsert(&mut *conn, table, &change.row_id, data).await {
289 - warn!(table = *table, row_id = %change.row_id, error = %e,
290 - "Skipping un-appliable remote upsert");
291 - skipped += 1;
299 + match apply_upsert(&mut *conn, table, &change.row_id, data).await {
300 + Ok(()) => {
301 + record_committed_hlc(&mut *conn, table, &change.row_id, change.hlc).await?;
302 + }
303 + Err(e) => {
304 + warn!(table = *table, row_id = %change.row_id, error = %e,
305 + "Skipping un-appliable remote upsert");
306 + skipped += 1;
307 + }
292 308 }
293 309 }
294 310 }
@@ -299,10 +315,15 @@ pub(crate) async fn apply_changes_inner(
299 315 if change.table != *table {
300 316 continue;
301 317 }
302 - if let Err(e) = apply_delete(&mut *conn, table, &change.row_id).await {
303 - warn!(table = *table, row_id = %change.row_id, error = %e,
304 - "Skipping un-appliable remote delete");
305 - skipped += 1;
318 + match apply_delete(&mut *conn, table, &change.row_id).await {
319 + Ok(()) => {
320 + record_committed_hlc(&mut *conn, table, &change.row_id, change.hlc).await?;
321 + }
322 + Err(e) => {
323 + warn!(table = *table, row_id = %change.row_id, error = %e,
324 + "Skipping un-appliable remote delete");
325 + skipped += 1;
326 + }
306 327 }
307 328 }
308 329 }
M src-tauri/src/sync_service/tests.rs +34
@@ -167,6 +167,40 @@ use sqlx::SqlitePool;
167 167 assert_eq!(get_sync_state(&pool, "initial_snapshot_done").await.unwrap(), "0");
168 168 }
169 169
170 + // -- committed HLC store (CleanChanges gate backing) --
171 +
172 + #[tokio::test]
173 + async fn committed_hlc_records_keeps_max_and_loads() {
174 + use crate::sync_service::hlc::{load_committed_hlcs, record_committed_hlc};
175 + use synckit_client::Hlc;
176 +
177 + let pool = setup_test_db().await;
178 + let node = uuid::Uuid::new_v4();
179 + let lo = Hlc { wall_ms: 100, counter: 0, node };
180 + let hi = Hlc { wall_ms: 200, counter: 0, node };
181 + let key = ("tasks".to_string(), "r1".to_string());
182 +
183 + record_committed_hlc(&pool, "tasks", "r1", lo).await.unwrap();
184 + let m = load_committed_hlcs(&pool, std::slice::from_ref(&key)).await.unwrap();
185 + assert_eq!(m.get(&key).copied(), Some(lo));
186 +
187 + // A higher HLC overwrites; a subsequent lower one does NOT (max-keeping).
188 + record_committed_hlc(&pool, "tasks", "r1", hi).await.unwrap();
189 + record_committed_hlc(&pool, "tasks", "r1", lo).await.unwrap();
190 + let m = load_committed_hlcs(&pool, std::slice::from_ref(&key)).await.unwrap();
191 + assert_eq!(m.get(&key).copied(), Some(hi), "older HLC must not overwrite newer committed clock");
192 +
193 + // A row never recorded is absent (the gate then keeps that change).
194 + let missing = ("tasks".to_string(), "nope".to_string());
195 + let m = load_committed_hlcs(&pool, std::slice::from_ref(&missing)).await.unwrap();
196 + assert!(m.is_empty());
197 + }
198 +
199 + // (The clean-change gate itself — detect_conflicts -> CleanChanges::gated —
200 + // is exercised end-to-end in synckit-client's own conflict tests, which can
201 + // construct the non-exhaustive PulledChange internally. Here we only cover
202 + // GO's committed-clock store that backs the gate.)
203 +
170 204 // -- apply_upsert --
171 205
172 206 #[tokio::test]