1use std::sync::atomic::Ordering;
45
46use anyhow::{bail, Result};
47use serde::{Deserialize, Serialize};
48use serde_json::Value;
49
50use crate::db::Db;
51use crate::namespace;
52
53pub const BRANCHES: &str = "_nedb.branches";
55
56pub const BRANCH_WRITES: &str = "_nedb.branch_writes";
62
63#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
69#[serde(tag = "state", rename_all = "snake_case")]
70pub enum BranchStatus {
71 Active,
72 Merged { at_seq: u64 },
73 Abandoned,
74}
75
76impl BranchStatus {
77 pub fn is_live(&self) -> bool {
79 matches!(self, BranchStatus::Active)
80 }
81}
82
83#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
85pub struct BranchRecord {
86 pub name: String,
87 pub base_seq: u64,
91 pub created_seq: u64,
95 pub status: BranchStatus,
96}
97
98#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
100pub struct BranchWrite {
101 pub branch: String,
102 pub created_seq: u64,
104 pub coll: String,
105 pub id: String,
106 pub value: Option<Value>,
110 pub at_seq: u64,
116 #[serde(default)]
128 pub source_hash: String,
129}
130
131pub fn branch_key(name: &str, created_seq: u64) -> String {
161 format!("{}@{}", namespace::seq_id(created_seq), name)
162}
163
164pub fn validate_branch_name(name: &str) -> Result<()> {
172 namespace::validate_name(name)
173 .map_err(|e| anyhow::anyhow!("branch name {:?} is unusable: {}", name, e))?;
174 if namespace::is_reserved(name) {
175 bail!(
176 "branch name {:?} is reserved: everything under {:?} is engine-owned",
177 name, namespace::RESERVED_PREFIX
178 );
179 }
180 if !name.is_empty() && name.chars().all(|c| c.is_ascii_digit()) {
181 bail!(
182 "branch name {:?} is purely numeric — every surface that accepts a \
183 branch also accepts a sequence number, and this name cannot be told \
184 apart from one",
185 name
186 );
187 }
188 Ok(())
189}
190
191fn read_record(db: &Db, key: &str) -> Option<BranchRecord> {
194 let n = db.get(BRANCHES, key)?;
195 serde_json::from_value(n.data).ok()
196}
197
198fn write_record(db: &Db, rec: &BranchRecord) -> Result<()> {
199 let key = branch_key(&rec.name, rec.created_seq);
200 db.put_unchecked(BRANCHES, &key, serde_json::to_value(rec)?, vec![], None, None)?;
201 Ok(())
202}
203
204pub fn create_branch(db: &Db, name: &str, base_seq: u64) -> Result<BranchRecord> {
212 validate_branch_name(name)?;
213
214 let next = db.seq.load(Ordering::SeqCst);
215 if base_seq >= next {
216 bail!(
217 "cannot fork branch {:?} from sequence {}: the database has not reached \
218 it (next sequence is {})",
219 name, base_seq, next
220 );
221 }
222 let floor = db.history_floor();
223 if base_seq < floor {
224 bail!(
225 "cannot fork branch {:?} from sequence {}: history below {} has been \
226 compacted away, so the merge base for this branch no longer exists and \
227 it could never be reconciled",
228 name, base_seq, floor
229 );
230 }
231
232 if let Some(existing) = get_branch(db, name) {
233 if existing.status.is_live() {
234 bail!(
235 "branch {:?} already exists and is active (forked from sequence {} at \
236 sequence {}); a name may be reused only after the branch holding it \
237 is merged or abandoned",
238 name, existing.base_seq, existing.created_seq
239 );
240 }
241 }
242
243 let created_seq = db.seq.load(Ordering::SeqCst);
244 let rec = BranchRecord {
245 name: name.to_string(),
246 base_seq,
247 created_seq,
248 status: BranchStatus::Active,
249 };
250 write_record(db, &rec)?;
251 Ok(rec)
252}
253
254pub fn list_all_branches(db: &Db) -> Vec<BranchRecord> {
259 let mut ids = db.list_ids_including_deleted(BRANCHES);
260 ids.sort();
261 ids.into_iter().filter_map(|id| read_record(db, &id)).collect()
262}
263
264pub fn list_branches(db: &Db) -> Vec<BranchRecord> {
266 let mut out: Vec<BranchRecord> = list_all_branches(db)
267 .into_iter()
268 .filter(|b| b.status.is_live())
269 .collect();
270 out.sort_by(|a, b| a.name.cmp(&b.name));
271 out
272}
273
274pub fn get_branch(db: &Db, name: &str) -> Option<BranchRecord> {
279 list_all_branches(db)
280 .into_iter()
281 .filter(|b| b.name == name)
282 .next_back()
283}
284
285pub fn abandon_branch(db: &Db, name: &str) -> Result<bool> {
293 let Some(mut rec) = get_branch(db, name) else { return Ok(false) };
294 if !rec.status.is_live() {
295 return Ok(false);
296 }
297 rec.status = BranchStatus::Abandoned;
298 write_record(db, &rec)?;
299 Ok(true)
300}
301
302pub(crate) fn mark_merged(db: &Db, name: &str, at_seq: u64) -> Result<()> {
304 let Some(mut rec) = get_branch(db, name) else {
305 bail!("branch {:?} does not exist", name)
306 };
307 rec.status = BranchStatus::Merged { at_seq };
308 write_record(db, &rec)
309}
310
311pub fn pinning_branches(db: &Db) -> Vec<(String, u64)> {
315 list_branches(db).into_iter().map(|b| (b.name, b.base_seq)).collect()
316}
317
318pub fn minimum_pinned_seq(db: &Db) -> Option<u64> {
326 list_branches(db).into_iter().map(|b| b.base_seq).min()
327}
328
329pub(crate) fn compaction_refusal(db: &Db, pinned: u64) -> String {
332 let names: Vec<String> = pinning_branches(db)
333 .into_iter()
334 .map(|(n, s)| format!("{:?} (base seq {})", n, s))
335 .collect();
336 format!(
337 "refusing to compact: {} live branch(es) pin history at or above sequence {} \
338 — {}. Compaction is all-or-nothing to the tip, so proceeding would discard \
339 the merge base these branches will need and leave them permanently \
340 unmergeable. Merge or abandon them first.",
341 names.len(), pinned, names.join(", ")
342 )
343}
344
345fn write_key(branch_key: &str, coll: &str, id: &str) -> String {
356 use blake2::{Blake2b512, Digest};
357 let mut h = Blake2b512::new();
358 for part in [branch_key, coll, id] {
359 h.update((part.len() as u64).to_be_bytes());
360 h.update(part.as_bytes());
361 }
362 hex::encode(&h.finalize()[..32])
363}
364
365fn live_branch(db: &Db, name: &str) -> Result<BranchRecord> {
366 let Some(rec) = get_branch(db, name) else {
367 bail!("branch {:?} does not exist", name)
368 };
369 if !rec.status.is_live() {
370 bail!(
371 "branch {:?} is {:?}, not active — a branch that has been merged or \
372 abandoned is closed history and cannot take new writes",
373 name, rec.status
374 );
375 }
376 Ok(rec)
377}
378
379fn record_branch_write(db: &Db, rec: &BranchRecord, coll: &str, id: &str, value: Option<Value>)
380 -> Result<BranchWrite>
381{
382 namespace::validate_writable(coll)?;
385 let at_seq = db.seq.load(Ordering::SeqCst);
386 let w = BranchWrite {
387 branch: rec.name.clone(),
388 created_seq: rec.created_seq,
389 coll: coll.to_string(),
390 id: id.to_string(),
391 value,
392 at_seq,
393 source_hash: String::new(),
394 };
395 let key = write_key(&branch_key(&rec.name, rec.created_seq), coll, id);
396 let first = db.put_unchecked(
417 BRANCH_WRITES, &key, serde_json::to_value(&w)?, vec![], None, None)?;
418 let w = BranchWrite { source_hash: first.hash.clone(), ..w };
419 db.put_unchecked(
420 BRANCH_WRITES, &key, serde_json::to_value(&w)?,
421 vec![first.hash], None, None)?;
425 Ok(w)
426}
427
428pub fn branch_put(db: &Db, branch: &str, coll: &str, id: &str, data: Value) -> Result<BranchWrite> {
434 let rec = live_branch(db, branch)?;
435 record_branch_write(db, &rec, coll, id, Some(data))
436}
437
438pub fn branch_delete(db: &Db, branch: &str, coll: &str, id: &str) -> Result<BranchWrite> {
444 let rec = live_branch(db, branch)?;
445 record_branch_write(db, &rec, coll, id, None)
446}
447
448pub fn branch_get(db: &Db, branch: &str, coll: &str, id: &str) -> Option<Value> {
456 let rec = get_branch(db, branch)?;
457 let key = write_key(&branch_key(&rec.name, rec.created_seq), coll, id);
458 if let Some(n) = db.get(BRANCH_WRITES, &key) {
459 let w: BranchWrite = serde_json::from_value(n.data).ok()?;
460 return w.value;
461 }
462 db.get_as_of(coll, id, rec.base_seq).map(|n| n.data)
463}
464
465pub fn branch_writes(db: &Db, branch: &str) -> Vec<BranchWrite> {
472 let Some(rec) = get_branch(db, branch) else { return Vec::new() };
473 let mut out: Vec<BranchWrite> = db
474 .list_ids_including_deleted(BRANCH_WRITES)
475 .into_iter()
476 .filter_map(|k| db.get(BRANCH_WRITES, &k))
477 .filter_map(|n| serde_json::from_value::<BranchWrite>(n.data).ok())
478 .filter(|w| w.branch == rec.name && w.created_seq == rec.created_seq)
479 .collect();
480 out.sort_by(|a, b| (&a.coll, &a.id).cmp(&(&b.coll, &b.id)));
481 out
482}
483
484#[cfg(test)]
485mod tests {
486 use super::*;
487 use tempfile::tempdir;
488
489 fn j(v: u64) -> Value { serde_json::json!({ "v": v }) }
490
491 fn seeded(n: u64) -> Db {
493 let db = Db::in_memory();
494 for i in 0..n {
495 db.put("orders", &i.to_string(), j(i), vec![], None, None).unwrap();
496 }
497 db
498 }
499
500 fn tip(db: &Db) -> u64 {
501 db.seq.load(Ordering::SeqCst).saturating_sub(1)
502 }
503
504 #[test]
505 fn create_get_list() {
506 let db = seeded(3);
507 let base = tip(&db);
508 let made = create_branch(&db, "fix-pricing", base).unwrap();
509 assert_eq!(made.name, "fix-pricing");
510 assert_eq!(made.base_seq, base);
511 assert_eq!(made.status, BranchStatus::Active);
512
513 let got = get_branch(&db, "fix-pricing").expect("branch is readable back");
514 assert_eq!(got, made);
515
516 create_branch(&db, "audit", base).unwrap();
517 let names: Vec<String> = list_branches(&db).into_iter().map(|b| b.name).collect();
518 assert_eq!(names, vec!["audit".to_string(), "fix-pricing".to_string()],
519 "active branches come back sorted by name");
520 }
521
522 #[test]
523 fn get_branch_is_none_for_a_name_never_used() {
524 let db = seeded(2);
525 assert!(get_branch(&db, "nope").is_none());
526 }
527
528 #[test]
529 fn a_base_the_database_has_not_reached_is_refused() {
530 let db = seeded(3);
531 let next = db.seq.load(Ordering::SeqCst);
532 let err = create_branch(&db, "future", next).unwrap_err().to_string();
533 assert!(err.contains("has not reached"), "{}", err);
534 assert!(create_branch(&db, "way-future", next + 1000).is_err());
535 create_branch(&db, "present", next - 1).unwrap();
537 }
538
539 #[test]
540 fn forking_from_pruned_history_is_refused() {
541 let db = seeded(4);
542 let old = 1u64;
543 db.compact().expect("no branches yet, so compaction proceeds");
549 db.set_history_floor(3).unwrap();
550 let floor = db.history_floor();
551 assert!(floor > old, "the database is in the pruned state past {}", old);
552
553 let err = create_branch(&db, "archaeology", old).unwrap_err().to_string();
554 assert!(err.contains("compacted away"), "{}", err);
555 assert!(err.contains("never be reconciled"), "{}", err);
556
557 create_branch(&db, "from-the-floor", floor).unwrap();
559 }
560
561 #[test]
562 fn unusable_names_are_refused_not_sanitised() {
563 let db = seeded(2);
564 let base = tip(&db);
565 for bad in ["", "a/b", "..", "with\0nul", " lead", "trail ", "1234", "_nedb.x", "_nedb"] {
566 assert!(
567 create_branch(&db, bad, base).is_err(),
568 "{:?} must not be usable as a branch name", bad
569 );
570 }
571 let long = "x".repeat(256);
572 assert!(create_branch(&db, &long, base).is_err(), "256 bytes is over the limit");
573 for ok in ["fix-pricing", "v2-rollout", "release-2026", "Ünicode"] {
575 create_branch(&db, ok, base).unwrap_or_else(|e| panic!("{:?} refused: {}", ok, e));
576 }
577 }
578
579 #[test]
580 fn a_live_name_cannot_be_taken_twice() {
581 let db = seeded(3);
582 let base = tip(&db);
583 create_branch(&db, "dup", base).unwrap();
584 let err = create_branch(&db, "dup", base).unwrap_err().to_string();
585 assert!(err.contains("already exists and is active"), "{}", err);
586 }
587
588 #[test]
591 fn a_closed_name_may_be_reused_without_losing_the_first_generation() {
592 let db = seeded(3);
593 let first = create_branch(&db, "recycle", 1).unwrap();
594 assert!(abandon_branch(&db, "recycle").unwrap());
595
596 db.put("orders", "x", j(99), vec![], None, None).unwrap();
597 let second = create_branch(&db, "recycle", tip(&db)).unwrap();
598 assert_ne!(first.created_seq, second.created_seq);
599
600 let gens: Vec<BranchRecord> = list_all_branches(&db)
601 .into_iter().filter(|b| b.name == "recycle").collect();
602 assert_eq!(gens.len(), 2, "both generations stay durable and auditable");
603 assert_eq!(gens[0].status, BranchStatus::Abandoned);
604 assert_eq!(gens[1].status, BranchStatus::Active);
605 assert_eq!(get_branch(&db, "recycle").unwrap(), second,
606 "the name resolves to the newest generation");
607 }
608
609 #[test]
610 fn abandoning_is_idempotent_and_honest_about_it() {
611 let db = seeded(3);
612 create_branch(&db, "gone", 1).unwrap();
613 assert!(abandon_branch(&db, "gone").unwrap(), "first abandon changes something");
614 assert!(!abandon_branch(&db, "gone").unwrap(), "second one does not");
615 assert!(!abandon_branch(&db, "never-existed").unwrap());
616 }
617
618 #[test]
619 fn a_closed_branch_refuses_new_writes() {
620 let db = seeded(3);
621 create_branch(&db, "closed", 1).unwrap();
622 abandon_branch(&db, "closed").unwrap();
623 let err = branch_put(&db, "closed", "orders", "1", j(7)).unwrap_err().to_string();
624 assert!(err.contains("not active"), "{}", err);
625 }
626
627 #[test]
630 fn nothing_pins_when_there_are_no_branches() {
631 let db = seeded(3);
632 assert_eq!(minimum_pinned_seq(&db), None);
633 }
634
635 #[test]
636 fn one_branch_pins_its_own_base() {
637 let db = seeded(5);
638 create_branch(&db, "one", 2).unwrap();
639 assert_eq!(minimum_pinned_seq(&db), Some(2));
640 }
641
642 #[test]
643 fn several_branches_pin_the_oldest_base() {
644 let db = seeded(9);
645 create_branch(&db, "a", 5).unwrap();
646 create_branch(&db, "b", 1).unwrap();
647 create_branch(&db, "c", 7).unwrap();
648 assert_eq!(minimum_pinned_seq(&db), Some(1));
649 }
650
651 #[test]
652 fn abandoned_branches_do_not_pin() {
653 let db = seeded(9);
654 create_branch(&db, "old", 1).unwrap();
655 create_branch(&db, "new", 6).unwrap();
656 assert_eq!(minimum_pinned_seq(&db), Some(1));
657 abandon_branch(&db, "old").unwrap();
658 assert_eq!(minimum_pinned_seq(&db), Some(6),
659 "an abandoned branch will never reconcile, so it needs nothing");
660 abandon_branch(&db, "new").unwrap();
661 assert_eq!(minimum_pinned_seq(&db), None);
662 }
663
664 #[test]
665 fn merged_branches_do_not_pin() {
666 let db = seeded(9);
667 create_branch(&db, "done", 2).unwrap();
668 assert_eq!(minimum_pinned_seq(&db), Some(2));
669 mark_merged(&db, "done", tip(&db)).unwrap();
670 assert_eq!(minimum_pinned_seq(&db), None);
671 }
672
673 #[test]
676 fn compaction_refuses_while_a_live_branch_pins_history() {
677 let db = seeded(6);
678 create_branch(&db, "keepme", 2).unwrap();
679 let err = db.compact().unwrap_err().to_string();
680 assert!(err.contains("refusing to compact"), "{}", err);
681 assert!(err.contains("keepme"), "the error must name the branch: {}", err);
682 assert!(err.contains("base seq 2"), "the error must name the pinned seq: {}", err);
683 assert_eq!(db.history_floor(), 0, "a refused compaction must not move the floor");
684 }
685
686 #[test]
687 fn compaction_proceeds_once_the_branch_is_abandoned() {
688 let db = seeded(6);
689 create_branch(&db, "keepme", 2).unwrap();
690 assert!(db.compact().is_err());
691 abandon_branch(&db, "keepme").unwrap();
692 let stats = db.compact().expect("nothing pins any more");
693 assert_eq!(stats.dropped_objects, 0, "v2 compaction prunes nothing");
700 assert_eq!(db.history_floor(), 0,
701 "and so it must not claim history was discarded");
702 }
703
704 #[test]
705 fn the_compaction_interlock_holds_on_disk_too() {
706 let dir = tempdir().unwrap();
707 let db = Db::open(dir.path(), None).unwrap();
708 for i in 0..4u64 {
709 db.put("orders", &i.to_string(), j(i), vec![], None, None).unwrap();
710 }
711 create_branch(&db, "ondisk", 1).unwrap();
712 assert!(db.compact().is_err(), "the refusal is a property of the engine, not of memory mode");
713 abandon_branch(&db, "ondisk").unwrap();
714 db.compact().unwrap();
715 }
716
717 #[test]
720 fn a_branch_read_falls_through_to_the_parent_at_the_fork_point() {
721 let db = seeded(0);
722 db.put("orders", "a", j(1), vec![], None, None).unwrap();
723 let base = tip(&db);
724 create_branch(&db, "b", base).unwrap();
725 assert_eq!(branch_get(&db, "b", "orders", "a"), Some(j(1)),
726 "untouched on the branch → the parent's value at the fork");
727 }
728
729 #[test]
730 fn the_fall_through_is_pinned_to_the_fork_not_to_the_parent_tip() {
731 let db = seeded(0);
732 db.put("orders", "a", j(1), vec![], None, None).unwrap();
733 let base = tip(&db);
734 create_branch(&db, "b", base).unwrap();
735 db.put("orders", "a", j(2), vec![], None, None).unwrap();
736 assert_eq!(branch_get(&db, "b", "orders", "a"), Some(j(1)),
737 "a branch whose base drifts has nothing stable to merge against");
738 }
739
740 #[test]
741 fn a_branch_write_is_invisible_to_the_destination_and_vice_versa() {
742 let db = seeded(0);
743 db.put("orders", "a", j(1), vec![], None, None).unwrap();
744 create_branch(&db, "b", tip(&db)).unwrap();
745
746 branch_put(&db, "b", "orders", "a", j(42)).unwrap();
747 assert_eq!(branch_get(&db, "b", "orders", "a"), Some(j(42)));
748 assert_eq!(db.get("orders", "a").unwrap().data, j(1),
749 "the destination must not see an unmerged branch write");
750
751 db.put("orders", "a", j(7), vec![], None, None).unwrap();
752 assert_eq!(branch_get(&db, "b", "orders", "a"), Some(j(42)),
753 "the branch must not see a destination write");
754 }
755
756 #[test]
757 fn a_branch_delete_is_a_recorded_fact_not_an_absent_record() {
758 let db = seeded(0);
759 db.put("orders", "a", j(1), vec![], None, None).unwrap();
760 create_branch(&db, "b", tip(&db)).unwrap();
761 branch_delete(&db, "b", "orders", "a").unwrap();
762 assert_eq!(branch_get(&db, "b", "orders", "a"), None);
763 let ws = branch_writes(&db, "b");
764 assert_eq!(ws.len(), 1);
765 assert_eq!(ws[0].value, None);
766 assert_eq!(ws[0].coll, "orders");
767 }
768
769 #[test]
770 fn branch_writes_are_scoped_to_one_branch_generation() {
771 let db = seeded(0);
772 db.put("orders", "a", j(1), vec![], None, None).unwrap();
773 let base = tip(&db);
774 create_branch(&db, "x", base).unwrap();
775 create_branch(&db, "y", base).unwrap();
776 branch_put(&db, "x", "orders", "a", j(10)).unwrap();
777 branch_put(&db, "y", "orders", "a", j(20)).unwrap();
778 assert_eq!(branch_get(&db, "x", "orders", "a"), Some(j(10)));
779 assert_eq!(branch_get(&db, "y", "orders", "a"), Some(j(20)));
780 assert_eq!(branch_writes(&db, "x").len(), 1);
781 assert_eq!(branch_writes(&db, "y").len(), 1);
782
783 abandon_branch(&db, "x").unwrap();
785 create_branch(&db, "x", base).unwrap();
786 assert!(branch_writes(&db, "x").is_empty());
787 assert_eq!(branch_get(&db, "x", "orders", "a"), Some(j(1)));
788 }
789
790 #[test]
791 fn a_branch_cannot_write_to_a_reserved_collection() {
792 let db = seeded(3);
793 create_branch(&db, "sneaky", 1).unwrap();
794 assert!(branch_put(&db, "sneaky", namespace::COLLECTIONS, "orders", j(1)).is_err());
795 assert!(branch_put(&db, "sneaky", BRANCHES, "x", j(1)).is_err());
796 }
797
798 #[test]
799 fn write_keys_cannot_be_forged_by_a_clever_id() {
800 let k = branch_key("b", 1);
802 assert_ne!(write_key(&k, "a", "b|c"), write_key(&k, "a|b", "c"));
803 assert_ne!(write_key(&k, "ab", "c"), write_key(&k, "a", "bc"));
804 assert_eq!(write_key(&k, "a", "b"), write_key(&k, "a", "b"), "…and it is stable");
805 }
806}