Skip to main content

max / synckit

52.5 KB · 1584 lines History Blame Raw
1 //! Tests for [`super`].
2
3 use super::super::apply::apply_remote_changes;
4 use super::super::db::configure_connection;
5 use super::super::schema::{SyncSchema, SyncTable};
6 use super::*;
7 use rusqlite::Connection;
8
9 fn node(n: u128) -> DeviceId {
10 DeviceId::new(Uuid::from_u128(n))
11 }
12
13 fn schema() -> SyncSchema {
14 SyncSchema::new(vec![SyncTable::full("note", &["id", "name"])])
15 }
16
17 fn server_order_schema() -> SyncSchema {
18 schema().conflict_strategy(ConflictStrategy::ServerOrder)
19 }
20
21 /// A device: in-memory DB with the note table + migration, and a node id.
22 fn device(n: u128) -> (Connection, DeviceId) {
23 let conn = Connection::open_in_memory().unwrap();
24 configure_connection(&conn).unwrap();
25 conn.execute_batch("CREATE TABLE note (id TEXT PRIMARY KEY, name TEXT);")
26 .unwrap();
27 conn.execute_batch(&schema().migration_sql()).unwrap();
28 (conn, node(n))
29 }
30
31 /// Make a local edit (domain write → trigger captures it), stamp it at
32 /// `now_ms`, and return it as a pulled change from `node` (as a peer would
33 /// receive it via the server).
34 fn local_edit_as_pulled(
35 conn: &Connection,
36 node: DeviceId,
37 id: &str,
38 name: &str,
39 now_ms: i64,
40 seq: i64,
41 ) -> PulledChange {
42 conn.execute(
43 "INSERT INTO note (id, name) VALUES (?1, ?2) \
44 ON CONFLICT(id) DO UPDATE SET name = excluded.name",
45 (id, name),
46 )
47 .unwrap();
48 stamp_pending(conn, node, now_ms).unwrap();
49 let entry = load_local_pending(conn, node)
50 .unwrap()
51 .into_iter()
52 .find(|e| e.row_id == id)
53 .unwrap();
54 PulledChange {
55 storage_version: None,
56 entry,
57 device_id: node,
58 seq,
59 }
60 }
61
62 /// A bare entry for the collapse tests: only table, row_id, op and HLC
63 /// matter there, so the payload names the entry for the assertion message.
64 fn entry(row_id: &str, op: ChangeOp, wall_ms: i64, node_n: u128, label: &str) -> ChangeEntry {
65 ChangeEntry {
66 table: "note".into(),
67 op,
68 row_id: row_id.into(),
69 timestamp: Utc::now(),
70 hlc: Hlc {
71 wall_ms,
72 counter: 0,
73 node: node(node_n),
74 },
75 data: Some(serde_json::json!({ "name": label })),
76 extra: serde_json::Map::default(),
77 }
78 }
79
80 fn labels(entries: &[ChangeEntry]) -> Vec<String> {
81 entries
82 .iter()
83 .map(|e| e.data.as_ref().unwrap()["name"].as_str().unwrap().into())
84 .collect()
85 }
86
87 /// The collapse keeps the highest HLC per row, and it has to do so whichever
88 /// order the entries arrive in. Both directions are asserted because the
89 /// obvious way to get this wrong, comparing in the wrong direction, is
90 /// invisible when only the already-sorted order is tested: it then keeps the
91 /// last entry, which is also the newest.
92 #[test]
93 fn collapse_keeps_the_highest_hlc_per_row_in_either_order() {
94 let older = entry("r1", ChangeOp::Update, 100, 1, "older");
95 let newer = entry("r1", ChangeOp::Update, 200, 1, "newer");
96
97 let ascending = collapse_max_hlc(vec![older.clone(), newer.clone()]);
98 assert_eq!(labels(&ascending), ["newer"], "newest lost, arriving last");
99
100 let descending = collapse_max_hlc(vec![newer, older]);
101 assert_eq!(
102 labels(&descending),
103 ["newer"],
104 "newest lost, arriving first"
105 );
106 }
107
108 /// Operation-agnostic: a newer delete beats an older edit and an older
109 /// delete loses to a newer edit. The HLC decides, never the operation.
110 #[test]
111 fn collapse_ignores_the_operation() {
112 let newer_delete = collapse_max_hlc(vec![
113 entry("r1", ChangeOp::Update, 100, 1, "edit"),
114 entry("r1", ChangeOp::Delete, 200, 1, "delete"),
115 ]);
116 assert_eq!(newer_delete.len(), 1);
117 assert_eq!(newer_delete[0].op, ChangeOp::Delete);
118
119 let older_delete = collapse_max_hlc(vec![
120 entry("r1", ChangeOp::Delete, 100, 1, "delete"),
121 entry("r1", ChangeOp::Update, 200, 1, "edit"),
122 ]);
123 assert_eq!(older_delete.len(), 1);
124 assert_eq!(older_delete[0].op, ChangeOp::Update);
125 }
126
127 /// The collapse is per row: distinct rows all survive, and first-seen order
128 /// is preserved, which is what the doc comment promises the apply engine.
129 #[test]
130 fn collapse_is_per_row_and_keeps_first_seen_order() {
131 let out = collapse_max_hlc(vec![
132 entry("r2", ChangeOp::Update, 100, 1, "r2-old"),
133 entry("r1", ChangeOp::Update, 100, 1, "r1-only"),
134 entry("r2", ChangeOp::Update, 200, 1, "r2-new"),
135 ]);
136 assert_eq!(labels(&out), ["r2-new", "r1-only"]);
137 }
138
139 /// The case the shared order exists for. Two changes for one row at an
140 /// exact HLC tie: the winner has to be the same on every device, and the
141 /// only thing every device agrees on is the payload bytes. Arrival order is
142 /// not that thing, so the two orders must agree here.
143 #[test]
144 fn collapse_breaks_an_exact_hlc_tie_on_payload_not_arrival_order() {
145 let a = entry("r1", ChangeOp::Update, 100, 1, "aaa");
146 let b = entry("r1", ChangeOp::Update, 100, 1, "bbb");
147 assert_eq!(a.hlc, b.hlc, "the tie is the premise of this test");
148
149 let forwards = collapse_max_hlc(vec![a.clone(), b.clone()]);
150 let backwards = collapse_max_hlc(vec![b, a]);
151 assert_eq!(
152 labels(&forwards),
153 labels(&backwards),
154 "two devices disagreed at an exact tie because they saw the batch in different orders"
155 );
156 assert_eq!(labels(&forwards), ["bbb"], "higher payload bytes win");
157 }
158
159 fn note_name(conn: &Connection, id: &str) -> Option<String> {
160 conn.query_row("SELECT name FROM note WHERE id = ?1", [id], |r| r.get(0))
161 .optional()
162 .unwrap()
163 }
164
165 fn pull_apply(conn: &mut Connection, s: &SyncSchema, node: DeviceId, pulled: Vec<PulledChange>) {
166 let now = Utc::now();
167 let resolved = resolve_pull(conn, s, node, pulled, now, "").unwrap();
168 apply_remote_changes(conn, s, &resolved, "").unwrap();
169 record_committed(conn, resolved.as_slice()).unwrap();
170 }
171
172 #[test]
173 fn stamp_pending_assigns_monotonic_hlcs() {
174 let (conn, n) = device(1);
175 conn.execute("INSERT INTO note (id, name) VALUES ('a', '1')", [])
176 .unwrap();
177 conn.execute("INSERT INTO note (id, name) VALUES ('b', '2')", [])
178 .unwrap();
179 assert_eq!(stamp_pending(&conn, n, 1000).unwrap(), 2);
180 // Re-stamping is a no-op (both already stamped).
181 assert_eq!(stamp_pending(&conn, n, 2000).unwrap(), 0);
182 let stamps: Vec<(i64, i64)> = {
183 let mut s = conn
184 .prepare("SELECT hlc_wall, hlc_counter FROM sync_changelog ORDER BY id")
185 .unwrap();
186 s.query_map([], |r| Ok((r.get(0)?, r.get(1)?)))
187 .unwrap()
188 .map(|r| r.unwrap())
189 .collect()
190 };
191 assert!(stamps.iter().all(|(w, _)| *w == 1000));
192 assert_eq!(
193 stamps[0].1 + 1,
194 stamps[1].1,
195 "counter increments within a wall_ms"
196 );
197 }
198
199 #[test]
200 fn committed_ledger_advances_only() {
201 let (conn, _) = device(1);
202 let older = Hlc {
203 wall_ms: 100,
204 counter: 0,
205 node: node(9),
206 };
207 let newer = Hlc {
208 wall_ms: 200,
209 counter: 0,
210 node: node(9),
211 };
212 set_committed(&conn, "note", "r", &newer).unwrap();
213 set_committed(&conn, "note", "r", &older).unwrap(); // must not regress
214 assert_eq!(
215 committed_hlc(&conn, "note", "r").unwrap().unwrap().wall_ms,
216 200
217 );
218 }
219
220 #[test]
221 fn conflicting_edits_converge_to_higher_hlc_both_directions() {
222 // A edits at t=100, B edits at t=200 → B wins everywhere.
223 let (mut a, an) = device(1);
224 let (mut b, bn) = device(2);
225 let a_change = local_edit_as_pulled(&a, an, "r", "from-A", 100, 1);
226 let b_change = local_edit_as_pulled(&b, bn, "r", "from-B", 200, 1);
227
228 pull_apply(&mut a, &schema(), an, vec![b_change]); // A pulls B (newer) → adopts B
229 pull_apply(&mut b, &schema(), bn, vec![a_change]); // B pulls A (older) → keeps B
230
231 assert_eq!(note_name(&a, "r").as_deref(), Some("from-B"));
232 assert_eq!(note_name(&b, "r").as_deref(), Some("from-B"));
233 }
234
235 #[test]
236 fn gate_drops_repulled_older_change() {
237 let (mut a, an) = device(1);
238 let (b, bn) = device(2);
239 // B's change is applied on A.
240 let b_change = local_edit_as_pulled(&b, bn, "r", "v-old", 100, 1);
241 pull_apply(&mut a, &schema(), an, vec![b_change.clone()]);
242 // A then makes a NEWER local edit and commits it.
243 let _a_new = local_edit_as_pulled(&a, an, "r", "v-new", 300, 2);
244 // Mark A's edit committed (as a push would) so the gate has a committed clock.
245 let a_pending = load_local_pending(&a, an);
246 record_committed(&a, &a_pending.unwrap()).unwrap();
247 // Re-pulling B's OLD change must be gated out (older than committed).
248 let resolved = resolve_pull(&a, &schema(), an, vec![b_change], Utc::now(), "").unwrap();
249 assert!(
250 resolved.iter().all(|e| e.row_id != "r"),
251 "stale re-pull must be gated"
252 );
253 }
254
255 #[test]
256 fn newer_delete_beats_older_edit() {
257 let (mut a, an) = device(1);
258 let (b, bn) = device(2);
259 // A has an older edit locally.
260 local_edit_as_pulled(&a, an, "r", "edit", 100, 1);
261 // B deletes the same row, newer.
262 b.execute("INSERT INTO note (id, name) VALUES ('r', 'x')", [])
263 .unwrap();
264 stamp_pending(&b, bn, 50).unwrap();
265 b.execute("DELETE FROM note WHERE id = 'r'", []).unwrap();
266 stamp_pending(&b, bn, 200).unwrap();
267 let del = load_local_pending(&b, bn)
268 .unwrap()
269 .into_iter()
270 .find(|e| e.op == ChangeOp::Delete)
271 .unwrap();
272 let pulled = PulledChange {
273 storage_version: None,
274 entry: del,
275 device_id: bn,
276 seq: 2,
277 };
278 pull_apply(&mut a, &schema(), an, vec![pulled]);
279 assert_eq!(
280 note_name(&a, "r"),
281 None,
282 "newer delete wins over older edit"
283 );
284 }
285
286 #[test]
287 fn server_order_applies_last_delivered_no_hlc() {
288 let (mut a, an) = device(1);
289 let s = server_order_schema();
290 // Two pulled changes for the same row; server order = last wins, HLC ignored.
291 let (src, sn) = device(2);
292 let first = local_edit_as_pulled(&src, sn, "r", "first", 999, 1); // higher wall
293 let (src2, sn2) = device(3);
294 let second = local_edit_as_pulled(&src2, sn2, "r", "second", 1, 2); // lower wall, later seq
295 let resolved = resolve_pull(&a, &s, an, vec![first, second], Utc::now(), "").unwrap();
296 apply_remote_changes(&mut a, &s, &resolved, "").unwrap();
297 assert_eq!(
298 note_name(&a, "r").as_deref(),
299 Some("second"),
300 "server order: last delivered wins"
301 );
302 }
303
304 /// The two guarantees `ResolvedChanges` documents, asserted on the same
305 /// input so the difference between them is the only variable. Under the HLC
306 /// strategy a batch carrying two changes for one row resolves to one entry;
307 /// under `ServerOrder` it deliberately stays two, because last-delivered-wins
308 /// is what an app choosing that strategy asked for.
309 #[test]
310 fn resolve_pull_collapses_a_row_under_hlc_and_does_not_under_server_order() {
311 let (src, sn) = device(2);
312 let first = local_edit_as_pulled(&src, sn, "r", "first", 100, 1);
313 let second = local_edit_as_pulled(&src, sn, "r", "second", 200, 2);
314
315 let (hlc_device, hn) = device(1);
316 let under_hlc = resolve_pull(
317 &hlc_device,
318 &schema(),
319 hn,
320 vec![first.clone(), second.clone()],
321 Utc::now(),
322 "",
323 )
324 .unwrap();
325 assert_eq!(
326 under_hlc.len(),
327 1,
328 "the HLC strategy promises one entry per row; the apply order would \
329 otherwise decide the value"
330 );
331
332 let (server_device, svn) = device(3);
333 let under_server_order = resolve_pull(
334 &server_device,
335 &server_order_schema(),
336 svn,
337 vec![first, second],
338 Utc::now(),
339 "",
340 )
341 .unwrap();
342 assert_eq!(
343 under_server_order.len(),
344 2,
345 "ServerOrder must not collapse: last delivered wins is the strategy"
346 );
347 }
348
349 /// A model of the pull pipeline.
350 ///
351 /// Aimed at [`resolve_pull`] with a real `Connection`, deliberately, and not
352 /// at the pure conflict layer one step down. The one-entry-per-row invariant
353 /// only exists after the collapse, so `resolve_pull` is the lowest layer
354 /// where a max-HLC-wins specification is an honest thing to assert.
355 /// `CleanChanges::gated_at`, one layer down, promises only committed-clock
356 /// filtering, so three of these four properties do not hold there even
357 /// against correct code. See wiki `testing-posture`, Phase 3.
358 ///
359 /// The four properties are the ones a sync engine lives or dies on, and each
360 /// is a different way for two devices to end up holding different bytes.
361 mod model {
362 use super::*;
363 use proptest::prelude::*;
364
365 /// A fixed instant, so the poisoning guard is deterministic. Generated
366 /// walls sit near it and well inside `MAX_HLC_DRIFT_MS`; poisoning is
367 /// covered by its own example test, and letting it fire here would mean
368 /// the properties were quietly asserting over an empty batch.
369 const BASE_MS: i64 = 1_700_000_000_000;
370
371 fn now() -> DateTime<Utc> {
372 DateTime::from_timestamp_millis(BASE_MS).unwrap()
373 }
374
375 /// Three rows, three devices, three wall readings, and every range here
376 /// is narrow on purpose.
377 ///
378 /// The interesting case is two changes for one row at an *exact* HLC
379 /// tie with differing payloads, because that is the only case the
380 /// payload tiebreak in `change_order` serves. A wider generator makes it
381 /// unreachable: a first attempt drew walls from a 100ms window and
382 /// produced roughly one tie across a whole 256-case run, few enough that
383 /// deleting the tiebreak outright left every property passing. Ties have
384 /// to be common for these properties to observe anything, so the clock
385 /// is generated with almost no entropy in it and the payload carries the
386 /// variation instead.
387 fn batch() -> impl Strategy<Value = Vec<PulledChange>> {
388 let one = (0u8..3, 0u8..3, 0i64..3, 0u32..2, 0u8..4).prop_map(
389 |(row, dev, wall_off, counter, payload)| {
390 let node = node(u128::from(dev));
391 PulledChange {
392 storage_version: None,
393 entry: ChangeEntry {
394 table: "note".into(),
395 op: ChangeOp::Update,
396 row_id: format!("r{row}"),
397 timestamp: now(),
398 hlc: Hlc {
399 wall_ms: BASE_MS + wall_off,
400 counter,
401 node,
402 },
403 data: Some(serde_json::json!({
404 "id": format!("r{row}"),
405 "name": format!("v{payload}"),
406 })),
407 extra: serde_json::Map::default(),
408 },
409 device_id: node,
410 seq: 0,
411 }
412 },
413 );
414 proptest::collection::vec(one, 0..6)
415 }
416
417 /// The whole observable state of a device: what each row holds, and what
418 /// the committed ledger says about it. Both matter. A pipeline that
419 /// wrote the right value but recorded the wrong committed clock would
420 /// gate its own next pull incorrectly, and comparing only the rows would
421 /// not see it.
422 fn state(conn: &Connection) -> Vec<(String, Option<String>, Option<Hlc>)> {
423 ["r0", "r1", "r2"]
424 .iter()
425 .map(|r| {
426 (
427 (*r).to_string(),
428 note_name(conn, r),
429 committed_hlc(conn, "note", r).unwrap(),
430 )
431 })
432 .collect()
433 }
434
435 /// Run a batch through the real pipeline on a fresh device.
436 fn pull(conn: &mut Connection, node: DeviceId, pulled: Vec<PulledChange>) {
437 let s = schema();
438 let resolved = resolve_pull(conn, &s, node, pulled, now(), "").unwrap();
439 apply_remote_changes(conn, &s, &resolved, "").unwrap();
440 record_committed(conn, resolved.as_slice()).unwrap();
441 }
442
443 /// The specification the pipeline is supposed to implement: per row, the
444 /// highest wall clock wins, then the highest counter, then the highest
445 /// device, then the highest payload bytes.
446 ///
447 /// Spelled out rather than delegated to `change_order`, and that is the
448 /// whole point of it. An oracle that called `change_order` would agree
449 /// with a broken `change_order`, which is the imitation-oracle failure
450 /// from wiki `testing-posture` wearing a different hat: it was the first
451 /// version of this function, and deleting the payload tiebreak left all
452 /// four properties passing. This version fails when the rule changes,
453 /// because it is a second statement of the rule rather than a reference
454 /// to the first.
455 fn expected_winner(pulled: &[PulledChange], row: &str) -> Option<String> {
456 fn rank(e: &ChangeEntry) -> (i64, u32, Uuid, Vec<u8>) {
457 (
458 e.hlc.wall_ms,
459 e.hlc.counter,
460 e.hlc.node.as_uuid(),
461 serde_json::to_vec(e.data.as_ref().unwrap()).unwrap(),
462 )
463 }
464 pulled
465 .iter()
466 .map(|p| &p.entry)
467 .filter(|e| e.row_id == row)
468 .max_by_key(|e| rank(e))
469 .map(|e| e.data.as_ref().unwrap()["name"].as_str().unwrap().into())
470 }
471
472 proptest! {
473 /// **The pipeline implements max-HLC-wins.** The value each row ends
474 /// up holding is the one from the change that wins under
475 /// `change_order`, whatever else the batch contained.
476 #[test]
477 fn final_value_is_the_winner_under_change_order(pulled in batch()) {
478 let (mut conn, n) = device(1);
479 pull(&mut conn, n, pulled.clone());
480 for row in ["r0", "r1", "r2"] {
481 prop_assert_eq!(
482 note_name(&conn, row),
483 expected_winner(&pulled, row),
484 "row {} does not hold the winner",
485 row
486 );
487 }
488 }
489
490 /// **Batch order does not change the final state.** The server may
491 /// deliver a batch in any order; two devices that see the same
492 /// changes in different orders must agree afterwards. This is the
493 /// property the whole change-ordering unification was for.
494 #[test]
495 fn order_within_a_batch_does_not_matter(pulled in batch()) {
496 let (mut forwards, fnode) = device(1);
497 pull(&mut forwards, fnode, pulled.clone());
498
499 let mut reversed_batch = pulled;
500 reversed_batch.reverse();
501 let (mut backwards, bnode) = device(1);
502 pull(&mut backwards, bnode, reversed_batch);
503
504 prop_assert_eq!(
505 state(&forwards),
506 state(&backwards),
507 "two devices diverged on batch order alone"
508 );
509 }
510
511 /// **Replaying a batch is a no-op.** A pull that is retried, or a
512 /// cursor that rewinds, must not change anything the first pass
513 /// already settled. This is what the committed ledger exists for.
514 #[test]
515 fn replaying_a_batch_changes_nothing(pulled in batch()) {
516 let (mut conn, n) = device(1);
517 pull(&mut conn, n, pulled.clone());
518 let after_first = state(&conn);
519 pull(&mut conn, n, pulled);
520 prop_assert_eq!(after_first, state(&conn), "a replayed batch moved the state");
521 }
522
523 /// **Committed clocks never go backwards.** The ledger is what gates
524 /// stale re-pulls, so a regression there un-gates a change the device
525 /// already superseded, and an old value overwrites a newer one.
526 ///
527 /// Two independent mechanisms hold this up, and this property
528 /// observes the pair rather than either one: the committed-HLC gate
529 /// drops a stale change before `set_committed` sees it, and
530 /// `set_committed` refuses to regress even if one reaches it.
531 /// Breaking either alone leaves this passing, which is what
532 /// defense in depth means and is worth knowing before trusting a
533 /// green run here; breaking both fails it. The individual clause in
534 /// `set_committed` has its own example test,
535 /// `committed_ledger_advances_only`.
536 #[test]
537 fn committed_clocks_only_advance(first in batch(), second in batch()) {
538 let (mut conn, n) = device(1);
539 pull(&mut conn, n, first);
540 let before: Vec<Option<Hlc>> =
541 state(&conn).into_iter().map(|(_, _, h)| h).collect();
542 pull(&mut conn, n, second);
543 let after: Vec<Option<Hlc>> =
544 state(&conn).into_iter().map(|(_, _, h)| h).collect();
545
546 for (before, after) in before.into_iter().zip(after) {
547 match (before, after) {
548 (Some(b), Some(a)) => prop_assert!(
549 a >= b,
550 "committed clock went backwards: {:?} -> {:?}",
551 b, a
552 ),
553 (Some(b), None) => {
554 prop_assert!(false, "committed clock for a row disappeared: {:?}", b);
555 }
556 _ => {}
557 }
558 }
559 }
560 }
561 }
562
563 // ── Conflict stash ──
564 //
565 // LWW always discards one side; these pin that the discarded bytes are kept
566 // rather than dropped. The one that matters most is the superseded case,
567 // which never becomes a ConflictPair and so is invisible to resolve_lww.
568
569 #[derive(Debug, PartialEq)]
570 struct StashRow {
571 table_name: String,
572 row_id: String,
573 scope: String,
574 losing_side: String,
575 losing_payload: Option<String>,
576 }
577
578 fn stash_rows(conn: &Connection) -> Vec<StashRow> {
579 let mut stmt = conn
580 .prepare(
581 "SELECT table_name, row_id, scope, losing_side, losing_payload
582 FROM sync_conflict_stash ORDER BY id",
583 )
584 .unwrap();
585 stmt.query_map([], |r| {
586 Ok(StashRow {
587 table_name: r.get(0)?,
588 row_id: r.get(1)?,
589 scope: r.get(2)?,
590 losing_side: r.get(3)?,
591 losing_payload: r.get(4)?,
592 })
593 })
594 .unwrap()
595 .map(|r| r.unwrap())
596 .collect()
597 }
598
599 /// A remote change with an explicit HLC and payload, as a peer would send it.
600 fn remote_change(from: DeviceId, id: &str, name: &str, hlc: Hlc, seq: i64) -> PulledChange {
601 PulledChange {
602 storage_version: None,
603 entry: ChangeEntry {
604 table: "note".to_string(),
605 op: ChangeOp::Update,
606 row_id: id.to_string(),
607 timestamp: Utc::now(),
608 hlc,
609 data: Some(serde_json::json!({ "id": id, "name": name })),
610 extra: serde_json::Map::default(),
611 },
612 device_id: from,
613 seq,
614 }
615 }
616
617 /// Site 1: a newer remote change beats our pending edit, so our edit is the
618 /// one that vanishes from the row. It must be recoverable.
619 #[test]
620 fn stash_keeps_our_own_edit_when_the_remote_wins() {
621 let (mut conn, n) = device(1);
622 let peer = node(2);
623
624 // Our pending edit, stamped early so it loses.
625 conn.execute("INSERT INTO note (id, name) VALUES ('n1', 'mine')", [])
626 .unwrap();
627 stamp_pending(&conn, n, 1_000).unwrap();
628
629 let newer = Hlc {
630 wall_ms: 9_000,
631 counter: 0,
632 node: peer,
633 };
634 let resolved = resolve_pull(
635 &conn,
636 &schema(),
637 n,
638 vec![remote_change(peer, "n1", "theirs", newer, 1)],
639 Utc::now(),
640 "",
641 )
642 .unwrap();
643 apply_remote_changes(&mut conn, &schema(), &resolved, "").unwrap();
644
645 assert_eq!(note_name(&conn, "n1").as_deref(), Some("theirs"));
646 let rows = stash_rows(&conn);
647 assert_eq!(
648 rows.len(),
649 1,
650 "our discarded edit must be stashed: {rows:?}"
651 );
652 assert_eq!(rows[0].losing_side, "local");
653 assert_eq!(rows[0].row_id, "n1");
654 assert!(
655 rows[0].losing_payload.as_deref().unwrap().contains("mine"),
656 "the stash must hold the discarded value, not a placeholder: {rows:?}"
657 );
658 }
659
660 /// Site 2: our pending edit wins, so the other writer's change is discarded.
661 /// Their bytes are the ones that need keeping.
662 #[test]
663 fn stash_keeps_the_remote_edit_when_ours_wins() {
664 let (conn, n) = device(1);
665 let peer = node(2);
666
667 conn.execute("INSERT INTO note (id, name) VALUES ('n1', 'mine')", [])
668 .unwrap();
669 stamp_pending(&conn, n, 9_000).unwrap();
670
671 let older = Hlc {
672 wall_ms: 1_000,
673 counter: 0,
674 node: peer,
675 };
676 let resolved = resolve_pull(
677 &conn,
678 &schema(),
679 n,
680 vec![remote_change(peer, "n1", "theirs", older, 1)],
681 Utc::now(),
682 "",
683 )
684 .unwrap();
685 assert!(
686 resolved.is_empty(),
687 "the older remote change must not apply"
688 );
689
690 let rows = stash_rows(&conn);
691 assert_eq!(
692 rows.len(),
693 1,
694 "their discarded edit must be stashed: {rows:?}"
695 );
696 assert_eq!(rows[0].losing_side, "remote");
697 assert!(
698 rows[0]
699 .losing_payload
700 .as_deref()
701 .unwrap()
702 .contains("theirs")
703 );
704 }
705
706 /// Site 3, the quiet one: no local pending edit contests the row, so no
707 /// ConflictPair is ever built. The change is dropped by the committed-HLC
708 /// gate alone, which is why a stash wired only into the match arms misses it.
709 #[test]
710 fn stash_keeps_a_change_superseded_by_the_committed_clock() {
711 let (mut conn, n) = device(1);
712 let peer = node(2);
713
714 // Apply and commit a newer value, with nothing left pending afterwards.
715 let newer = Hlc {
716 wall_ms: 9_000,
717 counter: 0,
718 node: peer,
719 };
720 let resolved = resolve_pull(
721 &conn,
722 &schema(),
723 n,
724 vec![remote_change(peer, "n1", "current", newer, 1)],
725 Utc::now(),
726 "",
727 )
728 .unwrap();
729 apply_remote_changes(&mut conn, &schema(), &resolved, "").unwrap();
730 record_committed(&conn, resolved.as_slice()).unwrap();
731 assert!(stash_rows(&conn).is_empty(), "nothing lost yet");
732
733 // Now pull an older change for the same row. No pending edit contests it.
734 let older = Hlc {
735 wall_ms: 1_000,
736 counter: 0,
737 node: peer,
738 };
739 let resolved = resolve_pull(
740 &conn,
741 &schema(),
742 n,
743 vec![remote_change(peer, "n1", "stale", older, 2)],
744 Utc::now(),
745 "",
746 )
747 .unwrap();
748 assert!(resolved.is_empty(), "the superseded change must not apply");
749 assert_eq!(note_name(&conn, "n1").as_deref(), Some("current"));
750
751 let rows = stash_rows(&conn);
752 assert_eq!(
753 rows.len(),
754 1,
755 "the superseded change must be stashed: {rows:?}"
756 );
757 assert_eq!(rows[0].losing_side, "remote");
758 assert!(rows[0].losing_payload.as_deref().unwrap().contains("stale"));
759 }
760
761 /// An echo carries the same bytes as the winner, so nothing was lost. Without
762 /// this the stash fills with no-ops and stops being worth reading.
763 #[test]
764 fn stash_ignores_a_conflict_whose_payloads_are_identical() {
765 let (conn, n) = device(1);
766 let peer = node(2);
767
768 conn.execute("INSERT INTO note (id, name) VALUES ('n1', 'same')", [])
769 .unwrap();
770 stamp_pending(&conn, n, 1_000).unwrap();
771 let local = load_local_pending(&conn, n).unwrap().pop().unwrap();
772
773 // Byte-identical payload, newer clock: the remote wins and our identical
774 // value goes away, which costs nothing.
775 let mut remote = remote_change(
776 peer,
777 "n1",
778 "x",
779 Hlc {
780 wall_ms: 9_000,
781 counter: 0,
782 node: peer,
783 },
784 1,
785 );
786 remote.entry.data = local.data.clone();
787
788 resolve_pull(&conn, &schema(), n, vec![remote], Utc::now(), "").unwrap();
789 assert!(
790 stash_rows(&conn).is_empty(),
791 "an identical payload is not a lost edit: {:?}",
792 stash_rows(&conn)
793 );
794 }
795
796 /// A clock-poisoned entry is refused as hostile, not outvoted. Stashing it
797 /// would hand whoever sent it a way to fill the stash.
798 #[test]
799 fn stash_ignores_a_clock_poisoned_drop() {
800 let (conn, n) = device(1);
801 let peer = node(2);
802 let now = Utc::now();
803 let poisoned = Hlc {
804 wall_ms: now.timestamp_millis() + crate::conflict::MAX_HLC_DRIFT_MS * 10,
805 counter: 0,
806 node: peer,
807 };
808
809 let resolved = resolve_pull(
810 &conn,
811 &schema(),
812 n,
813 vec![remote_change(peer, "n1", "from the future", poisoned, 1)],
814 now,
815 "",
816 )
817 .unwrap();
818 assert!(resolved.is_empty(), "a poisoned entry must not apply");
819 assert!(
820 stash_rows(&conn).is_empty(),
821 "a poisoned entry is refused, not outvoted: {:?}",
822 stash_rows(&conn)
823 );
824 }
825
826 /// The scope a loss happened in is recorded, so a consuming app can tell a
827 /// personal conflict from one inside a shared group.
828 #[test]
829 fn stash_records_the_scope() {
830 let (conn, n) = device(1);
831 let peer = node(2);
832
833 conn.execute("INSERT INTO note (id, name) VALUES ('n1', 'mine')", [])
834 .unwrap();
835 stamp_pending(&conn, n, 9_000).unwrap();
836
837 let older = Hlc {
838 wall_ms: 1_000,
839 counter: 0,
840 node: peer,
841 };
842 resolve_pull(
843 &conn,
844 &schema(),
845 n,
846 vec![remote_change(peer, "n1", "theirs", older, 1)],
847 Utc::now(),
848 "group-7",
849 )
850 .unwrap();
851
852 assert_eq!(stash_rows(&conn)[0].scope, "group-7");
853 }
854
855 /// ServerOrder does not compare versions, so it has no loser to name.
856 #[test]
857 fn server_order_stashes_nothing() {
858 let (mut conn, n) = device(1);
859 let peer = node(2);
860 let s = server_order_schema();
861
862 conn.execute("INSERT INTO note (id, name) VALUES ('n1', 'mine')", [])
863 .unwrap();
864 stamp_pending(&conn, n, 9_000).unwrap();
865
866 let older = Hlc {
867 wall_ms: 1_000,
868 counter: 0,
869 node: peer,
870 };
871 let resolved = resolve_pull(
872 &conn,
873 &s,
874 n,
875 vec![remote_change(peer, "n1", "theirs", older, 1)],
876 Utc::now(),
877 "",
878 )
879 .unwrap();
880 apply_remote_changes(&mut conn, &s, &resolved, "").unwrap();
881
882 assert!(stash_rows(&conn).is_empty());
883 }
884
885 // ── Field merge ──
886 //
887 // The merge semantics themselves are pinned in `conflict.rs`; what is new
888 // here is the wiring, which is where every one of these can go wrong
889 // independently of a correct merge: the base has to be recorded at the right
890 // moments, read back for the right tables, refused for counters, and left
891 // entirely alone for a table that did not opt in.
892
893 /// A four-column row, which is the point: a merge is only interesting when a
894 /// row has more columns than two devices are likely to both touch. `minutes`
895 /// is the counter.
896 const CARD_COLS: &[&str] = &["id", "title", "due", "minutes"];
897
898 fn card_schema(merges: bool) -> SyncSchema {
899 let table = SyncTable::full("card", CARD_COLS);
900 SyncSchema::new(vec![if merges {
901 table.field_merge(&["minutes"])
902 } else {
903 table
904 }])
905 }
906
907 fn card_device(n: u128, merges: bool) -> (Connection, DeviceId) {
908 let conn = Connection::open_in_memory().unwrap();
909 configure_connection(&conn).unwrap();
910 conn.execute_batch(
911 "CREATE TABLE card (id TEXT PRIMARY KEY, title TEXT, due TEXT, minutes INTEGER);",
912 )
913 .unwrap();
914 conn.execute_batch(&card_schema(merges).migration_sql())
915 .unwrap();
916 (conn, node(n))
917 }
918
919 /// A wall reading to hang a card test's clocks off.
920 ///
921 /// Real time, not a toy constant, because `resolve_pull` advances the local
922 /// clock to `now` on every pull it observes: an HLC of `2_000` would sit
923 /// decades below the clock it is meant to be newer than, and every one of
924 /// these tests would silently assert over a conflict that never happened.
925 fn card_t0() -> i64 {
926 Utc::now().timestamp_millis()
927 }
928
929 /// A remote change for the card table with an explicit payload and wall clock.
930 fn card_change(from: DeviceId, wall_ms: i64, data: serde_json::Value) -> PulledChange {
931 PulledChange {
932 storage_version: None,
933 entry: ChangeEntry {
934 table: "card".into(),
935 op: ChangeOp::Update,
936 row_id: "c1".into(),
937 timestamp: Utc::now(),
938 hlc: Hlc {
939 wall_ms,
940 counter: 0,
941 node: from,
942 },
943 data: Some(data),
944 extra: serde_json::Map::default(),
945 },
946 device_id: from,
947 seq: 1,
948 }
949 }
950
951 /// The state both devices start from, established the way it really is: a
952 /// remote change applied and committed, which is also what records the base.
953 fn seed_base(conn: &mut Connection, node: DeviceId, peer: DeviceId, merges: bool, t0: i64) {
954 let s = card_schema(merges);
955 let base = card_change(
956 peer,
957 t0,
958 serde_json::json!({"id": "c1", "title": "base", "due": null, "minutes": 0}),
959 );
960 pull_apply_with(conn, &s, node, vec![base]);
961 }
962
963 fn pull_apply_with(
964 conn: &mut Connection,
965 s: &SyncSchema,
966 node: DeviceId,
967 pulled: Vec<PulledChange>,
968 ) {
969 let resolved = resolve_pull(conn, s, node, pulled, Utc::now(), "").unwrap();
970 apply_remote_changes(conn, s, &resolved, "").unwrap();
971 record_committed(conn, resolved.as_slice()).unwrap();
972 }
973
974 fn card(conn: &Connection) -> (Option<String>, Option<String>, Option<i64>) {
975 conn.query_row(
976 "SELECT title, due, minutes FROM card WHERE id = 'c1'",
977 [],
978 |r| Ok((r.get(0)?, r.get(1)?, r.get(2)?)),
979 )
980 .unwrap()
981 }
982
983 /// The case the whole feature exists for. Two devices edit different columns
984 /// of one row; both edits survive, where LWW would have discarded one whole
985 /// edit for touching a row it never contested.
986 #[test]
987 fn opted_in_table_merges_edits_to_different_fields() {
988 let (mut conn, n) = card_device(1, true);
989 let peer = node(2);
990 let t0 = card_t0();
991 seed_base(&mut conn, n, peer, true, t0);
992
993 // Local: set the due date. Remote, newer: retitle.
994 conn.execute("UPDATE card SET due = '2026-09-01' WHERE id = 'c1'", [])
995 .unwrap();
996 stamp_pending(&conn, n, t0 + 1_000).unwrap();
997 let remote = card_change(
998 peer,
999 t0 + 2_000,
1000 serde_json::json!({"id": "c1", "title": "theirs", "due": null, "minutes": 0}),
1001 );
1002
1003 pull_apply_with(&mut conn, &card_schema(true), n, vec![remote]);
1004
1005 let (title, due, _) = card(&conn);
1006 assert_eq!(title.as_deref(), Some("theirs"), "remote's field was lost");
1007 assert_eq!(
1008 due.as_deref(),
1009 Some("2026-09-01"),
1010 "the local edit was discarded for contesting a field it never touched"
1011 );
1012 assert!(
1013 stash_rows(&conn).is_empty(),
1014 "a merge that contested nothing lost nothing, so it must not stash: {:?}",
1015 stash_rows(&conn)
1016 );
1017 }
1018
1019 /// A field both sides moved is a real contest. The merge hands it to the HLC
1020 /// winner (the same rule LWW would apply), and the loser's version is stashed,
1021 /// because a value did go away.
1022 #[test]
1023 fn a_contested_field_goes_to_the_hlc_winner_and_stashes_the_loser() {
1024 let (mut conn, n) = card_device(1, true);
1025 let peer = node(2);
1026 let t0 = card_t0();
1027 seed_base(&mut conn, n, peer, true, t0);
1028
1029 conn.execute(
1030 "UPDATE card SET title = 'mine', due = '2026-09-01' WHERE id = 'c1'",
1031 [],
1032 )
1033 .unwrap();
1034 stamp_pending(&conn, n, t0 + 1_000).unwrap();
1035 let remote = card_change(
1036 peer,
1037 t0 + 2_000,
1038 serde_json::json!({"id": "c1", "title": "theirs", "due": null, "minutes": 0}),
1039 );
1040
1041 pull_apply_with(&mut conn, &card_schema(true), n, vec![remote]);
1042
1043 let (title, due, _) = card(&conn);
1044 assert_eq!(
1045 title.as_deref(),
1046 Some("theirs"),
1047 "the contested field must go to the newer HLC"
1048 );
1049 assert_eq!(
1050 due.as_deref(),
1051 Some("2026-09-01"),
1052 "an uncontested field must survive even when the same row lost a contest"
1053 );
1054
1055 let rows = stash_rows(&conn);
1056 assert_eq!(
1057 rows.len(),
1058 1,
1059 "the losing title must be recoverable: {rows:?}"
1060 );
1061 assert_eq!(rows[0].losing_side, "local");
1062 assert!(rows[0].losing_payload.as_deref().unwrap().contains("mine"));
1063 }
1064
1065 /// The failure a value merge cannot see. Two devices each log thirty minutes;
1066 /// merging takes one side's absolute total and the other half-hour is gone
1067 /// with nothing recording that it existed. Refusing to merge is worse for the
1068 /// uncontested fields and better for the truth: LWW discards one whole edit
1069 /// and stashes it, so the loss stays visible and recoverable.
1070 #[test]
1071 fn a_contested_counter_refuses_to_merge_and_falls_back_to_lww() {
1072 let (mut conn, n) = card_device(1, true);
1073 let peer = node(2);
1074 let t0 = card_t0();
1075 seed_base(&mut conn, n, peer, true, t0);
1076
1077 // Both sides increment `minutes`, and each also moves a field the other
1078 // did not, so a merge would visibly have kept both.
1079 conn.execute(
1080 "UPDATE card SET minutes = minutes + 30, due = '2026-09-01' WHERE id = 'c1'",
1081 [],
1082 )
1083 .unwrap();
1084 stamp_pending(&conn, n, t0 + 1_000).unwrap();
1085 let remote = card_change(
1086 peer,
1087 t0 + 2_000,
1088 serde_json::json!({"id": "c1", "title": "theirs", "due": null, "minutes": 30}),
1089 );
1090
1091 pull_apply_with(&mut conn, &card_schema(true), n, vec![remote]);
1092
1093 let (title, due, minutes) = card(&conn);
1094 assert_eq!(minutes, Some(30), "a merged counter would still read 30");
1095 assert_eq!(title.as_deref(), Some("theirs"));
1096 assert_eq!(
1097 due, None,
1098 "the row must hold one side whole, not a merge of both"
1099 );
1100
1101 let rows = stash_rows(&conn);
1102 assert_eq!(
1103 rows.len(),
1104 1,
1105 "the discarded half-hour must stay visible in the stash: {rows:?}"
1106 );
1107 assert!(
1108 rows[0]
1109 .losing_payload
1110 .as_deref()
1111 .unwrap()
1112 .contains("2026-09-01"),
1113 "the stash must hold the whole discarded edit: {rows:?}"
1114 );
1115 }
1116
1117 /// A counter only one side moved is not the counter problem: nothing has to
1118 /// be reconstructed, so the merge runs and that side's value carries across.
1119 /// Without this the counter declaration would cost every row on the table its
1120 /// merge, which is most of the feature.
1121 #[test]
1122 fn an_uncontested_counter_does_not_block_the_merge() {
1123 let (mut conn, n) = card_device(1, true);
1124 let peer = node(2);
1125 let t0 = card_t0();
1126 seed_base(&mut conn, n, peer, true, t0);
1127
1128 conn.execute("UPDATE card SET minutes = minutes + 30 WHERE id = 'c1'", [])
1129 .unwrap();
1130 stamp_pending(&conn, n, t0 + 1_000).unwrap();
1131 let remote = card_change(
1132 peer,
1133 t0 + 2_000,
1134 serde_json::json!({"id": "c1", "title": "theirs", "due": null, "minutes": 0}),
1135 );
1136
1137 pull_apply_with(&mut conn, &card_schema(true), n, vec![remote]);
1138
1139 let (title, _, minutes) = card(&conn);
1140 assert_eq!(
1141 minutes,
1142 Some(30),
1143 "an uncontested counter must carry across"
1144 );
1145 assert_eq!(title.as_deref(), Some("theirs"));
1146 }
1147
1148 /// The default must be no change in behaviour. The identical conflict on a
1149 /// table that did not opt in resolves the way it always has: one whole edit
1150 /// wins, the other is stashed.
1151 #[test]
1152 fn a_table_that_did_not_opt_in_still_resolves_by_lww() {
1153 let (mut conn, n) = card_device(1, false);
1154 let peer = node(2);
1155 let t0 = card_t0();
1156 seed_base(&mut conn, n, peer, false, t0);
1157
1158 conn.execute("UPDATE card SET due = '2026-09-01' WHERE id = 'c1'", [])
1159 .unwrap();
1160 stamp_pending(&conn, n, t0 + 1_000).unwrap();
1161 let remote = card_change(
1162 peer,
1163 t0 + 2_000,
1164 serde_json::json!({"id": "c1", "title": "theirs", "due": null, "minutes": 0}),
1165 );
1166
1167 pull_apply_with(&mut conn, &card_schema(false), n, vec![remote]);
1168
1169 let (title, due, _) = card(&conn);
1170 assert_eq!(title.as_deref(), Some("theirs"));
1171 assert_eq!(
1172 due, None,
1173 "without the opt-in the whole remote edit wins, exactly as before"
1174 );
1175 assert_eq!(stash_rows(&conn).len(), 1, "and the loser is still stashed");
1176
1177 let bases: i64 = conn
1178 .query_row("SELECT COUNT(*) FROM sync_row_snapshot", [], |r| r.get(0))
1179 .unwrap();
1180 assert_eq!(bases, 0, "a table that did not opt in must store no base");
1181 }
1182
1183 /// A row whose base was never recorded, or was dropped by a delete. There is
1184 /// nothing to merge against, so the conflict falls through to LWW rather than
1185 /// merging against a base it invented.
1186 #[test]
1187 fn a_missing_base_falls_back_to_lww_without_failing() {
1188 let (mut conn, n) = card_device(1, true);
1189 let peer = node(2);
1190 let t0 = card_t0();
1191 seed_base(&mut conn, n, peer, true, t0);
1192 conn.execute("DELETE FROM sync_row_snapshot", []).unwrap();
1193
1194 conn.execute("UPDATE card SET due = '2026-09-01' WHERE id = 'c1'", [])
1195 .unwrap();
1196 stamp_pending(&conn, n, t0 + 1_000).unwrap();
1197 let remote = card_change(
1198 peer,
1199 t0 + 2_000,
1200 serde_json::json!({"id": "c1", "title": "theirs", "due": null, "minutes": 0}),
1201 );
1202
1203 pull_apply_with(&mut conn, &card_schema(true), n, vec![remote]);
1204
1205 let (title, due, _) = card(&conn);
1206 assert_eq!(title.as_deref(), Some("theirs"));
1207 assert_eq!(due, None, "with no base the whole remote edit wins");
1208 assert_eq!(stash_rows(&conn).len(), 1);
1209 }
1210
1211 /// Applying a remote change re-bases the row, so the *next* conflict measures
1212 /// both sides against what this device last received rather than against the
1213 /// version it first saw. Without this the base would go stale and every later
1214 /// merge would report fields as changed that nobody touched.
1215 #[test]
1216 fn applying_a_remote_change_rebases_the_row() {
1217 let (mut conn, n) = card_device(1, true);
1218 let peer = node(2);
1219 let t0 = card_t0();
1220 seed_base(&mut conn, n, peer, true, t0);
1221
1222 pull_apply_with(
1223 &mut conn,
1224 &card_schema(true),
1225 n,
1226 vec![card_change(
1227 peer,
1228 t0 + 2_000,
1229 serde_json::json!({"id": "c1", "title": "second", "due": null, "minutes": 0}),
1230 )],
1231 );
1232
1233 let base = super::snapshot::load(&conn, "card", "c1");
1234 assert_eq!(
1235 base["title"], "second",
1236 "the base must track the latest applied version, not the first"
1237 );
1238 }
1239
1240 /// A delete drops the base with the row. A stale base would otherwise be
1241 /// handed to a merge for whatever next occupies the key, describing a version
1242 /// of a different row.
1243 #[test]
1244 fn deleting_a_row_drops_its_base() {
1245 let (mut conn, n) = card_device(1, true);
1246 let peer = node(2);
1247 let t0 = card_t0();
1248 seed_base(&mut conn, n, peer, true, t0);
1249 assert!(super::snapshot::load(&conn, "card", "c1").is_object());
1250
1251 let mut delete = card_change(peer, t0 + 4_000, serde_json::json!({"id": "c1"}));
1252 delete.entry.op = ChangeOp::Delete;
1253 pull_apply_with(&mut conn, &card_schema(true), n, vec![delete]);
1254
1255 assert_eq!(
1256 super::snapshot::load(&conn, "card", "c1"),
1257 serde_json::Value::Null
1258 );
1259 }
1260
1261 // ── Dependent columns ──
1262
1263 /// A row whose `state` carries a `state_at` derived from it, the shape
1264 /// GoingsOn's `status`/`completed_at` has.
1265 fn dep_schema() -> SyncSchema {
1266 SyncSchema::new(vec![
1267 SyncTable::full("job", &["id", "state", "state_at", "note"])
1268 .field_merge(&[])
1269 .dependent_columns(&[&["state", "state_at"]]),
1270 ])
1271 }
1272
1273 fn dep_device(n: u128) -> (Connection, DeviceId) {
1274 let conn = Connection::open_in_memory().unwrap();
1275 configure_connection(&conn).unwrap();
1276 conn.execute_batch(
1277 "CREATE TABLE job (id TEXT PRIMARY KEY, state TEXT, state_at TEXT, note TEXT);",
1278 )
1279 .unwrap();
1280 conn.execute_batch(&dep_schema().migration_sql()).unwrap();
1281 (conn, node(n))
1282 }
1283
1284 fn job_change(from: DeviceId, wall_ms: i64, data: serde_json::Value) -> PulledChange {
1285 PulledChange {
1286 storage_version: None,
1287 entry: ChangeEntry {
1288 table: "job".into(),
1289 op: ChangeOp::Update,
1290 row_id: "j1".into(),
1291 timestamp: Utc::now(),
1292 hlc: Hlc {
1293 wall_ms,
1294 counter: 0,
1295 node: from,
1296 },
1297 data: Some(data),
1298 extra: serde_json::Map::default(),
1299 },
1300 device_id: from,
1301 seq: 1,
1302 }
1303 }
1304
1305 fn job(conn: &Connection) -> (Option<String>, Option<String>, Option<String>) {
1306 conn.query_row(
1307 "SELECT state, state_at, note FROM job WHERE id = 'j1'",
1308 [],
1309 |r| Ok((r.get(0)?, r.get(1)?, r.get(2)?)),
1310 )
1311 .unwrap()
1312 }
1313
1314 /// The failure the declaration exists for, and the reason it cannot be fixed
1315 /// in the app: one device moves the row to `started` (leaving `state_at`
1316 /// alone), the other to `done` (stamping it). Only the second moved
1317 /// `state_at`, so a column-by-column merge treats it as uncontested and keeps
1318 /// it whichever way `state` falls, producing a started job with a completion
1319 /// time. The group forces both from one side.
1320 #[test]
1321 fn a_contested_dependent_group_is_taken_whole_from_one_side() {
1322 for (local_newer, expect) in [(true, ("started", None)), (false, ("done", Some("T")))] {
1323 let (mut conn, n) = dep_device(1);
1324 let peer = node(2);
1325 let t0 = card_t0();
1326 let s = dep_schema();
1327
1328 pull_apply_with(
1329 &mut conn,
1330 &s,
1331 n,
1332 vec![job_change(
1333 peer,
1334 t0,
1335 serde_json::json!({"id": "j1", "state": "pending", "state_at": null, "note": "n"}),
1336 )],
1337 );
1338
1339 // Local moves state only; remote moves state and its derived stamp.
1340 conn.execute("UPDATE job SET state = 'started' WHERE id = 'j1'", [])
1341 .unwrap();
1342 let (local_ms, remote_ms) = if local_newer {
1343 (t0 + 3_000, t0 + 2_000)
1344 } else {
1345 (t0 + 1_000, t0 + 2_000)
1346 };
1347 stamp_pending(&conn, n, local_ms).unwrap();
1348 pull_apply_with(
1349 &mut conn,
1350 &s,
1351 n,
1352 vec![job_change(
1353 peer,
1354 remote_ms,
1355 serde_json::json!({"id": "j1", "state": "done", "state_at": "T", "note": "n"}),
1356 )],
1357 );
1358
1359 let (state, state_at, _) = job(&conn);
1360 assert_eq!(
1361 (state.as_deref(), state_at.as_deref()),
1362 (Some(expect.0), expect.1),
1363 "the group must land as one device's version of it (local_newer = {local_newer})"
1364 );
1365 }
1366 }
1367
1368 /// The same split by the other route, and the one that was briefly wrong.
1369 ///
1370 /// Here no single column is contested: one device moves `state`, the other
1371 /// moves only `state_at`. A field-by-field pass sees two disjoint changes,
1372 /// merges both, and lands one column from each device, which is the same
1373 /// broken pair as the contested case. So the trigger is both sides having
1374 /// touched the group *anywhere*, not both having touched the same column.
1375 #[test]
1376 fn a_group_touched_by_both_sides_on_different_columns_is_still_taken_whole() {
1377 let (mut conn, n) = dep_device(1);
1378 let peer = node(2);
1379 let t0 = card_t0();
1380 let s = dep_schema();
1381
1382 pull_apply_with(
1383 &mut conn,
1384 &s,
1385 n,
1386 vec![job_change(
1387 peer,
1388 t0,
1389 serde_json::json!({"id": "j1", "state": "pending", "state_at": null, "note": "n"}),
1390 )],
1391 );
1392
1393 // Local moves only `state`; remote moves only `state_at`. Disjoint.
1394 conn.execute("UPDATE job SET state = 'started' WHERE id = 'j1'", [])
1395 .unwrap();
1396 stamp_pending(&conn, n, t0 + 3_000).unwrap();
1397 pull_apply_with(
1398 &mut conn,
1399 &s,
1400 n,
1401 vec![job_change(
1402 peer,
1403 t0 + 2_000,
1404 serde_json::json!({"id": "j1", "state": "pending", "state_at": "T", "note": "n"}),
1405 )],
1406 );
1407
1408 let (state, state_at, _) = job(&conn);
1409 assert_eq!(
1410 (state.as_deref(), state_at.as_deref()),
1411 (Some("started"), None),
1412 "the group took one column from each device: a state neither one held"
1413 );
1414 }
1415
1416 /// The group only fires when it is contested. A device that moves the group
1417 /// while the other moves an unrelated column still gets a merge, which is the
1418 /// whole reason the table opted in.
1419 #[test]
1420 fn an_uncontested_dependent_group_still_merges() {
1421 let (mut conn, n) = dep_device(1);
1422 let peer = node(2);
1423 let t0 = card_t0();
1424 let s = dep_schema();
1425
1426 pull_apply_with(
1427 &mut conn,
1428 &s,
1429 n,
1430 vec![job_change(
1431 peer,
1432 t0,
1433 serde_json::json!({"id": "j1", "state": "pending", "state_at": null, "note": "n"}),
1434 )],
1435 );
1436
1437 // Local edits the unrelated column; only remote touches the group.
1438 conn.execute("UPDATE job SET note = 'mine' WHERE id = 'j1'", [])
1439 .unwrap();
1440 stamp_pending(&conn, n, t0 + 1_000).unwrap();
1441 pull_apply_with(
1442 &mut conn,
1443 &s,
1444 n,
1445 vec![job_change(
1446 peer,
1447 t0 + 2_000,
1448 serde_json::json!({"id": "j1", "state": "done", "state_at": "T", "note": "n"}),
1449 )],
1450 );
1451
1452 let (state, state_at, note) = job(&conn);
1453 assert_eq!(
1454 (state.as_deref(), state_at.as_deref()),
1455 (Some("done"), Some("T")),
1456 "an uncontested group must carry across intact"
1457 );
1458 assert_eq!(
1459 note.as_deref(),
1460 Some("mine"),
1461 "declaring a group must not cost the table its merge on other columns"
1462 );
1463 }
1464
1465 /// The stash is bounded. Unbounded, a pathological sync loop grows it without
1466 /// limit.
1467 #[test]
1468 fn stash_is_trimmed_to_its_ceiling() {
1469 let (conn, _) = device(1);
1470 for i in 0..(super::stash::MAX_STASH_ROWS + 50) {
1471 conn.execute(
1472 "INSERT INTO sync_conflict_stash
1473 (table_name, row_id, losing_side, losing_payload, losing_hlc, losing_device, winning_hlc)
1474 VALUES ('note', ?1, 'remote', '{}', '1:0:x', 'dev', '2:0:y')",
1475 [i.to_string()],
1476 )
1477 .unwrap();
1478 }
1479 super::stash::trim_stash(&conn).unwrap();
1480
1481 let kept: i64 = conn
1482 .query_row("SELECT COUNT(*) FROM sync_conflict_stash", [], |r| r.get(0))
1483 .unwrap();
1484 assert_eq!(kept, super::stash::MAX_STASH_ROWS);
1485
1486 // The newest survive: the oldest losses are the least likely to be acted on.
1487 let oldest_kept: String = conn
1488 .query_row(
1489 "SELECT row_id FROM sync_conflict_stash ORDER BY id ASC LIMIT 1",
1490 [],
1491 |r| r.get(0),
1492 )
1493 .unwrap();
1494 assert_eq!(oldest_kept, "50");
1495 }
1496
1497 /// `stamp_pending` has to leave the advanced clock on disk, not only on the
1498 /// changelog rows. The wall component is high (5_000_000) and the row count
1499 /// is three, so the persisted clock is a value no fresh `Hlc::zero` can
1500 /// coincide with, and the second stamp runs at a LOWER now_ms (1000) so a
1501 /// clock that failed to persist would visibly restart at 1000 instead of
1502 /// continuing the counter at 5_000_000.
1503 #[test]
1504 fn stamp_pending_persists_the_advanced_clock() {
1505 let (conn, n) = device(1);
1506 for id in ["r1", "r2", "r3"] {
1507 conn.execute("INSERT INTO note (id, name) VALUES (?1, 'v')", [id])
1508 .unwrap();
1509 }
1510 assert_eq!(stamp_pending(&conn, n, 5_000_000).unwrap(), 3);
1511
1512 // Three ticks at one wall component: adopt 5_000_000 with counter 0,
1513 // then bump twice.
1514 assert_eq!(
1515 load_clock(&conn, n).unwrap(),
1516 Hlc {
1517 wall_ms: 5_000_000,
1518 counter: 2,
1519 node: n
1520 },
1521 "the clock reached by stamping must survive a reload"
1522 );
1523
1524 conn.execute("INSERT INTO note (id, name) VALUES ('r4', 'v')", [])
1525 .unwrap();
1526 assert_eq!(stamp_pending(&conn, n, 1000).unwrap(), 1);
1527 let r4 = load_local_pending(&conn, n)
1528 .unwrap()
1529 .into_iter()
1530 .find(|e| e.row_id == "r4")
1531 .unwrap();
1532 assert_eq!(
1533 r4.hlc,
1534 Hlc {
1535 wall_ms: 5_000_000,
1536 counter: 3,
1537 node: n
1538 },
1539 "a later stamp at an earlier now_ms must continue the reloaded clock"
1540 );
1541 }
1542
1543 /// `observe` has to leave the merged clock on disk: the whole point is that
1544 /// a subsequent local write outranks the remote it observed. The remote sits
1545 /// far in the future (9_000_000_000) and the local stamp that follows runs
1546 /// at now_ms 2000, so if the merge were not persisted the new stamp would
1547 /// land at wall 2000 and lose to the remote by seven orders of magnitude.
1548 #[test]
1549 fn observe_persists_the_merged_clock() {
1550 let (conn, n) = device(1);
1551 let remote = Hlc {
1552 wall_ms: 9_000_000_000,
1553 counter: 5,
1554 node: node(2),
1555 };
1556 observe(&conn, n, [remote], 1000).unwrap();
1557 assert_eq!(
1558 load_clock(&conn, n).unwrap(),
1559 Hlc {
1560 wall_ms: 9_000_000_000,
1561 counter: 6,
1562 node: n
1563 },
1564 "observe must persist the remote wall and a strictly greater counter"
1565 );
1566
1567 conn.execute("INSERT INTO note (id, name) VALUES ('r', 'v')", [])
1568 .unwrap();
1569 assert_eq!(stamp_pending(&conn, n, 2000).unwrap(), 1);
1570 let stamped = load_local_pending(&conn, n).unwrap().remove(0).hlc;
1571 assert_eq!(
1572 stamped,
1573 Hlc {
1574 wall_ms: 9_000_000_000,
1575 counter: 7,
1576 node: n
1577 }
1578 );
1579 assert!(
1580 stamped > remote,
1581 "a local write after observing must outrank the observed remote"
1582 );
1583 }
1584