1use std::collections::{HashMap, HashSet, VecDeque};
2
3use crate::entry::{Entry, GraphOp, Hash};
4
5pub struct OpLog {
11 entries: HashMap<Hash, Entry>,
13 heads: HashSet<Hash>,
15 children: HashMap<Hash, HashSet<Hash>>,
18 len: usize,
20}
21
22impl OpLog {
23 pub fn new(genesis: Entry) -> Self {
25 let hash = genesis.hash;
26 let mut entries = HashMap::new();
27 entries.insert(hash, genesis);
28 let mut heads = HashSet::new();
29 heads.insert(hash);
30 Self {
31 entries,
32 heads,
33 children: HashMap::new(),
34 len: 1,
35 }
36 }
37
38 pub fn append(&mut self, entry: Entry) -> Result<bool, OpLogError> {
45 if !entry.verify_hash() {
46 return Err(OpLogError::InvalidHash);
47 }
48
49 if self.entries.contains_key(&entry.hash) {
51 return Ok(false);
52 }
53
54 if entry.next.is_empty()
57 && !self.entries.is_empty()
58 && matches!(entry.payload, GraphOp::Checkpoint { .. })
59 {
60 self.replace_with_checkpoint(entry);
61 return Ok(true);
62 }
63
64 for parent_hash in &entry.next {
66 if !self.entries.contains_key(parent_hash) {
67 return Err(OpLogError::MissingParent(hex::encode(parent_hash)));
68 }
69 }
70
71 let hash = entry.hash;
72
73 for parent_hash in &entry.next {
75 self.heads.remove(parent_hash);
76 self.children.entry(*parent_hash).or_default().insert(hash);
77 }
78
79 self.heads.insert(hash);
81 self.entries.insert(hash, entry);
82 self.len += 1;
83
84 Ok(true)
85 }
86
87 pub fn heads(&self) -> Vec<Hash> {
89 self.heads.iter().copied().collect()
90 }
91
92 pub(crate) fn iter_entries(&self) -> impl Iterator<Item = (&Hash, &Entry)> {
95 self.entries.iter()
96 }
97
98 pub fn get(&self, hash: &Hash) -> Option<&Entry> {
99 self.entries.get(hash)
100 }
101
102 pub fn len(&self) -> usize {
104 self.len
105 }
106
107 pub fn is_empty(&self) -> bool {
109 self.len == 0
110 }
111
112 pub fn estimated_memory_bytes(&self) -> usize {
117 let mut total = 0;
118 for entry in self.entries.values() {
120 total += entry.to_bytes().len() + 32 + 64;
121 }
122 total += self.heads.len() * (32 + 16);
124 for children in self.children.values() {
126 total += 32 + 16 + children.len() * (32 + 16);
127 }
128 total
129 }
130
131 pub fn verify_integrity(&self) -> Vec<String> {
135 let mut errors = Vec::new();
136
137 for (hash, entry) in &self.entries {
139 if !entry.verify_hash() {
140 errors.push(format!(
141 "I-01 violated: entry {} has invalid hash",
142 hex::encode(hash)
143 ));
144 }
145 }
146
147 for (hash, entry) in &self.entries {
149 for parent in &entry.next {
150 if !self.entries.contains_key(parent) {
151 errors.push(format!(
152 "I-02 violated: entry {} references missing parent {}",
153 hex::encode(hash),
154 hex::encode(parent)
155 ));
156 }
157 }
158 }
159
160 let mut computed_heads = HashSet::new();
162 let mut has_successor: HashSet<Hash> = HashSet::new();
163 for entry in self.entries.values() {
164 for parent in &entry.next {
165 has_successor.insert(*parent);
166 }
167 }
168 for hash in self.entries.keys() {
169 if !has_successor.contains(hash) {
170 computed_heads.insert(*hash);
171 }
172 }
173 if computed_heads != self.heads {
174 let extra: Vec<_> = self
175 .heads
176 .difference(&computed_heads)
177 .map(hex::encode)
178 .collect();
179 let missing: Vec<_> = computed_heads
180 .difference(&self.heads)
181 .map(hex::encode)
182 .collect();
183 if !extra.is_empty() {
184 errors.push(format!(
185 "I-04 violated: spurious heads: {}",
186 extra.join(", ")
187 ));
188 }
189 if !missing.is_empty() {
190 errors.push(format!(
191 "I-04 violated: missing heads: {}",
192 missing.join(", ")
193 ));
194 }
195 }
196
197 errors
198 }
199
200 pub fn entries_since(&self, known_hash: Option<&Hash>) -> Vec<&Entry> {
208 let all_from_heads = self.reachable_from(&self.heads.iter().copied().collect::<Vec<_>>());
210
211 match known_hash {
212 None => {
213 self.topo_sort(&all_from_heads)
215 }
216 Some(kh) => {
217 let known_set = self.reachable_from(&[*kh]);
219 let delta: HashSet<Hash> = all_from_heads.difference(&known_set).copied().collect();
221 self.topo_sort(&delta)
222 }
223 }
224 }
225
226 pub fn entries_since_heads(&self, heads: &[Hash]) -> Result<Vec<&Entry>, OpLogError> {
238 for h in heads {
240 if !self.entries.contains_key(h) {
241 return Err(OpLogError::MissingParent(hex::encode(h)));
242 }
243 }
244
245 let all_from_heads = self.reachable_from(&self.heads.iter().copied().collect::<Vec<_>>());
247
248 let known_set = if heads.is_empty() {
250 HashSet::new()
251 } else {
252 self.reachable_from(heads)
253 };
254
255 let delta: HashSet<Hash> = all_from_heads.difference(&known_set).copied().collect();
257 Ok(self.topo_sort(&delta))
258 }
259
260 pub fn heads_known(&self, heads: &[Hash]) -> bool {
263 heads.iter().all(|h| self.entries.contains_key(h))
264 }
265
266 pub fn topo_sort(&self, hashes: &HashSet<Hash>) -> Vec<&Entry> {
269 let mut in_degree: HashMap<Hash, usize> = HashMap::new();
271 for &h in hashes {
272 let entry = &self.entries[&h];
273 let deg = entry.next.iter().filter(|p| hashes.contains(*p)).count();
274 in_degree.insert(h, deg);
275 }
276
277 let mut queue: VecDeque<Hash> = in_degree
278 .iter()
279 .filter(|(_, °)| deg == 0)
280 .map(|(&h, _)| h)
281 .collect();
282
283 let mut sorted_queue: Vec<Hash> = queue.drain(..).collect();
285 sorted_queue.sort_by(|a, b| {
286 let ea = &self.entries[a];
287 let eb = &self.entries[b];
288 ea.clock
289 .as_tuple()
290 .cmp(&eb.clock.as_tuple())
291 .then_with(|| a.cmp(b))
292 });
293 queue = sorted_queue.into();
294
295 let mut result = Vec::new();
296 while let Some(h) = queue.pop_front() {
297 result.push(&self.entries[&h]);
298 if let Some(ch) = self.children.get(&h) {
300 let mut ready = Vec::new();
301 for &child in ch {
302 if !hashes.contains(&child) {
303 continue;
304 }
305 if let Some(deg) = in_degree.get_mut(&child) {
306 *deg -= 1;
307 if *deg == 0 {
308 ready.push(child);
309 }
310 }
311 }
312 ready.sort_by(|a, b| {
314 let ea = &self.entries[a];
315 let eb = &self.entries[b];
316 ea.clock
317 .as_tuple()
318 .cmp(&eb.clock.as_tuple())
319 .then_with(|| a.cmp(b))
320 });
321 for r in ready {
322 queue.push_back(r);
323 }
324 }
325 }
326
327 result
328 }
329
330 pub fn entries_as_of(&self, cutoff_physical: u64, cutoff_logical: u32) -> Vec<&Entry> {
333 let cutoff = (cutoff_physical, cutoff_logical);
334 let filtered: HashSet<Hash> = self
335 .entries
336 .iter()
337 .filter(|(_, e)| e.clock.as_tuple() <= cutoff)
338 .map(|(h, _)| *h)
339 .collect();
340 self.topo_sort(&filtered)
341 }
342
343 pub fn replace_with_checkpoint(&mut self, checkpoint: Entry) {
347 self.entries.clear();
348 self.heads.clear();
349 self.children.clear();
350 let hash = checkpoint.hash;
351 self.entries.insert(hash, checkpoint);
352 self.heads.insert(hash);
353 self.len = 1;
354 }
355
356 fn reachable_from(&self, starts: &[Hash]) -> HashSet<Hash> {
359 let mut visited = HashSet::new();
360 let mut queue: VecDeque<Hash> = starts.iter().copied().collect();
361 while let Some(h) = queue.pop_front() {
362 if !visited.insert(h) {
363 continue;
364 }
365 if let Some(entry) = self.entries.get(&h) {
366 for parent in &entry.next {
367 if !visited.contains(parent) {
368 queue.push_back(*parent);
369 }
370 }
371 for r in &entry.refs {
373 if !visited.contains(r) {
374 queue.push_back(*r);
375 }
376 }
377 }
378 }
379 visited
380 }
381}
382
383#[derive(Debug, PartialEq)]
385pub enum OpLogError {
386 InvalidHash,
387 MissingParent(String),
388}
389
390impl std::fmt::Display for OpLogError {
391 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
392 match self {
393 OpLogError::InvalidHash => write!(f, "entry hash verification failed"),
394 OpLogError::MissingParent(h) => write!(f, "missing parent entry: {h}"),
395 }
396 }
397}
398
399impl std::error::Error for OpLogError {}
400
401#[cfg(test)]
402mod tests {
403 use super::*;
404 use crate::clock::LamportClock;
405 use crate::entry::GraphOp;
406 use crate::ontology::{EdgeTypeDef, NodeTypeDef, Ontology};
407 use std::collections::BTreeMap;
408
409 fn test_ontology() -> Ontology {
410 Ontology {
411 node_types: BTreeMap::from([(
412 "entity".into(),
413 NodeTypeDef {
414 description: None,
415 properties: BTreeMap::new(),
416 subtypes: None,
417 parent_type: None,
418 },
419 )]),
420 edge_types: BTreeMap::from([(
421 "LINKS".into(),
422 EdgeTypeDef {
423 description: None,
424 source_types: vec!["entity".into()],
425 target_types: vec!["entity".into()],
426 properties: BTreeMap::new(),
427 },
428 )]),
429 }
430 }
431
432 fn genesis() -> Entry {
433 Entry::new(
434 GraphOp::DefineOntology {
435 ontology: test_ontology(),
436 },
437 vec![],
438 vec![],
439 LamportClock::new("test"),
440 "test",
441 )
442 }
443
444 fn add_node_op(id: &str) -> GraphOp {
445 GraphOp::AddNode {
446 node_id: id.into(),
447 node_type: "entity".into(),
448 label: id.into(),
449 properties: BTreeMap::new(),
450 subtype: None,
451 }
452 }
453
454 fn make_entry(op: GraphOp, next: Vec<Hash>, clock_time: u64) -> Entry {
455 Entry::new(
456 op,
457 next,
458 vec![],
459 LamportClock::with_values("test", clock_time, 0),
460 "test",
461 )
462 }
463
464 #[test]
469 fn append_single_entry() {
470 let g = genesis();
471 let mut log = OpLog::new(g.clone());
472 assert_eq!(log.len(), 1);
473 assert_eq!(log.heads().len(), 1);
474 assert_eq!(log.heads()[0], g.hash);
475
476 let e1 = make_entry(add_node_op("n1"), vec![g.hash], 2);
477 assert!(log.append(e1.clone()).unwrap());
478 assert_eq!(log.len(), 2);
479 assert_eq!(log.heads().len(), 1);
480 assert_eq!(log.heads()[0], e1.hash);
481 }
482
483 #[test]
484 fn append_chain() {
485 let g = genesis();
487 let mut log = OpLog::new(g.clone());
488
489 let a = make_entry(add_node_op("a"), vec![g.hash], 2);
490 let b = make_entry(add_node_op("b"), vec![a.hash], 3);
491 let c = make_entry(add_node_op("c"), vec![b.hash], 4);
492
493 log.append(a).unwrap();
494 log.append(b).unwrap();
495 log.append(c.clone()).unwrap();
496
497 assert_eq!(log.len(), 4); assert_eq!(log.heads().len(), 1);
499 assert_eq!(log.heads()[0], c.hash);
500 }
501
502 #[test]
503 fn append_fork() {
504 let g = genesis();
506 let mut log = OpLog::new(g.clone());
507
508 let a = make_entry(add_node_op("a"), vec![g.hash], 2);
509 log.append(a.clone()).unwrap();
510
511 let b = make_entry(add_node_op("b"), vec![a.hash], 3);
512 let c = make_entry(add_node_op("c"), vec![a.hash], 3);
513 log.append(b.clone()).unwrap();
514 log.append(c.clone()).unwrap();
515
516 assert_eq!(log.len(), 4);
517 let heads = log.heads();
518 assert_eq!(heads.len(), 2);
519 assert!(heads.contains(&b.hash));
520 assert!(heads.contains(&c.hash));
521 }
522
523 #[test]
524 fn append_merge() {
525 let g = genesis();
527 let mut log = OpLog::new(g.clone());
528
529 let a = make_entry(add_node_op("a"), vec![g.hash], 2);
530 log.append(a.clone()).unwrap();
531
532 let b = make_entry(add_node_op("b"), vec![a.hash], 3);
533 let c = make_entry(add_node_op("c"), vec![a.hash], 3);
534 log.append(b.clone()).unwrap();
535 log.append(c.clone()).unwrap();
536 assert_eq!(log.heads().len(), 2);
537
538 let d = make_entry(add_node_op("d"), vec![b.hash, c.hash], 4);
540 log.append(d.clone()).unwrap();
541
542 assert_eq!(log.heads().len(), 1);
543 assert_eq!(log.heads()[0], d.hash);
544 }
545
546 #[test]
547 fn heads_updated_on_append() {
548 let g = genesis();
549 let mut log = OpLog::new(g.clone());
550 assert!(log.heads().contains(&g.hash));
551
552 let e1 = make_entry(add_node_op("n1"), vec![g.hash], 2);
553 log.append(e1.clone()).unwrap();
554 assert!(!log.heads().contains(&g.hash));
555 assert!(log.heads().contains(&e1.hash));
556 }
557
558 #[test]
559 fn entries_since_returns_delta() {
560 let g = genesis();
563 let mut log = OpLog::new(g.clone());
564
565 let a = make_entry(add_node_op("a"), vec![g.hash], 2);
566 let b = make_entry(add_node_op("b"), vec![a.hash], 3);
567 let c = make_entry(add_node_op("c"), vec![b.hash], 4);
568
569 log.append(a.clone()).unwrap();
570 log.append(b.clone()).unwrap();
571 log.append(c.clone()).unwrap();
572
573 let delta = log.entries_since(Some(&a.hash));
574 let delta_hashes: Vec<Hash> = delta.iter().map(|e| e.hash).collect();
575 assert_eq!(delta_hashes.len(), 2);
576 assert!(delta_hashes.contains(&b.hash));
577 assert!(delta_hashes.contains(&c.hash));
578 assert_eq!(delta_hashes[0], b.hash);
580 assert_eq!(delta_hashes[1], c.hash);
581 }
582
583 #[test]
584 fn entries_since_empty_returns_all() {
585 let g = genesis();
586 let mut log = OpLog::new(g.clone());
587 let a = make_entry(add_node_op("a"), vec![g.hash], 2);
588 log.append(a).unwrap();
589
590 let all = log.entries_since(None);
591 assert_eq!(all.len(), 2); }
593
594 #[test]
595 fn topological_sort_respects_causality() {
596 let g = genesis();
598 let mut log = OpLog::new(g.clone());
599
600 let a = make_entry(add_node_op("a"), vec![g.hash], 2);
601 log.append(a.clone()).unwrap();
602 let b = make_entry(add_node_op("b"), vec![a.hash], 3);
603 let c = make_entry(add_node_op("c"), vec![a.hash], 4);
604 log.append(b.clone()).unwrap();
605 log.append(c.clone()).unwrap();
606
607 let all = log.entries_since(None);
608 assert_eq!(all[0].hash, g.hash);
610 assert_eq!(all[1].hash, a.hash);
611 let last_two: HashSet<Hash> = all[2..].iter().map(|e| e.hash).collect();
613 assert!(last_two.contains(&b.hash));
614 assert!(last_two.contains(&c.hash));
615 }
616
617 #[test]
618 fn duplicate_entry_ignored() {
619 let g = genesis();
620 let mut log = OpLog::new(g.clone());
621
622 let e1 = make_entry(add_node_op("n1"), vec![g.hash], 2);
623 assert!(log.append(e1.clone()).unwrap()); assert!(!log.append(e1.clone()).unwrap()); assert_eq!(log.len(), 2); }
627
628 #[test]
629 fn entry_not_found_error() {
630 let g = genesis();
631 let log = OpLog::new(g.clone());
632 let fake_hash = [0xffu8; 32];
633 assert!(log.get(&fake_hash).is_none());
634 }
635
636 #[test]
637 fn invalid_hash_rejected() {
638 let g = genesis();
639 let mut log = OpLog::new(g.clone());
640 let mut bad = make_entry(add_node_op("n1"), vec![g.hash], 2);
641 bad.author = "tampered".into(); assert_eq!(log.append(bad), Err(OpLogError::InvalidHash));
643 }
644
645 #[test]
646 fn missing_parent_rejected() {
647 let g = genesis();
648 let mut log = OpLog::new(g.clone());
649 let fake_parent = [0xaau8; 32];
650 let bad = make_entry(add_node_op("n1"), vec![fake_parent], 2);
651 match log.append(bad) {
652 Err(OpLogError::MissingParent(_)) => {} other => panic!("expected MissingParent, got {:?}", other),
654 }
655 }
656
657 #[test]
660 fn entries_since_heads_empty_returns_all() {
661 let g = genesis();
662 let mut log = OpLog::new(g.clone());
663 let e1 = make_entry(add_node_op("n1"), vec![g.hash], 2);
664 let e2 = make_entry(add_node_op("n2"), vec![e1.hash], 3);
665 log.append(e1.clone()).unwrap();
666 log.append(e2.clone()).unwrap();
667
668 let result = log.entries_since_heads(&[]).unwrap();
669 let hashes: Vec<Hash> = result.iter().map(|e| e.hash).collect();
670 assert_eq!(hashes, vec![g.hash, e1.hash, e2.hash]);
671 }
672
673 #[test]
674 fn entries_since_heads_current_heads_returns_empty() {
675 let g = genesis();
676 let mut log = OpLog::new(g.clone());
677 let e1 = make_entry(add_node_op("n1"), vec![g.hash], 2);
678 log.append(e1.clone()).unwrap();
679
680 let result = log.entries_since_heads(&[e1.hash]).unwrap();
682 assert!(result.is_empty());
683 }
684
685 #[test]
686 fn entries_since_heads_partial_cursor_returns_delta() {
687 let g = genesis();
689 let mut log = OpLog::new(g.clone());
690 let e1 = make_entry(add_node_op("n1"), vec![g.hash], 2);
691 let e2 = make_entry(add_node_op("n2"), vec![e1.hash], 3);
692 let e3 = make_entry(add_node_op("n3"), vec![e2.hash], 4);
693 log.append(e1.clone()).unwrap();
694 log.append(e2.clone()).unwrap();
695 log.append(e3.clone()).unwrap();
696
697 let result = log.entries_since_heads(&[e1.hash]).unwrap();
698 let hashes: Vec<Hash> = result.iter().map(|e| e.hash).collect();
699 assert_eq!(hashes, vec![e2.hash, e3.hash]);
700 }
701
702 #[test]
703 fn entries_since_heads_multiple_heads_concurrent_dag() {
704 let g = genesis();
706 let mut log = OpLog::new(g.clone());
707 let e1 = make_entry(add_node_op("n1"), vec![g.hash], 2);
708 let e2 = make_entry(add_node_op("n2"), vec![e1.hash], 3);
709 let e3 = make_entry(add_node_op("n3"), vec![e1.hash], 3);
710 log.append(e1.clone()).unwrap();
711 log.append(e2.clone()).unwrap();
712 log.append(e3.clone()).unwrap();
713
714 let result = log.entries_since_heads(&[e2.hash]).unwrap();
715 let hashes: Vec<Hash> = result.iter().map(|e| e.hash).collect();
716 assert_eq!(hashes, vec![e3.hash]);
717 }
718
719 #[test]
720 fn entries_since_heads_multiple_cursor_heads() {
721 let g = genesis();
723 let mut log = OpLog::new(g.clone());
724 let e1 = make_entry(add_node_op("n1"), vec![g.hash], 2);
725 let e2 = make_entry(add_node_op("n2"), vec![e1.hash], 3);
726 let e3 = make_entry(add_node_op("n3"), vec![e1.hash], 3);
727 log.append(e1.clone()).unwrap();
728 log.append(e2.clone()).unwrap();
729 log.append(e3.clone()).unwrap();
730
731 let result = log.entries_since_heads(&[e2.hash, e3.hash]).unwrap();
732 assert!(result.is_empty());
733 }
734
735 #[test]
736 fn entries_since_heads_unknown_hash_returns_error() {
737 let g = genesis();
738 let log = OpLog::new(g.clone());
739 let fake = [0xcdu8; 32];
740 let result = log.entries_since_heads(&[fake]);
741 assert!(result.is_err());
742 }
743
744 #[test]
745 fn heads_known_true_for_valid_cursor() {
746 let g = genesis();
747 let mut log = OpLog::new(g.clone());
748 let e1 = make_entry(add_node_op("n1"), vec![g.hash], 2);
749 log.append(e1.clone()).unwrap();
750
751 assert!(log.heads_known(&[]));
752 assert!(log.heads_known(&[g.hash]));
753 assert!(log.heads_known(&[e1.hash]));
754 assert!(log.heads_known(&[g.hash, e1.hash]));
755 }
756
757 #[test]
758 fn heads_known_false_for_unknown_hash() {
759 let g = genesis();
760 let log = OpLog::new(g.clone());
761 let fake = [0xabu8; 32];
762 assert!(!log.heads_known(&[fake]));
763 assert!(!log.heads_known(&[g.hash, fake]));
764 }
765
766 #[test]
767 fn entries_since_heads_topological_order() {
768 let g = genesis();
770 let mut log = OpLog::new(g.clone());
771 let e1 = make_entry(add_node_op("n1"), vec![g.hash], 2);
772 let e2 = make_entry(add_node_op("n2"), vec![e1.hash], 3);
773 let e3 = make_entry(add_node_op("n3"), vec![e2.hash], 4);
774 log.append(e1.clone()).unwrap();
775 log.append(e2.clone()).unwrap();
776 log.append(e3.clone()).unwrap();
777
778 let result = log.entries_since_heads(&[]).unwrap();
779 let hashes: Vec<Hash> = result.iter().map(|e| e.hash).collect();
781 let pos_g = hashes.iter().position(|h| *h == g.hash).unwrap();
782 let pos_e1 = hashes.iter().position(|h| *h == e1.hash).unwrap();
783 let pos_e2 = hashes.iter().position(|h| *h == e2.hash).unwrap();
784 let pos_e3 = hashes.iter().position(|h| *h == e3.hash).unwrap();
785 assert!(pos_g < pos_e1);
786 assert!(pos_e1 < pos_e2);
787 assert!(pos_e2 < pos_e3);
788 }
789}