Skip to main content

max / audiofiles

Sync: collapse pulled changes to max-HLC per row before apply Ultra-fuzz run-6 remediation, Sync & Plugins axis. Within one pull batch the SyncKit server orders changes by arrival seq, not HLC (the HLC lives inside the E2E-encrypted payload the server can't read), and apply_remote_changes applies in that order. CleanChanges::gated compares every clean change against the same pre-batch committed clock, so two changes to the same (table, row_id) from two devices can both survive the gate -- and the later-in-seq one would clobber the other regardless of HLC. A re-pull from cursor 0 could then yield a different final value than the incremental path. apply_pulled now collapses to_apply to the max-HLC entry per (table, row_id) before applying, so last-writer-wins is decided by the clock, not arrival order, and convergence holds across incremental vs full re-pull. PulledChange is #[non_exhaustive] with no downstream constructor, so the fix is unit-tested at the collapse helper (ChangeEntry is constructible): the higher-HLC entry wins regardless of vec order, and distinct rows are preserved. 63 sync tests pass (2 new); clippy clean. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
Author: Max Johnson <me@maxj.phd> · 2026-06-23 15:40 UTC
Commit: 0e3cf3a86c731299d074ac0f6fbf3a2df7e6f175
Parent: d07a34c
1 file changed, +61 insertions, -0 deletions
M crates/audiofiles-sync/src/service/resolve.rs +61
@@ -53,9 +53,39 @@ pub(crate) fn apply_pulled(
53 53 }
54 54 }
55 55
56 + // Collapse to the max-HLC entry per (table, row_id). Within one pull batch the
57 + // server orders changes by arrival `seq`, not HLC (the HLC lives inside the
58 + // E2E-encrypted payload the server can't read), and apply_remote_changes
59 + // applies in that order. Gating compares every clean change against the same
60 + // pre-batch committed clock, so two changes to the same row from two devices
61 + // can both survive — and the later-in-seq one would win regardless of HLC.
62 + // Deciding the survivor by HLC here makes last-writer-wins convergent: a
63 + // re-pull from cursor 0 yields the same final value as the incremental path.
64 + let to_apply = collapse_to_max_hlc(to_apply);
65 +
56 66 apply_remote_changes(conn, &to_apply)
57 67 }
58 68
69 + /// Keep only the highest-HLC change per `(table, row_id)`. `Hlc` orders
70 + /// lexicographically on `(wall_ms, counter, node)`, so the winner is identical
71 + /// on every device. Order of the result does not matter: `apply_remote_changes`
72 + /// re-sorts by `UPSERT_ORDER`/`DELETE_ORDER`, and there is now at most one entry
73 + /// per row.
74 + fn collapse_to_max_hlc(changes: Vec<ChangeEntry>) -> Vec<ChangeEntry> {
75 + use std::collections::HashMap;
76 + let mut best: HashMap<(String, String), ChangeEntry> = HashMap::new();
77 + for change in changes {
78 + let key = (change.table.clone(), change.row_id.clone());
79 + match best.get(&key) {
80 + Some(existing) if existing.hlc >= change.hlc => {}
81 + _ => {
82 + best.insert(key, change);
83 + }
84 + }
85 + }
86 + best.into_values().collect()
87 + }
88 +
59 89 /// Build the local pending changes (un-pushed changelog rows) as `ChangeEntry`s
60 90 /// for conflict detection. `stamp_pending` must have run first so `hlc` is set.
61 91 fn load_local_pending(conn: &Connection) -> Result<Vec<ChangeEntry>> {
@@ -403,6 +433,37 @@ mod tests {
403 433 // so an end-to-end apply_pulled test can't be built from this crate.
404 434
405 435 #[test]
436 + fn collapse_keeps_max_hlc_regardless_of_order() {
437 + // Two changes to the same row within one batch (e.g. two devices), the
438 + // HLC-newer one arriving FIRST in seq order. Collapsing must keep the
439 + // newer HLC, not the later-in-vec entry.
440 + let older = Hlc { wall_ms: 100, counter: 0, node: Uuid::from_u128(1) };
441 + let newer = Hlc { wall_ms: 200, counter: 0, node: Uuid::from_u128(2) };
442 +
443 + let mut newer_entry = entry("k1", newer);
444 + newer_entry.data = Some(serde_json::json!({"sample_hash": "h1", "tag": "newer"}));
445 + let mut older_entry = entry("k1", older);
446 + older_entry.data = Some(serde_json::json!({"sample_hash": "h1", "tag": "older"}));
447 +
448 + // newer first, older second: naive last-wins would pick "older".
449 + let out = collapse_to_max_hlc(vec![newer_entry.clone(), older_entry.clone()]);
450 + assert_eq!(out.len(), 1);
451 + assert_eq!(out[0].hlc, newer);
452 +
453 + // Reverse order: same winner.
454 + let out = collapse_to_max_hlc(vec![older_entry, newer_entry]);
455 + assert_eq!(out.len(), 1);
456 + assert_eq!(out[0].hlc, newer);
457 + }
458 +
459 + #[test]
460 + fn collapse_preserves_distinct_rows() {
461 + let h = Hlc { wall_ms: 100, counter: 0, node: Uuid::from_u128(1) };
462 + let out = collapse_to_max_hlc(vec![entry("k1", h), entry("k2", h)]);
463 + assert_eq!(out.len(), 2);
464 + }
465 +
466 + #[test]
406 467 fn apply_records_committed_hlc() {
407 468 // The new behavior: applying a remote change records its HLC as that row's
408 469 // committed clock, so a later stale remote is gated against it.