use std::sync::atomic::Ordering;
use anyhow::{bail, Result};
use serde::{Deserialize, Serialize};
use serde_json::Value;
use crate::branch::{self, BranchStatus};
use crate::conflict::{Conflict, ConflictKind};
use crate::db::Db;
use crate::namespace;
pub const MERGES: &str = "_nedb.merges";
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum ChangeKind {
Add,
Update,
Delete,
}
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
pub struct PlannedChange {
pub coll: String,
pub id: String,
pub kind: ChangeKind,
pub base: Option<Value>,
pub value: Option<Value>,
#[serde(default)]
pub source_hash: String,
}
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
pub struct MergePlan {
pub branch: String,
pub base_seq: u64,
pub into_seq: u64,
pub changes: Vec<PlannedChange>,
pub conflicts: Vec<Conflict>,
}
impl MergePlan {
pub fn is_empty(&self) -> bool {
self.changes.is_empty() && self.conflicts.is_empty()
}
pub fn is_clean(&self) -> bool {
self.conflicts.is_empty()
}
}
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
pub struct MergeRecord {
pub branch: String,
pub base_seq: u64,
pub merged_at_seq: u64,
pub replayed: usize,
pub state_root: Option<String>,
}
pub fn plan(db: &Db, branch_name: &str) -> Result<MergePlan> {
let Some(rec) = branch::get_branch(db, branch_name) else {
bail!("branch {:?} does not exist", branch_name)
};
match rec.status {
BranchStatus::Active => {}
BranchStatus::Merged { at_seq } => bail!(
"branch {:?} was already merged at sequence {} — merging it again would \
replay changes that are already in the destination's history",
branch_name, at_seq
),
BranchStatus::Abandoned => bail!(
"branch {:?} was abandoned; revive it by cutting a new branch rather than \
merging a line of work the registry records as given up",
branch_name
),
}
let into_seq = db.seq.load(Ordering::SeqCst).saturating_sub(1);
let mut changes = Vec::new();
let mut conflicts = Vec::new();
for w in branch::branch_writes(db, branch_name) {
let base = db.get_as_of(&w.coll, &w.id, rec.base_seq).map(|n| n.data);
let ours = db.get(&w.coll, &w.id).map(|n| n.data);
let theirs = w.value;
if theirs == base {
continue;
}
if ours == base {
let kind = if theirs.is_none() {
ChangeKind::Delete
} else if ours.is_none() {
ChangeKind::Add
} else {
ChangeKind::Update
};
changes.push(PlannedChange {
coll: w.coll, id: w.id, kind, base, value: theirs,
source_hash: w.source_hash,
});
continue;
}
if ours == theirs {
continue;
}
let kind = match (&ours, &theirs, &base) {
(None, Some(_), _) => ConflictKind::DeletedModified,
(Some(_), None, _) => ConflictKind::ModifiedDeleted,
(Some(_), Some(_), None) => ConflictKind::BothAdded,
_ => ConflictKind::BothModified,
};
let c = Conflict {
branch: rec.name.clone(),
branch_created_seq: rec.created_seq,
coll: w.coll, id: w.id, base, ours, theirs, kind,
};
if crate::conflict::is_settled(db, &c) {
continue;
}
conflicts.push(c);
}
changes.sort_by(|a, b| (&a.coll, &a.id).cmp(&(&b.coll, &b.id)));
conflicts.sort_by(|a, b| (&a.coll, &a.id).cmp(&(&b.coll, &b.id)));
Ok(MergePlan {
branch: branch_name.to_string(),
base_seq: rec.base_seq,
into_seq,
changes,
conflicts,
})
}
pub fn execute(db: &Db, plan: &MergePlan) -> Result<MergeRecord> {
if !plan.conflicts.is_empty() {
let names: Vec<String> = plan.conflicts.iter()
.map(|c| format!("{}/{}", c.coll, c.id))
.collect();
bail!(
"refusing to merge branch {:?}: {} unresolved conflict(s) — {}. Settle \
each one with conflict::resolve and re-plan; there is no side the engine \
may pick on your behalf.",
plan.branch, names.len(), names.join(", ")
);
}
let Some(rec) = branch::get_branch(db, &plan.branch) else {
bail!("branch {:?} does not exist", plan.branch)
};
if !rec.status.is_live() {
bail!("branch {:?} is {:?}, not active", plan.branch, rec.status);
}
if rec.base_seq != plan.base_seq {
bail!(
"plan for branch {:?} was computed against base sequence {}, but the \
branch forked at {}",
plan.branch, plan.base_seq, rec.base_seq
);
}
let tip = db.seq.load(Ordering::SeqCst).saturating_sub(1);
if tip != plan.into_seq {
bail!(
"plan for branch {:?} is stale: it was computed against destination \
sequence {}, which is now {}. Re-plan.",
plan.branch, plan.into_seq, tip
);
}
let mut replayed = 0usize;
for ch in &plan.changes {
namespace::validate_writable(&ch.coll)?;
match &ch.value {
Some(v) => {
let cause = if ch.source_hash.is_empty() {
eprintln!(
"nedb: merge replay of {}/{} has no source hash — the \
destination node will carry no causal edge to the \
branch write that caused it (overlay record predates \
source-hash capture)",
ch.coll, ch.id
);
vec![]
} else {
vec![ch.source_hash.clone()]
};
db.put(&ch.coll, &ch.id, v.clone(), cause, None, None)?;
}
None => { db.delete(&ch.coll, &ch.id)?; }
}
replayed += 1;
}
let merged_at_seq = db.seq.load(Ordering::SeqCst).saturating_sub(1);
let state_root = db.state_root().ok().map(|r| r.state_root);
let record = MergeRecord {
branch: plan.branch.clone(),
base_seq: plan.base_seq,
merged_at_seq,
replayed,
state_root,
};
db.put_unchecked(
MERGES,
&namespace::seq_id(merged_at_seq),
serde_json::to_value(&record)?,
vec![], None, None,
)?;
branch::mark_merged(db, &plan.branch, merged_at_seq)?;
Ok(record)
}
pub fn get_merge(db: &Db, merged_at_seq: u64) -> Option<MergeRecord> {
let n = db.get(MERGES, &namespace::seq_id(merged_at_seq))?;
serde_json::from_value(n.data).ok()
}
pub fn list_merges(db: &Db) -> Vec<MergeRecord> {
let mut ids = db.list_ids_including_deleted(MERGES);
ids.sort();
ids.into_iter()
.filter_map(|id| db.get(MERGES, &id))
.filter_map(|n| serde_json::from_value(n.data).ok())
.collect()
}
#[cfg(test)]
mod tests {
use super::*;
use crate::branch::{abandon_branch, branch_delete, branch_put, create_branch};
use crate::conflict::Resolution;
use tempfile::tempdir;
fn j(v: u64) -> Value { serde_json::json!({ "v": v }) }
fn tip(db: &Db) -> u64 { db.seq.load(Ordering::SeqCst).saturating_sub(1) }
fn forked() -> Db {
let db = Db::in_memory();
db.put("orders", "a", j(1), vec![], None, None).unwrap();
db.put("orders", "b", j(1), vec![], None, None).unwrap();
create_branch(&db, "b1", tip(&db)).unwrap();
db
}
fn fork_point(db: &Db) -> u64 {
branch::get_branch(db, "b1").expect("the fixture forked").base_seq
}
#[test]
fn plan_writes_nothing() {
let db = forked();
branch_put(&db, "b1", "orders", "a", j(2)).unwrap();
db.put("orders", "b", j(9), vec![], None, None).unwrap();
let before = db.seq.load(Ordering::SeqCst);
let p = plan(&db, "b1").unwrap();
let after = db.seq.load(Ordering::SeqCst);
assert_eq!(before, after, "planning moved the sequence counter — it wrote something");
assert!(!p.changes.is_empty(), "…and it did produce a real plan");
let p2 = plan(&db, "b1").unwrap();
assert_eq!(db.seq.load(Ordering::SeqCst), after);
assert_eq!(p, p2);
}
#[test]
fn a_branch_with_no_writes_plans_nothing() {
let db = forked();
let p = plan(&db, "b1").unwrap();
assert!(p.is_empty());
assert!(p.is_clean());
assert_eq!(p.base_seq, fork_point(&db));
assert_eq!(p.base_seq, tip(&db) - 1,
"the branch record itself advanced the tip past the fork point");
}
#[test]
fn both_sides_unchanged_is_no_change_even_when_the_branch_rewrote_the_value() {
let db = forked();
branch_put(&db, "b1", "orders", "a", j(1)).unwrap(); let p = plan(&db, "b1").unwrap();
assert!(p.changes.is_empty(), "a write that changed nothing carries nothing over");
assert!(p.conflicts.is_empty());
}
#[test]
fn a_one_sided_branch_change_is_a_clean_fast_forward() {
let db = forked();
branch_put(&db, "b1", "orders", "a", j(2)).unwrap();
let p = plan(&db, "b1").unwrap();
assert!(p.is_clean());
assert_eq!(p.changes.len(), 1);
assert_eq!(p.changes[0].kind, ChangeKind::Update);
assert_eq!(p.changes[0].base, Some(j(1)));
assert_eq!(p.changes[0].value, Some(j(2)));
}
#[test]
fn a_one_sided_destination_change_produces_no_plan_entry() {
let db = forked();
db.put("orders", "a", j(5), vec![], None, None).unwrap();
let p = plan(&db, "b1").unwrap();
assert!(p.is_empty(), "the destination already holds its own change");
}
#[test]
fn a_branch_add_and_a_branch_delete_are_classified() {
let db = forked();
branch_put(&db, "b1", "orders", "new", j(1)).unwrap();
branch_delete(&db, "b1", "orders", "b").unwrap();
let p = plan(&db, "b1").unwrap();
assert!(p.is_clean());
let kinds: Vec<(String, ChangeKind)> = p.changes.iter()
.map(|c| (c.id.clone(), c.kind)).collect();
assert_eq!(kinds, vec![
("b".to_string(), ChangeKind::Delete),
("new".to_string(), ChangeKind::Add),
]);
}
#[test]
fn a_convergent_identical_edit_is_not_a_conflict() {
let db = forked();
branch_put(&db, "b1", "orders", "a", j(7)).unwrap();
db.put("orders", "a", j(7), vec![], None, None).unwrap();
let p = plan(&db, "b1").unwrap();
assert!(p.conflicts.is_empty(), "agreeing is not disagreeing");
assert!(p.changes.is_empty(), "and there is nothing left to write");
}
#[test]
fn a_divergent_edit_is_a_conflict_carrying_all_three_sides() {
let db = forked();
branch_put(&db, "b1", "orders", "a", j(7)).unwrap();
db.put("orders", "a", j(8), vec![], None, None).unwrap();
let p = plan(&db, "b1").unwrap();
assert!(p.changes.is_empty(), "nothing may be replayed while a conflict stands");
assert_eq!(p.conflicts.len(), 1);
let c = &p.conflicts[0];
assert_eq!(c.kind, ConflictKind::BothModified);
assert_eq!(c.base, Some(j(1)));
assert_eq!(c.ours, Some(j(8)));
assert_eq!(c.theirs, Some(j(7)));
}
#[test]
fn delete_against_modify_is_classified_from_the_destinations_point_of_view() {
let db = forked();
branch_delete(&db, "b1", "orders", "a").unwrap();
db.put("orders", "a", j(8), vec![], None, None).unwrap();
assert_eq!(plan(&db, "b1").unwrap().conflicts[0].kind, ConflictKind::ModifiedDeleted);
let db = forked();
branch_put(&db, "b1", "orders", "a", j(8)).unwrap();
db.delete("orders", "a").unwrap();
assert_eq!(plan(&db, "b1").unwrap().conflicts[0].kind, ConflictKind::DeletedModified);
}
#[test]
fn two_creations_of_the_same_id_are_both_added() {
let db = forked();
branch_put(&db, "b1", "orders", "fresh", j(1)).unwrap();
db.put("orders", "fresh", j(2), vec![], None, None).unwrap();
let p = plan(&db, "b1").unwrap();
assert_eq!(p.conflicts.len(), 1);
assert_eq!(p.conflicts[0].kind, ConflictKind::BothAdded);
assert_eq!(p.conflicts[0].base, None);
}
#[test]
fn planning_a_closed_branch_is_refused() {
let db = forked();
abandon_branch(&db, "b1").unwrap();
assert!(plan(&db, "b1").unwrap_err().to_string().contains("abandoned"));
assert!(plan(&db, "never-existed").is_err());
}
#[test]
fn execute_refuses_while_conflicts_stand() {
let db = forked();
branch_put(&db, "b1", "orders", "a", j(7)).unwrap();
db.put("orders", "a", j(8), vec![], None, None).unwrap();
let p = plan(&db, "b1").unwrap();
let before = db.seq.load(Ordering::SeqCst);
let err = execute(&db, &p).unwrap_err().to_string();
assert!(err.contains("unresolved conflict"), "{}", err);
assert!(err.contains("orders/a"), "the refusal must name the document: {}", err);
assert_eq!(db.seq.load(Ordering::SeqCst), before, "a refused merge writes nothing");
assert_eq!(db.get("orders", "a").unwrap().data, j(8), "…and changes nothing");
assert!(crate::branch::get_branch(&db, "b1").unwrap().status.is_live(),
"…and leaves the branch open");
}
#[test]
fn execute_replays_a_clean_plan_and_records_it() {
let db = forked();
branch_put(&db, "b1", "orders", "a", j(2)).unwrap();
branch_put(&db, "b1", "orders", "new", j(3)).unwrap();
branch_delete(&db, "b1", "orders", "b").unwrap();
let p = plan(&db, "b1").unwrap();
assert_eq!(p.changes.len(), 3);
let rec = execute(&db, &p).unwrap();
assert_eq!(db.get("orders", "a").unwrap().data, j(2));
assert_eq!(db.get("orders", "new").unwrap().data, j(3));
assert!(db.get("orders", "b").is_none());
assert_eq!(rec.branch, "b1");
assert_eq!(rec.replayed, 3);
assert_eq!(rec.base_seq, p.base_seq);
assert!(rec.state_root.is_some());
assert_eq!(get_merge(&db, rec.merged_at_seq).as_ref(), Some(&rec));
assert_eq!(list_merges(&db), vec![rec.clone()]);
}
#[test]
fn execute_flips_the_branch_to_merged_and_releases_its_pin() {
let db = forked();
branch_put(&db, "b1", "orders", "a", j(2)).unwrap();
assert!(crate::branch::minimum_pinned_seq(&db).is_some());
let p = plan(&db, "b1").unwrap();
let rec = execute(&db, &p).unwrap();
let b = crate::branch::get_branch(&db, "b1").unwrap();
assert_eq!(b.status, BranchStatus::Merged { at_seq: rec.merged_at_seq });
assert_eq!(crate::branch::minimum_pinned_seq(&db), None);
db.compact().expect("a merged branch no longer blocks compaction");
}
#[test]
fn merged_writes_are_new_history_and_the_base_version_survives() {
let db = forked();
let base = tip(&db);
branch_put(&db, "b1", "orders", "a", j(2)).unwrap();
let p = plan(&db, "b1").unwrap();
execute(&db, &p).unwrap();
assert_eq!(db.get("orders", "a").unwrap().data, j(2), "the merge landed");
assert_eq!(db.get_as_of("orders", "a", base).unwrap().data, j(1),
"the pre-merge value is still readable at the fork point");
let now = db.get("orders", "a").unwrap();
assert!(now.seq > base, "the replayed write has a destination sequence");
assert!(now.prev.is_some(), "it is a continuation of the destination's chain");
}
#[test]
fn a_deleted_document_is_still_readable_before_the_merge_that_removed_it() {
let db = forked();
let base = tip(&db);
branch_delete(&db, "b1", "orders", "b").unwrap();
let p = plan(&db, "b1").unwrap();
execute(&db, &p).unwrap();
assert!(db.get("orders", "b").is_none());
assert_eq!(db.get_as_of("orders", "b", base).unwrap().data, j(1));
}
#[test]
fn an_empty_merge_is_allowed_and_still_closes_the_branch() {
let db = forked();
let p = plan(&db, "b1").unwrap();
let rec = execute(&db, &p).unwrap();
assert_eq!(rec.replayed, 0);
assert!(matches!(crate::branch::get_branch(&db, "b1").unwrap().status,
BranchStatus::Merged { .. }));
}
#[test]
fn a_branch_cannot_be_merged_twice() {
let db = forked();
branch_put(&db, "b1", "orders", "a", j(2)).unwrap();
let p = plan(&db, "b1").unwrap();
execute(&db, &p).unwrap();
assert!(execute(&db, &p).is_err(), "the branch is closed");
assert!(plan(&db, "b1").unwrap_err().to_string().contains("already merged"));
}
#[test]
fn a_stale_plan_is_refused_rather_than_silently_overwriting() {
let db = forked();
branch_put(&db, "b1", "orders", "a", j(2)).unwrap();
let p = plan(&db, "b1").unwrap();
db.put("orders", "a", j(99), vec![], None, None).unwrap();
let err = execute(&db, &p).unwrap_err().to_string();
assert!(err.contains("stale"), "{}", err);
assert_eq!(db.get("orders", "a").unwrap().data, j(99), "their write survived");
let p2 = plan(&db, "b1").unwrap();
assert_eq!(p2.conflicts.len(), 1);
}
#[test]
fn resolving_toward_the_branch_clears_the_conflict_and_the_merge_proceeds() {
let db = forked();
branch_put(&db, "b1", "orders", "a", j(7)).unwrap();
db.put("orders", "a", j(8), vec![], None, None).unwrap();
let p = plan(&db, "b1").unwrap();
assert_eq!(p.conflicts.len(), 1);
crate::conflict::resolve(&db, &p.conflicts[0], Resolution::TakeTheirs).unwrap();
let p2 = plan(&db, "b1").unwrap();
assert!(p2.is_clean(), "the decision settled it");
execute(&db, &p2).unwrap();
assert_eq!(db.get("orders", "a").unwrap().data, j(7));
assert_eq!(crate::conflict::resolutions(&db).len(), 1, "and it is auditable");
}
#[test]
fn resolving_toward_the_destination_also_clears_the_conflict() {
let db = forked();
branch_put(&db, "b1", "orders", "a", j(7)).unwrap();
db.put("orders", "a", j(8), vec![], None, None).unwrap();
let p = plan(&db, "b1").unwrap();
crate::conflict::resolve(&db, &p.conflicts[0], Resolution::TakeOurs).unwrap();
let p2 = plan(&db, "b1").unwrap();
assert!(p2.is_clean(), "a decision to keep ours is still a decision");
execute(&db, &p2).unwrap();
assert_eq!(db.get("orders", "a").unwrap().data, j(8));
}
#[test]
fn a_hand_merged_third_value_settles_it_too() {
let db = forked();
branch_put(&db, "b1", "orders", "a", j(7)).unwrap();
db.put("orders", "a", j(8), vec![], None, None).unwrap();
let p = plan(&db, "b1").unwrap();
let both = serde_json::json!({ "v": 15, "note": "summed by hand" });
crate::conflict::resolve(&db, &p.conflicts[0], Resolution::TakeValue(both.clone())).unwrap();
let p2 = plan(&db, "b1").unwrap();
assert!(p2.is_clean());
execute(&db, &p2).unwrap();
assert_eq!(db.get("orders", "a").unwrap().data, both);
}
#[test]
fn two_branches_from_one_fork_merge_independently() {
let db = Db::in_memory();
db.put("orders", "a", j(1), vec![], None, None).unwrap();
db.put("orders", "b", j(1), vec![], None, None).unwrap();
let base = tip(&db);
create_branch(&db, "x", base).unwrap();
create_branch(&db, "y", base).unwrap();
branch_put(&db, "x", "orders", "a", j(2)).unwrap();
branch_put(&db, "y", "orders", "b", j(2)).unwrap();
let px = plan(&db, "x").unwrap();
execute(&db, &px).unwrap();
let py = plan(&db, "y").unwrap();
assert!(py.is_clean(), "disjoint documents do not conflict");
execute(&db, &py).unwrap();
assert_eq!(db.get("orders", "a").unwrap().data, j(2));
assert_eq!(db.get("orders", "b").unwrap().data, j(2));
assert_eq!(list_merges(&db).len(), 2);
assert_eq!(crate::branch::minimum_pinned_seq(&db), None);
}
#[test]
fn a_merge_cannot_reach_a_reserved_collection() {
let db = forked();
assert!(branch_put(&db, "b1", namespace::ROOTS, "x", j(1)).is_err());
let bad = MergePlan {
branch: "b1".into(),
base_seq: crate::branch::get_branch(&db, "b1").unwrap().base_seq,
into_seq: tip(&db),
changes: vec![PlannedChange {
coll: namespace::ROOTS.into(), id: "x".into(),
kind: ChangeKind::Add, base: None, value: Some(j(1)),
source_hash: String::new(),
}],
conflicts: vec![],
};
assert!(execute(&db, &bad).is_err());
}
#[test]
fn the_whole_cycle_works_on_disk() {
let dir = tempdir().unwrap();
let db = Db::open(dir.path(), None).unwrap();
db.put("orders", "a", j(1), vec![], None, None).unwrap();
create_branch(&db, "d1", tip(&db)).unwrap();
branch_put(&db, "d1", "orders", "a", j(2)).unwrap();
let p = plan(&db, "d1").unwrap();
let rec = execute(&db, &p).unwrap();
db.flush_all();
assert_eq!(db.get("orders", "a").unwrap().data, j(2));
assert_eq!(get_merge(&db, rec.merged_at_seq).unwrap().replayed, 1);
}
}