1use std::sync::atomic::Ordering;
47
48use anyhow::{bail, Result};
49use serde::{Deserialize, Serialize};
50use serde_json::Value;
51
52use crate::branch::{self, BranchStatus};
53use crate::conflict::{Conflict, ConflictKind};
54use crate::db::Db;
55use crate::namespace;
56
57pub const MERGES: &str = "_nedb.merges";
60
61#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
63#[serde(rename_all = "snake_case")]
64pub enum ChangeKind {
65 Add,
67 Update,
69 Delete,
71}
72
73#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
79pub struct PlannedChange {
80 pub coll: String,
81 pub id: String,
82 pub kind: ChangeKind,
83 pub base: Option<Value>,
84 pub value: Option<Value>,
86 #[serde(default)]
97 pub source_hash: String,
98}
99
100#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
102pub struct MergePlan {
103 pub branch: String,
104 pub base_seq: u64,
105 pub into_seq: u64,
108 pub changes: Vec<PlannedChange>,
109 pub conflicts: Vec<Conflict>,
110}
111
112impl MergePlan {
113 pub fn is_empty(&self) -> bool {
115 self.changes.is_empty() && self.conflicts.is_empty()
116 }
117 pub fn is_clean(&self) -> bool {
119 self.conflicts.is_empty()
120 }
121}
122
123#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
125pub struct MergeRecord {
126 pub branch: String,
127 pub base_seq: u64,
128 pub merged_at_seq: u64,
131 pub replayed: usize,
132 pub state_root: Option<String>,
137}
138
139pub fn plan(db: &Db, branch_name: &str) -> Result<MergePlan> {
146 let Some(rec) = branch::get_branch(db, branch_name) else {
147 bail!("branch {:?} does not exist", branch_name)
148 };
149 match rec.status {
150 BranchStatus::Active => {}
151 BranchStatus::Merged { at_seq } => bail!(
152 "branch {:?} was already merged at sequence {} — merging it again would \
153 replay changes that are already in the destination's history",
154 branch_name, at_seq
155 ),
156 BranchStatus::Abandoned => bail!(
157 "branch {:?} was abandoned; revive it by cutting a new branch rather than \
158 merging a line of work the registry records as given up",
159 branch_name
160 ),
161 }
162
163 let into_seq = db.seq.load(Ordering::SeqCst).saturating_sub(1);
164 let mut changes = Vec::new();
165 let mut conflicts = Vec::new();
166
167 for w in branch::branch_writes(db, branch_name) {
172 let base = db.get_as_of(&w.coll, &w.id, rec.base_seq).map(|n| n.data);
173 let ours = db.get(&w.coll, &w.id).map(|n| n.data);
174 let theirs = w.value;
175
176 if theirs == base {
177 continue;
180 }
181 if ours == base {
182 let kind = if theirs.is_none() {
184 ChangeKind::Delete
185 } else if ours.is_none() {
186 ChangeKind::Add
187 } else {
188 ChangeKind::Update
189 };
190 changes.push(PlannedChange {
191 coll: w.coll, id: w.id, kind, base, value: theirs,
192 source_hash: w.source_hash,
193 });
194 continue;
195 }
196 if ours == theirs {
197 continue;
200 }
201
202 let kind = match (&ours, &theirs, &base) {
203 (None, Some(_), _) => ConflictKind::DeletedModified,
204 (Some(_), None, _) => ConflictKind::ModifiedDeleted,
205 (Some(_), Some(_), None) => ConflictKind::BothAdded,
206 _ => ConflictKind::BothModified,
207 };
208 let c = Conflict {
209 branch: rec.name.clone(),
210 branch_created_seq: rec.created_seq,
211 coll: w.coll, id: w.id, base, ours, theirs, kind,
212 };
213 if crate::conflict::is_settled(db, &c) {
217 continue;
218 }
219 conflicts.push(c);
220 }
221
222 changes.sort_by(|a, b| (&a.coll, &a.id).cmp(&(&b.coll, &b.id)));
223 conflicts.sort_by(|a, b| (&a.coll, &a.id).cmp(&(&b.coll, &b.id)));
224
225 Ok(MergePlan {
226 branch: branch_name.to_string(),
227 base_seq: rec.base_seq,
228 into_seq,
229 changes,
230 conflicts,
231 })
232}
233
234pub fn execute(db: &Db, plan: &MergePlan) -> Result<MergeRecord> {
240 if !plan.conflicts.is_empty() {
241 let names: Vec<String> = plan.conflicts.iter()
244 .map(|c| format!("{}/{}", c.coll, c.id))
245 .collect();
246 bail!(
247 "refusing to merge branch {:?}: {} unresolved conflict(s) — {}. Settle \
248 each one with conflict::resolve and re-plan; there is no side the engine \
249 may pick on your behalf.",
250 plan.branch, names.len(), names.join(", ")
251 );
252 }
253
254 let Some(rec) = branch::get_branch(db, &plan.branch) else {
255 bail!("branch {:?} does not exist", plan.branch)
256 };
257 if !rec.status.is_live() {
258 bail!("branch {:?} is {:?}, not active", plan.branch, rec.status);
259 }
260 if rec.base_seq != plan.base_seq {
261 bail!(
262 "plan for branch {:?} was computed against base sequence {}, but the \
263 branch forked at {}",
264 plan.branch, plan.base_seq, rec.base_seq
265 );
266 }
267
268 let tip = db.seq.load(Ordering::SeqCst).saturating_sub(1);
275 if tip != plan.into_seq {
276 bail!(
277 "plan for branch {:?} is stale: it was computed against destination \
278 sequence {}, which is now {}. Re-plan.",
279 plan.branch, plan.into_seq, tip
280 );
281 }
282
283 let mut replayed = 0usize;
287 for ch in &plan.changes {
288 namespace::validate_writable(&ch.coll)?;
289 match &ch.value {
290 Some(v) => {
291 let cause = if ch.source_hash.is_empty() {
300 eprintln!(
305 "nedb: merge replay of {}/{} has no source hash — the \
306 destination node will carry no causal edge to the \
307 branch write that caused it (overlay record predates \
308 source-hash capture)",
309 ch.coll, ch.id
310 );
311 vec![]
312 } else {
313 vec![ch.source_hash.clone()]
314 };
315 db.put(&ch.coll, &ch.id, v.clone(), cause, None, None)?;
316 }
317 None => { db.delete(&ch.coll, &ch.id)?; }
318 }
319 replayed += 1;
320 }
321
322 let merged_at_seq = db.seq.load(Ordering::SeqCst).saturating_sub(1);
323 let state_root = db.state_root().ok().map(|r| r.state_root);
324
325 let record = MergeRecord {
326 branch: plan.branch.clone(),
327 base_seq: plan.base_seq,
328 merged_at_seq,
329 replayed,
330 state_root,
331 };
332 db.put_unchecked(
333 MERGES,
334 &namespace::seq_id(merged_at_seq),
335 serde_json::to_value(&record)?,
336 vec![], None, None,
337 )?;
338
339 branch::mark_merged(db, &plan.branch, merged_at_seq)?;
344
345 Ok(record)
346}
347
348pub fn get_merge(db: &Db, merged_at_seq: u64) -> Option<MergeRecord> {
350 let n = db.get(MERGES, &namespace::seq_id(merged_at_seq))?;
351 serde_json::from_value(n.data).ok()
352}
353
354pub fn list_merges(db: &Db) -> Vec<MergeRecord> {
356 let mut ids = db.list_ids_including_deleted(MERGES);
357 ids.sort();
358 ids.into_iter()
359 .filter_map(|id| db.get(MERGES, &id))
360 .filter_map(|n| serde_json::from_value(n.data).ok())
361 .collect()
362}
363
364#[cfg(test)]
365mod tests {
366 use super::*;
367 use crate::branch::{abandon_branch, branch_delete, branch_put, create_branch};
368 use crate::conflict::Resolution;
369 use tempfile::tempdir;
370
371 fn j(v: u64) -> Value { serde_json::json!({ "v": v }) }
372
373 fn tip(db: &Db) -> u64 { db.seq.load(Ordering::SeqCst).saturating_sub(1) }
374
375 fn forked() -> Db {
378 let db = Db::in_memory();
379 db.put("orders", "a", j(1), vec![], None, None).unwrap();
380 db.put("orders", "b", j(1), vec![], None, None).unwrap();
381 create_branch(&db, "b1", tip(&db)).unwrap();
382 db
383 }
384
385 fn fork_point(db: &Db) -> u64 {
394 branch::get_branch(db, "b1").expect("the fixture forked").base_seq
395 }
396
397 #[test]
400 fn plan_writes_nothing() {
401 let db = forked();
402 branch_put(&db, "b1", "orders", "a", j(2)).unwrap();
403 db.put("orders", "b", j(9), vec![], None, None).unwrap();
404
405 let before = db.seq.load(Ordering::SeqCst);
406 let p = plan(&db, "b1").unwrap();
407 let after = db.seq.load(Ordering::SeqCst);
408 assert_eq!(before, after, "planning moved the sequence counter — it wrote something");
409 assert!(!p.changes.is_empty(), "…and it did produce a real plan");
410
411 let p2 = plan(&db, "b1").unwrap();
413 assert_eq!(db.seq.load(Ordering::SeqCst), after);
414 assert_eq!(p, p2);
415 }
416
417 #[test]
418 fn a_branch_with_no_writes_plans_nothing() {
419 let db = forked();
420 let p = plan(&db, "b1").unwrap();
421 assert!(p.is_empty());
422 assert!(p.is_clean());
423 assert_eq!(p.base_seq, fork_point(&db));
424 assert_eq!(p.base_seq, tip(&db) - 1,
425 "the branch record itself advanced the tip past the fork point");
426 }
427
428 #[test]
429 fn both_sides_unchanged_is_no_change_even_when_the_branch_rewrote_the_value() {
430 let db = forked();
431 branch_put(&db, "b1", "orders", "a", j(1)).unwrap(); let p = plan(&db, "b1").unwrap();
433 assert!(p.changes.is_empty(), "a write that changed nothing carries nothing over");
434 assert!(p.conflicts.is_empty());
435 }
436
437 #[test]
438 fn a_one_sided_branch_change_is_a_clean_fast_forward() {
439 let db = forked();
440 branch_put(&db, "b1", "orders", "a", j(2)).unwrap();
441 let p = plan(&db, "b1").unwrap();
442 assert!(p.is_clean());
443 assert_eq!(p.changes.len(), 1);
444 assert_eq!(p.changes[0].kind, ChangeKind::Update);
445 assert_eq!(p.changes[0].base, Some(j(1)));
446 assert_eq!(p.changes[0].value, Some(j(2)));
447 }
448
449 #[test]
450 fn a_one_sided_destination_change_produces_no_plan_entry() {
451 let db = forked();
452 db.put("orders", "a", j(5), vec![], None, None).unwrap();
453 let p = plan(&db, "b1").unwrap();
454 assert!(p.is_empty(), "the destination already holds its own change");
455 }
456
457 #[test]
458 fn a_branch_add_and_a_branch_delete_are_classified() {
459 let db = forked();
460 branch_put(&db, "b1", "orders", "new", j(1)).unwrap();
461 branch_delete(&db, "b1", "orders", "b").unwrap();
462 let p = plan(&db, "b1").unwrap();
463 assert!(p.is_clean());
464 let kinds: Vec<(String, ChangeKind)> = p.changes.iter()
465 .map(|c| (c.id.clone(), c.kind)).collect();
466 assert_eq!(kinds, vec![
467 ("b".to_string(), ChangeKind::Delete),
468 ("new".to_string(), ChangeKind::Add),
469 ]);
470 }
471
472 #[test]
473 fn a_convergent_identical_edit_is_not_a_conflict() {
474 let db = forked();
475 branch_put(&db, "b1", "orders", "a", j(7)).unwrap();
476 db.put("orders", "a", j(7), vec![], None, None).unwrap();
477 let p = plan(&db, "b1").unwrap();
478 assert!(p.conflicts.is_empty(), "agreeing is not disagreeing");
479 assert!(p.changes.is_empty(), "and there is nothing left to write");
480 }
481
482 #[test]
483 fn a_divergent_edit_is_a_conflict_carrying_all_three_sides() {
484 let db = forked();
485 branch_put(&db, "b1", "orders", "a", j(7)).unwrap();
486 db.put("orders", "a", j(8), vec![], None, None).unwrap();
487 let p = plan(&db, "b1").unwrap();
488 assert!(p.changes.is_empty(), "nothing may be replayed while a conflict stands");
489 assert_eq!(p.conflicts.len(), 1);
490 let c = &p.conflicts[0];
491 assert_eq!(c.kind, ConflictKind::BothModified);
492 assert_eq!(c.base, Some(j(1)));
493 assert_eq!(c.ours, Some(j(8)));
494 assert_eq!(c.theirs, Some(j(7)));
495 }
496
497 #[test]
498 fn delete_against_modify_is_classified_from_the_destinations_point_of_view() {
499 let db = forked();
500 branch_delete(&db, "b1", "orders", "a").unwrap();
501 db.put("orders", "a", j(8), vec![], None, None).unwrap();
502 assert_eq!(plan(&db, "b1").unwrap().conflicts[0].kind, ConflictKind::ModifiedDeleted);
503
504 let db = forked();
505 branch_put(&db, "b1", "orders", "a", j(8)).unwrap();
506 db.delete("orders", "a").unwrap();
507 assert_eq!(plan(&db, "b1").unwrap().conflicts[0].kind, ConflictKind::DeletedModified);
508 }
509
510 #[test]
511 fn two_creations_of_the_same_id_are_both_added() {
512 let db = forked();
513 branch_put(&db, "b1", "orders", "fresh", j(1)).unwrap();
514 db.put("orders", "fresh", j(2), vec![], None, None).unwrap();
515 let p = plan(&db, "b1").unwrap();
516 assert_eq!(p.conflicts.len(), 1);
517 assert_eq!(p.conflicts[0].kind, ConflictKind::BothAdded);
518 assert_eq!(p.conflicts[0].base, None);
519 }
520
521 #[test]
522 fn planning_a_closed_branch_is_refused() {
523 let db = forked();
524 abandon_branch(&db, "b1").unwrap();
525 assert!(plan(&db, "b1").unwrap_err().to_string().contains("abandoned"));
526 assert!(plan(&db, "never-existed").is_err());
527 }
528
529 #[test]
532 fn execute_refuses_while_conflicts_stand() {
533 let db = forked();
534 branch_put(&db, "b1", "orders", "a", j(7)).unwrap();
535 db.put("orders", "a", j(8), vec![], None, None).unwrap();
536 let p = plan(&db, "b1").unwrap();
537 let before = db.seq.load(Ordering::SeqCst);
538
539 let err = execute(&db, &p).unwrap_err().to_string();
540 assert!(err.contains("unresolved conflict"), "{}", err);
541 assert!(err.contains("orders/a"), "the refusal must name the document: {}", err);
542 assert_eq!(db.seq.load(Ordering::SeqCst), before, "a refused merge writes nothing");
543 assert_eq!(db.get("orders", "a").unwrap().data, j(8), "…and changes nothing");
544 assert!(crate::branch::get_branch(&db, "b1").unwrap().status.is_live(),
545 "…and leaves the branch open");
546 }
547
548 #[test]
549 fn execute_replays_a_clean_plan_and_records_it() {
550 let db = forked();
551 branch_put(&db, "b1", "orders", "a", j(2)).unwrap();
552 branch_put(&db, "b1", "orders", "new", j(3)).unwrap();
553 branch_delete(&db, "b1", "orders", "b").unwrap();
554
555 let p = plan(&db, "b1").unwrap();
556 assert_eq!(p.changes.len(), 3);
557 let rec = execute(&db, &p).unwrap();
558
559 assert_eq!(db.get("orders", "a").unwrap().data, j(2));
560 assert_eq!(db.get("orders", "new").unwrap().data, j(3));
561 assert!(db.get("orders", "b").is_none());
562
563 assert_eq!(rec.branch, "b1");
564 assert_eq!(rec.replayed, 3);
565 assert_eq!(rec.base_seq, p.base_seq);
566 assert!(rec.state_root.is_some());
567 assert_eq!(get_merge(&db, rec.merged_at_seq).as_ref(), Some(&rec));
568 assert_eq!(list_merges(&db), vec![rec.clone()]);
569 }
570
571 #[test]
572 fn execute_flips_the_branch_to_merged_and_releases_its_pin() {
573 let db = forked();
574 branch_put(&db, "b1", "orders", "a", j(2)).unwrap();
575 assert!(crate::branch::minimum_pinned_seq(&db).is_some());
576
577 let p = plan(&db, "b1").unwrap();
578 let rec = execute(&db, &p).unwrap();
579
580 let b = crate::branch::get_branch(&db, "b1").unwrap();
581 assert_eq!(b.status, BranchStatus::Merged { at_seq: rec.merged_at_seq });
582 assert_eq!(crate::branch::minimum_pinned_seq(&db), None);
583 db.compact().expect("a merged branch no longer blocks compaction");
584 }
585
586 #[test]
588 fn merged_writes_are_new_history_and_the_base_version_survives() {
589 let db = forked();
590 let base = tip(&db);
591 branch_put(&db, "b1", "orders", "a", j(2)).unwrap();
592 let p = plan(&db, "b1").unwrap();
593 execute(&db, &p).unwrap();
594
595 assert_eq!(db.get("orders", "a").unwrap().data, j(2), "the merge landed");
596 assert_eq!(db.get_as_of("orders", "a", base).unwrap().data, j(1),
597 "the pre-merge value is still readable at the fork point");
598
599 let now = db.get("orders", "a").unwrap();
602 assert!(now.seq > base, "the replayed write has a destination sequence");
603 assert!(now.prev.is_some(), "it is a continuation of the destination's chain");
604 }
605
606 #[test]
607 fn a_deleted_document_is_still_readable_before_the_merge_that_removed_it() {
608 let db = forked();
609 let base = tip(&db);
610 branch_delete(&db, "b1", "orders", "b").unwrap();
611 let p = plan(&db, "b1").unwrap();
612 execute(&db, &p).unwrap();
613 assert!(db.get("orders", "b").is_none());
614 assert_eq!(db.get_as_of("orders", "b", base).unwrap().data, j(1));
615 }
616
617 #[test]
618 fn an_empty_merge_is_allowed_and_still_closes_the_branch() {
619 let db = forked();
620 let p = plan(&db, "b1").unwrap();
621 let rec = execute(&db, &p).unwrap();
622 assert_eq!(rec.replayed, 0);
623 assert!(matches!(crate::branch::get_branch(&db, "b1").unwrap().status,
624 BranchStatus::Merged { .. }));
625 }
626
627 #[test]
628 fn a_branch_cannot_be_merged_twice() {
629 let db = forked();
630 branch_put(&db, "b1", "orders", "a", j(2)).unwrap();
631 let p = plan(&db, "b1").unwrap();
632 execute(&db, &p).unwrap();
633 assert!(execute(&db, &p).is_err(), "the branch is closed");
634 assert!(plan(&db, "b1").unwrap_err().to_string().contains("already merged"));
635 }
636
637 #[test]
638 fn a_stale_plan_is_refused_rather_than_silently_overwriting() {
639 let db = forked();
640 branch_put(&db, "b1", "orders", "a", j(2)).unwrap();
641 let p = plan(&db, "b1").unwrap();
642 db.put("orders", "a", j(99), vec![], None, None).unwrap();
644
645 let err = execute(&db, &p).unwrap_err().to_string();
646 assert!(err.contains("stale"), "{}", err);
647 assert_eq!(db.get("orders", "a").unwrap().data, j(99), "their write survived");
648
649 let p2 = plan(&db, "b1").unwrap();
651 assert_eq!(p2.conflicts.len(), 1);
652 }
653
654 #[test]
657 fn resolving_toward_the_branch_clears_the_conflict_and_the_merge_proceeds() {
658 let db = forked();
659 branch_put(&db, "b1", "orders", "a", j(7)).unwrap();
660 db.put("orders", "a", j(8), vec![], None, None).unwrap();
661
662 let p = plan(&db, "b1").unwrap();
663 assert_eq!(p.conflicts.len(), 1);
664 crate::conflict::resolve(&db, &p.conflicts[0], Resolution::TakeTheirs).unwrap();
665
666 let p2 = plan(&db, "b1").unwrap();
667 assert!(p2.is_clean(), "the decision settled it");
668 execute(&db, &p2).unwrap();
669 assert_eq!(db.get("orders", "a").unwrap().data, j(7));
670 assert_eq!(crate::conflict::resolutions(&db).len(), 1, "and it is auditable");
671 }
672
673 #[test]
677 fn resolving_toward_the_destination_also_clears_the_conflict() {
678 let db = forked();
679 branch_put(&db, "b1", "orders", "a", j(7)).unwrap();
680 db.put("orders", "a", j(8), vec![], None, None).unwrap();
681
682 let p = plan(&db, "b1").unwrap();
683 crate::conflict::resolve(&db, &p.conflicts[0], Resolution::TakeOurs).unwrap();
684
685 let p2 = plan(&db, "b1").unwrap();
686 assert!(p2.is_clean(), "a decision to keep ours is still a decision");
687 execute(&db, &p2).unwrap();
688 assert_eq!(db.get("orders", "a").unwrap().data, j(8));
689 }
690
691 #[test]
692 fn a_hand_merged_third_value_settles_it_too() {
693 let db = forked();
694 branch_put(&db, "b1", "orders", "a", j(7)).unwrap();
695 db.put("orders", "a", j(8), vec![], None, None).unwrap();
696 let p = plan(&db, "b1").unwrap();
697 let both = serde_json::json!({ "v": 15, "note": "summed by hand" });
698 crate::conflict::resolve(&db, &p.conflicts[0], Resolution::TakeValue(both.clone())).unwrap();
699
700 let p2 = plan(&db, "b1").unwrap();
701 assert!(p2.is_clean());
702 execute(&db, &p2).unwrap();
703 assert_eq!(db.get("orders", "a").unwrap().data, both);
704 }
705
706 #[test]
707 fn two_branches_from_one_fork_merge_independently() {
708 let db = Db::in_memory();
709 db.put("orders", "a", j(1), vec![], None, None).unwrap();
710 db.put("orders", "b", j(1), vec![], None, None).unwrap();
711 let base = tip(&db);
712 create_branch(&db, "x", base).unwrap();
713 create_branch(&db, "y", base).unwrap();
714 branch_put(&db, "x", "orders", "a", j(2)).unwrap();
715 branch_put(&db, "y", "orders", "b", j(2)).unwrap();
716
717 let px = plan(&db, "x").unwrap();
718 execute(&db, &px).unwrap();
719 let py = plan(&db, "y").unwrap();
721 assert!(py.is_clean(), "disjoint documents do not conflict");
722 execute(&db, &py).unwrap();
723
724 assert_eq!(db.get("orders", "a").unwrap().data, j(2));
725 assert_eq!(db.get("orders", "b").unwrap().data, j(2));
726 assert_eq!(list_merges(&db).len(), 2);
727 assert_eq!(crate::branch::minimum_pinned_seq(&db), None);
728 }
729
730 #[test]
731 fn a_merge_cannot_reach_a_reserved_collection() {
732 let db = forked();
733 assert!(branch_put(&db, "b1", namespace::ROOTS, "x", j(1)).is_err());
735 let bad = MergePlan {
737 branch: "b1".into(),
738 base_seq: crate::branch::get_branch(&db, "b1").unwrap().base_seq,
739 into_seq: tip(&db),
740 changes: vec![PlannedChange {
741 coll: namespace::ROOTS.into(), id: "x".into(),
742 kind: ChangeKind::Add, base: None, value: Some(j(1)),
743 source_hash: String::new(),
744 }],
745 conflicts: vec![],
746 };
747 assert!(execute(&db, &bad).is_err());
748 }
749
750 #[test]
751 fn the_whole_cycle_works_on_disk() {
752 let dir = tempdir().unwrap();
753 let db = Db::open(dir.path(), None).unwrap();
754 db.put("orders", "a", j(1), vec![], None, None).unwrap();
755 create_branch(&db, "d1", tip(&db)).unwrap();
756 branch_put(&db, "d1", "orders", "a", j(2)).unwrap();
757 let p = plan(&db, "d1").unwrap();
758 let rec = execute(&db, &p).unwrap();
759 db.flush_all();
760 assert_eq!(db.get("orders", "a").unwrap().data, j(2));
761 assert_eq!(get_merge(&db, rec.merged_at_seq).unwrap().replayed, 1);
762 }
763}