Skip to main content

max / synckit

Carry the pipeline invariant in a type resolve_pull established one entry per row and apply_remote_changes depended on it, but nothing connected the two. The invariant lived in the order the pipeline's steps were written in, and applying a batch straight off the wire, skipping conflict detection, the committed gate and the collapse, was a function call that compiled. resolve_pull now returns ResolvedChanges and apply_remote_changes takes nothing else. Its field is private to store::hlc, so the pipeline is the only constructor outside tests. Same technique as CleanChanges one layer up. What the type guarantees depends on the strategy, and the doc says so rather than overclaiming: HybridLogicalClock gives exactly one entry per row, ServerOrder deliberately does not, because last-delivered-wins is what an app choosing it asked for. That asymmetry was not in the write-up and is the reason the type documents two contracts instead of one. record_committed keeps taking a slice. The ledger also records this device's own pushed edits, which never pass through the pull pipeline, so requiring the resolved type there would seal the wrong door. Tested at the pipeline level: the same two same-row changes resolve to one entry under HLC and stay two under ServerOrder. Verified by breaking each half separately (dropping the collapse, then collapsing ServerOrder too) and confirming each failed on its own assertion. The seal itself was verified by trying to construct a ResolvedChanges in store::sync and confirming it does not compile.
Co-Authored-By
Claude Opus 5 (1M context) <noreply@anthropic.com>
Author: Max Johnson <me@maxj.phd> · 2026-08-05 16:05 UTC
Signed with PGP, not checked
Commit: 5b1a9b3d4a35105ed7c4905dbb51271dc0f5fabd
Parent: 43e5812
5 files changed, +152 insertions, -19 deletions
M synckit-client/src/store/apply.rs +8 -2
@@ -21,6 +21,7 @@
21 21 use tracing::warn;
22 22
23 23 use super::db::with_applying_remote;
24 + use super::hlc::ResolvedChanges;
24 25 use super::migrate::row_id_expr;
25 26 use super::schema::{DeleteMode, SyncMode, SyncSchema, SyncTable};
26 27 use crate::error::Result;
@@ -110,10 +111,11 @@
110 111 pub fn apply_remote_changes(
111 112 conn: &mut Connection,
112 113 schema: &SyncSchema,
113 - changes: &[ChangeEntry],
114 + changes: &ResolvedChanges,
114 115 scope: &str,
115 116 ) -> Result<ApplyOutcome> {
116 117 let by_name: HashMap<&str, &SyncTable> = schema.tables.iter().map(|t| (t.name, t)).collect();
118 + let changes = changes.as_slice();
117 119
118 120 let fk_off = changes.iter().any(|c| {
119 121 by_name
@@ -770,8 +772,12 @@
770 772 conn
771 773 }
772 774
775 + /// These tests exercise the applier, not the pipeline that feeds it, so they
776 + /// build the resolved batch directly rather than routing every case through
777 + /// `resolve_pull`.
773 778 fn apply(conn: &mut Connection, changes: &[ChangeEntry]) -> ApplyOutcome {
774 - apply_remote_changes(conn, &schema(), changes, "").unwrap()
779 + let changes = ResolvedChanges::for_test(changes.to_vec());
780 + apply_remote_changes(conn, &schema(), &changes, "").unwrap()
775 781 }
776 782
777 783 #[test]
M synckit-client/src/store/config.rs +7 -6
@@ -96,6 +96,7 @@
96 96 mod tests {
97 97 use super::*;
98 98 use crate::store::apply::apply_remote_changes;
99 + use crate::store::hlc::ResolvedChanges;
99 100 use crate::store::schema::SyncSchema;
100 101 use crate::types::{ChangeEntry, ChangeOp, hlc_legacy_floor};
101 102 use synckit_config::{ConfigStore, Posture};
@@ -226,7 +227,7 @@
226 227 let outcome = apply_remote_changes(
227 228 &mut conn,
228 229 &SyncSchema::new(vec![config_sync_table(&SPEC)]),
229 - &[inbound("theme", "akari-night")],
230 + &ResolvedChanges::for_test(vec![inbound("theme", "akari-night")]),
230 231 "",
231 232 )
232 233 .unwrap();
@@ -251,7 +252,7 @@
251 252 let outcome = apply_remote_changes(
252 253 &mut conn,
253 254 &SyncSchema::new(vec![config_sync_table(&SPEC)]),
254 - &[inbound("mirror_path", "/tmp/attacker")],
255 + &ResolvedChanges::for_test(vec![inbound("mirror_path", "/tmp/attacker")]),
255 256 "",
256 257 )
257 258 .unwrap();
@@ -277,7 +278,7 @@
277 278 let outcome = apply_remote_changes(
278 279 &mut conn,
279 280 &SyncSchema::new(vec![config_sync_table(&SPEC)]),
280 - &[inbound("some_new_local_path", "/etc/secret")],
281 + &ResolvedChanges::for_test(vec![inbound("some_new_local_path", "/etc/secret")]),
281 282 "",
282 283 )
283 284 .unwrap();
@@ -303,10 +304,10 @@
303 304 let outcome = apply_remote_changes(
304 305 &mut conn,
305 306 &SyncSchema::new(vec![config_sync_table(&SPEC)]),
306 - &[ChangeEntry {
307 + &ResolvedChanges::for_test(vec![ChangeEntry {
307 308 op: ChangeOp::Delete,
308 309 ..inbound("mirror_path", "/home/max/samples")
309 - }],
310 + }]),
310 311 "",
311 312 )
312 313 .unwrap();
@@ -332,7 +333,7 @@
332 333 let outcome = apply_remote_changes(
333 334 &mut conn,
334 335 &SyncSchema::new(vec![config_sync_table(&SPEC)]),
335 - &[inbound("theme", "akari-night")],
336 + &ResolvedChanges::for_test(vec![inbound("theme", "akari-night")]),
336 337 "",
337 338 )
338 339 .unwrap();
M synckit-client/src/store/hlc.rs +134 -8
@@ -9,9 +9,11 @@
9 9 //! - the **committed ledger** (`sync_committed_hlc`), the HLC last applied for a
10 10 //! row, which the gate uses to drop a re-pulled older change.
11 11 //!
12 - //! [`resolve_pull`] is the entry point: it turns a pulled batch into the resolved
13 - //! `Vec<ChangeEntry>` the apply engine writes, dispatching on the schema's
14 - //! [`ConflictStrategy`].
12 + //! [`resolve_pull`] is the entry point: it turns a pulled batch into the
13 + //! [`ResolvedChanges`] the apply engine writes, dispatching on the schema's
14 + //! [`ConflictStrategy`]. That return type is the pipeline's boundary: the apply
15 + //! engine takes nothing else, so a batch cannot reach the database without
16 + //! having been resolved.
15 17
16 18 use std::collections::HashMap;
17 19
@@ -147,6 +149,10 @@
147 149 }
148 150
149 151 /// Advance the committed ledger for a batch of applied entries.
152 + ///
153 + /// Takes a plain slice rather than [`ResolvedChanges`]: the ledger also records
154 + /// this device's own pushed edits, which never pass through the pull pipeline,
155 + /// so requiring the resolved type here would be sealing the wrong door.
150 156 pub fn record_committed(conn: &Connection, entries: &[ChangeEntry]) -> Result<()> {
151 157 for e in entries {
152 158 set_committed(conn, &e.table, &e.row_id, &e.hlc)?;
@@ -156,6 +162,79 @@
156 162
157 163 // pull resolution
158 164
165 + /// Changes that have been through [`resolve_pull`], and the only thing
166 + /// [`apply_remote_changes`](super::apply::apply_remote_changes) accepts.
167 + ///
168 + /// The point is what it makes unrepresentable. Applying a batch straight off
169 + /// the wire, skipping conflict detection, the committed-HLC gate and the
170 + /// collapse, used to be a plain function call that compiled; the invariant
171 + /// lived in the order the pipeline's steps happened to be written in, and
172 + /// nothing carried it to the consumers that depend on it. Now the only way to
173 + /// obtain one is to run the pipeline. Same technique as
174 + /// [`CleanChanges`](crate::conflict::CleanChanges) one layer up, and as MNW's
175 + /// `S3DeleteAuthority`.
176 + ///
177 + /// What it guarantees depends on the strategy that produced it, and the
178 + /// difference is deliberate:
179 + ///
180 + /// - [`ConflictStrategy::HybridLogicalClock`]: **exactly one entry per
181 + /// `(table, row_id)`**, the winner under
182 + /// [`change_order`](crate::conflict::change_order). This is the one the
183 + /// consumers care about. Applying two entries for one row would make the
184 + /// final value depend on apply order, and the committed ledger, which
185 + /// advances only, would record the highest HLC of the batch rather than the
186 + /// HLC of the value actually written, leaving the two disagreeing about what
187 + /// this device holds.
188 + /// - [`ConflictStrategy::ServerOrder`]: entries verbatim in server sequence
189 + /// order, which **may hold several per row**. That is the whole meaning of
190 + /// the strategy (last delivered wins), so it cannot be collapsed without
191 + /// changing what an app opted into. Its consumers are safe for a different
192 + /// reason: applying in order leaves the last one standing, and the committed
193 + /// ledger it writes is never read back, because only the HLC strategy gates
194 + /// against it.
195 + #[derive(Debug, Clone)]
196 + pub struct ResolvedChanges(Vec<ChangeEntry>);
197 +
198 + impl ResolvedChanges {
199 + /// The resolved entries, in apply order.
200 + pub fn iter(&self) -> impl Iterator<Item = &ChangeEntry> {
201 + self.0.iter()
202 + }
203 +
204 + /// The resolved entries as a slice, for callers that only read them.
205 + pub fn as_slice(&self) -> &[ChangeEntry] {
206 + &self.0
207 + }
208 +
209 + /// How many changes will be applied.
210 + pub fn len(&self) -> usize {
211 + self.0.len()
212 + }
213 +
214 + /// Whether the pull resolved to nothing at all, every change gated out or
215 + /// lost its conflict.
216 + pub fn is_empty(&self) -> bool {
217 + self.0.is_empty()
218 + }
219 +
220 + /// Build one directly, for tests that exercise a consumer rather than the
221 + /// pipeline that feeds it. Test-only on purpose: outside tests the pipeline
222 + /// is the only constructor, which is the entire point of the type.
223 + #[cfg(test)]
224 + pub(crate) fn for_test(entries: Vec<ChangeEntry>) -> Self {
225 + ResolvedChanges(entries)
226 + }
227 + }
228 +
229 + impl<'a> IntoIterator for &'a ResolvedChanges {
230 + type Item = &'a ChangeEntry;
231 + type IntoIter = std::slice::Iter<'a, ChangeEntry>;
232 +
233 + fn into_iter(self) -> Self::IntoIter {
234 + self.0.iter()
235 + }
236 + }
237 +
159 238 /// Turn a pulled batch into the resolved changes to apply, per the schema's
160 239 /// [`ConflictStrategy`].
161 240 ///
@@ -183,9 +262,11 @@
183 262 pulled: Vec<PulledChange>,
184 263 now: DateTime<Utc>,
185 264 scope: &str,
186 - ) -> Result<Vec<ChangeEntry>> {
265 + ) -> Result<ResolvedChanges> {
187 266 match schema.conflict {
188 - ConflictStrategy::ServerOrder => Ok(pulled.into_iter().map(|p| p.entry).collect()),
267 + ConflictStrategy::ServerOrder => Ok(ResolvedChanges(
268 + pulled.into_iter().map(|p| p.entry).collect(),
269 + )),
189 270 ConflictStrategy::HybridLogicalClock => {
190 271 let now_ms = now.timestamp_millis();
191 272 observe(conn, node, pulled.iter().map(|p| p.entry.hlc), now_ms)?;
@@ -253,7 +334,7 @@
253 334 tracing::warn!("could not trim the conflict stash: {e}");
254 335 }
255 336
256 - Ok(collapse_max_hlc(resolved))
337 + Ok(ResolvedChanges(collapse_max_hlc(resolved)))
257 338 }
258 339 }
259 340 }
@@ -532,7 +613,7 @@
532 613 let now = Utc::now();
533 614 let resolved = resolve_pull(conn, s, node, pulled, now, "").unwrap();
534 615 apply_remote_changes(conn, s, &resolved, "").unwrap();
535 - record_committed(conn, &resolved).unwrap();
616 + record_committed(conn, resolved.as_slice()).unwrap();
536 617 }
537 618
538 619 #[test]
@@ -666,6 +747,51 @@
666 747 );
667 748 }
668 749
750 + /// The two guarantees `ResolvedChanges` documents, asserted on the same
751 + /// input so the difference between them is the only variable. Under the HLC
752 + /// strategy a batch carrying two changes for one row resolves to one entry;
753 + /// under `ServerOrder` it deliberately stays two, because last-delivered-wins
754 + /// is what an app choosing that strategy asked for.
755 + #[test]
756 + fn resolve_pull_collapses_a_row_under_hlc_and_does_not_under_server_order() {
757 + let (src, sn) = device(2);
758 + let first = local_edit_as_pulled(&src, sn, "r", "first", 100, 1);
759 + let second = local_edit_as_pulled(&src, sn, "r", "second", 200, 2);
760 +
761 + let (hlc_device, hn) = device(1);
762 + let under_hlc = resolve_pull(
763 + &hlc_device,
764 + &schema(),
765 + hn,
766 + vec![first.clone(), second.clone()],
767 + Utc::now(),
768 + "",
769 + )
770 + .unwrap();
771 + assert_eq!(
772 + under_hlc.len(),
773 + 1,
774 + "the HLC strategy promises one entry per row; the apply order would \
775 + otherwise decide the value"
776 + );
777 +
778 + let (server_device, svn) = device(3);
779 + let under_server_order = resolve_pull(
780 + &server_device,
781 + &server_order_schema(),
782 + svn,
783 + vec![first, second],
784 + Utc::now(),
785 + "",
786 + )
787 + .unwrap();
788 + assert_eq!(
789 + under_server_order.len(),
790 + 2,
791 + "ServerOrder must not collapse: last delivered wins is the strategy"
792 + );
793 + }
794 +
669 795 // ── Conflict stash ──
670 796 //
671 797 // LWW always discards one side; these pin that the discarded bytes are kept
@@ -832,7 +958,7 @@
832 958 )
833 959 .unwrap();
834 960 apply_remote_changes(&mut conn, &schema(), &resolved, "").unwrap();
835 - record_committed(&conn, &resolved).unwrap();
961 + record_committed(&conn, resolved.as_slice()).unwrap();
836 962 assert!(stash_rows(&conn).is_empty(), "nothing lost yet");
837 963
838 964 // Now pull an older change for the same row. No pending edit contests it.
M synckit-client/src/store/mod.rs +2 -2
@@ -43,8 +43,8 @@
43 43 pub use deferred::{HeldEntry, HoldCounts, HoldState, MAX_ATTEMPTS};
44 44 pub use facade::{SchedulerHandle, SyncConfig, SyncOutcome, SyncStore, SyncStoreBuilder};
45 45 pub use hlc::{
46 - committed_hlc, load_clock, observe, record_committed, resolve_pull, set_committed,
47 - stamp_pending,
46 + ResolvedChanges, committed_hlc, load_clock, observe, record_committed, resolve_pull,
47 + set_committed, stamp_pending,
48 48 };
49 49 pub use scheduler::{NoopObserver, SyncObserver, SyncState};
50 50 pub use schema::{ConflictStrategy, DeleteMode, RowIdScheme, SyncMode, SyncSchema, SyncTable};
M synckit-client/src/store/sync.rs +1 -1
@@ -547,7 +547,7 @@
547 547 .chain(outcome.rejected.iter())
548 548 .map(|u| (u.table.clone(), u.row_id.clone()))
549 549 .collect();
550 - for entry in &resolved {
550 + for entry in resolved.iter() {
551 551 if !unapplied.contains(&deferred::key_of(entry)) {
552 552 set_committed(conn, &entry.table, &entry.row_id, &entry.hlc)?;
553 553 }