Skip to main content

max / audiofiles

sync: adopt synckit 0.6 DeviceId-typed HLC node Lockstep update for the synckit-client breaking change. Type the sync HLC layer on DeviceId end to end: load_clock/stamp_pending/observe now take a DeviceId (dropping the as_uuid() downcast at the resolve.rs call site); wrap the remaining Uuid boundaries in upload.rs; a pending row with no HLC floors to Hlc::zero(DeviceId::nil()) instead of the removed Hlc::default(). No behavior change. Gate: 88 sync tests + full workspace check + 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: b8036fe149f03ef5042e0c859afc4a22775c3fd2
Parent: 65498ed
4 files changed, +26 insertions, -24 deletions
M crates/audiofiles-sync/src/service/hlc.rs +10 -10
@@ -16,8 +16,7 @@
16 16 //! `device_id`s are the design assumption.
17 17
18 18 use rusqlite::{Connection, OptionalExtension};
19 - use synckit_client::Hlc;
20 - use uuid::Uuid;
19 + use synckit_client::{DeviceId, Hlc};
21 20
22 21 use super::{get_sync_state, set_sync_state};
23 22 use crate::error::Result;
@@ -30,7 +29,7 @@ fn now_ms() -> i64 {
30 29 /// Load the persisted device clock. Returns the zero clock on first sync (no
31 30 /// persisted value); on a *corrupt* persisted value it seeds from physical time
32 31 /// rather than zero — see below.
33 - pub(crate) fn load_clock(conn: &Connection, node: Uuid) -> Hlc {
32 + pub(crate) fn load_clock(conn: &Connection, node: DeviceId) -> Hlc {
34 33 match get_sync_state(conn, "last_hlc") {
35 34 Ok(s) if !s.is_empty() => serde_json::from_str(&s).unwrap_or_else(|e| {
36 35 // Corrupt persisted clock: do NOT reset to zero. Zero sends the clock
@@ -87,7 +86,7 @@ pub(crate) fn set_committed(conn: &Connection, table: &str, row_id: &str, hlc: &
87 86 ///
88 87 /// Idempotent: a row that already carries an HLC keeps it, so the value is stable
89 88 /// whether it is read first by conflict detection (pull) or by push.
90 - pub(crate) fn stamp_pending(conn: &Connection, node: Uuid) -> Result<()> {
89 + pub(crate) fn stamp_pending(conn: &Connection, node: DeviceId) -> Result<()> {
91 90 let ids: Vec<i64> = {
92 91 let mut stmt = conn.prepare(
93 92 "SELECT id FROM sync_changelog WHERE pushed = 0 AND hlc IS NULL ORDER BY id",
@@ -114,7 +113,7 @@ pub(crate) fn stamp_pending(conn: &Connection, node: Uuid) -> Result<()> {
114 113
115 114 /// Advance the device clock by observing a remote HLC, so our next local edit is
116 115 /// ordered after every change we've seen (standard HLC receive rule).
117 - pub(crate) fn observe(conn: &Connection, node: Uuid, remote: Hlc) -> Result<()> {
116 + pub(crate) fn observe(conn: &Connection, node: DeviceId, remote: Hlc) -> Result<()> {
118 117 let next = Hlc::observe(load_clock(conn, node), remote, now_ms(), node);
119 118 store_clock(conn, &next)
120 119 }
@@ -122,6 +121,7 @@ pub(crate) fn observe(conn: &Connection, node: Uuid, remote: Hlc) -> Result<()>
122 121 #[cfg(test)]
123 122 mod tests {
124 123 use super::*;
124 + use uuid::Uuid;
125 125
126 126 fn mem() -> Connection {
127 127 // A standalone in-memory DB with just the tables the HLC layer touches.
@@ -142,7 +142,7 @@ mod tests {
142 142 #[test]
143 143 fn stamp_pending_assigns_monotonic_hlcs() {
144 144 let conn = mem();
145 - let node = Uuid::from_u128(1);
145 + let node = DeviceId::new(Uuid::from_u128(1));
146 146 for i in 0..3 {
147 147 conn.execute(
148 148 "INSERT INTO sync_changelog (table_name, op, row_id) VALUES ('tags', 'INSERT', ?1)",
@@ -178,7 +178,7 @@ mod tests {
178 178 #[test]
179 179 fn corrupt_clock_seeds_forward_not_zero() {
180 180 let conn = mem();
181 - let node = Uuid::from_u128(1);
181 + let node = DeviceId::new(Uuid::from_u128(1));
182 182 // First sync (empty) is genuinely the zero clock.
183 183 assert_eq!(load_clock(&conn, node), Hlc::zero(node));
184 184 // A corrupt persisted clock must NOT reset to zero (that moves the clock
@@ -194,7 +194,7 @@ mod tests {
194 194 #[test]
195 195 fn ledger_roundtrip_and_gate() {
196 196 let conn = mem();
197 - let node = Uuid::from_u128(7);
197 + let node = DeviceId::new(Uuid::from_u128(7));
198 198 assert!(committed_hlc(&conn, "tags", "r1").is_none());
199 199 let older = Hlc { wall_ms: 100, counter: 0, node };
200 200 let newer = Hlc { wall_ms: 200, counter: 0, node };
@@ -210,8 +210,8 @@ mod tests {
210 210 #[test]
211 211 fn observe_keeps_clock_ahead() {
212 212 let conn = mem();
213 - let node = Uuid::from_u128(2);
214 - let remote = Hlc { wall_ms: i64::MAX / 2, counter: 5, node: Uuid::from_u128(99) };
213 + let node = DeviceId::new(Uuid::from_u128(2));
214 + let remote = Hlc { wall_ms: i64::MAX / 2, counter: 5, node: DeviceId::new(Uuid::from_u128(99)) };
215 215 observe(&conn, node, remote).unwrap();
216 216 let clock = load_clock(&conn, node);
217 217 assert!(clock >= remote, "local clock must dominate the observed remote");
M crates/audiofiles-sync/src/service/resolve.rs +9 -7
@@ -28,7 +28,7 @@ pub(crate) fn apply_pulled(
28 28 if remote.is_empty() {
29 29 return Ok(0);
30 30 }
31 - let node = our_device.as_uuid();
31 + let node = our_device;
32 32
33 33 // Advance our clock past every remote change (HLC receive rule), and stamp any
34 34 // un-pushed local changes so conflict detection has clocks to compare.
@@ -111,9 +111,11 @@ fn load_local_pending(conn: &Connection) -> Result<Vec<ChangeEntry>> {
111 111 .map(|dt| dt.with_timezone(&chrono::Utc))
112 112 .unwrap_or_else(|_| chrono::Utc::now());
113 113 let data = data.and_then(|d| serde_json::from_str(&d).ok());
114 + // A pending row lacking an HLC is a legacy/pre-HLC entry: floor it (nil
115 + // node, always loses) — it is re-stamped by `stamp_pending` before push.
114 116 let hlc = hlc_json
115 117 .and_then(|s| serde_json::from_str(&s).ok())
116 - .unwrap_or_default();
118 + .unwrap_or_else(|| synckit_client::Hlc::zero(synckit_client::DeviceId::nil()));
117 119 out.push(ChangeEntry { table, op, row_id, timestamp: ts, hlc, data, extra: Default::default() });
118 120 }
119 121 Ok(out)
@@ -523,8 +525,8 @@ mod tests {
523 525 // Two changes to the same row within one batch (e.g. two devices), the
524 526 // HLC-newer one arriving FIRST in seq order. Collapsing must keep the
525 527 // newer HLC, not the later-in-vec entry.
526 - let older = Hlc { wall_ms: 100, counter: 0, node: Uuid::from_u128(1) };
527 - let newer = Hlc { wall_ms: 200, counter: 0, node: Uuid::from_u128(2) };
528 + let older = Hlc { wall_ms: 100, counter: 0, node: DeviceId::new(Uuid::from_u128(1)) };
529 + let newer = Hlc { wall_ms: 200, counter: 0, node: DeviceId::new(Uuid::from_u128(2)) };
528 530
529 531 let mut newer_entry = entry("k1", newer);
530 532 newer_entry.data = Some(serde_json::json!({"sample_hash": "h1", "tag": "newer"}));
@@ -544,7 +546,7 @@ mod tests {
544 546
545 547 #[test]
546 548 fn collapse_preserves_distinct_rows() {
547 - let h = Hlc { wall_ms: 100, counter: 0, node: Uuid::from_u128(1) };
549 + let h = Hlc { wall_ms: 100, counter: 0, node: DeviceId::new(Uuid::from_u128(1)) };
548 550 let out = collapse_to_max_hlc(vec![entry("k1", h), entry("k2", h)]);
549 551 assert_eq!(out.len(), 2);
550 552 }
@@ -554,7 +556,7 @@ mod tests {
554 556 // The new behavior: applying a remote change records its HLC as that row's
555 557 // committed clock, so a later stale remote is gated against it.
556 558 let conn = setup_conn();
557 - let hlc = Hlc { wall_ms: 300, counter: 0, node: Uuid::from_u128(2) };
559 + let hlc = Hlc { wall_ms: 300, counter: 0, node: DeviceId::new(Uuid::from_u128(2)) };
558 560 let applied = apply_remote_changes(&conn, &[entry("k1", hlc)]).unwrap();
559 561 assert_eq!(applied, 1);
560 562 let tag: String = conn
@@ -692,7 +694,7 @@ mod tests {
692 694 )
693 695 .unwrap();
694 696
695 - let hlc = Hlc { wall_ms: 100, counter: 0, node: Uuid::from_u128(1) };
697 + let hlc = Hlc { wall_ms: 100, counter: 0, node: DeviceId::new(Uuid::from_u128(1)) };
696 698 let mk = |sample_hash: &str, tag: &str| ChangeEntry {
697 699 table: "tags".into(),
698 700 op: ChangeOp::Insert,
M crates/audiofiles-sync/src/service/state.rs +2 -2
@@ -271,7 +271,7 @@ mod tests {
271 271 op,
272 272 row_id: row_id.to_string(),
273 273 timestamp: chrono::Utc::now(),
274 - hlc: Default::default(),
274 + hlc: synckit_client::Hlc::zero(synckit_client::DeviceId::nil()),
275 275 data,
276 276 extra: Default::default(),
277 277 }
@@ -1612,7 +1612,7 @@ mod tests {
1612 1612 },
1613 1613 row_id,
1614 1614 timestamp: chrono::Utc::now(),
1615 - hlc: Default::default(),
1615 + hlc: synckit_client::Hlc::zero(synckit_client::DeviceId::nil()),
1616 1616 data: data.and_then(|d| serde_json::from_str(&d).ok()),
1617 1617 extra: Default::default(),
1618 1618 })
M crates/audiofiles-sync/src/service/upload.rs +5 -5
@@ -249,7 +249,7 @@ fn prepare_batch(rows: Vec<ChangelogRow>, node: Uuid) -> PreparedBatch {
249 249 // if a row somehow lacks one (defensive — keeps the push going).
250 250 let hlc = hlc_json
251 251 .and_then(|s| serde_json::from_str(&s).ok())
252 - .unwrap_or_else(|| synckit_client::Hlc::zero(node));
252 + .unwrap_or_else(|| synckit_client::Hlc::zero(DeviceId::new(node)));
253 253
254 254 pushed_ids.push(id);
255 255 Some(ChangeEntry {
@@ -313,7 +313,7 @@ async fn push_one_batch(
313 313 // Mint an HLC for every un-stamped pending change before reading them, so
314 314 // each pushed ChangeEntry carries a real clock (was Default/zero, which
315 315 // made every local edit lose conflict resolution).
316 - super::hlc::stamp_pending(&conn, node)?;
316 + super::hlc::stamp_pending(&conn, DeviceId::new(node))?;
317 317 let mut stmt = conn.prepare(
318 318 "SELECT id, table_name, op, row_id, timestamp, data, hlc
319 319 FROM sync_changelog
@@ -385,7 +385,7 @@ mod tests {
385 385 use synckit_client::Hlc;
386 386
387 387 fn hlc(wall_ms: i64, node: Uuid) -> Hlc {
388 - Hlc { wall_ms, counter: 0, node }
388 + Hlc { wall_ms, counter: 0, node: DeviceId::new(node) }
389 389 }
390 390
391 391 /// (id, table, op, row_id, timestamp, data, hlc) with sane defaults.
@@ -465,8 +465,8 @@ mod tests {
465 465 ];
466 466 let b = prepare_batch(rows, node);
467 467 assert_eq!(b.changes.len(), 2);
468 - assert_eq!(b.changes[0].hlc, Hlc::zero(node));
469 - assert_eq!(b.changes[1].hlc, Hlc::zero(node));
468 + assert_eq!(b.changes[0].hlc, Hlc::zero(DeviceId::new(node)));
469 + assert_eq!(b.changes[1].hlc, Hlc::zero(DeviceId::new(node)));
470 470 }
471 471
472 472 #[test]