Skip to main content

max / audiofiles

10.9 KB · 273 lines History Blame Raw
1 //! Hybrid logical clock (HLC) state for sync conflict resolution.
2 //!
3 //! AF stamps each local pending change with a minted HLC and records the
4 //! committed HLC per `(table, row_id)` in `hlc_ledger`, so synckit's conflict
5 //! resolution can drop a stale remote change instead of clobbering a newer local
6 //! value (replacing the prior blind last-writer-wins in `apply_remote_changes`).
7 //!
8 //! All HLC logic lives here in the sync layer; the content store and DB schema
9 //! stay sync-agnostic. The device's `node` is its registered `device_id`; the
10 //! running clock is persisted as `sync_state.last_hlc`.
11 //!
12 //! Convergence rests on `device_id` (the HLC `node`) being unique per install: it
13 //! is the final, deterministic tiebreak when two HLCs share `wall_ms`+`counter`.
14 //! A restored backup that copies the `device_id` violates this; synckit mitigates
15 //! the exact-HLC collision with a canonical-payload tiebreak, but unique
16 //! `device_id`s are the design assumption.
17
18 use rusqlite::{Connection, OptionalExtension};
19 use synckit_client::{DeviceId, Hlc};
20
21 use super::{get_sync_state, set_sync_state};
22 use crate::error::Result;
23
24 /// Wall-clock milliseconds since the Unix epoch (the HLC physical component).
25 fn now_ms() -> i64 {
26 chrono::Utc::now().timestamp_millis()
27 }
28
29 /// Load the persisted device clock. Returns the zero clock on first sync (no
30 /// persisted value); on a *corrupt* persisted value it seeds from physical time
31 /// rather than zero, see below.
32 pub(crate) fn load_clock(conn: &Connection, node: DeviceId) -> Hlc {
33 match get_sync_state(conn, "last_hlc") {
34 Ok(s) if !s.is_empty() => serde_json::from_str(&s).unwrap_or_else(|e| {
35 // Corrupt persisted clock: do NOT reset to zero. Zero sends the clock
36 // backward, so a subsequent local edit could mint an HLC that loses to
37 // an already-committed row, the one place HLC monotonicity isn't
38 // structurally guaranteed. Seed from physical time instead (forward of
39 // any sanely-timestamped past HLC); `observe()` advances it past
40 // anything seen on the next pull.
41 tracing::warn!("corrupt last_hlc ({e}); seeding clock from physical time");
42 Hlc::tick(Hlc::zero(node), now_ms(), node)
43 }),
44 _ => Hlc::zero(node), // first sync: genuinely no prior clock
45 }
46 }
47
48 fn store_clock(conn: &Connection, clock: &Hlc) -> Result<()> {
49 // Hlc is a plain derive-Serialize struct, so serialization cannot fail; make
50 // that an invariant (panic on the impossible) rather than silently persisting
51 // an empty string that load_clock would then treat as a corrupt clock.
52 let json = serde_json::to_string(clock).expect("Hlc always serializes");
53 set_sync_state(conn, "last_hlc", &json)
54 }
55
56 /// The committed HLC for a row, if any, the oracle `CleanChanges::gated` uses to
57 /// drop remote changes that are not newer than what this device already holds.
58 ///
59 /// A genuine query error propagates. It was previously swallowed as "no committed
60 /// HLC", which is the one answer that gates a remote change *in* (`is_none_or`), so
61 /// a failed ledger read could let a stale remote value clobber a newer local one.
62 /// A *corrupt* stored HLC still degrades to `None` with a warning, matching how
63 /// `load_clock` treats a corrupt `last_hlc`, one bad ledger row must not wedge the
64 /// entire sync.
65 pub(crate) fn committed_hlc(conn: &Connection, table: &str, row_id: &str) -> Result<Option<Hlc>> {
66 let stored: Option<String> = conn
67 .query_row(
68 "SELECT hlc FROM hlc_ledger WHERE table_name = ?1 AND row_id = ?2",
69 rusqlite::params![table, row_id],
70 |r| r.get::<_, String>(0),
71 )
72 .optional()?;
73 Ok(stored.and_then(|s| match serde_json::from_str(&s) {
74 Ok(hlc) => Some(hlc),
75 Err(e) => {
76 tracing::warn!("corrupt committed HLC for {table}/{row_id} ({e}); treating as absent");
77 None
78 }
79 }))
80 }
81
82 /// Record the committed HLC for a row (one of our pushes, or an applied remote).
83 pub(crate) fn set_committed(conn: &Connection, table: &str, row_id: &str, hlc: &Hlc) -> Result<()> {
84 conn.execute(
85 "INSERT INTO hlc_ledger (table_name, row_id, hlc) VALUES (?1, ?2, ?3)
86 ON CONFLICT(table_name, row_id) DO UPDATE SET hlc = excluded.hlc",
87 rusqlite::params![
88 table,
89 row_id,
90 serde_json::to_string(hlc).expect("Hlc always serializes")
91 ],
92 )?;
93 Ok(())
94 }
95
96 /// Assign an HLC to every un-stamped pending changelog row (`pushed = 0` and
97 /// `hlc IS NULL`), in `id` order, advancing and persisting the device clock.
98 ///
99 /// Idempotent: a row that already carries an HLC keeps it, so the value is stable
100 /// whether it is read first by conflict detection (pull) or by push.
101 pub(crate) fn stamp_pending(conn: &Connection, node: DeviceId) -> Result<()> {
102 let ids: Vec<i64> = {
103 let mut stmt = conn.prepare(
104 "SELECT id FROM sync_changelog WHERE pushed = 0 AND hlc IS NULL ORDER BY id",
105 )?;
106 let rows = stmt.query_map([], |r| r.get(0))?;
107 rows.collect::<std::result::Result<_, _>>()?
108 };
109 if ids.is_empty() {
110 return Ok(());
111 }
112 let mut clock = load_clock(conn, node);
113 for id in ids {
114 clock = Hlc::tick(clock, now_ms(), node);
115 conn.execute(
116 "UPDATE sync_changelog SET hlc = ?1 WHERE id = ?2",
117 rusqlite::params![
118 serde_json::to_string(&clock).expect("Hlc always serializes"),
119 id
120 ],
121 )?;
122 }
123 store_clock(conn, &clock)
124 }
125
126 /// Advance the device clock by observing a remote HLC, so our next local edit is
127 /// ordered after every change we've seen (standard HLC receive rule).
128 pub(crate) fn observe(conn: &Connection, node: DeviceId, remote: Hlc) -> Result<()> {
129 let next = Hlc::observe(load_clock(conn, node), remote, now_ms(), node);
130 store_clock(conn, &next)
131 }
132
133 #[cfg(test)]
134 mod tests {
135 use super::*;
136 use uuid::Uuid;
137
138 fn mem() -> Connection {
139 // A standalone in-memory DB with just the tables the HLC layer touches.
140 let conn = Connection::open_in_memory().unwrap();
141 conn.execute_batch(
142 "CREATE TABLE sync_state (key TEXT PRIMARY KEY, value TEXT NOT NULL);
143 CREATE TABLE sync_changelog (id INTEGER PRIMARY KEY AUTOINCREMENT,
144 table_name TEXT, op TEXT, row_id TEXT, timestamp TEXT, data TEXT,
145 pushed INTEGER NOT NULL DEFAULT 0, hlc TEXT);
146 CREATE TABLE hlc_ledger (table_name TEXT, row_id TEXT, hlc TEXT,
147 PRIMARY KEY (table_name, row_id));
148 INSERT INTO sync_state (key, value) VALUES ('last_hlc', '');",
149 )
150 .unwrap();
151 conn
152 }
153
154 #[test]
155 fn stamp_pending_assigns_monotonic_hlcs() {
156 let conn = mem();
157 let node = DeviceId::new(Uuid::from_u128(1));
158 for i in 0..3 {
159 conn.execute(
160 "INSERT INTO sync_changelog (table_name, op, row_id) VALUES ('tags', 'INSERT', ?1)",
161 rusqlite::params![format!("r{i}")],
162 )
163 .unwrap();
164 }
165 stamp_pending(&conn, node).unwrap();
166 let hlcs: Vec<Hlc> = {
167 let mut stmt = conn
168 .prepare("SELECT hlc FROM sync_changelog ORDER BY id")
169 .unwrap();
170 stmt.query_map([], |r| r.get::<_, String>(0))
171 .unwrap()
172 .map(|s| serde_json::from_str(&s.unwrap()).unwrap())
173 .collect()
174 };
175 assert_eq!(hlcs.len(), 3);
176 assert!(
177 hlcs[0] < hlcs[1] && hlcs[1] < hlcs[2],
178 "HLCs must be strictly increasing"
179 );
180 // Re-stamping is a no-op: existing HLCs are stable.
181 stamp_pending(&conn, node).unwrap();
182 let again: Hlc = serde_json::from_str(
183 &conn
184 .query_row("SELECT hlc FROM sync_changelog WHERE id = 1", [], |r| {
185 r.get::<_, String>(0)
186 })
187 .unwrap(),
188 )
189 .unwrap();
190 assert_eq!(again, hlcs[0]);
191 }
192
193 #[test]
194 fn corrupt_clock_seeds_forward_not_zero() {
195 let conn = mem();
196 let node = DeviceId::new(Uuid::from_u128(1));
197 // First sync (empty) is genuinely the zero clock.
198 assert_eq!(load_clock(&conn, node), Hlc::zero(node));
199 // A corrupt persisted clock must NOT reset to zero (that moves the clock
200 // backward); it seeds from physical time, so it's well ahead of zero.
201 set_sync_state(&conn, "last_hlc", "{not valid json").unwrap();
202 let recovered = load_clock(&conn, node);
203 assert!(
204 recovered.wall_ms > 0 && recovered > Hlc::zero(node),
205 "corrupt clock seeded backward: {recovered:?}"
206 );
207 }
208
209 #[test]
210 fn ledger_roundtrip_and_gate() {
211 let conn = mem();
212 let node = DeviceId::new(Uuid::from_u128(7));
213 assert!(committed_hlc(&conn, "tags", "r1").unwrap().is_none());
214 let older = Hlc {
215 wall_ms: 100,
216 counter: 0,
217 node,
218 };
219 let newer = Hlc {
220 wall_ms: 200,
221 counter: 0,
222 node,
223 };
224 set_committed(&conn, "tags", "r1", &newer).unwrap();
225 assert_eq!(committed_hlc(&conn, "tags", "r1").unwrap(), Some(newer));
226 // A stale remote (older HLC) must be recognisable as not-newer.
227 assert!(older < committed_hlc(&conn, "tags", "r1").unwrap().unwrap());
228 // Upsert keeps the latest.
229 set_committed(&conn, "tags", "r1", &older).unwrap();
230 assert_eq!(committed_hlc(&conn, "tags", "r1").unwrap(), Some(older));
231 }
232
233 #[test]
234 fn committed_hlc_propagates_query_error() {
235 // A genuine read failure must surface as an error, not collapse to "no
236 // committed HLC" (which gates a stale remote change *in*).
237 let conn = mem();
238 conn.execute_batch("DROP TABLE hlc_ledger").unwrap();
239 assert!(committed_hlc(&conn, "tags", "r1").is_err());
240 }
241
242 #[test]
243 fn committed_hlc_degrades_corrupt_row_to_none() {
244 // A corrupt stored HLC is not a query error: it degrades to None (with a
245 // warning) so one bad ledger row can't wedge the whole sync.
246 let conn = mem();
247 conn.execute(
248 "INSERT INTO hlc_ledger (table_name, row_id, hlc) VALUES ('tags', 'r1', ?1)",
249 rusqlite::params!["{not valid json"],
250 )
251 .unwrap();
252 assert_eq!(committed_hlc(&conn, "tags", "r1").unwrap(), None);
253 }
254
255 #[test]
256 fn observe_keeps_clock_ahead() {
257 let conn = mem();
258 let node = DeviceId::new(Uuid::from_u128(2));
259 let remote = Hlc {
260 wall_ms: i64::MAX / 2,
261 counter: 5,
262 node: DeviceId::new(Uuid::from_u128(99)),
263 };
264 observe(&conn, node, remote).unwrap();
265 let clock = load_clock(&conn, node);
266 assert!(
267 clock >= remote,
268 "local clock must dominate the observed remote"
269 );
270 assert_eq!(clock.node, node, "our clock carries our node");
271 }
272 }
273