1use std::sync::atomic::Ordering;
35
36use anyhow::Result;
37use serde::{Deserialize, Serialize};
38use serde_json::Value;
39
40use crate::db::Db;
41
42pub const CONFLICTS: &str = "_nedb.conflicts";
46
47#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
52#[serde(rename_all = "snake_case")]
53pub enum ConflictKind {
54 BothModified,
56 ModifiedDeleted,
58 DeletedModified,
60 BothAdded,
62}
63
64#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
66pub struct Conflict {
67 pub branch: String,
86 pub branch_created_seq: u64,
87 pub coll: String,
88 pub id: String,
89 pub base: Option<Value>,
91 pub ours: Option<Value>,
93 pub theirs: Option<Value>,
95 pub kind: ConflictKind,
96}
97
98#[derive(Debug, Clone, PartialEq)]
100pub enum Resolution {
101 TakeOurs,
102 TakeTheirs,
103 TakeValue(Value),
107}
108
109#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
113#[serde(rename_all = "snake_case")]
114pub enum Choice {
115 Ours,
116 Theirs,
117 Value,
118}
119
120#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
122pub struct ResolutionRecord {
123 pub branch: String,
124 pub branch_created_seq: u64,
125 pub coll: String,
126 pub id: String,
127 pub kind: ConflictKind,
128 pub base: Option<Value>,
129 pub ours: Option<Value>,
130 pub theirs: Option<Value>,
131 pub choice: Choice,
132 pub chosen: Option<Value>,
134 pub at_seq: u64,
136}
137
138fn conflict_key(branch: &str, created_seq: u64, coll: &str, id: &str) -> String {
145 use blake2::{Blake2b512, Digest};
146 let mut h = Blake2b512::new();
147 for part in [branch, coll, id] {
149 h.update((part.len() as u64).to_be_bytes());
150 h.update(part.as_bytes());
151 }
152 h.update(created_seq.to_be_bytes());
155 hex::encode(&h.finalize()[..32])
156}
157
158pub fn resolve(db: &Db, c: &Conflict, r: Resolution) -> Result<()> {
166 let (choice, chosen) = match r {
167 Resolution::TakeOurs => (Choice::Ours, c.ours.clone()),
168 Resolution::TakeTheirs => (Choice::Theirs, c.theirs.clone()),
169 Resolution::TakeValue(v) => (Choice::Value, Some(v)),
170 };
171
172 match &chosen {
173 Some(v) => {
174 db.put(&c.coll, &c.id, v.clone(), vec![], None, None)?;
175 }
176 None => {
177 db.delete(&c.coll, &c.id)?;
182 }
183 }
184
185 let at_seq = db.seq.load(Ordering::SeqCst).saturating_sub(1);
186 let rec = ResolutionRecord {
187 branch: c.branch.clone(),
188 branch_created_seq: c.branch_created_seq,
189 coll: c.coll.clone(),
190 id: c.id.clone(),
191 kind: c.kind,
192 base: c.base.clone(),
193 ours: c.ours.clone(),
194 theirs: c.theirs.clone(),
195 choice,
196 chosen,
197 at_seq,
198 };
199 db.put_unchecked(
200 CONFLICTS,
201 &conflict_key(&c.branch, c.branch_created_seq, &c.coll, &c.id),
202 serde_json::to_value(&rec)?,
203 vec![], None, None,
204 )?;
205 Ok(())
206}
207
208pub fn resolution_for(db: &Db, branch: &str, created_seq: u64, coll: &str, id: &str)
210 -> Option<ResolutionRecord>
211{
212 let n = db.get(CONFLICTS, &conflict_key(branch, created_seq, coll, id))?;
213 serde_json::from_value(n.data).ok()
214}
215
216pub fn resolutions(db: &Db) -> Vec<ResolutionRecord> {
222 let mut out: Vec<ResolutionRecord> = db
223 .list_ids_including_deleted(CONFLICTS)
224 .into_iter()
225 .filter_map(|k| db.get(CONFLICTS, &k))
226 .filter_map(|n| serde_json::from_value::<ResolutionRecord>(n.data).ok())
227 .collect();
228 out.sort_by(|a, b| (&a.coll, &a.id).cmp(&(&b.coll, &b.id)));
229 out
230}
231
232pub(crate) fn is_settled(db: &Db, c: &Conflict) -> bool {
242 match resolution_for(db, &c.branch, c.branch_created_seq, &c.coll, &c.id) {
243 Some(rec) => rec.theirs == c.theirs,
249 None => false,
250 }
251}
252
253#[cfg(test)]
254mod tests {
255 use super::*;
256 use tempfile::tempdir;
257
258 fn j(v: u64) -> Value { serde_json::json!({ "v": v }) }
259
260 fn a_conflict() -> Conflict {
261 Conflict {
262 branch: "b".into(),
263 branch_created_seq: 0,
264 coll: "orders".into(),
265 id: "42".into(),
266 base: Some(j(1)),
267 ours: Some(j(2)),
268 theirs: Some(j(3)),
269 kind: ConflictKind::BothModified,
270 }
271 }
272
273 #[test]
274 fn taking_theirs_writes_their_value_as_a_new_version() {
275 let db = Db::in_memory();
276 db.put("orders", "42", j(1), vec![], None, None).unwrap();
277 let base_seq = db.seq.load(Ordering::SeqCst) - 1;
278 db.put("orders", "42", j(2), vec![], None, None).unwrap();
279
280 resolve(&db, &a_conflict(), Resolution::TakeTheirs).unwrap();
281 assert_eq!(db.get("orders", "42").unwrap().data, j(3));
282 assert_eq!(db.get_as_of("orders", "42", base_seq).unwrap().data, j(1));
284 }
285
286 #[test]
287 fn taking_ours_still_writes_a_version_rather_than_doing_nothing() {
288 let db = Db::in_memory();
289 db.put("orders", "42", j(1), vec![], None, None).unwrap();
290 db.put("orders", "42", j(2), vec![], None, None).unwrap();
291 let before = db.seq.load(Ordering::SeqCst);
292
293 resolve(&db, &a_conflict(), Resolution::TakeOurs).unwrap();
294 assert_eq!(db.get("orders", "42").unwrap().data, j(2));
295 assert!(db.seq.load(Ordering::SeqCst) > before,
296 "a decision is an event; it has to land in history to be auditable");
297 }
298
299 #[test]
300 fn a_third_value_can_be_chosen() {
301 let db = Db::in_memory();
302 db.put("orders", "42", j(2), vec![], None, None).unwrap();
303 let merged = serde_json::json!({ "v": 2, "note": "hand-merged" });
304 resolve(&db, &a_conflict(), Resolution::TakeValue(merged.clone())).unwrap();
305 assert_eq!(db.get("orders", "42").unwrap().data, merged);
306 }
307
308 #[test]
309 fn resolving_toward_a_delete_removes_the_live_document() {
310 let db = Db::in_memory();
311 db.put("orders", "42", j(2), vec![], None, None).unwrap();
312 let at = db.seq.load(Ordering::SeqCst) - 1;
313 let c = Conflict { theirs: None, kind: ConflictKind::ModifiedDeleted, ..a_conflict() };
314 resolve(&db, &c, Resolution::TakeTheirs).unwrap();
315 assert!(db.get("orders", "42").is_none());
316 assert_eq!(db.get_as_of("orders", "42", at).unwrap().data, j(2),
317 "a delete is a tombstone; the value before it is still readable");
318 }
319
320 #[test]
321 fn resolving_a_delete_that_already_happened_is_not_an_error() {
322 let db = Db::in_memory();
323 db.put("orders", "42", j(2), vec![], None, None).unwrap();
324 db.delete("orders", "42").unwrap();
325 let c = Conflict { ours: None, theirs: None, kind: ConflictKind::ModifiedDeleted, ..a_conflict() };
326 resolve(&db, &c, Resolution::TakeTheirs).unwrap();
327 assert!(db.get("orders", "42").is_none());
328 assert_eq!(resolutions(&db).len(), 1, "the decision is still recorded");
329 }
330
331 #[test]
332 fn every_resolution_leaves_an_audit_record_with_all_three_sides() {
333 let db = Db::in_memory();
334 db.put("orders", "42", j(2), vec![], None, None).unwrap();
335 resolve(&db, &a_conflict(), Resolution::TakeTheirs).unwrap();
336
337 let all = resolutions(&db);
338 assert_eq!(all.len(), 1);
339 let r = &all[0];
340 assert_eq!(r.coll, "orders");
341 assert_eq!(r.id, "42");
342 assert_eq!(r.kind, ConflictKind::BothModified);
343 assert_eq!(r.base, Some(j(1)));
344 assert_eq!(r.ours, Some(j(2)));
345 assert_eq!(r.theirs, Some(j(3)));
346 assert_eq!(r.choice, Choice::Theirs);
347 assert_eq!(r.chosen, Some(j(3)));
348 assert_eq!(resolution_for(&db, "b", 0, "orders", "42").as_ref(), Some(r));
349 }
350
351 #[test]
352 fn a_second_decision_supersedes_the_first_without_erasing_it() {
353 let db = Db::in_memory();
354 db.put("orders", "42", j(2), vec![], None, None).unwrap();
355 resolve(&db, &a_conflict(), Resolution::TakeOurs).unwrap();
356 let after_first = db.seq.load(Ordering::SeqCst) - 1;
357 resolve(&db, &a_conflict(), Resolution::TakeTheirs).unwrap();
358
359 assert_eq!(resolution_for(&db, "b", 0, "orders", "42").unwrap().choice, Choice::Theirs);
360 assert_eq!(resolutions(&db).len(), 1, "one live record per document");
361 let old = db.get_as_of(CONFLICTS, &conflict_key("b", 0, "orders", "42"), after_first).unwrap();
363 let old: ResolutionRecord = serde_json::from_value(old.data).unwrap();
364 assert_eq!(old.choice, Choice::Ours);
365 }
366
367 #[test]
368 fn a_decision_settles_the_branch_claim_it_was_taken_against_and_no_other() {
369 let db = Db::in_memory();
370 db.put("orders", "42", j(2), vec![], None, None).unwrap();
371 let c = a_conflict();
372 assert!(!is_settled(&db, &c), "nothing is settled before it is decided");
373 resolve(&db, &c, Resolution::TakeOurs).unwrap();
374 assert!(is_settled(&db, &c));
375
376 let moved_on = Conflict { theirs: Some(j(99)), ..c };
377 assert!(!is_settled(&db, &moved_on),
378 "a new claim from the branch is a new disagreement");
379 }
380
381 #[test]
382 fn conflict_keys_cannot_be_forged_by_a_clever_id() {
383 assert_ne!(conflict_key("br", 0, "a", "b|c"), conflict_key("br", 0, "a|b", "c"));
384 assert_ne!(conflict_key("br", 0, "ab", "c"), conflict_key("br", 0, "a", "bc"));
385 assert_eq!(conflict_key("br", 0, "a", "b"), conflict_key("br", 0, "a", "b"));
386 assert_ne!(conflict_key("x", 0, "a", "b"), conflict_key("y", 0, "a", "b"));
388 assert_ne!(conflict_key("x", 0, "a", "b"), conflict_key("x", 1, "a", "b"));
389 assert_ne!(conflict_key("xa", 0, "b", "c"), conflict_key("x", 0, "ab", "c"));
391 }
392
393 #[test]
394 fn resolution_works_on_disk_too() {
395 let dir = tempdir().unwrap();
396 let db = Db::open(dir.path(), None).unwrap();
397 db.put("orders", "42", j(2), vec![], None, None).unwrap();
398 resolve(&db, &a_conflict(), Resolution::TakeTheirs).unwrap();
399 db.flush_all();
400 assert_eq!(db.get("orders", "42").unwrap().data, j(3));
401 assert_eq!(resolutions(&db).len(), 1);
402 }
403}
404
405#[cfg(test)]
408mod scoped_to_the_branch_that_raised_it {
409 use super::*;
410 use crate::branch::{branch_put, create_branch, get_branch};
411 use crate::merge;
412
413 fn j(v: u64) -> Value { serde_json::json!({ "v": v }) }
414
415 #[test]
426 fn resolving_one_branch_does_not_settle_another_making_the_same_claim() {
427 let db = Db::in_memory();
428 db.put("orders", "42", j(1), vec![], None, None).unwrap();
429 let base = db.seq.load(Ordering::SeqCst) - 1;
430
431 create_branch(&db, "x", base).unwrap();
432 create_branch(&db, "y", base).unwrap();
433 branch_put(&db, "x", "orders", "42", j(7)).unwrap();
434 branch_put(&db, "y", "orders", "42", j(7)).unwrap();
435 db.put("orders", "42", j(8), vec![], None, None).unwrap();
436
437 let px = merge::plan(&db, "x").unwrap();
438 let py = merge::plan(&db, "y").unwrap();
439 assert_eq!(px.conflicts.len(), 1, "X disagrees with the destination");
440 assert_eq!(py.conflicts.len(), 1, "so does Y");
441
442 resolve(&db, &px.conflicts[0], Resolution::TakeOurs).unwrap();
443
444 assert!(is_settled(&db, &px.conflicts[0]), "X was decided");
445 assert!(
446 !is_settled(&db, &py.conflicts[0]),
447 "NOBODY decided Y — a human decision about one line of history must \
448 not implicitly authorise another"
449 );
450 assert_eq!(
451 merge::plan(&db, "y").unwrap().conflicts.len(), 1,
452 "and Y must still be planned as conflicted"
453 );
454 }
455
456 #[test]
458 fn a_new_generation_of_a_name_starts_unresolved() {
459 let db = Db::in_memory();
460 db.put("orders", "42", j(1), vec![], None, None).unwrap();
461 let base = db.seq.load(Ordering::SeqCst) - 1;
462
463 create_branch(&db, "fix", base).unwrap();
464 branch_put(&db, "fix", "orders", "42", j(7)).unwrap();
465 db.put("orders", "42", j(8), vec![], None, None).unwrap();
466 let first = merge::plan(&db, "fix").unwrap().conflicts.remove(0);
467 resolve(&db, &first, Resolution::TakeOurs).unwrap();
468 assert!(is_settled(&db, &first));
469 crate::branch::abandon_branch(&db, "fix").unwrap();
470
471 let base2 = db.seq.load(Ordering::SeqCst) - 1;
473 create_branch(&db, "fix", base2).unwrap();
474 branch_put(&db, "fix", "orders", "42", j(7)).unwrap();
475 db.put("orders", "42", j(9), vec![], None, None).unwrap();
476
477 let again = merge::plan(&db, "fix").unwrap();
478 assert_eq!(again.conflicts.len(), 1);
479 assert!(
480 !is_settled(&db, &again.conflicts[0]),
481 "a name is a working label; the decision belonged to the generation"
482 );
483 assert_ne!(
484 get_branch(&db, "fix").unwrap().created_seq, first.branch_created_seq,
485 "precondition: this really is a different generation"
486 );
487 }
488
489 #[test]
490 fn a_resolution_records_which_branch_it_was_taken_against() {
491 let db = Db::in_memory();
492 db.put("orders", "42", j(1), vec![], None, None).unwrap();
493 let base = db.seq.load(Ordering::SeqCst) - 1;
494 create_branch(&db, "x", base).unwrap();
495 branch_put(&db, "x", "orders", "42", j(7)).unwrap();
496 db.put("orders", "42", j(8), vec![], None, None).unwrap();
497
498 let c = merge::plan(&db, "x").unwrap().conflicts.remove(0);
499 resolve(&db, &c, Resolution::TakeTheirs).unwrap();
500
501 let gen = get_branch(&db, "x").unwrap().created_seq;
502 let rec = resolution_for(&db, "x", gen, "orders", "42")
503 .expect("the decision is recorded under the branch that raised it");
504 assert_eq!(rec.branch, "x");
505 assert_eq!(rec.branch_created_seq, gen);
506 assert!(resolution_for(&db, "y", gen, "orders", "42").is_none());
508 }
509}
510
511#[cfg(test)]
513mod replay_carries_its_cause {
514 use super::*;
515 use crate::branch::{branch_put, create_branch};
516 use crate::merge;
517
518 fn j(v: u64) -> Value { serde_json::json!({ "v": v }) }
519
520 #[test]
521 fn a_replayed_write_points_back_at_the_branch_write_that_caused_it() {
522 let db = Db::in_memory();
523 db.put("orders", "a", j(1), vec![], None, None).unwrap();
524 let base = db.seq.load(Ordering::SeqCst) - 1;
525 create_branch(&db, "x", base).unwrap();
526 let bw = branch_put(&db, "x", "orders", "a", j(2)).unwrap();
527 assert!(!bw.source_hash.is_empty(), "the branch write is addressable");
528
529 let plan = merge::plan(&db, "x").unwrap();
530 assert!(plan.is_clean());
531 assert_eq!(plan.changes.len(), 1);
532 assert_eq!(plan.changes[0].source_hash, bw.source_hash,
533 "the plan carries the source identity through");
534
535 merge::execute(&db, &plan).unwrap();
536
537 let landed = db.get("orders", "a").expect("the replay landed");
538 assert_eq!(landed.data, j(2));
539 assert_eq!(
540 landed.caused_by, vec![bw.source_hash.clone()],
541 "the destination node names the branch write that caused it"
542 );
543
544 let traced = db.trace(&landed.hash, false, 10);
546 assert!(
547 traced.iter().any(|n| n.hash == bw.source_hash),
548 "TRACE must reach the branch write from the merged node"
549 );
550 }
551
552 #[test]
553 fn every_replayed_change_carries_a_cause() {
554 let db = Db::in_memory();
555 for i in 0..4u64 {
556 db.put("orders", &i.to_string(), j(1), vec![], None, None).unwrap();
557 }
558 let base = db.seq.load(Ordering::SeqCst) - 1;
559 create_branch(&db, "x", base).unwrap();
560 for i in 0..4u64 {
561 branch_put(&db, "x", "orders", &i.to_string(), j(2)).unwrap();
562 }
563 let plan = merge::plan(&db, "x").unwrap();
564 assert_eq!(plan.changes.len(), 4);
565 assert!(
566 plan.changes.iter().all(|c| !c.source_hash.is_empty()),
567 "a plan with an anonymous change would replay an anonymous node"
568 );
569 merge::execute(&db, &plan).unwrap();
570 for i in 0..4u64 {
571 let n = db.get("orders", &i.to_string()).unwrap();
572 assert_eq!(n.caused_by.len(), 1, "doc {} lost its causal edge", i);
573 }
574 }
575}