use std::sync::atomic::Ordering;
use anyhow::Result;
use serde::{Deserialize, Serialize};
use serde_json::Value;
use crate::db::Db;
pub const CONFLICTS: &str = "_nedb.conflicts";
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum ConflictKind {
BothModified,
ModifiedDeleted,
DeletedModified,
BothAdded,
}
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
pub struct Conflict {
pub branch: String,
pub branch_created_seq: u64,
pub coll: String,
pub id: String,
pub base: Option<Value>,
pub ours: Option<Value>,
pub theirs: Option<Value>,
pub kind: ConflictKind,
}
#[derive(Debug, Clone, PartialEq)]
pub enum Resolution {
TakeOurs,
TakeTheirs,
TakeValue(Value),
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum Choice {
Ours,
Theirs,
Value,
}
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
pub struct ResolutionRecord {
pub branch: String,
pub branch_created_seq: u64,
pub coll: String,
pub id: String,
pub kind: ConflictKind,
pub base: Option<Value>,
pub ours: Option<Value>,
pub theirs: Option<Value>,
pub choice: Choice,
pub chosen: Option<Value>,
pub at_seq: u64,
}
fn conflict_key(branch: &str, created_seq: u64, coll: &str, id: &str) -> String {
use blake2::{Blake2b512, Digest};
let mut h = Blake2b512::new();
for part in [branch, coll, id] {
h.update((part.len() as u64).to_be_bytes());
h.update(part.as_bytes());
}
h.update(created_seq.to_be_bytes());
hex::encode(&h.finalize()[..32])
}
pub fn resolve(db: &Db, c: &Conflict, r: Resolution) -> Result<()> {
let (choice, chosen) = match r {
Resolution::TakeOurs => (Choice::Ours, c.ours.clone()),
Resolution::TakeTheirs => (Choice::Theirs, c.theirs.clone()),
Resolution::TakeValue(v) => (Choice::Value, Some(v)),
};
match &chosen {
Some(v) => {
db.put(&c.coll, &c.id, v.clone(), vec![], None, None)?;
}
None => {
db.delete(&c.coll, &c.id)?;
}
}
let at_seq = db.seq.load(Ordering::SeqCst).saturating_sub(1);
let rec = ResolutionRecord {
branch: c.branch.clone(),
branch_created_seq: c.branch_created_seq,
coll: c.coll.clone(),
id: c.id.clone(),
kind: c.kind,
base: c.base.clone(),
ours: c.ours.clone(),
theirs: c.theirs.clone(),
choice,
chosen,
at_seq,
};
db.put_unchecked(
CONFLICTS,
&conflict_key(&c.branch, c.branch_created_seq, &c.coll, &c.id),
serde_json::to_value(&rec)?,
vec![], None, None,
)?;
Ok(())
}
pub fn resolution_for(db: &Db, branch: &str, created_seq: u64, coll: &str, id: &str)
-> Option<ResolutionRecord>
{
let n = db.get(CONFLICTS, &conflict_key(branch, created_seq, coll, id))?;
serde_json::from_value(n.data).ok()
}
pub fn resolutions(db: &Db) -> Vec<ResolutionRecord> {
let mut out: Vec<ResolutionRecord> = db
.list_ids_including_deleted(CONFLICTS)
.into_iter()
.filter_map(|k| db.get(CONFLICTS, &k))
.filter_map(|n| serde_json::from_value::<ResolutionRecord>(n.data).ok())
.collect();
out.sort_by(|a, b| (&a.coll, &a.id).cmp(&(&b.coll, &b.id)));
out
}
pub(crate) fn is_settled(db: &Db, c: &Conflict) -> bool {
match resolution_for(db, &c.branch, c.branch_created_seq, &c.coll, &c.id) {
Some(rec) => rec.theirs == c.theirs,
None => false,
}
}
#[cfg(test)]
mod tests {
use super::*;
use tempfile::tempdir;
fn j(v: u64) -> Value { serde_json::json!({ "v": v }) }
fn a_conflict() -> Conflict {
Conflict {
branch: "b".into(),
branch_created_seq: 0,
coll: "orders".into(),
id: "42".into(),
base: Some(j(1)),
ours: Some(j(2)),
theirs: Some(j(3)),
kind: ConflictKind::BothModified,
}
}
#[test]
fn taking_theirs_writes_their_value_as_a_new_version() {
let db = Db::in_memory();
db.put("orders", "42", j(1), vec![], None, None).unwrap();
let base_seq = db.seq.load(Ordering::SeqCst) - 1;
db.put("orders", "42", j(2), vec![], None, None).unwrap();
resolve(&db, &a_conflict(), Resolution::TakeTheirs).unwrap();
assert_eq!(db.get("orders", "42").unwrap().data, j(3));
assert_eq!(db.get_as_of("orders", "42", base_seq).unwrap().data, j(1));
}
#[test]
fn taking_ours_still_writes_a_version_rather_than_doing_nothing() {
let db = Db::in_memory();
db.put("orders", "42", j(1), vec![], None, None).unwrap();
db.put("orders", "42", j(2), vec![], None, None).unwrap();
let before = db.seq.load(Ordering::SeqCst);
resolve(&db, &a_conflict(), Resolution::TakeOurs).unwrap();
assert_eq!(db.get("orders", "42").unwrap().data, j(2));
assert!(db.seq.load(Ordering::SeqCst) > before,
"a decision is an event; it has to land in history to be auditable");
}
#[test]
fn a_third_value_can_be_chosen() {
let db = Db::in_memory();
db.put("orders", "42", j(2), vec![], None, None).unwrap();
let merged = serde_json::json!({ "v": 2, "note": "hand-merged" });
resolve(&db, &a_conflict(), Resolution::TakeValue(merged.clone())).unwrap();
assert_eq!(db.get("orders", "42").unwrap().data, merged);
}
#[test]
fn resolving_toward_a_delete_removes_the_live_document() {
let db = Db::in_memory();
db.put("orders", "42", j(2), vec![], None, None).unwrap();
let at = db.seq.load(Ordering::SeqCst) - 1;
let c = Conflict { theirs: None, kind: ConflictKind::ModifiedDeleted, ..a_conflict() };
resolve(&db, &c, Resolution::TakeTheirs).unwrap();
assert!(db.get("orders", "42").is_none());
assert_eq!(db.get_as_of("orders", "42", at).unwrap().data, j(2),
"a delete is a tombstone; the value before it is still readable");
}
#[test]
fn resolving_a_delete_that_already_happened_is_not_an_error() {
let db = Db::in_memory();
db.put("orders", "42", j(2), vec![], None, None).unwrap();
db.delete("orders", "42").unwrap();
let c = Conflict { ours: None, theirs: None, kind: ConflictKind::ModifiedDeleted, ..a_conflict() };
resolve(&db, &c, Resolution::TakeTheirs).unwrap();
assert!(db.get("orders", "42").is_none());
assert_eq!(resolutions(&db).len(), 1, "the decision is still recorded");
}
#[test]
fn every_resolution_leaves_an_audit_record_with_all_three_sides() {
let db = Db::in_memory();
db.put("orders", "42", j(2), vec![], None, None).unwrap();
resolve(&db, &a_conflict(), Resolution::TakeTheirs).unwrap();
let all = resolutions(&db);
assert_eq!(all.len(), 1);
let r = &all[0];
assert_eq!(r.coll, "orders");
assert_eq!(r.id, "42");
assert_eq!(r.kind, ConflictKind::BothModified);
assert_eq!(r.base, Some(j(1)));
assert_eq!(r.ours, Some(j(2)));
assert_eq!(r.theirs, Some(j(3)));
assert_eq!(r.choice, Choice::Theirs);
assert_eq!(r.chosen, Some(j(3)));
assert_eq!(resolution_for(&db, "b", 0, "orders", "42").as_ref(), Some(r));
}
#[test]
fn a_second_decision_supersedes_the_first_without_erasing_it() {
let db = Db::in_memory();
db.put("orders", "42", j(2), vec![], None, None).unwrap();
resolve(&db, &a_conflict(), Resolution::TakeOurs).unwrap();
let after_first = db.seq.load(Ordering::SeqCst) - 1;
resolve(&db, &a_conflict(), Resolution::TakeTheirs).unwrap();
assert_eq!(resolution_for(&db, "b", 0, "orders", "42").unwrap().choice, Choice::Theirs);
assert_eq!(resolutions(&db).len(), 1, "one live record per document");
let old = db.get_as_of(CONFLICTS, &conflict_key("b", 0, "orders", "42"), after_first).unwrap();
let old: ResolutionRecord = serde_json::from_value(old.data).unwrap();
assert_eq!(old.choice, Choice::Ours);
}
#[test]
fn a_decision_settles_the_branch_claim_it_was_taken_against_and_no_other() {
let db = Db::in_memory();
db.put("orders", "42", j(2), vec![], None, None).unwrap();
let c = a_conflict();
assert!(!is_settled(&db, &c), "nothing is settled before it is decided");
resolve(&db, &c, Resolution::TakeOurs).unwrap();
assert!(is_settled(&db, &c));
let moved_on = Conflict { theirs: Some(j(99)), ..c };
assert!(!is_settled(&db, &moved_on),
"a new claim from the branch is a new disagreement");
}
#[test]
fn conflict_keys_cannot_be_forged_by_a_clever_id() {
assert_ne!(conflict_key("br", 0, "a", "b|c"), conflict_key("br", 0, "a|b", "c"));
assert_ne!(conflict_key("br", 0, "ab", "c"), conflict_key("br", 0, "a", "bc"));
assert_eq!(conflict_key("br", 0, "a", "b"), conflict_key("br", 0, "a", "b"));
assert_ne!(conflict_key("x", 0, "a", "b"), conflict_key("y", 0, "a", "b"));
assert_ne!(conflict_key("x", 0, "a", "b"), conflict_key("x", 1, "a", "b"));
assert_ne!(conflict_key("xa", 0, "b", "c"), conflict_key("x", 0, "ab", "c"));
}
#[test]
fn resolution_works_on_disk_too() {
let dir = tempdir().unwrap();
let db = Db::open(dir.path(), None).unwrap();
db.put("orders", "42", j(2), vec![], None, None).unwrap();
resolve(&db, &a_conflict(), Resolution::TakeTheirs).unwrap();
db.flush_all();
assert_eq!(db.get("orders", "42").unwrap().data, j(3));
assert_eq!(resolutions(&db).len(), 1);
}
}
#[cfg(test)]
mod scoped_to_the_branch_that_raised_it {
use super::*;
use crate::branch::{branch_put, create_branch, get_branch};
use crate::merge;
fn j(v: u64) -> Value { serde_json::json!({ "v": v }) }
#[test]
fn resolving_one_branch_does_not_settle_another_making_the_same_claim() {
let db = Db::in_memory();
db.put("orders", "42", j(1), vec![], None, None).unwrap();
let base = db.seq.load(Ordering::SeqCst) - 1;
create_branch(&db, "x", base).unwrap();
create_branch(&db, "y", base).unwrap();
branch_put(&db, "x", "orders", "42", j(7)).unwrap();
branch_put(&db, "y", "orders", "42", j(7)).unwrap();
db.put("orders", "42", j(8), vec![], None, None).unwrap();
let px = merge::plan(&db, "x").unwrap();
let py = merge::plan(&db, "y").unwrap();
assert_eq!(px.conflicts.len(), 1, "X disagrees with the destination");
assert_eq!(py.conflicts.len(), 1, "so does Y");
resolve(&db, &px.conflicts[0], Resolution::TakeOurs).unwrap();
assert!(is_settled(&db, &px.conflicts[0]), "X was decided");
assert!(
!is_settled(&db, &py.conflicts[0]),
"NOBODY decided Y — a human decision about one line of history must \
not implicitly authorise another"
);
assert_eq!(
merge::plan(&db, "y").unwrap().conflicts.len(), 1,
"and Y must still be planned as conflicted"
);
}
#[test]
fn a_new_generation_of_a_name_starts_unresolved() {
let db = Db::in_memory();
db.put("orders", "42", j(1), vec![], None, None).unwrap();
let base = db.seq.load(Ordering::SeqCst) - 1;
create_branch(&db, "fix", base).unwrap();
branch_put(&db, "fix", "orders", "42", j(7)).unwrap();
db.put("orders", "42", j(8), vec![], None, None).unwrap();
let first = merge::plan(&db, "fix").unwrap().conflicts.remove(0);
resolve(&db, &first, Resolution::TakeOurs).unwrap();
assert!(is_settled(&db, &first));
crate::branch::abandon_branch(&db, "fix").unwrap();
let base2 = db.seq.load(Ordering::SeqCst) - 1;
create_branch(&db, "fix", base2).unwrap();
branch_put(&db, "fix", "orders", "42", j(7)).unwrap();
db.put("orders", "42", j(9), vec![], None, None).unwrap();
let again = merge::plan(&db, "fix").unwrap();
assert_eq!(again.conflicts.len(), 1);
assert!(
!is_settled(&db, &again.conflicts[0]),
"a name is a working label; the decision belonged to the generation"
);
assert_ne!(
get_branch(&db, "fix").unwrap().created_seq, first.branch_created_seq,
"precondition: this really is a different generation"
);
}
#[test]
fn a_resolution_records_which_branch_it_was_taken_against() {
let db = Db::in_memory();
db.put("orders", "42", j(1), vec![], None, None).unwrap();
let base = db.seq.load(Ordering::SeqCst) - 1;
create_branch(&db, "x", base).unwrap();
branch_put(&db, "x", "orders", "42", j(7)).unwrap();
db.put("orders", "42", j(8), vec![], None, None).unwrap();
let c = merge::plan(&db, "x").unwrap().conflicts.remove(0);
resolve(&db, &c, Resolution::TakeTheirs).unwrap();
let gen = get_branch(&db, "x").unwrap().created_seq;
let rec = resolution_for(&db, "x", gen, "orders", "42")
.expect("the decision is recorded under the branch that raised it");
assert_eq!(rec.branch, "x");
assert_eq!(rec.branch_created_seq, gen);
assert!(resolution_for(&db, "y", gen, "orders", "42").is_none());
}
}
#[cfg(test)]
mod replay_carries_its_cause {
use super::*;
use crate::branch::{branch_put, create_branch};
use crate::merge;
fn j(v: u64) -> Value { serde_json::json!({ "v": v }) }
#[test]
fn a_replayed_write_points_back_at_the_branch_write_that_caused_it() {
let db = Db::in_memory();
db.put("orders", "a", j(1), vec![], None, None).unwrap();
let base = db.seq.load(Ordering::SeqCst) - 1;
create_branch(&db, "x", base).unwrap();
let bw = branch_put(&db, "x", "orders", "a", j(2)).unwrap();
assert!(!bw.source_hash.is_empty(), "the branch write is addressable");
let plan = merge::plan(&db, "x").unwrap();
assert!(plan.is_clean());
assert_eq!(plan.changes.len(), 1);
assert_eq!(plan.changes[0].source_hash, bw.source_hash,
"the plan carries the source identity through");
merge::execute(&db, &plan).unwrap();
let landed = db.get("orders", "a").expect("the replay landed");
assert_eq!(landed.data, j(2));
assert_eq!(
landed.caused_by, vec![bw.source_hash.clone()],
"the destination node names the branch write that caused it"
);
let traced = db.trace(&landed.hash, false, 10);
assert!(
traced.iter().any(|n| n.hash == bw.source_hash),
"TRACE must reach the branch write from the merged node"
);
}
#[test]
fn every_replayed_change_carries_a_cause() {
let db = Db::in_memory();
for i in 0..4u64 {
db.put("orders", &i.to_string(), j(1), vec![], None, None).unwrap();
}
let base = db.seq.load(Ordering::SeqCst) - 1;
create_branch(&db, "x", base).unwrap();
for i in 0..4u64 {
branch_put(&db, "x", "orders", &i.to_string(), j(2)).unwrap();
}
let plan = merge::plan(&db, "x").unwrap();
assert_eq!(plan.changes.len(), 4);
assert!(
plan.changes.iter().all(|c| !c.source_hash.is_empty()),
"a plan with an anonymous change would replay an anonymous node"
);
merge::execute(&db, &plan).unwrap();
for i in 0..4u64 {
let n = db.get("orders", &i.to_string()).unwrap();
assert_eq!(n.caused_by.len(), 1, "doc {} lost its causal edge", i);
}
}
}