Skip to main content

max / goingson

sync: adopt synckit 0.6 DeviceId-typed HLC node Lockstep update for the synckit-client breaking change: Hlc.node is now a DeviceId and the Hlc constructors take DeviceId. Wrap the local device_id at each Hlc construction/tick/observe boundary in the sync_service HLC layer; hlc.node.to_string() persistence is unchanged (DeviceId: Display). No behavior change. Gate: 62 sync_service tests + clippy clean. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Author: Max Johnson <me@maxj.phd> · 2026-07-07 18:36 UTC
Commit: 29e6ffc612149b78e4a57b8275772a320c7e796c
Parent: fc62ae0
4 files changed, +19 insertions, -18 deletions
M src-tauri/src/sync_service/hlc.rs +7 -6
@@ -15,7 +15,7 @@ use chrono::Utc;
15 15 use goingson_core::CoreError;
16 16 use sqlx::SqlitePool;
17 17 use std::collections::{HashMap, HashSet};
18 - use synckit_client::Hlc;
18 + use synckit_client::{DeviceId, Hlc};
19 19 use uuid::Uuid;
20 20
21 21 /// Load the persistent clock, reattaching `node` (which is not stored).
@@ -25,7 +25,7 @@ async fn load_clock(pool: &SqlitePool, node: Uuid) -> Result<Hlc, CoreError> {
25 25 .fetch_one(pool)
26 26 .await
27 27 .map_err(CoreError::database)?;
28 - Ok(Hlc { wall_ms, counter: counter as u32, node })
28 + Ok(Hlc { wall_ms, counter: counter as u32, node: DeviceId::new(node) })
29 29 }
30 30
31 31 /// Stamp every unpushed changelog row that lacks an HLC, in id order, advancing the
@@ -53,10 +53,11 @@ pub async fn assign_pending_hlcs(pool: &SqlitePool, device_id: Uuid) -> Result<(
53 53 .fetch_one(&mut *tx)
54 54 .await
55 55 .map_err(CoreError::database)?;
56 - let mut clock = Hlc { wall_ms, counter: counter as u32, node: device_id };
56 + let node = DeviceId::new(device_id);
57 + let mut clock = Hlc { wall_ms, counter: counter as u32, node };
57 58
58 59 for (id, table_name, row_id) in ids {
59 - clock = Hlc::tick(clock, Utc::now().timestamp_millis(), device_id);
60 + clock = Hlc::tick(clock, Utc::now().timestamp_millis(), node);
60 61 sqlx::query("UPDATE sync_changelog SET hlc_wall = ?, hlc_counter = ? WHERE id = ?")
61 62 .bind(clock.wall_ms)
62 63 .bind(clock.counter as i64)
@@ -86,7 +87,7 @@ pub async fn assign_pending_hlcs(pool: &SqlitePool, device_id: Uuid) -> Result<(
86 87 #[tracing::instrument(skip_all)]
87 88 pub async fn observe_remote(pool: &SqlitePool, remote: Hlc, device_id: Uuid) -> Result<(), CoreError> {
88 89 let clock = load_clock(pool, device_id).await?;
89 - let advanced = Hlc::observe(clock, remote, Utc::now().timestamp_millis(), device_id);
90 + let advanced = Hlc::observe(clock, remote, Utc::now().timestamp_millis(), DeviceId::new(device_id));
90 91 sqlx::query("UPDATE hlc_state SET wall_ms = ?, counter = ? WHERE id = 1")
91 92 .bind(advanced.wall_ms)
92 93 .bind(advanced.counter as i64)
@@ -168,7 +169,7 @@ pub(crate) async fn load_committed_hlcs(
168 169 {
169 170 map.insert(
170 171 (table, row_id),
171 - Hlc { wall_ms: wall, counter: counter as u32, node },
172 + Hlc { wall_ms: wall, counter: counter as u32, node: DeviceId::new(node) },
172 173 );
173 174 }
174 175 }
M src-tauri/src/sync_service/pull.rs +2 -2
@@ -167,8 +167,8 @@ async fn read_local_pending(pool: &SqlitePool, device_id: Uuid) -> Result<Vec<Ch
167 167 let json_data = data.and_then(|d| serde_json::from_str(&d).ok());
168 168
169 169 let hlc = match (hlc_wall, hlc_counter) {
170 - (Some(wall), Some(counter)) => Hlc { wall_ms: wall, counter: counter as u32, node: device_id },
171 - _ => Hlc::from_legacy(ts.timestamp_millis(), device_id),
170 + (Some(wall), Some(counter)) => Hlc { wall_ms: wall, counter: counter as u32, node: DeviceId::new(device_id) },
171 + _ => Hlc::from_legacy(ts.timestamp_millis(), DeviceId::new(device_id)),
172 172 };
173 173
174 174 Some(ChangeEntry {
M src-tauri/src/sync_service/push.rs +2 -2
@@ -64,8 +64,8 @@ pub async fn push_changes(
64 64 // Stamped by assign_pending_hlcs above; fall back to a
65 65 // timestamp-derived HLC if a row was somehow left unstamped.
66 66 let hlc = match (hlc_wall, hlc_counter) {
67 - (Some(wall), Some(counter)) => Hlc { wall_ms: wall, counter: counter as u32, node: device_id },
68 - _ => Hlc::from_legacy(ts.timestamp_millis(), device_id),
67 + (Some(wall), Some(counter)) => Hlc { wall_ms: wall, counter: counter as u32, node: DeviceId::new(device_id) },
68 + _ => Hlc::from_legacy(ts.timestamp_millis(), DeviceId::new(device_id)),
69 69 };
70 70
71 71 Some(ChangeEntry {
M src-tauri/src/sync_service/tests.rs +8 -8
@@ -45,7 +45,7 @@ use sqlx::SqlitePool;
45 45 op,
46 46 row_id: row_id.to_string(),
47 47 timestamp: chrono::Utc::now(),
48 - hlc: Default::default(),
48 + hlc: synckit_client::Hlc::zero(synckit_client::DeviceId::nil()),
49 49 data,
50 50 extra: Default::default(),
51 51 }
@@ -177,8 +177,8 @@ use sqlx::SqlitePool;
177 177
178 178 let pool = setup_test_db().await;
179 179 let node = uuid::Uuid::new_v4();
180 - let lo = Hlc { wall_ms: 100, counter: 0, node };
181 - let hi = Hlc { wall_ms: 200, counter: 0, node };
180 + let lo = Hlc { wall_ms: 100, counter: 0, node: synckit_client::DeviceId::new(node) };
181 + let hi = Hlc { wall_ms: 200, counter: 0, node: synckit_client::DeviceId::new(node) };
182 182 let key = ("tasks".to_string(), "r1".to_string());
183 183
184 184 record_committed_hlc(&pool, "tasks", "r1", lo).await.unwrap();
@@ -1918,7 +1918,7 @@ use sqlx::SqlitePool;
1918 1918 .unwrap_or_else(|| panic!("no committed HLC for {row_id}"));
1919 1919 assert_eq!(hlc.wall_ms, *wall);
1920 1920 assert_eq!(hlc.counter as i64, *ctr);
1921 - assert_eq!(hlc.node, device_id, "committed HLC carries this device's node");
1921 + assert_eq!(hlc.node.as_uuid(), device_id, "committed HLC carries this device's node");
1922 1922 }
1923 1923
1924 1924 // The persistent clock advanced exactly to the last stamp.
@@ -1952,7 +1952,7 @@ use sqlx::SqlitePool;
1952 1952 let annotation_id = uuid::Uuid::new_v4().to_string();
1953 1953 let now = now_sql();
1954 1954 let node = uuid::Uuid::new_v4();
1955 - let hlc = Hlc { wall_ms: 5000, counter: 2, node };
1955 + let hlc = Hlc { wall_ms: 5000, counter: 2, node: synckit_client::DeviceId::new(node) };
1956 1956
1957 1957 let task_data = json!({
1958 1958 "id": task_id, "project_id": null, "description": "Idempotent",
@@ -2031,17 +2031,17 @@ use sqlx::SqlitePool;
2031 2031 let node = uuid::Uuid::new_v4();
2032 2032
2033 2033 // An honest local write at ~now is never flagged.
2034 - let local = Hlc { wall_ms: now_ms, counter: 0, node };
2034 + let local = Hlc { wall_ms: now_ms, counter: 0, node: synckit_client::DeviceId::new(node) };
2035 2035 assert!(!is_clock_poisoned(&local, now), "an at-now HLC must not be poisoned");
2036 2036
2037 2037 // Just inside the drift window is still honest (inter-device skew tolerated).
2038 - let within = Hlc { wall_ms: now_ms + MAX_HLC_DRIFT_MS - 1_000, counter: 0, node };
2038 + let within = Hlc { wall_ms: now_ms + MAX_HLC_DRIFT_MS - 1_000, counter: 0, node: synckit_client::DeviceId::new(node) };
2039 2039 assert!(!is_clock_poisoned(&within, now), "skew within the drift cap is allowed");
2040 2040
2041 2041 // A remote HLC far beyond the drift cap is poisoned. It sorts ABOVE the local
2042 2042 // write, so it would win raw LWW ordering -- but the poison guard makes it lose,
2043 2043 // so the local value survives.
2044 - let poisoned = Hlc { wall_ms: now_ms + MAX_HLC_DRIFT_MS + 60_000, counter: 0, node };
2044 + let poisoned = Hlc { wall_ms: now_ms + MAX_HLC_DRIFT_MS + 60_000, counter: 0, node: synckit_client::DeviceId::new(node) };
2045 2045 assert!(is_clock_poisoned(&poisoned, now), "a far-future remote HLC must be flagged");
2046 2046 assert!(poisoned > local, "the poisoned HLC would otherwise win raw LWW ordering");
2047 2047 }