max / synckit

Wire field merge: last-synced snapshots, per-table opt-in, counters excluded resolve_field_merge has always been able to merge two edits that touched different columns and nothing could call it, because a three-way merge needs the version both devices started from and the engine kept none. Under plain LWW one whole edit wins and the loser goes to the stash, so on a wide table (GO's tasks has 29 synced columns) most of what the stash collects is not conflict but collateral, and that volume is what stops anyone reading it. Adds sync_row_snapshot, the last-synced payload per row, written the two moments a row becomes common ground: a remote change applied, and a local change the server acknowledged. Payload rather than a row read, since a row carries preserve_local secrets and group provenance the wire never sends and the merge would report those as changes. Opt-in per table via SyncTable::field_merge(counters). A table that does not opt in stores no snapshot and resolves exactly as before. The counter list is an argument rather than a flag on purpose. A merge works on values: base 0 and two absolutes of 30 cannot tell it two devices each added thirty minutes, so it would keep 30 where the user did 60. A conflict that moved a declared counter on both sides refuses to merge and falls back to LWW-and-stash, where the loss stays visible and recoverable. Every other fallback (no opt-in, clock-poisoned side, no object payload, no base) also lands on today's path, so a merge is never a weaker guarantee than LWW. A field both sides moved to different values still costs one side its value, so the loser is stashed as LWW would have stashed it; a merge that contested nothing stashes nothing. push_scope/push_changes now take &SyncSchema for the push-side re-base.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

Author: Max Johnson <me@maxj.phd> · 2026-08-06 19:48 UTC

Signed with PGP, not checked

Commit: 0c6a52c9ada80415cba33ba407caf90ef2e6ca9e

Parent: 92358af

12 files changed, +962 insertions, -57 deletions

M synckit-client/docs/architecture.md +34

			@@ -459,6 +459,7 @@
459	459		\| Row-id privacy hashing on the wire \| AF content tables (`hash_row_id(salt,…)`) \| `RowIdScheme::Hashed` \|
460	460		\| Exclude some rows/keys from sync \| AF `user_config` `sync_*`, `loose_files` \| `exclude_where: Option<&str>` (a SQL predicate, see AF example; must be SQL because it also compiles into the generated trigger's WHEN clause, not just the apply guard) \|
461	461		\| Conflict model \| GO/AF (HLC), BB (server-order) \| `ConflictStrategy::{HybridLogicalClock, ServerOrder}` \|
	462	+	\| Merge edits to different columns of one row \| wide tables (GO `tasks` has 29 synced columns) \| `field_merge(&[counter_cols])` \|
462	463		\| Blob-bearing tables \| AF only \| `BlobPolicy` trait (below) \|
463	464		\| Retention cap, cleanup window, batch, interval \| all (different numbers) \| `SyncConfig` \|
464	465		\| Device-name source, status sink, error mapping \| all (framework-specific) \| `SyncObserver` trait \|
			@@ -625,6 +626,39 @@
625	626		Default `HybridLogicalClock`; BB can adopt it (it is the roadmap intent anyway) or
626	627		stay on `ServerOrder` during migration.
627	628
	629	+	### Field merge, per table
	630	+
	631	+	Under plain LWW one whole edit wins and the other goes to the conflict stash. On a
	632	+	wide table that is mostly wrong: GO's `tasks` has 29 synced columns, so two edits
	633	+	landing on the same one is the uncommon case, and the stash fills with edits that
	634	+	never conflicted. Volume is what stops anyone reading it.
	635	+
	636	+	A table opts out of that with `field_merge`:
	637	+
	638	+	```rust
	639	+	SyncTable::full("tasks", TASK_COLUMNS)
	640	+	.field_merge(&["actual_minutes"]) // the counter columns, checked below
	641	+	```
	642	+
	643	+	That stores a last-synced snapshot per row (`sync_row_snapshot`, roughly doubling
	644	+	the table's storage) and resolves its conflicts with `resolve_field_merge` against
	645	+	that base. Fields only one side changed carry across; a field both sides changed
	646	+	still goes to the HLC winner and the loser is still stashed. A table that does not
	647	+	opt in is untouched: no snapshot rows, no behaviour change.
	648	+
	649	+	Two things to check before opting a table in, neither of which the engine can see:
	650	+
	651	+	- Counter columns, written as `SET n = n + ?`. A merge works on values, so a
	652	+	base of `0` and two absolutes of `30` cannot tell it two devices each added
	653	+	thirty minutes. It would keep `30` where the user did `60`. Declaring them makes
	654	+	a conflict that moved one refuse to merge and fall back to LWW-and-stash, where
	655	+	the loss is at least visible. This is why `field_merge` takes the list as an
	656	+	argument rather than being a bare flag.
	657	+	- Columns only valid together, like a `status` and the `completed_at` derived
	658	+	from it. If one write site sets `status` without `completed_at`, a merge can take
	659	+	each from a different device and produce a pair neither one wrote. Enumerate the
	660	+	table's writers and make dependent columns move together at every site first.
	661	+
628	662		## Blob policy
629	663
630	664		Only audiofiles syncs blobs, but the shape is general: some rows own a

M synckit-client/src/conflict.rs +80 -44

			@@ -448,50 +448,8 @@
448	448		};
449	449		};
450	450
451		-	// Compute diffs: keys where local/remote differ from base
452		-	let mut local_changed: HashMap<&str, Option<&serde_json::Value>> = HashMap::new();
453		-	let mut remote_changed: HashMap<&str, Option<&serde_json::Value>> = HashMap::new();
454		-
455		-	// Check keys in base for changes or deletions
456		-	for key in base_obj.keys() {
457		-	let base_val = &base_obj[key];
458		-
459		-	match local_obj.get(key) {
460		-	Some(local_val) if local_val != base_val => {
461		-	local_changed.insert(key, Some(local_val));
462		-	}
463		-	None => {
464		-	// Key deleted on local side
465		-	local_changed.insert(key, None);
466		-	}
467		-	_ => {}
468		-	}
469		-
470		-	match remote_obj.get(key) {
471		-	Some(remote_val) if remote_val != base_val => {
472		-	remote_changed.insert(key, Some(remote_val));
473		-	}
474		-	None => {
475		-	// Key deleted on remote side
476		-	remote_changed.insert(key, None);
477		-	}
478		-	_ => {}
479		-	}
480		-	}
481		-
482		-	// Check for new keys added by local (not in base)
483		-	for (key, val) in local_obj {
484		-	if !base_obj.contains_key(key) {
485		-	local_changed.insert(key, Some(val));
486		-	}
487		-	}
488		-
489		-	// Check for new keys added by remote (not in base)
490		-	for (key, val) in remote_obj {
491		-	if !base_obj.contains_key(key) {
492		-	remote_changed.insert(key, Some(val));
493		-	}
494		-	}
	451	+	let local_changed = diff_against_base(local_obj, base_obj);
	452	+	let remote_changed = diff_against_base(remote_obj, base_obj);
495	453
496	454		// Build merged result starting from base
497	455		let mut result = base_obj.clone();
			@@ -540,6 +498,84 @@
540	498		Resolution::Merged(serde_json::Value::Object(result))
541	499		}
542	500
	501	+	/// One side's top-level diff against the base: every key it changed, added, or
	502	+	/// dropped, mapped to its new value (`None` for a key the side no longer carries).
	503	+	///
	504	+	/// Factored out rather than inlined because [`contested_fields`] has to compute
	505	+	/// the same diff to answer which fields both sides moved, and the two answers
	506	+	/// disagreeing would mean a caller refusing (or permitting) a merge on a
	507	+	/// different set of keys than the merge itself acts on.
	508	+	type SideDiff<'a> = HashMap<&'a str, Option<&'a serde_json::Value>>;
	509	+
	510	+	fn diff_against_base<'a>(
	511	+	side: &'a serde_json::Map<String, serde_json::Value>,
	512	+	base: &'a serde_json::Map<String, serde_json::Value>,
	513	+	) -> SideDiff<'a> {
	514	+	let mut changed: SideDiff<'a> = HashMap::new();
	515	+	for (key, base_val) in base {
	516	+	match side.get(key) {
	517	+	Some(val) if val != base_val => {
	518	+	changed.insert(key, Some(val));
	519	+	}
	520	+	// Key dropped on this side.
	521	+	None => {
	522	+	changed.insert(key, None);
	523	+	}
	524	+	_ => {}
	525	+	}
	526	+	}
	527	+	for (key, val) in side {
	528	+	if !base.contains_key(key) {
	529	+	changed.insert(key, Some(val));
	530	+	}
	531	+	}
	532	+	changed
	533	+	}
	534	+
	535	+	/// The top-level keys both sides moved away from `base`: the fields a
	536	+	/// [`resolve_field_merge`] has to hand to a winner rather than simply carrying
	537	+	/// across.
	538	+	///
	539	+	/// Exists because the merged object cannot tell its caller what it had to decide.
	540	+	/// Two callers need that. The counter rule, since a base plus two absolute totals
	541	+	/// cannot be merged at all, and a row whose counter both devices moved has to
	542	+	/// fall back to LWW-and-stash instead. And the stash, since a merge that
	543	+	/// contested nothing lost nothing and filling the stash with non-conflicts is the
	544	+	/// exact failure field merge exists to fix.
	545	+	///
	546	+	/// Both sides moving a field to the same value is still reported here, and
	547	+	/// that is not an oversight: it is the shape of the counter failure, where two
	548	+	/// devices each add thirty minutes and arrive at the identical total. A caller
	549	+	/// that only cares whether a value was lost compares the two sides' values for
	550	+	/// the returned keys; the counter rule must not.
	551	+	///
	552	+	/// Returns empty when any input is not a JSON object: with no usable base there
	553	+	/// is no diff to take, and the merge falls back to LWW on its own.
	554	+	pub fn contested_fields(
	555	+	local: &serde_json::Value,
	556	+	remote: &serde_json::Value,
	557	+	base: &serde_json::Value,
	558	+	) -> Vec<String> {
	559	+	let (Some(local_obj), Some(remote_obj), Some(base_obj)) =
	560	+	(local.as_object(), remote.as_object(), base.as_object())
	561	+	else {
	562	+	return Vec::new();
	563	+	};
	564	+	let local_changed = diff_against_base(local_obj, base_obj);
	565	+	let remote_changed = diff_against_base(remote_obj, base_obj);
	566	+
	567	+	let mut contested: Vec<String> = local_changed
	568	+	.keys()
	569	+	.filter(\|k\| remote_changed.contains_key(*k))
	570	+	.map(\|k\| (*k).to_string())
	571	+	.collect();
	572	+	// HashMap iteration order is not stable, and this is logged and matched
	573	+	// against a declared counter list; an unstable order would read differently
	574	+	// on every run.
	575	+	contested.sort();
	576	+	contested
	577	+	}
	578	+
543	579		#[cfg(test)]
544	580		mod tests {
545	581		use super::*;

M synckit-client/src/lib.rs +1 -1

			@@ -90,7 +90,7 @@
90	90		SecretToken, SessionInfo, SyncKitClient, SyncKitConfig, SyncNotifyStream, validate_api_key,
91	91		};
92	92		pub use conflict::{
93		-	CleanChanges, ConflictPair, ConflictResolver, Resolution, detect_conflicts,
	93	+	CleanChanges, ConflictPair, ConflictResolver, Resolution, contested_fields, detect_conflicts,
94	94		resolve_field_merge, resolve_lww,
95	95		};
96	96		pub use error::{Result, SyncKitError};

M synckit-client/src/store/apply.rs +41 -1

			@@ -403,7 +403,7 @@
403	403		if is_excluded(tx, table, Some(data))? {
404	404		return Ok(RowOutcome::Filtered);
405	405		}
406		-	match &table.mode {
	406	+	let outcome = match &table.mode {
407	407		SyncMode::PartialUpdate { set } => apply_partial_update(tx, table, data, set),
408	408		SyncMode::Full => {
409	409		let nn = not_null
			@@ -411,6 +411,31 @@
411	411		.expect("Full table not_null precomputed");
412	412		apply_full_upsert(tx, table, data, nn, scope)
413	413		}
	414	+	}?;
	415	+
	416	+	// A remote change this device just wrote is, by construction, a version both
	417	+	// devices have seen: it is the base a later three-way merge measures both
	418	+	// sides against. Recorded only for tables that opted into field merge, and
	419	+	// only for a row that actually landed, so a rejected row does not re-base the
	420	+	// merge onto a version the database never held.
	421	+	if table.field_merge && matches!(outcome, RowOutcome::Applied) {
	422	+	record_snapshot(tx, table, &change.row_id, data);
	423	+	}
	424	+	Ok(outcome)
	425	+	}
	426	+
	427	+	/// Record a merge base, logging rather than failing.
	428	+	///
	429	+	/// A missing base is not a correctness problem: `resolve_field_merge` treats an
	430	+	/// absent base as "no usable base" and falls back to LWW, the behaviour of every
	431	+	/// table that never opted in. Failing the whole apply to protect an optimisation
	432	+	/// would trade a worse outcome for a better one.
	433	+	fn record_snapshot(tx: &Transaction<'_>, table: &SyncTable, row_id: &str, data: &Value) {
	434	+	if let Err(e) = super::snapshot::record(tx, table.name, row_id, data) {
	435	+	warn!(
	436	+	table = table.name,
	437	+	row_id, "could not record the field-merge base: {e}"
	438	+	);
414	439		}
415	440		}
416	441
			@@ -607,6 +632,21 @@
607	632		DeleteMode::Ignore => unreachable!("returned above"),
608	633		};
609	634		exec(tx, &sql, &params)?;
	635	+
	636	+	// The row is gone (or tombstoned, which leaves nothing the wire describes),
	637	+	// so its merge base describes a version that no longer exists. Dropped for
	638	+	// tombstones too: a delete payload carries only the primary key, so there is
	639	+	// no post-delete base to record in its place, and a stale one would be handed
	640	+	// to a merge for whatever next occupies the key.
	641	+	if table.field_merge
	642	+	&& let Err(e) = super::snapshot::forget(tx, table.name, &change.row_id)
	643	+	{
	644	+	warn!(
	645	+	table = table.name,
	646	+	row_id = %change.row_id,
	647	+	"could not drop the field-merge base of a deleted row: {e}"
	648	+	);
	649	+	}
610	650		Ok(RowOutcome::Applied)
611	651		}
612	652

M synckit-client/src/store/db.rs +4

			@@ -129,6 +129,10 @@
129	129		// install that predates it gets the table on the next connection open rather
130	130		// than on the next time the app happens to re-run its migration.
131	131		conn.execute_batch(super::deferred::DEFERRED_DDL)?;
	132	+	// The field-merge base, same reasoning: an install that predates field merge
	133	+	// gets the table on the next connection open. It stays empty until a table
	134	+	// opts in, so creating it unconditionally costs nothing.
	135	+	conn.execute_batch(super::snapshot::SNAPSHOT_DDL)?;
132	136		Ok(())
133	137		}
134	138

M synckit-client/src/store/facade.rs +2 -2

			@@ -170,7 +170,7 @@
170	170		.await?;
171	171		self.maybe_snapshot().await?;
172	172
173		-	let mut pushed = push_changes(&self.db, &*self.client, device_id).await?;
	173	+	let mut pushed = push_changes(&self.db, &*self.client, &self.schema, device_id).await?;
174	174		let pull = pull_changes(&self.db, &*self.client, &self.schema, device_id).await?;
175	175		let mut pulled = pull.applied;
176	176		let mut changed_tables = pull.changed_tables;
			@@ -180,7 +180,7 @@
180	180		// loop is a no-op for personal-only apps.
181	181		for (id, gck_version) in self.client.list_group_scopes().await? {
182	182		let scope = SyncScope::Group { id, gck_version };
183		-	pushed += push_scope(&self.db, &*self.client, device_id, scope).await?;
	183	+	pushed += push_scope(&self.db, &*self.client, &self.schema, device_id, scope).await?;
184	184		let gp = pull_scope(&self.db, &*self.client, &self.schema, device_id, scope).await?;
185	185		pulled += gp.applied;
186	186		changed_tables.extend(gp.changed_tables);

M synckit-client/src/store/hlc.rs +469 -3

			@@ -23,8 +23,12 @@
23	23
24	24		use super::db::{get_sync_state, set_sync_state};
25	25		use super::schema::{ConflictStrategy, SyncSchema};
	26	+	use super::snapshot;
26	27		use super::stash;
27		-	use crate::conflict::{Resolution, change_order, detect_conflicts, resolve_lww_at};
	28	+	use crate::conflict::{
	29	+	Resolution, change_order, contested_fields, detect_conflicts, is_clock_poisoned,
	30	+	resolve_field_merge, resolve_lww_at,
	31	+	};
28	32		use crate::error::Result;
29	33		use crate::ids::DeviceId;
30	34		use crate::types::{ChangeEntry, ChangeOp, Hlc, PulledChange};
			@@ -290,7 +294,12 @@
290	294		);
291	295
292	296		for pair in conflicts {
293		-	match resolve_lww_at(&pair.local, &pair.remote, now) {
	297	+	// A table that opted into field merge tries that first; every
	298	+	// other table, and every merge that cannot be based, gets the
	299	+	// LWW-and-stash path unchanged.
	300	+	let resolution = try_field_merge(conn, schema, &pair, now, scope)
	301	+	.unwrap_or_else(\|\| resolve_lww_at(&pair.local, &pair.remote, now));
	302	+	match resolution {
294	303		// Site 1: our own edit is discarded.
295	304		Resolution::KeepRemote => {
296	305		stash_or_warn(
			@@ -314,7 +323,9 @@
314	323		&pair.local,
315	324		);
316	325		}
317		-	// A merge keeps both sides' fields, so nothing was thrown away.
	326	+	// A merge keeps each side's uncontested fields, so nothing is
	327	+	// thrown away here. A contested field does lose one side's
	328	+	// value, and `try_field_merge` has already stashed it.
318	329		Resolution::Merged(data) => {
319	330		let hlc = pair.local.hlc.max(pair.remote.entry.hlc);
320	331		resolved.push(ChangeEntry {
			@@ -339,6 +350,114 @@
339	350		}
340	351		}
341	352
	353	+	/// Try to resolve one conflict as a three-way field merge, or return `None` to
	354	+	/// let the caller fall back to [`resolve_lww_at`].
	355	+	///
	356	+	/// Every `None` below is a deliberate fallback to today's behaviour rather than a
	357	+	/// failure, and they are worth reading as a list, because the design is that a
	358	+	/// merge is an optimisation over LWW and never a weaker guarantee than it:
	359	+	///
	360	+	/// - the table did not opt in, so nothing about it changes;
	361	+	/// - either side is clock-poisoned, so the poisoning guard in [`resolve_lww_at`]
	362	+	/// must decide instead. Merging here would hand a hostile far-future entry
	363	+	/// the contested fields it wants while looking like cooperation;
	364	+	/// - either side carries no object payload (a delete), which has no fields to
	365	+	/// merge;
	366	+	/// - no base, the first-ever conflict for a row, or one whose base was
	367	+	/// dropped by a delete. `resolve_field_merge` would fall back internally, but
	368	+	/// routing through LWW keeps the stash behaviour byte-identical to today
	369	+	/// rather than depending on the fallback's phrasing;
	370	+	/// - a declared counter moved on both sides. This is the one that is not an
	371	+	/// absence of information but a refusal. Two devices each logging thirty
	372	+	/// minutes produce a base of `0` and two absolutes of `30`; a value merge takes
	373	+	/// one of them and the row reads `30` where the user did `60`. Under LWW the
	374	+	/// same loss is at least whole, stashed, and recoverable, so LWW is the better
	375	+	/// answer and the merge declines. Reconstructing the increment is not an
	376	+	/// option: a base plus two absolutes cannot distinguish two additions from one
	377	+	/// overwrite.
	378	+	///
	379	+	/// When the merge does run, a field both sides moved to different values still
	380	+	/// costs one side its value, so the loser is stashed exactly as LWW would have
	381	+	/// stashed it. A merge that contested nothing stashes nothing, which is the whole
	382	+	/// point: under plain LWW that case fills the stash with edits that never
	383	+	/// conflicted, and the volume is what stops anyone reading it.
	384	+	fn try_field_merge(
	385	+	conn: &Connection,
	386	+	schema: &SyncSchema,
	387	+	pair: &crate::conflict::ConflictPair,
	388	+	now: DateTime<Utc>,
	389	+	scope: &str,
	390	+	) -> Option<Resolution> {
	391	+	let entry = &pair.remote.entry;
	392	+	let table = schema.tables.iter().find(\|t\| t.name() == entry.table)?;
	393	+	if !table.merges_fields() {
	394	+	return None;
	395	+	}
	396	+	if is_clock_poisoned(&pair.local.hlc, now) \|\| is_clock_poisoned(&entry.hlc, now) {
	397	+	return None;
	398	+	}
	399	+
	400	+	let local_data = pair.local.data.as_ref().filter(\|v\| v.is_object())?;
	401	+	let remote_data = entry.data.as_ref().filter(\|v\| v.is_object())?;
	402	+	let base = snapshot::load(conn, &entry.table, &entry.row_id);
	403	+	if !base.is_object() {
	404	+	tracing::debug!(
	405	+	table = %entry.table,
	406	+	row_id = %entry.row_id,
	407	+	"no field-merge base for this row; resolving by LWW"
	408	+	);
	409	+	return None;
	410	+	}
	411	+
	412	+	let contested = contested_fields(local_data, remote_data, &base);
	413	+	if let Some(counter) = contested
	414	+	.iter()
	415	+	.find(\|c\| table.counters().contains(&c.as_str()))
	416	+	{
	417	+	tracing::debug!(
	418	+	table = %entry.table,
	419	+	row_id = %entry.row_id,
	420	+	counter,
	421	+	"both sides moved a counter column; refusing to merge so the loss stays visible"
	422	+	);
	423	+	return None;
	424	+	}
	425	+
	426	+	let merged = resolve_field_merge(local_data, remote_data, &base, &pair.local.hlc, &entry.hlc);
	427	+
	428	+	// Only a field both sides moved to different values loses anything. Both
	429	+	// arriving at the same value is a contest with no loser, and stashing it
	430	+	// would refill the stash with the non-conflicts this exists to drain.
	431	+	let lost = contested
	432	+	.iter()
	433	+	.any(\|k\| local_data.get(k) != remote_data.get(k));
	434	+	if lost && matches!(merged, Resolution::Merged(_)) {
	435	+	match resolve_lww_at(&pair.local, &pair.remote, now) {
	436	+	Resolution::KeepLocal \| Resolution::Skip => stash_or_warn(
	437	+	conn,
	438	+	scope,
	439	+	stash::LosingSide::Remote,
	440	+	entry,
	441	+	pair.remote.device_id,
	442	+	&pair.local,
	443	+	),
	444	+	Resolution::KeepRemote => stash_or_warn(
	445	+	conn,
	446	+	scope,
	447	+	stash::LosingSide::Local,
	448	+	&pair.local,
	449	+	// Always this device: `load_local_pending` stamps every entry it
	450	+	// builds with the local node.
	451	+	pair.local.hlc.node,
	452	+	entry,
	453	+	),
	454	+	Resolution::Merged(_) => {}
	455	+	}
	456	+	}
	457	+
	458	+	Some(merged)
	459	+	}
	460	+
342	461		/// Stash one discarded side, logging rather than failing. Same reasoning as the
343	462		/// gate's sink: the stash is evidence about a sync, never a reason to fail one.
344	463		fn stash_or_warn(
			@@ -1326,6 +1445,381 @@
1326	1445		assert!(stash_rows(&conn).is_empty());
1327	1446		}
1328	1447
	1448	+	// ── Field merge ──
	1449	+	//
	1450	+	// The merge semantics themselves are pinned in `conflict.rs`; what is new
	1451	+	// here is the wiring, which is where every one of these can go wrong
	1452	+	// independently of a correct merge: the base has to be recorded at the right
	1453	+	// moments, read back for the right tables, refused for counters, and left
	1454	+	// entirely alone for a table that did not opt in.
	1455	+
	1456	+	/// A four-column row, which is the point: a merge is only interesting when a
	1457	+	/// row has more columns than two devices are likely to both touch. `minutes`
	1458	+	/// is the counter.
	1459	+	const CARD_COLS: &[&str] = &["id", "title", "due", "minutes"];
	1460	+
	1461	+	fn card_schema(merges: bool) -> SyncSchema {
	1462	+	let table = SyncTable::full("card", CARD_COLS);
	1463	+	SyncSchema::new(vec![if merges {
	1464	+	table.field_merge(&["minutes"])
	1465	+	} else {
	1466	+	table
	1467	+	}])
	1468	+	}
	1469	+
	1470	+	fn card_device(n: u128, merges: bool) -> (Connection, DeviceId) {
	1471	+	let conn = Connection::open_in_memory().unwrap();
	1472	+	configure_connection(&conn).unwrap();
	1473	+	conn.execute_batch(
	1474	+	"CREATE TABLE card (id TEXT PRIMARY KEY, title TEXT, due TEXT, minutes INTEGER);",
	1475	+	)
	1476	+	.unwrap();
	1477	+	conn.execute_batch(&card_schema(merges).migration_sql())
	1478	+	.unwrap();
	1479	+	(conn, node(n))
	1480	+	}
	1481	+
	1482	+	/// A wall reading to hang a card test's clocks off.
	1483	+	///
	1484	+	/// Real time, not a toy constant, because `resolve_pull` advances the local
	1485	+	/// clock to `now` on every pull it observes: an HLC of `2_000` would sit
	1486	+	/// decades below the clock it is meant to be newer than, and every one of
	1487	+	/// these tests would silently assert over a conflict that never happened.
	1488	+	fn card_t0() -> i64 {
	1489	+	Utc::now().timestamp_millis()
	1490	+	}
	1491	+
	1492	+	/// A remote change for the card table with an explicit payload and wall clock.
	1493	+	fn card_change(from: DeviceId, wall_ms: i64, data: serde_json::Value) -> PulledChange {
	1494	+	PulledChange {
	1495	+	entry: ChangeEntry {
	1496	+	table: "card".into(),
	1497	+	op: ChangeOp::Update,
	1498	+	row_id: "c1".into(),
	1499	+	timestamp: Utc::now(),
	1500	+	hlc: Hlc {
	1501	+	wall_ms,
	1502	+	counter: 0,
	1503	+	node: from,
	1504	+	},
	1505	+	data: Some(data),
	1506	+	extra: serde_json::Map::default(),
	1507	+	},
	1508	+	device_id: from,
	1509	+	seq: 1,
	1510	+	}
	1511	+	}
	1512	+
	1513	+	/// The state both devices start from, established the way it really is: a
	1514	+	/// remote change applied and committed, which is also what records the base.
	1515	+	fn seed_base(conn: &mut Connection, node: DeviceId, peer: DeviceId, merges: bool, t0: i64) {
	1516	+	let s = card_schema(merges);
	1517	+	let base = card_change(
	1518	+	peer,
	1519	+	t0,
	1520	+	serde_json::json!({"id": "c1", "title": "base", "due": null, "minutes": 0}),
	1521	+	);
	1522	+	pull_apply_with(conn, &s, node, vec![base]);
	1523	+	}
	1524	+
	1525	+	fn pull_apply_with(
	1526	+	conn: &mut Connection,
	1527	+	s: &SyncSchema,
	1528	+	node: DeviceId,
	1529	+	pulled: Vec<PulledChange>,
	1530	+	) {
	1531	+	let resolved = resolve_pull(conn, s, node, pulled, Utc::now(), "").unwrap();
	1532	+	apply_remote_changes(conn, s, &resolved, "").unwrap();
	1533	+	record_committed(conn, resolved.as_slice()).unwrap();
	1534	+	}
	1535	+
	1536	+	fn card(conn: &Connection) -> (Option<String>, Option<String>, Option<i64>) {
	1537	+	conn.query_row(
	1538	+	"SELECT title, due, minutes FROM card WHERE id = 'c1'",
	1539	+	[],
	1540	+	\|r\| Ok((r.get(0)?, r.get(1)?, r.get(2)?)),
	1541	+	)
	1542	+	.unwrap()
	1543	+	}
	1544	+
	1545	+	/// The case the whole feature exists for. Two devices edit different columns
	1546	+	/// of one row; both edits survive, where LWW would have discarded one whole
	1547	+	/// edit for touching a row it never contested.
	1548	+	#[test]
	1549	+	fn opted_in_table_merges_edits_to_different_fields() {
	1550	+	let (mut conn, n) = card_device(1, true);
	1551	+	let peer = node(2);
	1552	+	let t0 = card_t0();
	1553	+	seed_base(&mut conn, n, peer, true, t0);
	1554	+
	1555	+	// Local: set the due date. Remote, newer: retitle.
	1556	+	conn.execute("UPDATE card SET due = '2026-09-01' WHERE id = 'c1'", [])
	1557	+	.unwrap();
	1558	+	stamp_pending(&conn, n, t0 + 1_000).unwrap();
	1559	+	let remote = card_change(
	1560	+	peer,
	1561	+	t0 + 2_000,
	1562	+	serde_json::json!({"id": "c1", "title": "theirs", "due": null, "minutes": 0}),
	1563	+	);
	1564	+
	1565	+	pull_apply_with(&mut conn, &card_schema(true), n, vec![remote]);
	1566	+
	1567	+	let (title, due, _) = card(&conn);
	1568	+	assert_eq!(title.as_deref(), Some("theirs"), "remote's field was lost");
	1569	+	assert_eq!(
	1570	+	due.as_deref(),
	1571	+	Some("2026-09-01"),
	1572	+	"the local edit was discarded for contesting a field it never touched"
	1573	+	);
	1574	+	assert!(
	1575	+	stash_rows(&conn).is_empty(),
	1576	+	"a merge that contested nothing lost nothing, so it must not stash: {:?}",
	1577	+	stash_rows(&conn)
	1578	+	);
	1579	+	}
	1580	+
	1581	+	/// A field both sides moved is a real contest. The merge hands it to the HLC
	1582	+	/// winner (the same rule LWW would apply), and the loser's version is stashed,
	1583	+	/// because a value did go away.
	1584	+	#[test]
	1585	+	fn a_contested_field_goes_to_the_hlc_winner_and_stashes_the_loser() {
	1586	+	let (mut conn, n) = card_device(1, true);
	1587	+	let peer = node(2);
	1588	+	let t0 = card_t0();
	1589	+	seed_base(&mut conn, n, peer, true, t0);
	1590	+
	1591	+	conn.execute(
	1592	+	"UPDATE card SET title = 'mine', due = '2026-09-01' WHERE id = 'c1'",
	1593	+	[],
	1594	+	)
	1595	+	.unwrap();
	1596	+	stamp_pending(&conn, n, t0 + 1_000).unwrap();
	1597	+	let remote = card_change(
	1598	+	peer,
	1599	+	t0 + 2_000,
	1600	+	serde_json::json!({"id": "c1", "title": "theirs", "due": null, "minutes": 0}),
	1601	+	);
	1602	+
	1603	+	pull_apply_with(&mut conn, &card_schema(true), n, vec![remote]);
	1604	+
	1605	+	let (title, due, _) = card(&conn);
	1606	+	assert_eq!(
	1607	+	title.as_deref(),
	1608	+	Some("theirs"),
	1609	+	"the contested field must go to the newer HLC"
	1610	+	);
	1611	+	assert_eq!(
	1612	+	due.as_deref(),
	1613	+	Some("2026-09-01"),
	1614	+	"an uncontested field must survive even when the same row lost a contest"
	1615	+	);
	1616	+
	1617	+	let rows = stash_rows(&conn);
	1618	+	assert_eq!(
	1619	+	rows.len(),
	1620	+	1,
	1621	+	"the losing title must be recoverable: {rows:?}"
	1622	+	);
	1623	+	assert_eq!(rows[0].losing_side, "local");
	1624	+	assert!(rows[0].losing_payload.as_deref().unwrap().contains("mine"));
	1625	+	}
	1626	+
	1627	+	/// The failure a value merge cannot see. Two devices each log thirty minutes;
	1628	+	/// merging takes one side's absolute total and the other half-hour is gone
	1629	+	/// with nothing recording that it existed. Refusing to merge is worse for the
	1630	+	/// uncontested fields and better for the truth: LWW discards one whole edit
	1631	+	/// and stashes it, so the loss stays visible and recoverable.
	1632	+	#[test]
	1633	+	fn a_contested_counter_refuses_to_merge_and_falls_back_to_lww() {
	1634	+	let (mut conn, n) = card_device(1, true);
	1635	+	let peer = node(2);
	1636	+	let t0 = card_t0();
	1637	+	seed_base(&mut conn, n, peer, true, t0);
	1638	+
	1639	+	// Both sides increment `minutes`, and each also moves a field the other
	1640	+	// did not, so a merge would visibly have kept both.
	1641	+	conn.execute(
	1642	+	"UPDATE card SET minutes = minutes + 30, due = '2026-09-01' WHERE id = 'c1'",
	1643	+	[],
	1644	+	)
	1645	+	.unwrap();
	1646	+	stamp_pending(&conn, n, t0 + 1_000).unwrap();
	1647	+	let remote = card_change(
	1648	+	peer,
	1649	+	t0 + 2_000,
	1650	+	serde_json::json!({"id": "c1", "title": "theirs", "due": null, "minutes": 30}),
	1651	+	);
	1652	+
	1653	+	pull_apply_with(&mut conn, &card_schema(true), n, vec![remote]);
	1654	+
	1655	+	let (title, due, minutes) = card(&conn);
	1656	+	assert_eq!(minutes, Some(30), "a merged counter would still read 30");
	1657	+	assert_eq!(title.as_deref(), Some("theirs"));
	1658	+	assert_eq!(
	1659	+	due, None,
	1660	+	"the row must hold one side whole, not a merge of both"
	1661	+	);
	1662	+
	1663	+	let rows = stash_rows(&conn);
	1664	+	assert_eq!(
	1665	+	rows.len(),
	1666	+	1,
	1667	+	"the discarded half-hour must stay visible in the stash: {rows:?}"
	1668	+	);
	1669	+	assert!(
	1670	+	rows[0]
	1671	+	.losing_payload
	1672	+	.as_deref()
	1673	+	.unwrap()
	1674	+	.contains("2026-09-01"),
	1675	+	"the stash must hold the whole discarded edit: {rows:?}"
	1676	+	);
	1677	+	}
	1678	+
	1679	+	/// A counter only one side moved is not the counter problem: nothing has to
	1680	+	/// be reconstructed, so the merge runs and that side's value carries across.
	1681	+	/// Without this the counter declaration would cost every row on the table its
	1682	+	/// merge, which is most of the feature.
	1683	+	#[test]
	1684	+	fn an_uncontested_counter_does_not_block_the_merge() {
	1685	+	let (mut conn, n) = card_device(1, true);
	1686	+	let peer = node(2);
	1687	+	let t0 = card_t0();
	1688	+	seed_base(&mut conn, n, peer, true, t0);
	1689	+
	1690	+	conn.execute("UPDATE card SET minutes = minutes + 30 WHERE id = 'c1'", [])
	1691	+	.unwrap();
	1692	+	stamp_pending(&conn, n, t0 + 1_000).unwrap();
	1693	+	let remote = card_change(
	1694	+	peer,
	1695	+	t0 + 2_000,
	1696	+	serde_json::json!({"id": "c1", "title": "theirs", "due": null, "minutes": 0}),
	1697	+	);
	1698	+
	1699	+	pull_apply_with(&mut conn, &card_schema(true), n, vec![remote]);
	1700	+
	1701	+	let (title, _, minutes) = card(&conn);
	1702	+	assert_eq!(
	1703	+	minutes,
	1704	+	Some(30),
	1705	+	"an uncontested counter must carry across"
	1706	+	);
	1707	+	assert_eq!(title.as_deref(), Some("theirs"));
	1708	+	}
	1709	+
	1710	+	/// The default must be no change in behaviour. The identical conflict on a
	1711	+	/// table that did not opt in resolves the way it always has: one whole edit
	1712	+	/// wins, the other is stashed.
	1713	+	#[test]
	1714	+	fn a_table_that_did_not_opt_in_still_resolves_by_lww() {
	1715	+	let (mut conn, n) = card_device(1, false);
	1716	+	let peer = node(2);
	1717	+	let t0 = card_t0();
	1718	+	seed_base(&mut conn, n, peer, false, t0);
	1719	+
	1720	+	conn.execute("UPDATE card SET due = '2026-09-01' WHERE id = 'c1'", [])
	1721	+	.unwrap();
	1722	+	stamp_pending(&conn, n, t0 + 1_000).unwrap();
	1723	+	let remote = card_change(
	1724	+	peer,
	1725	+	t0 + 2_000,
	1726	+	serde_json::json!({"id": "c1", "title": "theirs", "due": null, "minutes": 0}),
	1727	+	);
	1728	+
	1729	+	pull_apply_with(&mut conn, &card_schema(false), n, vec![remote]);
	1730	+
	1731	+	let (title, due, _) = card(&conn);
	1732	+	assert_eq!(title.as_deref(), Some("theirs"));
	1733	+	assert_eq!(
	1734	+	due, None,
	1735	+	"without the opt-in the whole remote edit wins, exactly as before"
	1736	+	);
	1737	+	assert_eq!(stash_rows(&conn).len(), 1, "and the loser is still stashed");
	1738	+
	1739	+	let bases: i64 = conn
	1740	+	.query_row("SELECT COUNT(*) FROM sync_row_snapshot", [], \|r\| r.get(0))
	1741	+	.unwrap();
	1742	+	assert_eq!(bases, 0, "a table that did not opt in must store no base");
	1743	+	}
	1744	+
	1745	+	/// A row whose base was never recorded, or was dropped by a delete. There is
	1746	+	/// nothing to merge against, so the conflict falls through to LWW rather than
	1747	+	/// merging against a base it invented.
	1748	+	#[test]
	1749	+	fn a_missing_base_falls_back_to_lww_without_failing() {
	1750	+	let (mut conn, n) = card_device(1, true);
	1751	+	let peer = node(2);
	1752	+	let t0 = card_t0();
	1753	+	seed_base(&mut conn, n, peer, true, t0);
	1754	+	conn.execute("DELETE FROM sync_row_snapshot", []).unwrap();
	1755	+
	1756	+	conn.execute("UPDATE card SET due = '2026-09-01' WHERE id = 'c1'", [])
	1757	+	.unwrap();
	1758	+	stamp_pending(&conn, n, t0 + 1_000).unwrap();
	1759	+	let remote = card_change(
	1760	+	peer,
	1761	+	t0 + 2_000,
	1762	+	serde_json::json!({"id": "c1", "title": "theirs", "due": null, "minutes": 0}),
	1763	+	);
	1764	+
	1765	+	pull_apply_with(&mut conn, &card_schema(true), n, vec![remote]);
	1766	+
	1767	+	let (title, due, _) = card(&conn);
	1768	+	assert_eq!(title.as_deref(), Some("theirs"));
	1769	+	assert_eq!(due, None, "with no base the whole remote edit wins");
	1770	+	assert_eq!(stash_rows(&conn).len(), 1);
	1771	+	}
	1772	+
	1773	+	/// Applying a remote change re-bases the row, so the next conflict measures
	1774	+	/// both sides against what this device last received rather than against the
	1775	+	/// version it first saw. Without this the base would go stale and every later
	1776	+	/// merge would report fields as changed that nobody touched.
	1777	+	#[test]
	1778	+	fn applying_a_remote_change_rebases_the_row() {
	1779	+	let (mut conn, n) = card_device(1, true);
	1780	+	let peer = node(2);
	1781	+	let t0 = card_t0();
	1782	+	seed_base(&mut conn, n, peer, true, t0);
	1783	+
	1784	+	pull_apply_with(
	1785	+	&mut conn,
	1786	+	&card_schema(true),
	1787	+	n,
	1788	+	vec![card_change(
	1789	+	peer,
	1790	+	t0 + 2_000,
	1791	+	serde_json::json!({"id": "c1", "title": "second", "due": null, "minutes": 0}),
	1792	+	)],
	1793	+	);
	1794	+

Lines truncated

M synckit-client/src/store/migrate.rs +1

			@@ -115,6 +115,7 @@
115	115		pub fn migration_sql(&self) -> String {
116	116		let mut out = String::from(BASE_TABLES);
117	117		out.push_str(super::deferred::DEFERRED_DDL);
	118	+	out.push_str(super::snapshot::SNAPSHOT_DDL);
118	119		if self.any_hashed() {
119	120		out.push_str(SALT_SEED);
120	121		}

M synckit-client/src/store/mod.rs +1

			@@ -27,6 +27,7 @@
27	27		pub mod migrate;
28	28		pub mod scheduler;
29	29		pub mod schema;
	30	+	pub(crate) mod snapshot;
30	31		pub(crate) mod stash;
31	32		pub mod sync;
32	33

M synckit-client/src/store/schema.rs +57

			@@ -91,6 +91,8 @@
91	91		pub(crate) references_unsynced: bool,
92	92		pub(crate) exclude_where: Option<&'static str>,
93	93		pub(crate) group_scope: Option<&'static str>,
	94	+	pub(crate) field_merge: bool,
	95	+	pub(crate) counters: &'static [&'static str],
94	96		}
95	97
96	98		impl SyncTable {
			@@ -99,6 +101,18 @@
99	101		self.name
100	102		}
101	103
	104	+	/// Whether this table opted into three-way field merge
	105	+	/// ([`field_merge`](Self::field_merge)).
	106	+	pub fn merges_fields(&self) -> bool {
	107	+	self.field_merge
	108	+	}
	109	+
	110	+	/// The columns this table declared as counters, never field-merged. Empty
	111	+	/// unless the table opted into field merge.
	112	+	pub fn counters(&self) -> &'static [&'static str] {
	113	+	self.counters
	114	+	}
	115	+
102	116		/// The local provenance column that routes this table's rows to a group
103	117		/// scope, or `None` if the table is personal-only. Public so a consumer can
104	118		/// assert exactly which tables are group-scoped (GoingsOn's M3 check).
			@@ -122,6 +136,8 @@
122	136		references_unsynced: false,
123	137		exclude_where: None,
124	138		group_scope: None,
	139	+	field_merge: false,
	140	+	counters: &[],
125	141		}
126	142		}
127	143
			@@ -170,6 +186,47 @@
170	186		self
171	187		}
172	188
	189	+	/// Merge concurrent edits to different columns of this row instead of
	190	+	/// discarding one of them, declaring the table's counter columns as you do.
	191	+	///
	192	+	/// Off by default, and a table that does not call this resolves conflicts
	193	+	/// exactly as it always has: one whole edit wins under
	194	+	/// [`resolve_lww`](crate::conflict::resolve_lww) and the other goes to the
	195	+	/// conflict stash. That is safe but blunt. A row with thirty columns makes
	196	+	/// two edits landing on the same one the uncommon case, so most of what the
	197	+	/// stash collects under plain LWW is not conflict, it is collateral, and the
	198	+	/// volume is what stops anyone reading it.
	199	+	///
	200	+	/// Opting in stores a last-synced snapshot of every row of this table (see
	201	+	/// [`snapshot`](super::snapshot)), roughly doubling its storage, and routes
	202	+	/// conflicts through [`resolve_field_merge`](crate::conflict::resolve_field_merge).
	203	+	/// Fields only one side changed are kept from that side; a field both sides
	204	+	/// changed still goes to the HLC winner and the loser is still stashed.
	205	+	///
	206	+	/// `counters` names the columns this table increments in place
	207	+	/// (`SET n = n + ?`). **They are the reason this takes an argument rather
	208	+	/// than being a bare flag.** A merge is defined over values: given a base and
	209	+	/// two absolute totals it cannot tell that two devices each meant to add
	210	+	/// thirty minutes rather than that one overwrote the other, so it would keep
	211	+	/// one side's total and silently lose the other's addition. Under plain LWW
	212	+	/// the same loss is at least visible and recoverable in the stash. So a
	213	+	/// conflict where both sides moved a declared counter refuses to merge at all
	214	+	/// and falls back to LWW-and-stash. Pass `&[]` only after checking there is
	215	+	/// no such column; the API asks because the question is easy to not ask.
	216	+	///
	217	+	/// One thing merge cannot check for you: **columns that are only valid
	218	+	/// together**. If `status` and `completed_at` are written as a pair
	219	+	/// everywhere but one, a merge can take `status` from one device and
	220	+	/// `completed_at` from the other and produce a pair neither device wrote.
	221	+	/// Enumerate the table's writers before opting in, and make dependent columns
	222	+	/// move together at every site.
	223	+	#[must_use]
	224	+	pub fn field_merge(mut self, counters: &'static [&'static str]) -> Self {
	225	+	self.field_merge = true;
	226	+	self.counters = counters;
	227	+	self
	228	+	}
	229	+
173	230		/// Values injected only on first INSERT (e.g. to satisfy a NOT NULL on a
174	231		/// preserved secret column).
175	232		#[must_use]

M synckit-client/src/store/sync.rs +112 -6

			@@ -26,6 +26,7 @@
26	26		use super::hlc::{resolve_pull, set_committed, stamp_pending};
27	27		use super::migrate::{json_object, row_id_expr};
28	28		use super::schema::{SyncMode, SyncSchema};
	29	+	use super::snapshot;
29	30		use crate::client::SyncKitClient;
30	31		use crate::error::{Result, SyncKitError};
31	32		use crate::ids::{DeviceId, GroupId};
			@@ -287,9 +288,10 @@
287	288		pub async fn push_changes<T: SyncTransport>(
288	289		db: &DbSource,
289	290		client: &T,
	291	+	schema: &SyncSchema,
290	292		device_id: DeviceId,
291	293		) -> Result<u64> {
292		-	push_scope(db, client, device_id, SyncScope::Personal).await
	294	+	push_scope(db, client, schema, device_id, SyncScope::Personal).await
293	295		}
294	296
295	297		/// Drain all unpushed local changes for one `scope` to the server, batch by batch.
			@@ -300,13 +302,28 @@
300	302		/// of this row is gated out). `device_id` doubles as the HLC node. The drain
301	303		/// reads only rows whose `sync_changelog.scope` matches, so a group's changes
302	304		/// never leak into the personal push and vice versa.
	305	+	///
	306	+	/// `schema` is read for one thing only: which tables opted into field merge, so
	307	+	/// an acknowledged push can re-base them (see [`snapshot`](super::snapshot)). A
	308	+	/// push the server accepted is the other moment a row becomes common ground, the
	309	+	/// mirror of an applied pull, and skipping it here would leave the base stuck at
	310	+	/// whatever this device last received while the row moved on.
303	311		pub async fn push_scope<T: SyncTransport>(
304	312		db: &DbSource,
305	313		client: &T,
	314	+	schema: &SyncSchema,
306	315		device_id: DeviceId,
307	316		scope: SyncScope,
308	317		) -> Result<u64> {
309	318		let scope_key = scope.key();
	319	+	// Owned, because the write step runs on a blocking thread and cannot borrow
	320	+	// the schema across it.
	321	+	let merging: HashSet<String> = schema
	322	+	.tables()
	323	+	.iter()
	324	+	.filter(\|t\| t.merges_fields())
	325	+	.map(\|t\| t.name().to_string())
	326	+	.collect();
310	327		let mut pushed_total = 0u64;
311	328		loop {
312	329		let db_read = db.clone();
			@@ -341,8 +358,9 @@
341	358		total_read,
342	359		} = batch;
343	360		let sent = wire.len() as u64;
	361	+	let merging = merging.clone();
344	362		tokio::task::spawn_blocking(move \|\| {
345		-	mark_pushed_committed(&mut db_write.open()?, &wire_ids, &skip_ids, &wire)
	363	+	mark_pushed_committed(&mut db_write.open()?, &wire_ids, &skip_ids, &wire, &merging)
346	364		})
347	365		.await
348	366		.map_err(\|e\| join_err(&e))??;
			@@ -415,6 +433,7 @@
415	433		wire_ids: &[i64],
416	434		skip_ids: &[i64],
417	435		wire: &[ChangeEntry],
	436	+	merging: &HashSet<String>,
418	437		) -> Result<()> {
419	438		let tx = conn.transaction()?;
420	439		for id in wire_ids.iter().chain(skip_ids) {
			@@ -425,6 +444,19 @@
425	444		for e in wire {
426	445		set_committed(&tx, &e.table, &e.row_id, &e.hlc)?;
427	446		}
	447	+	// Push-side re-base, for tables that opted into field merge. The server has
	448	+	// taken this edit, so it is the version a peer will pull and therefore the one
	449	+	// both devices next diverge from. A delete drops the base instead of setting
	450	+	// it: the payload is only the primary key, and the row it described is gone.
	451	+	for e in wire.iter().filter(\|e\| merging.contains(&e.table)) {
	452	+	match (&e.op, &e.data) {
	453	+	(ChangeOp::Insert \| ChangeOp::Update, Some(data)) => {
	454	+	snapshot::record(&tx, &e.table, &e.row_id, data)?;
	455	+	}
	456	+	(ChangeOp::Delete, _) => snapshot::forget(&tx, &e.table, &e.row_id)?,
	457	+	_ => {}
	458	+	}
	459	+	}
428	460		tx.commit()?;
429	461		Ok(())
430	462		}
			@@ -746,7 +778,7 @@
746	778		}
747	779
748	780		// Personal push drains only the personal row.
749		-	let pushed = push_scope(&db, &server, node, SyncScope::Personal)
	781	+	let pushed = push_scope(&db, &server, &group_schema(), node, SyncScope::Personal)
750	782		.await
751	783		.unwrap();
752	784		assert_eq!(pushed, 1);
			@@ -758,6 +790,7 @@
758	790		let pushed = push_scope(
759	791		&db,
760	792		&server,
	793	+	&group_schema(),
761	794		node,
762	795		SyncScope::Group {
763	796		id: gid,
			@@ -970,8 +1003,8 @@
970	1003		edit(&db_, "r", "from-B");
971	1004		stamp_at(&db_, nb, 200);
972	1005
973		-	push_changes(&da, &server, na).await.unwrap();
974		-	push_changes(&db_, &server, nb).await.unwrap();
	1006	+	push_changes(&da, &server, &schema(), na).await.unwrap();
	1007	+	push_changes(&db_, &server, &schema(), nb).await.unwrap();
975	1008
976	1009		pull_changes(&da, &server, &schema(), na).await.unwrap();
977	1010		pull_changes(&db_, &server, &schema(), nb).await.unwrap();
			@@ -988,7 +1021,7 @@
988	1021		edit(&da, "r1", "x");
989	1022		edit(&da, "r2", "y");
990	1023
991		-	let pushed = push_changes(&da, &server, na).await.unwrap();
	1024	+	let pushed = push_changes(&da, &server, &schema(), na).await.unwrap();
992	1025		assert_eq!(pushed, 2);
993	1026		// All local rows are marked pushed.
994	1027		let unpushed: i64 = da
			@@ -1012,6 +1045,79 @@
1012	1045		assert_eq!(cursor, 2);
1013	1046		}
1014	1047
	1048	+	/// The push half of the field-merge base.
	1049	+	///
	1050	+	/// A pull is the obvious moment a row becomes common ground and it is only
	1051	+	/// half of them: once the server takes an edit, that edit is what a peer will
	1052	+	/// pull, so it is the version the two devices next diverge from. Re-basing
	1053	+	/// only on pull would leave the base stuck at whatever this device last
	1054	+	/// received, and every merge afterwards would report this device's own
	1055	+	/// already-shared edits as changes, handing itself fields it never contested.
	1056	+	#[tokio::test]
	1057	+	async fn an_acknowledged_push_rebases_the_row() {
	1058	+	let dir = tempdir();
	1059	+	let db = DbSource::path(dir.join("a.db"));
	1060	+	{
	1061	+	let conn = db.open().unwrap();
	1062	+	conn.execute_batch("CREATE TABLE note (id TEXT PRIMARY KEY, name TEXT);")
	1063	+	.unwrap();
	1064	+	conn.execute_batch(&merge_schema().migration_sql()).unwrap();
	1065	+	}
	1066	+	let node = DeviceId::new(uuid::Uuid::from_u128(1));
	1067	+	let server = FakeServer::default();
	1068	+
	1069	+	edit(&db, "r1", "mine");
	1070	+	assert_eq!(
	1071	+	base(&db, "r1"),
	1072	+	serde_json::Value::Null,
	1073	+	"nothing shared yet"
	1074	+	);
	1075	+
	1076	+	push_changes(&db, &server, &merge_schema(), node)
	1077	+	.await
	1078	+	.unwrap();
	1079	+	assert_eq!(
	1080	+	base(&db, "r1")["name"],
	1081	+	"mine",
	1082	+	"the server took this edit, so it is the version a peer will pull"
	1083	+	);
	1084	+
	1085	+	// And a pushed delete drops the base rather than leaving it describing a
	1086	+	// row that no longer exists.
	1087	+	db.open()
	1088	+	.unwrap()
	1089	+	.execute("DELETE FROM note WHERE id = 'r1'", [])
	1090	+	.unwrap();
	1091	+	push_changes(&db, &server, &merge_schema(), node)
	1092	+	.await
	1093	+	.unwrap();
	1094	+	assert_eq!(base(&db, "r1"), serde_json::Value::Null);
	1095	+	}
	1096	+
	1097	+	/// A table that did not opt in must store no base, so the storage cost lands
	1098	+	/// only where someone asked for it.
	1099	+	#[tokio::test]
	1100	+	async fn a_push_stores_no_base_for_a_table_that_did_not_opt_in() {
	1101	+	let dir = tempdir();
	1102	+	let (db, node) = device(&dir.join("a.db"), 1);
	1103	+	let server = FakeServer::default();
	1104	+
	1105	+	edit(&db, "r1", "mine");
	1106	+	push_changes(&db, &server, &schema(), node).await.unwrap();
	1107	+
	1108	+	assert_eq!(base(&db, "r1"), serde_json::Value::Null);
	1109	+	}
	1110	+
	1111	+	fn merge_schema() -> SyncSchema {
	1112	+	SyncSchema::new(vec![
	1113	+	SyncTable::full("note", &["id", "name"]).field_merge(&[]),
	1114	+	])
	1115	+	}
	1116	+
	1117	+	fn base(db: &DbSource, row_id: &str) -> serde_json::Value {
	1118	+	super::snapshot::load(&db.open().unwrap(), "note", row_id)
	1119	+	}
	1120	+
1015	1121		#[test]
1016	1122		fn initial_snapshot_captures_existing_rows_once() {
1017	1123		let dir = tempdir();

A synckit-client/src/store/snapshot.rs +160

		@@ -1,0 +1,160 @@
1	+	//! Last-synced row snapshots: the base version a three-way field merge needs.
2	+	//!
3	+	//! [`resolve_field_merge`](crate::conflict::resolve_field_merge) has always been
4	+	//! able to merge two edits that touched different columns, and nothing could call
5	+	//! it, because a three-way merge needs the version both devices started from and
6	+	//! the engine kept no such version. This is that version.
7	+	//!
8	+	//! A snapshot is the wire payload of the last change for a row that **both
9	+	//! devices have seen**, which is exactly the two moments a row becomes common
10	+	//! ground: a remote change this device applied, and a local change the server
11	+	//! acknowledged. Written at both, so the base re-bases itself on every sync and
12	+	//! never needs a separate reconciliation pass.
13	+	//!
14	+	//! The payload is stored rather than the row read back, deliberately. A merge
15	+	//! compares two payloads against the base, and a row read carries columns the
16	+	//! wire never sends (`preserve_local` secrets, group provenance, anything outside
17	+	//! the manifest). Basing a merge on those would report a field as "changed" on
18	+	//! every device that has a different local secret. The payload is the only shape
19	+	//! all three sides share.
20	+	//!
21	+	//! Opt-in per table ([`SyncTable::field_merge`](super::schema::SyncTable::field_merge)):
22	+	//! a table nobody opted in stores nothing here and resolves conflicts exactly as
23	+	//! it did before. Storage roughly doubles for the tables that do opt in, which is
24	+	//! why it is not simply on for everything.
25	+
26	+	use rusqlite::{Connection, OptionalExtension};
27	+	use serde_json::Value;
28	+
29	+	use crate::error::Result;
30	+
31	+	/// DDL for the snapshot store, shared by
32	+	/// [`SyncSchema::migration_sql`](super::schema::SyncSchema::migration_sql) and the
33	+	/// per-connection upgrade in [`db::ensure_scope_schema`](super::db::ensure_scope_schema),
34	+	/// so an install that predates field merge gets the table on its next connection
35	+	/// open rather than waiting for the app to re-run its migration.
36	+	///
37	+	/// One row per `(table_name, row_id)`. Not scoped: a row lives in one scope at a
38	+	/// time and moving it between scopes rewrites the row, so a per-scope base would
39	+	/// be a second copy of the same answer.
40	+	pub(crate) const SNAPSHOT_DDL: &str = "\
41	+	CREATE TABLE IF NOT EXISTS sync_row_snapshot (
42	+	table_name TEXT NOT NULL,
43	+	row_id TEXT NOT NULL,
44	+	payload TEXT NOT NULL,
45	+	updated_at TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%fZ', 'now')),
46	+	PRIMARY KEY (table_name, row_id)
47	+	) WITHOUT ROWID;
48	+	";
49	+
50	+	/// Record `payload` as the last-synced base for a row.
51	+	///
52	+	/// Called on both sides of common ground: after a remote change is applied, and
53	+	/// after a local change is acknowledged by a push. A non-object payload is not
54	+	/// stored, since it could never serve as a merge base.
55	+	pub(crate) fn record(conn: &Connection, table: &str, row_id: &str, payload: &Value) -> Result<()> {
56	+	if !payload.is_object() {
57	+	return Ok(());
58	+	}
59	+	conn.execute(
60	+	"INSERT INTO sync_row_snapshot (table_name, row_id, payload) VALUES (?1, ?2, ?3) \
61	+	ON CONFLICT(table_name, row_id) DO UPDATE SET \
62	+	payload = excluded.payload, \
63	+	updated_at = strftime('%Y-%m-%dT%H:%M:%fZ', 'now')",
64	+	rusqlite::params![table, row_id, serde_json::to_string(payload)?],
65	+	)?;
66	+	Ok(())
67	+	}
68	+
69	+	/// Drop a row's base, because the row is gone.
70	+	///
71	+	/// A stale base outliving its row would be handed to the merge if the same key
72	+	/// were later recreated, and it would describe a version of a different row.
73	+	pub(crate) fn forget(conn: &Connection, table: &str, row_id: &str) -> Result<()> {
74	+	conn.execute(
75	+	"DELETE FROM sync_row_snapshot WHERE table_name = ?1 AND row_id = ?2",
76	+	rusqlite::params![table, row_id],
77	+	)?;
78	+	Ok(())
79	+	}
80	+
81	+	/// The base for a row, or [`Value::Null`] when there is none.
82	+	///
83	+	/// Null rather than an error or an `Option` because that is what the merge takes:
84	+	/// [`resolve_field_merge`](crate::conflict::resolve_field_merge) treats a
85	+	/// non-object base as "no usable base" and falls back on its own. A read error is
86	+	/// logged and reported as absent, since a merge this device cannot base is a
87	+	/// merge that should not happen, not a sync that should fail.
88	+	pub(crate) fn load(conn: &Connection, table: &str, row_id: &str) -> Value {
89	+	let stored: Option<String> = match conn
90	+	.query_row(
91	+	"SELECT payload FROM sync_row_snapshot WHERE table_name = ?1 AND row_id = ?2",
92	+	rusqlite::params![table, row_id],
93	+	\|r\| r.get(0),
94	+	)
95	+	.optional()
96	+	{
97	+	Ok(v) => v,
98	+	Err(e) => {
99	+	tracing::warn!(
100	+	table,
101	+	row_id,
102	+	"snapshot lookup failed, treating as absent: {e}"
103	+	);
104	+	None
105	+	}
106	+	};
107	+	stored
108	+	.and_then(\|s\| serde_json::from_str(&s).ok())
109	+	.unwrap_or(Value::Null)
110	+	}
111	+
112	+	#[cfg(test)]
113	+	mod tests {
114	+	use super::*;
115	+	use serde_json::json;
116	+
117	+	fn conn() -> Connection {
118	+	let conn = Connection::open_in_memory().unwrap();
119	+	conn.execute_batch(SNAPSHOT_DDL).unwrap();
120	+	conn
121	+	}
122	+
123	+	#[test]
124	+	fn round_trips_and_overwrites() {
125	+	let c = conn();
126	+	record(&c, "note", "r1", &json!({"name": "one"})).unwrap();
127	+	assert_eq!(load(&c, "note", "r1"), json!({"name": "one"}));
128	+
129	+	record(&c, "note", "r1", &json!({"name": "two"})).unwrap();
130	+	assert_eq!(
131	+	load(&c, "note", "r1"),
132	+	json!({"name": "two"}),
133	+	"a later sync must re-base the row, not accumulate versions"
134	+	);
135	+	}
136	+
137	+	#[test]
138	+	fn absent_and_forgotten_rows_read_as_null() {
139	+	let c = conn();
140	+	assert_eq!(load(&c, "note", "missing"), Value::Null);
141	+
142	+	record(&c, "note", "r1", &json!({"name": "one"})).unwrap();
143	+	forget(&c, "note", "r1").unwrap();
144	+	assert_eq!(
145	+	load(&c, "note", "r1"),
146	+	Value::Null,
147	+	"a deleted row's base must not survive to be merged against"
148	+	);
149	+	}
150	+
151	+	/// A delete carries no object payload, and an entry with no payload has no
152	+	/// base to offer. Storing one would put a scalar where the merge expects an
153	+	/// object.
154	+	#[test]
155	+	fn a_non_object_payload_is_not_stored() {
156	+	let c = conn();
157	+	record(&c, "note", "r1", &json!("not an object")).unwrap();
158	+	assert_eq!(load(&c, "note", "r1"), Value::Null);
159	+	}
160	+	}