Skip to main content

max / audiofiles

Propagate committed_hlc read errors instead of swallowing them committed_hlc did .optional().ok().flatten(): a genuine hlc_ledger query error collapsed to None, and None is the one answer that gates a remote change *in* (is_none_or), so a failed read could let a stale remote value clobber a newer local one. Return Result<Option<Hlc>> and propagate the query error; the resolve path captures it out of gated's infallible oracle (via a Cell) and aborts before applying anything. A corrupt stored HLC still degrades to None with a warning, matching load_clock's handling of a corrupt last_hlc, so one bad row can't wedge the whole sync. Adds tests for both paths. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Author: Max Johnson <me@maxj.phd> · 2026-07-23 20:48 UTC
Commit: 61240f874d924d4547e33dd3be4446febaee4064
Parent: 72be856
3 files changed, +63 insertions, -18 deletions
M crates/audiofiles-sync/src/service/hlc.rs +48 -14
@@ -55,16 +55,28 @@ fn store_clock(conn: &Connection, clock: &Hlc) -> Result<()> {
55 55
56 56 /// The committed HLC for a row, if any — the oracle `CleanChanges::gated` uses to
57 57 /// drop remote changes that are not newer than what this device already holds.
58 - pub(crate) fn committed_hlc(conn: &Connection, table: &str, row_id: &str) -> Option<Hlc> {
59 - conn.query_row(
60 - "SELECT hlc FROM hlc_ledger WHERE table_name = ?1 AND row_id = ?2",
61 - rusqlite::params![table, row_id],
62 - |r| r.get::<_, String>(0),
63 - )
64 - .optional()
65 - .ok()
66 - .flatten()
67 - .and_then(|s| serde_json::from_str(&s).ok())
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 + }))
68 80 }
69 81
70 82 /// Record the committed HLC for a row (one of our pushes, or an applied remote).
@@ -198,7 +210,7 @@ mod tests {
198 210 fn ledger_roundtrip_and_gate() {
199 211 let conn = mem();
200 212 let node = DeviceId::new(Uuid::from_u128(7));
201 - assert!(committed_hlc(&conn, "tags", "r1").is_none());
213 + assert!(committed_hlc(&conn, "tags", "r1").unwrap().is_none());
202 214 let older = Hlc {
203 215 wall_ms: 100,
204 216 counter: 0,
@@ -210,12 +222,34 @@ mod tests {
210 222 node,
211 223 };
212 224 set_committed(&conn, "tags", "r1", &newer).unwrap();
213 - assert_eq!(committed_hlc(&conn, "tags", "r1"), Some(newer));
225 + assert_eq!(committed_hlc(&conn, "tags", "r1").unwrap(), Some(newer));
214 226 // A stale remote (older HLC) must be recognisable as not-newer.
215 - assert!(older < committed_hlc(&conn, "tags", "r1").unwrap());
227 + assert!(older < committed_hlc(&conn, "tags", "r1").unwrap().unwrap());
216 228 // Upsert keeps the latest.
217 229 set_committed(&conn, "tags", "r1", &older).unwrap();
218 - assert_eq!(committed_hlc(&conn, "tags", "r1"), Some(older));
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);
219 253 }
220 254
221 255 #[test]
M crates/audiofiles-sync/src/service/resolve.rs +14 -3
@@ -41,8 +41,19 @@ pub(crate) fn apply_pulled(
41 41 let (clean, conflicts) = detect_conflicts(remote, &local_pending, our_device);
42 42
43 43 // Clean: keep only changes strictly newer than the row's committed HLC.
44 - let mut to_apply: Vec<ChangeEntry> =
45 - clean.gated(|table, row_id| super::hlc::committed_hlc(conn, table, row_id));
44 + // `gated`'s oracle is infallible, so capture any ledger-read error out of the
45 + // closure and abort before applying anything, rather than letting a failed read
46 + // masquerade as "no committed HLC" (which would gate the stale change *in*).
47 + let hlc_err: std::cell::Cell<Option<SyncError>> = std::cell::Cell::new(None);
48 + let mut to_apply: Vec<ChangeEntry> = clean.gated(|table, row_id| {
49 + super::hlc::committed_hlc(conn, table, row_id).unwrap_or_else(|e| {
50 + hlc_err.set(Some(e));
51 + None
52 + })
53 + });
54 + if let Some(e) = hlc_err.into_inner() {
55 + return Err(e);
56 + }
46 57
47 58 // Conflicts (a remote change contests an un-pushed local edit): HLC LWW. AF has
48 59 // no field-merge adapter, so resolve_lww only ever yields KeepLocal/KeepRemote.
@@ -590,7 +601,7 @@ mod tests {
590 601 .unwrap();
591 602 assert_eq!(tag, "drums");
592 603 assert_eq!(
593 - super::super::hlc::committed_hlc(&conn, "tags", "k1"),
604 + super::super::hlc::committed_hlc(&conn, "tags", "k1").unwrap(),
594 605 Some(hlc),
595 606 "applying a remote must commit its HLC to the ledger"
596 607 );
M crates/audiofiles-sync/src/service/upload.rs +1 -1
@@ -597,7 +597,7 @@ mod tests {
597 597
598 598 assert_eq!(pushed_flag(conn, id), 1);
599 599 assert_eq!(
600 - super::super::hlc::committed_hlc(conn, "tags", "k1"),
600 + super::super::hlc::committed_hlc(conn, "tags", "k1").unwrap(),
601 601 Some(clock),
602 602 "our own edit's HLC must be committed so a later stale remote is gated"
603 603 );