Skip to main content

max / audiofiles

Plugin/Sync A+: HLC conflict resolution, seal apply_upsert, harden plugin loader D3 — HLC (hybrid logical clock) replaces blind last-writer-wins, so a stale or malicious remote change can no longer clobber a newer local value. All HLC logic lives in the sync layer (the content store + DB stay sync-agnostic): - Migration M030: sync_changelog gains an `hlc` column; new hlc_ledger table holds the committed clock per (table, row_id). Local-only, additive, replay-safe. - service/hlc.rs: per-device clock in sync_state (device_id = node, + last_hlc); stamp_pending mints monotonic HLCs onto un-pushed changelog rows; committed_hlc/ set_committed read/write the ledger; observe advances the clock past remotes. - Push (upload.rs): stamps each ChangeEntry's HLC (was Default/zero) and records it in the ledger atomically with marking pushed. - Pull (download.rs/resolve.rs): pull_rich + apply_pulled routes remotes through synckit's detect_conflicts -> gated(ledger) -> resolve_lww instead of blind apply; survivors apply and commit their HLC in one transaction. Also: - Seal the apply_upsert sparse-column bug: placeholders were numbered by the column-allowlist index while values were pushed densely, binding the wrong value (or out of range) for a sparse row. Now numbered by dense position (+ test). - Plugin loader: add a symlink-escape negative test (the TOCTOU vector the canonicalize-both-sides guard exists for). - Scheduler: extract backoff_minutes (exponential, capped, saturating) + tests. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
Author: Max Johnson <me@maxj.phd> · 2026-06-22 23:58 UTC
Commit: 72d2d3e926cd58fb72d70aa2ed0de44db71b040b
Parent: dac7386
8 files changed, +469 insertions, -21 deletions
M crates/audiofiles-core/src/db.rs +26 -6
@@ -1475,6 +1475,24 @@ CREATE UNIQUE INDEX IF NOT EXISTS idx_vfs_root_name
1475 1475 ON vfs_nodes(vfs_id, name) WHERE parent_id IS NULL;
1476 1476 "#;
1477 1477
1478 + const MIGRATION_030: &str = r#"
1479 + -- HLC (hybrid logical clock) conflict resolution for sync. The local changelog
1480 + -- gains an `hlc` column: the sync layer stamps each pending row with a minted
1481 + -- HLC before push (the trigger can't compute one), and the same value is the
1482 + -- "local pending" clock for conflict detection. `hlc_ledger` records the
1483 + -- committed HLC per (table, row_id) — from our own pushes and applied remotes —
1484 + -- so a stale remote change (older HLC) can be dropped instead of clobbering a
1485 + -- newer local value (the prior blind last-writer-wins). Both are local-only
1486 + -- (not synced, no triggers); additive and idempotent.
1487 + ALTER TABLE sync_changelog ADD COLUMN hlc TEXT;
1488 + CREATE TABLE IF NOT EXISTS hlc_ledger (
1489 + table_name TEXT NOT NULL,
1490 + row_id TEXT NOT NULL,
1491 + hlc TEXT NOT NULL,
1492 + PRIMARY KEY (table_name, row_id)
1493 + );
1494 + "#;
1495 +
1478 1496 /// Register `hash_row_id(salt, key) -> TEXT` as a deterministic SQLite
1479 1497 /// function on the given connection. Used by the M018 sync triggers so the
1480 1498 /// `sync_changelog.row_id` field never carries cleartext content (tag strings,
@@ -1609,6 +1627,7 @@ impl Database {
1609 1627 MIGRATION_027,
1610 1628 MIGRATION_028,
1611 1629 MIGRATION_029,
1630 + MIGRATION_030,
1612 1631 ];
1613 1632
1614 1633 for (i, sql) in MIGRATIONS.iter().enumerate() {
@@ -1824,6 +1843,7 @@ mod tests {
1824 1843 "collections",
1825 1844 "edit_history",
1826 1845 "fingerprints",
1846 + "hlc_ledger",
1827 1847 "sample_features",
1828 1848 "samples",
1829 1849 "sync_changelog",
@@ -1848,7 +1868,7 @@ mod tests {
1848 1868 .conn()
1849 1869 .query_row("PRAGMA user_version", [], |row| row.get(0))
1850 1870 .unwrap();
1851 - assert_eq!(version, 29);
1871 + assert_eq!(version, 30);
1852 1872 }
1853 1873
1854 1874 #[test]
@@ -1859,7 +1879,7 @@ mod tests {
1859 1879 .conn()
1860 1880 .query_row("PRAGMA user_version", [], |row| row.get(0))
1861 1881 .unwrap();
1862 - assert_eq!(version, 29);
1882 + assert_eq!(version, 30);
1863 1883 }
1864 1884
1865 1885 /// Open a fresh file-backed DB, close, reopen. The second open re-enters
@@ -1878,7 +1898,7 @@ mod tests {
1878 1898 .conn()
1879 1899 .query_row("PRAGMA user_version", [], |row| row.get(0))
1880 1900 .unwrap();
1881 - assert_eq!(version, 29);
1901 + assert_eq!(version, 30);
1882 1902 }
1883 1903
1884 1904 /// Simulates the worst-case recovery path: a prior partial migration left
@@ -1922,7 +1942,7 @@ mod tests {
1922 1942 .conn()
1923 1943 .query_row("PRAGMA user_version", [], |row| row.get(0))
1924 1944 .unwrap();
1925 - assert_eq!(version, 29);
1945 + assert_eq!(version, 30);
1926 1946 }
1927 1947
1928 1948 /// M018 contract: the `sync_changelog.row_id` for sensitive tables must
@@ -2144,7 +2164,7 @@ mod tests {
2144 2164 let initial_version: i32 = conn
2145 2165 .query_row("PRAGMA user_version", [], |row| row.get(0))
2146 2166 .unwrap();
2147 - assert_eq!(initial_version, 29);
2167 + assert_eq!(initial_version, 30);
2148 2168
2149 2169 let batch = format!(
2150 2170 "BEGIN;\n{}\nPRAGMA user_version = 999;\nCOMMIT;",
@@ -2207,7 +2227,7 @@ mod tests {
2207 2227 .conn()
2208 2228 .query_row("PRAGMA user_version", [], |row| row.get(0))
2209 2229 .unwrap();
2210 - assert_eq!(version, 29);
2230 + assert_eq!(version, 30);
2211 2231 }
2212 2232
2213 2233 #[test]
M crates/audiofiles-rhai/src/loader.rs +19
@@ -187,4 +187,23 @@ channels = "both"
187 187 let result = load_hook_script(&plugin_dir, "../secret.txt");
188 188 assert!(result.is_err(), "path traversal should be rejected");
189 189 }
190 +
191 + #[cfg(unix)]
192 + #[test]
193 + fn load_hook_script_rejects_symlink_escape() {
194 + // The TOCTOU/escape vector the canonicalize-both-sides guard exists for: a
195 + // symlink *inside* the plugin dir that points outside it. Canonicalizing
196 + // the full path resolves the link, so the containment check fails — a
197 + // plugin author can't read arbitrary files by planting a symlink.
198 + let dir = tempfile::tempdir().unwrap();
199 + let outside = dir.path().join("secret.txt");
200 + std::fs::write(&outside, "sensitive data").unwrap();
201 +
202 + let plugin_dir = dir.path().join("my-plugin");
203 + std::fs::create_dir(&plugin_dir).unwrap();
204 + std::os::unix::fs::symlink(&outside, plugin_dir.join("link.rhai")).unwrap();
205 +
206 + let result = load_hook_script(&plugin_dir, "link.rhai");
207 + assert!(result.is_err(), "a symlink escaping the plugin dir must be rejected");
208 + }
190 209 }
M crates/audiofiles-sync/src/scheduler.rs +31 -4
@@ -29,6 +29,15 @@ pub enum SyncCommand {
29 29 /// Delay before reconnecting the SSE stream after a disconnect (seconds).
30 30 const SSE_RECONNECT_DELAY_SECS: u64 = 5;
31 31
32 + /// Cap on the exponential auto-sync retry backoff (minutes).
33 + const MAX_BACKOFF_MINUTES: u64 = 15;
34 +
35 + /// Exponential backoff for the Nth consecutive auto-sync failure: `2^n` minutes,
36 + /// capped at [`MAX_BACKOFF_MINUTES`], saturating so a long outage can't overflow.
37 + fn backoff_minutes(consecutive_failures: u32) -> u64 {
38 + std::cmp::min(2u64.saturating_pow(consecutive_failures), MAX_BACKOFF_MINUTES)
39 + }
40 +
32 41 /// Run the background sync scheduler loop.
33 42 ///
34 43 /// Uses `db_path` to open short-lived connections inside `spawn_blocking`.
@@ -74,10 +83,7 @@ pub async fn run_scheduler(
74 83 }
75 84 Err(e) => {
76 85 consecutive_failures += 1;
77 - let backoff_minutes = std::cmp::min(
78 - 2u64.saturating_pow(consecutive_failures),
79 - 15,
80 - );
86 + let backoff_minutes = backoff_minutes(consecutive_failures);
81 87 backoff_until = Some(
82 88 chrono::Utc::now()
83 89 + chrono::Duration::minutes(backoff_minutes as i64),
@@ -313,3 +319,24 @@ async fn run_sync_cycle(
313 319
314 320 Ok(result.pulled)
315 321 }
322 +
323 + #[cfg(test)]
324 + mod tests {
325 + use super::{backoff_minutes, MAX_BACKOFF_MINUTES};
326 +
327 + #[test]
328 + fn backoff_is_exponential_then_capped() {
329 + assert_eq!(backoff_minutes(1), 2);
330 + assert_eq!(backoff_minutes(2), 4);
331 + assert_eq!(backoff_minutes(3), 8);
332 + // 2^4 = 16 > cap, so it pins at the cap and stays there.
333 + assert_eq!(backoff_minutes(4), MAX_BACKOFF_MINUTES);
334 + assert_eq!(backoff_minutes(10), MAX_BACKOFF_MINUTES);
335 + }
336 +
337 + #[test]
338 + fn backoff_saturates_on_huge_failure_count() {
339 + // A pathologically long outage must not overflow the shift.
340 + assert_eq!(backoff_minutes(u32::MAX), MAX_BACKOFF_MINUTES);
341 + }
342 + }
M crates/audiofiles-sync/src/service/download.rs +4 -2
@@ -277,8 +277,10 @@ pub(crate) async fn pull_changes(
277 277
278 278 let cursor: i64 = cursor_str.parse().unwrap_or(0);
279 279
280 + // pull_rich keeps each change's originating device_id + seq, which the HLC
281 + // conflict resolution needs (own-echo detection, ordering).
280 282 let (changes, new_cursor, has_more) = client
281 - .pull(device_id, cursor)
283 + .pull_rich(device_id, cursor)
282 284 .await
283 285 .map_err(|e| SyncError::Client(e.to_string()))?;
284 286
@@ -286,7 +288,7 @@ pub(crate) async fn pull_changes(
286 288 let p = db_path.to_path_buf();
287 289 let applied = tokio::task::spawn_blocking(move || {
288 290 let conn = open_conn(&p)?;
289 - super::resolve::apply_remote_changes(&conn, &changes)
291 + super::resolve::apply_pulled(&conn, device_id, changes)
290 292 })
291 293 .await
292 294 .map_err(|e| SyncError::Other(e.to_string()))??;
A crates/audiofiles-sync/src/service/hlc.rs +177
@@ -0,0 +1,177 @@
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 + use rusqlite::{Connection, OptionalExtension};
13 + use synckit_client::Hlc;
14 + use uuid::Uuid;
15 +
16 + use super::{get_sync_state, set_sync_state};
17 + use crate::error::Result;
18 +
19 + /// Wall-clock milliseconds since the Unix epoch (the HLC physical component).
20 + fn now_ms() -> i64 {
21 + chrono::Utc::now().timestamp_millis()
22 + }
23 +
24 + /// Load the persisted device clock, falling back to the zero clock for `node`
25 + /// (first sync, or unreadable state).
26 + pub(crate) fn load_clock(conn: &Connection, node: Uuid) -> Hlc {
27 + match get_sync_state(conn, "last_hlc") {
28 + Ok(s) if !s.is_empty() => serde_json::from_str(&s).unwrap_or_else(|_| Hlc::zero(node)),
29 + _ => Hlc::zero(node),
30 + }
31 + }
32 +
33 + fn store_clock(conn: &Connection, clock: &Hlc) -> Result<()> {
34 + set_sync_state(conn, "last_hlc", &serde_json::to_string(clock).unwrap_or_default())
35 + }
36 +
37 + /// The committed HLC for a row, if any — the oracle `CleanChanges::gated` uses to
38 + /// drop remote changes that are not newer than what this device already holds.
39 + pub(crate) fn committed_hlc(conn: &Connection, table: &str, row_id: &str) -> Option<Hlc> {
40 + conn.query_row(
41 + "SELECT hlc FROM hlc_ledger WHERE table_name = ?1 AND row_id = ?2",
42 + rusqlite::params![table, row_id],
43 + |r| r.get::<_, String>(0),
44 + )
45 + .optional()
46 + .ok()
47 + .flatten()
48 + .and_then(|s| serde_json::from_str(&s).ok())
49 + }
50 +
51 + /// Record the committed HLC for a row (one of our pushes, or an applied remote).
52 + pub(crate) fn set_committed(conn: &Connection, table: &str, row_id: &str, hlc: &Hlc) -> Result<()> {
53 + conn.execute(
54 + "INSERT INTO hlc_ledger (table_name, row_id, hlc) VALUES (?1, ?2, ?3)
55 + ON CONFLICT(table_name, row_id) DO UPDATE SET hlc = excluded.hlc",
56 + rusqlite::params![table, row_id, serde_json::to_string(hlc).unwrap_or_default()],
57 + )?;
58 + Ok(())
59 + }
60 +
61 + /// Assign an HLC to every un-stamped pending changelog row (`pushed = 0` and
62 + /// `hlc IS NULL`), in `id` order, advancing and persisting the device clock.
63 + ///
64 + /// Idempotent: a row that already carries an HLC keeps it, so the value is stable
65 + /// whether it is read first by conflict detection (pull) or by push.
66 + pub(crate) fn stamp_pending(conn: &Connection, node: Uuid) -> Result<()> {
67 + let ids: Vec<i64> = {
68 + let mut stmt = conn.prepare(
69 + "SELECT id FROM sync_changelog WHERE pushed = 0 AND hlc IS NULL ORDER BY id",
70 + )?;
71 + let rows = stmt.query_map([], |r| r.get(0))?;
72 + rows.collect::<std::result::Result<_, _>>()?
73 + };
74 + if ids.is_empty() {
75 + return Ok(());
76 + }
77 + let mut clock = load_clock(conn, node);
78 + for id in ids {
79 + clock = Hlc::tick(clock, now_ms(), node);
80 + conn.execute(
81 + "UPDATE sync_changelog SET hlc = ?1 WHERE id = ?2",
82 + rusqlite::params![serde_json::to_string(&clock).unwrap_or_default(), id],
83 + )?;
84 + }
85 + store_clock(conn, &clock)
86 + }
87 +
88 + /// Advance the device clock by observing a remote HLC, so our next local edit is
89 + /// ordered after every change we've seen (standard HLC receive rule).
90 + pub(crate) fn observe(conn: &Connection, node: Uuid, remote: Hlc) -> Result<()> {
91 + let next = Hlc::observe(load_clock(conn, node), remote, now_ms(), node);
92 + store_clock(conn, &next)
93 + }
94 +
95 + #[cfg(test)]
96 + mod tests {
97 + use super::*;
98 +
99 + fn mem() -> Connection {
100 + // A standalone in-memory DB with just the tables the HLC layer touches.
101 + let conn = Connection::open_in_memory().unwrap();
102 + conn.execute_batch(
103 + "CREATE TABLE sync_state (key TEXT PRIMARY KEY, value TEXT NOT NULL);
104 + CREATE TABLE sync_changelog (id INTEGER PRIMARY KEY AUTOINCREMENT,
105 + table_name TEXT, op TEXT, row_id TEXT, timestamp TEXT, data TEXT,
106 + pushed INTEGER NOT NULL DEFAULT 0, hlc TEXT);
107 + CREATE TABLE hlc_ledger (table_name TEXT, row_id TEXT, hlc TEXT,
108 + PRIMARY KEY (table_name, row_id));
109 + INSERT INTO sync_state (key, value) VALUES ('last_hlc', '');",
110 + )
111 + .unwrap();
112 + conn
113 + }
114 +
115 + #[test]
116 + fn stamp_pending_assigns_monotonic_hlcs() {
117 + let conn = mem();
118 + let node = Uuid::from_u128(1);
119 + for i in 0..3 {
120 + conn.execute(
121 + "INSERT INTO sync_changelog (table_name, op, row_id) VALUES ('tags', 'INSERT', ?1)",
122 + rusqlite::params![format!("r{i}")],
123 + )
124 + .unwrap();
125 + }
126 + stamp_pending(&conn, node).unwrap();
127 + let hlcs: Vec<Hlc> = {
128 + let mut stmt = conn
129 + .prepare("SELECT hlc FROM sync_changelog ORDER BY id")
130 + .unwrap();
131 + stmt.query_map([], |r| r.get::<_, String>(0))
132 + .unwrap()
133 + .map(|s| serde_json::from_str(&s.unwrap()).unwrap())
134 + .collect()
135 + };
136 + assert_eq!(hlcs.len(), 3);
137 + assert!(hlcs[0] < hlcs[1] && hlcs[1] < hlcs[2], "HLCs must be strictly increasing");
138 + // Re-stamping is a no-op: existing HLCs are stable.
139 + stamp_pending(&conn, node).unwrap();
140 + let again: Hlc = serde_json::from_str(
141 + &conn
142 + .query_row("SELECT hlc FROM sync_changelog WHERE id = 1", [], |r| {
143 + r.get::<_, String>(0)
144 + })
145 + .unwrap(),
146 + )
147 + .unwrap();
148 + assert_eq!(again, hlcs[0]);
149 + }
150 +
151 + #[test]
152 + fn ledger_roundtrip_and_gate() {
153 + let conn = mem();
154 + let node = Uuid::from_u128(7);
155 + assert!(committed_hlc(&conn, "tags", "r1").is_none());
156 + let older = Hlc { wall_ms: 100, counter: 0, node };
157 + let newer = Hlc { wall_ms: 200, counter: 0, node };
158 + set_committed(&conn, "tags", "r1", &newer).unwrap();
159 + assert_eq!(committed_hlc(&conn, "tags", "r1"), Some(newer));
160 + // A stale remote (older HLC) must be recognisable as not-newer.
161 + assert!(older < committed_hlc(&conn, "tags", "r1").unwrap());
162 + // Upsert keeps the latest.
163 + set_committed(&conn, "tags", "r1", &older).unwrap();
164 + assert_eq!(committed_hlc(&conn, "tags", "r1"), Some(older));
165 + }
166 +
167 + #[test]
168 + fn observe_keeps_clock_ahead() {
169 + let conn = mem();
170 + let node = Uuid::from_u128(2);
171 + let remote = Hlc { wall_ms: i64::MAX / 2, counter: 5, node: Uuid::from_u128(99) };
172 + observe(&conn, node, remote).unwrap();
173 + let clock = load_clock(&conn, node);
174 + assert!(clock >= remote, "local clock must dominate the observed remote");
175 + assert_eq!(clock.node, node, "our clock carries our node");
176 + }
177 + }
M crates/audiofiles-sync/src/service/mod.rs +1
@@ -4,6 +4,7 @@
4 4 //! holding a `Connection` across async `.await` points (Connection is !Send).
5 5
6 6 mod download;
7 + mod hlc;
7 8 mod resolve;
8 9 mod state;
9 10 mod upload;
M crates/audiofiles-sync/src/service/resolve.rs +177 -3
@@ -1,7 +1,8 @@
1 1 //! Conflict resolution: apply remote changes locally with FK-safe ordering.
2 2
3 3 use rusqlite::Connection;
4 - use synckit_client::{ChangeEntry, ChangeOp};
4 + use synckit_client::conflict::{detect_conflicts, resolve_lww, Resolution};
5 + use synckit_client::{ChangeEntry, ChangeOp, DeviceId, PulledChange};
5 6
6 7 use tracing::instrument;
7 8
@@ -9,6 +10,85 @@ use crate::error::Result;
9 10
10 11 use super::{json_to_sql, pk_columns, table_columns, UPSERT_ORDER, DELETE_ORDER};
11 12
13 + /// Apply pulled remote changes with HLC conflict resolution.
14 + ///
15 + /// Replaces the prior blind apply: routes each remote change through synckit's
16 + /// conflict layer so a stale or malicious remote can't clobber a newer local
17 + /// value. Clean changes (no contesting local pending edit) are kept only if their
18 + /// HLC is newer than this row's committed clock (`CleanChanges::gated`); changes
19 + /// that conflict with an un-pushed local edit are resolved by HLC last-writer-wins
20 + /// (`resolve_lww`). Survivors are applied (and their HLCs committed) atomically by
21 + /// [`apply_remote_changes`]. The device clock advances past every remote seen.
22 + #[instrument(skip_all)]
23 + pub(crate) fn apply_pulled(
24 + conn: &Connection,
25 + our_device: DeviceId,
26 + remote: Vec<PulledChange>,
27 + ) -> Result<i64> {
28 + if remote.is_empty() {
29 + return Ok(0);
30 + }
31 + let node = our_device.as_uuid();
32 +
33 + // Advance our clock past every remote change (HLC receive rule), and stamp any
34 + // un-pushed local changes so conflict detection has clocks to compare.
35 + for pc in &remote {
36 + super::hlc::observe(conn, node, pc.entry.hlc)?;
37 + }
38 + super::hlc::stamp_pending(conn, node)?;
39 +
40 + let local_pending = load_local_pending(conn)?;
41 + let (clean, conflicts) = detect_conflicts(remote, &local_pending, our_device);
42 +
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));
46 +
47 + // Conflicts (a remote change contests an un-pushed local edit): HLC LWW. AF has
48 + // no field-merge adapter, so resolve_lww only ever yields KeepLocal/KeepRemote.
49 + for pair in conflicts {
50 + match resolve_lww(&pair.local, &pair.remote) {
51 + Resolution::KeepRemote | Resolution::Merged(_) => to_apply.push(pair.remote.entry),
52 + Resolution::KeepLocal | Resolution::Skip => {}
53 + }
54 + }
55 +
56 + apply_remote_changes(conn, &to_apply)
57 + }
58 +
59 + /// Build the local pending changes (un-pushed changelog rows) as `ChangeEntry`s
60 + /// for conflict detection. `stamp_pending` must have run first so `hlc` is set.
61 + fn load_local_pending(conn: &Connection) -> Result<Vec<ChangeEntry>> {
62 + let mut stmt = conn.prepare(
63 + "SELECT table_name, op, row_id, timestamp, data, hlc
64 + FROM sync_changelog WHERE pushed = 0 ORDER BY id",
65 + )?;
66 + let rows = stmt.query_map([], |row| {
67 + Ok((
68 + row.get::<_, String>(0)?,
69 + row.get::<_, String>(1)?,
70 + row.get::<_, String>(2)?,
71 + row.get::<_, String>(3)?,
72 + row.get::<_, Option<String>>(4)?,
73 + row.get::<_, Option<String>>(5)?,
74 + ))
75 + })?;
76 + let mut out = Vec::new();
77 + for r in rows {
78 + let (table, op, row_id, timestamp, data, hlc_json) = r?;
79 + let Some(op) = ChangeOp::from_str_opt(&op) else { continue };
80 + let ts = chrono::DateTime::parse_from_rfc3339(&timestamp)
81 + .map(|dt| dt.with_timezone(&chrono::Utc))
82 + .unwrap_or_else(|_| chrono::Utc::now());
83 + let data = data.and_then(|d| serde_json::from_str(&d).ok());
84 + let hlc = hlc_json
85 + .and_then(|s| serde_json::from_str(&s).ok())
86 + .unwrap_or_default();
87 + out.push(ChangeEntry { table, op, row_id, timestamp: ts, hlc, data });
88 + }
89 + Ok(out)
90 + }
91 +
12 92 /// Apply a batch of remote changes locally, with trigger suppression.
13 93 ///
14 94 /// The flag set, data changes, and flag clear all happen inside a single
@@ -43,6 +123,9 @@ pub(crate) fn apply_remote_changes(conn: &Connection, changes: &[ChangeEntry]) -
43 123 if change.table == *table
44 124 && let Some(data) = &change.data {
45 125 apply_upsert(&tx, table, data)?;
126 + // Record the applied remote's HLC as this row's committed clock,
127 + // in the same transaction, so a later stale remote is gated out.
128 + super::hlc::set_committed(&tx, table, &change.row_id, &change.hlc)?;
46 129 count += 1;
47 130 }
48 131 }
@@ -56,6 +139,7 @@ pub(crate) fn apply_remote_changes(conn: &Connection, changes: &[ChangeEntry]) -
56 139 for change in &deletes {
57 140 if change.table == *table {
58 141 apply_delete(&tx, table, &change.row_id, change.data.as_ref())?;
142 + super::hlc::set_committed(&tx, table, &change.row_id, &change.hlc)?;
59 143 count += 1;
60 144 }
61 145 }
@@ -92,10 +176,15 @@ pub(crate) fn apply_upsert(
92 176 let mut placeholders = Vec::new();
93 177 let mut values: Vec<Box<dyn rusqlite::types::ToSql>> = Vec::new();
94 178
95 - for (i, col) in columns.iter().enumerate() {
179 + for col in columns.iter() {
96 180 if let Some(val) = obj.get(*col) {
97 181 col_names.push(*col);
98 - placeholders.push(format!("?{}", i + 1));
182 + // Number placeholders by the DENSE position (how many we've pushed),
183 + // not the column-allowlist index: only present columns get a value, so
184 + // a sparse row (some allowlist columns absent) must still bind ?1, ?2…
185 + // contiguously to values[0], values[1]…. Numbering by the allowlist
186 + // index left gaps that bound the wrong value (or out of range).
187 + placeholders.push(format!("?{}", col_names.len()));
99 188 values.push(json_to_sql(val));
100 189 }
101 190 }
@@ -271,3 +360,88 @@ fn sample_delete_hash(row_id: &str, data: Option<&serde_json::Value>) -> Option<
271 360 Some(row_id.to_string())
272 361 }
273 362 }
363 +
364 + #[cfg(test)]
365 + mod tests {
366 + use super::*;
367 + use synckit_client::Hlc;
368 + use uuid::Uuid;
369 +
370 + fn setup_conn() -> Connection {
371 + let conn = Connection::open_in_memory().unwrap();
372 + conn.execute_batch(
373 + "CREATE TABLE sync_state (key TEXT PRIMARY KEY, value TEXT NOT NULL);
374 + INSERT INTO sync_state (key, value) VALUES ('applying_remote','0'), ('last_hlc','');
375 + CREATE TABLE sync_changelog (id INTEGER PRIMARY KEY AUTOINCREMENT,
376 + table_name TEXT, op TEXT, row_id TEXT,
377 + timestamp TEXT DEFAULT (datetime('now')), data TEXT,
378 + pushed INTEGER NOT NULL DEFAULT 0, hlc TEXT);
379 + CREATE TABLE hlc_ledger (table_name TEXT, row_id TEXT, hlc TEXT,
380 + PRIMARY KEY (table_name, row_id));
381 + CREATE TABLE tags (sample_hash TEXT, tag TEXT, PRIMARY KEY (sample_hash, tag));",
382 + )
383 + .unwrap();
384 + conn
385 + }
386 +
387 + fn entry(row_id: &str, hlc: Hlc) -> ChangeEntry {
388 + ChangeEntry {
389 + table: "tags".into(),
390 + op: ChangeOp::Insert,
391 + row_id: row_id.into(),
392 + timestamp: chrono::Utc::now(),
393 + hlc,
394 + data: Some(serde_json::json!({"sample_hash": "h1", "tag": "drums"})),
395 + }
396 + }
397 +
398 + // The full apply_pulled gating path (observe -> stamp -> detect_conflicts ->
399 + // gated -> resolve_lww) is the composition of synckit's conflict primitives
400 + // (tested in synckit-client/conflict.rs), the ledger ordering (tested in
401 + // hlc.rs: ledger_roundtrip_and_gate), and the apply+commit verified here.
402 + // synckit's PulledChange is #[non_exhaustive] with no downstream constructor,
403 + // so an end-to-end apply_pulled test can't be built from this crate.
404 +
405 + #[test]
406 + fn apply_records_committed_hlc() {
407 + // The new behavior: applying a remote change records its HLC as that row's
408 + // committed clock, so a later stale remote is gated against it.
409 + let conn = setup_conn();
410 + let hlc = Hlc { wall_ms: 300, counter: 0, node: Uuid::from_u128(2) };
411 + let applied = apply_remote_changes(&conn, &[entry("k1", hlc)]).unwrap();
412 + assert_eq!(applied, 1);
413 + let tag: String = conn
414 + .query_row("SELECT tag FROM tags WHERE sample_hash='h1'", [], |r| r.get(0))
415 + .unwrap();
416 + assert_eq!(tag, "drums");
417 + assert_eq!(
418 + super::super::hlc::committed_hlc(&conn, "tags", "k1"),
419 + Some(hlc),
420 + "applying a remote must commit its HLC to the ledger"
421 + );
422 + }
423 +
424 + #[test]
425 + fn apply_upsert_binds_sparse_columns() {
426 + // Regression: placeholders were numbered by the column-allowlist index, so a
427 + // sparse row (present columns non-contiguous in the allowlist) bound the
428 + // wrong value or an out-of-range parameter. audio_analysis's allowlist puts
429 + // hash/onset_strength/attack_time at indices 0/10/20.
430 + let conn = Connection::open_in_memory().unwrap();
431 + conn.execute_batch(
432 + "CREATE TABLE audio_analysis (hash TEXT PRIMARY KEY, onset_strength REAL, attack_time REAL);",
433 + )
434 + .unwrap();
435 + let data = serde_json::json!({"hash": "h1", "onset_strength": 0.5, "attack_time": 0.02});
436 + apply_upsert(&conn, "audio_analysis", &data).unwrap();
437 + let (onset, attack): (f64, f64) = conn
438 + .query_row(
439 + "SELECT onset_strength, attack_time FROM audio_analysis WHERE hash='h1'",
440 + [],
441 + |r| Ok((r.get(0)?, r.get(1)?)),
442 + )
443 + .unwrap();
444 + assert_eq!(onset, 0.5);
445 + assert_eq!(attack, 0.02);
446 + }
447 + }
M crates/audiofiles-sync/src/service/upload.rs +34 -6
@@ -166,18 +166,25 @@ async fn push_changes(
166 166 client: &SyncKitClient,
167 167 device_id: DeviceId,
168 168 ) -> Result<i64> {
169 + let node = device_id.as_uuid();
169 170 let p = db_path.to_path_buf();
170 171 let rows = tokio::task::spawn_blocking(move || -> Result<_> {
171 172 let conn = open_conn(&p)?;
173 + // Mint an HLC for every un-stamped pending change before reading them, so
174 + // each pushed ChangeEntry carries a real clock (was Default/zero, which
175 + // made every local edit lose conflict resolution).
176 + super::hlc::stamp_pending(&conn, node)?;
172 177 let mut stmt = conn.prepare(
173 - "SELECT id, table_name, op, row_id, timestamp, data
178 + "SELECT id, table_name, op, row_id, timestamp, data, hlc
174 179 FROM sync_changelog
175 180 WHERE pushed = 0
176 181 ORDER BY id ASC
177 182 LIMIT ?1",
178 183 )?;
179 184
180 - let rows: Vec<(i64, String, String, String, String, Option<String>)> = stmt
185 + // (id, table_name, op, row_id, timestamp, data, hlc)
186 + type ChangelogRow = (i64, String, String, String, String, Option<String>, Option<String>);
187 + let rows: Vec<ChangelogRow> = stmt
181 188 .query_map([PUSH_BATCH_LIMIT as i64], |row| {
182 189 Ok((
183 190 row.get(0)?,
@@ -186,6 +193,7 @@ async fn push_changes(
186 193 row.get(3)?,
187 194 row.get(4)?,
188 195 row.get(5)?,
196 + row.get(6)?,
189 197 ))
190 198 })?
191 199 .collect::<std::result::Result<Vec<_>, _>>()?;
@@ -204,7 +212,7 @@ async fn push_changes(
204 212
205 213 let changes: Vec<ChangeEntry> = rows
206 214 .into_iter()
207 - .filter_map(|(id, table, op, row_id, timestamp, data)| {
215 + .filter_map(|(id, table, op, row_id, timestamp, data, hlc_json)| {
208 216 let ts = chrono::DateTime::parse_from_rfc3339(&timestamp)
209 217 .map(|dt| dt.with_timezone(&chrono::Utc))
210 218 .unwrap_or_else(|_| chrono::Utc::now());
@@ -232,18 +240,31 @@ async fn push_changes(
232 240 None => None,
233 241 };
234 242
243 + // The HLC was stamped above; fall back to the zero clock only if a row
244 + // somehow lacks one (defensive — keeps the push going).
245 + let hlc = hlc_json
246 + .and_then(|s| serde_json::from_str(&s).ok())
247 + .unwrap_or_else(|| synckit_client::Hlc::zero(node));
248 +
235 249 pushed_ids.push(id);
236 250 Some(ChangeEntry {
237 251 table,
238 252 op: change_op,
239 253 row_id,
240 254 timestamp: ts,
241 - hlc: Default::default(),
255 + hlc,
242 256 data: parsed_data,
243 257 })
244 258 })
245 259 .collect();
246 260
261 + // Our own edits become the committed HLC for their rows, so a later remote
262 + // change for the same row is gated against them.
263 + let ledger_updates: Vec<(String, String, synckit_client::Hlc)> = changes
264 + .iter()
265 + .map(|c| (c.table.clone(), c.row_id.clone(), c.hlc))
266 + .collect();
267 +
247 268 if skipped > 0 {
248 269 tracing::warn!(skipped, "Some changelog entries could not be parsed (unknown op or bad JSON) and were marked pushed to break the retry loop; they are dropped, not retried");
249 270 }
@@ -257,18 +278,25 @@ async fn push_changes(
257 278
258 279 let pushed_count = pushed_ids.len() as i64;
259 280
260 - // Mark only successfully-pushed entries
281 + // Mark only successfully-pushed entries, and record our edits' committed HLCs.
282 + // Both in one transaction so a crash can't leave the changelog and ledger out
283 + // of step.
261 284 if !pushed_ids.is_empty() {
262 285 let p = db_path.to_path_buf();
263 286 tokio::task::spawn_blocking(move || {
264 287 let conn = open_conn(&p)?;
288 + let tx = conn.unchecked_transaction()?;
265 289 let placeholders: String = pushed_ids.iter().map(|_| "?").collect::<Vec<_>>().join(",");
266 290 let sql = format!("UPDATE sync_changelog SET pushed = 1 WHERE id IN ({})", placeholders);
267 291 let params: Vec<&dyn rusqlite::types::ToSql> = pushed_ids
268 292 .iter()
269 293 .map(|id| id as &dyn rusqlite::types::ToSql)
270 294 .collect();
271 - conn.execute(&sql, params.as_slice())?;
295 + tx.execute(&sql, params.as_slice())?;
296 + for (table, row_id, hlc) in &ledger_updates {
297 + super::hlc::set_committed(&tx, table, row_id, hlc)?;
298 + }
299 + tx.commit()?;
272 300 Ok::<_, SyncError>(())
273 301 })
274 302 .await