1use std::collections::{BTreeMap, HashMap, HashSet};
2
3use crate::clock::LamportClock;
4use crate::entry::{Entry, GraphOp, Hash, Value};
5use crate::ontology::Ontology;
6
7#[derive(Debug, Clone, PartialEq)]
9pub struct Node {
10 pub node_id: String,
11 pub node_type: String,
12 pub subtype: Option<String>,
13 pub label: String,
14 pub properties: BTreeMap<String, Value>,
15 pub property_clocks: HashMap<String, LamportClock>,
19 pub last_clock: LamportClock,
22 pub last_add_clock: LamportClock,
26 pub tombstoned: bool,
28}
29
30#[derive(Debug, Clone, PartialEq)]
32pub struct Edge {
33 pub edge_id: String,
34 pub edge_type: String,
35 pub source_id: String,
36 pub target_id: String,
37 pub properties: BTreeMap<String, Value>,
38 pub property_clocks: HashMap<String, LamportClock>,
40 pub last_clock: LamportClock,
41 pub last_add_clock: LamportClock,
43 pub tombstoned: bool,
44}
45
46pub struct MaterializedGraph {
57 pub nodes: HashMap<String, Node>,
59 pub edges: HashMap<String, Edge>,
61 pub outgoing: HashMap<String, HashSet<String>>,
63 pub incoming: HashMap<String, HashSet<String>>,
65 pub by_type: HashMap<String, HashSet<String>>,
67 pub ontology: Ontology,
69 pub quarantined: HashSet<Hash>,
76}
77
78impl MaterializedGraph {
79 pub fn new(ontology: Ontology) -> Self {
81 Self {
82 nodes: HashMap::new(),
83 edges: HashMap::new(),
84 outgoing: HashMap::new(),
85 incoming: HashMap::new(),
86 by_type: HashMap::new(),
87 ontology,
88 quarantined: HashSet::new(),
89 }
90 }
91
92 pub fn apply(&mut self, entry: &Entry) {
99 self.apply_entry(entry, false)
100 }
101
102 fn apply_entry(&mut self, entry: &Entry, trusted: bool) {
110 match &entry.payload {
111 GraphOp::Checkpoint { ops, op_clocks, .. } => {
112 for (i, op) in ops.iter().enumerate() {
115 if let GraphOp::DefineOntology { ontology } = op {
122 self.ontology = ontology.clone();
123 continue;
124 }
125 let clock = if i < op_clocks.len() {
126 LamportClock::with_values(&entry.author, op_clocks[i].0, op_clocks[i].1)
127 } else {
128 entry.clock.clone() };
130 let synthetic = Entry::new(op.clone(), vec![], vec![], clock, &entry.author);
131 self.apply_entry(&synthetic, true);
132 }
133 }
134 GraphOp::DefineOntology { .. } => {
135 }
137 GraphOp::ExtendOntology { extension } => {
138 if let Err(_e) = self.ontology.merge_extension(extension) {
139 if !trusted {
140 self.quarantined.insert(entry.hash);
141 }
142 }
143 }
144 GraphOp::AddNode {
145 node_id,
146 node_type,
147 subtype,
148 label,
149 properties,
150 } => {
151 if !trusted {
153 if let Err(_e) =
154 self.ontology
155 .validate_node(node_type, subtype.as_deref(), properties)
156 {
157 self.quarantined.insert(entry.hash);
158 return;
159 }
160 }
161 self.apply_add_node(
162 node_id,
163 node_type,
164 subtype.as_deref(),
165 label,
166 properties,
167 &entry.clock,
168 );
169 }
170 GraphOp::AddEdge {
171 edge_id,
172 edge_type,
173 source_id,
174 target_id,
175 properties,
176 } => {
177 if !trusted {
178 if !self.ontology.edge_types.contains_key(edge_type.as_str()) {
180 self.quarantined.insert(entry.hash);
181 return;
182 }
183 if let (Some(src), Some(tgt)) = (
187 self.nodes.get(source_id.as_str()),
188 self.nodes.get(target_id.as_str()),
189 ) {
190 if self
191 .ontology
192 .validate_edge(edge_type, &src.node_type, &tgt.node_type, properties)
193 .is_err()
194 {
195 self.quarantined.insert(entry.hash);
196 return;
197 }
198 }
199 }
200 self.apply_add_edge(
201 edge_id,
202 edge_type,
203 source_id,
204 target_id,
205 properties,
206 &entry.clock,
207 );
208 }
209 GraphOp::UpdateProperty {
210 entity_id,
211 key,
212 value,
213 } => {
214 self.apply_update_property(entity_id, key, value, &entry.clock);
215 }
216 GraphOp::RemoveNode { node_id } => {
217 self.apply_remove_node(node_id, &entry.clock);
218 }
219 GraphOp::RemoveEdge { edge_id } => {
220 self.apply_remove_edge(edge_id, &entry.clock);
221 }
222 GraphOp::DefineLens { .. } => {
223 }
225 }
226 }
227
228 pub fn apply_all(&mut self, entries: &[&Entry]) {
230 for entry in entries {
231 self.apply(entry);
232 }
233 }
234
235 pub fn rebuild(&mut self, entries: &[&Entry]) {
237 self.nodes.clear();
238 self.edges.clear();
239 self.outgoing.clear();
240 self.incoming.clear();
241 self.by_type.clear();
242 self.quarantined.clear();
243 self.apply_all(entries);
244 }
245
246 pub fn get_node(&self, node_id: &str) -> Option<&Node> {
250 self.nodes.get(node_id).filter(|n| !n.tombstoned)
251 }
252
253 pub fn get_edge(&self, edge_id: &str) -> Option<&Edge> {
255 self.edges.get(edge_id).filter(|e| !e.tombstoned)
256 }
257
258 pub fn nodes_by_type(&self, node_type: &str) -> Vec<&Node> {
261 let mut types = vec![node_type.to_string()];
262 types.extend(
263 self.ontology
264 .descendants(node_type)
265 .into_iter()
266 .map(|s| s.to_string()),
267 );
268 types
269 .iter()
270 .flat_map(|t| self.by_type.get(t.as_str()))
271 .flatten()
272 .filter_map(|id| self.get_node(id))
273 .collect()
274 }
275
276 pub fn nodes_by_subtype(&self, subtype: &str) -> Vec<&Node> {
278 self.nodes
279 .values()
280 .filter(|n| !n.tombstoned && n.subtype.as_deref() == Some(subtype))
281 .collect()
282 }
283
284 pub fn nodes_by_property(&self, key: &str, value: &Value) -> Vec<&Node> {
286 self.nodes
287 .values()
288 .filter(|n| !n.tombstoned && n.properties.get(key) == Some(value))
289 .collect()
290 }
291
292 pub fn outgoing_edges(&self, node_id: &str) -> Vec<&Edge> {
294 match self.outgoing.get(node_id) {
295 Some(edge_ids) => edge_ids
296 .iter()
297 .filter_map(|eid| self.get_edge(eid))
298 .filter(|e| self.is_node_live(&e.target_id))
299 .collect(),
300 None => vec![],
301 }
302 }
303
304 pub fn incoming_edges(&self, node_id: &str) -> Vec<&Edge> {
306 match self.incoming.get(node_id) {
307 Some(edge_ids) => edge_ids
308 .iter()
309 .filter_map(|eid| self.get_edge(eid))
310 .filter(|e| self.is_node_live(&e.source_id))
311 .collect(),
312 None => vec![],
313 }
314 }
315
316 pub fn estimated_memory_bytes(&self) -> usize {
321 let mut total = 0;
322 for node in self.nodes.values() {
324 total += node.node_id.len() + node.node_type.len() + node.label.len();
325 total += node.subtype.as_ref().map_or(0, |s| s.len());
326 for (k, v) in &node.properties {
328 total += k.len() + std::mem::size_of_val(v) + 48; }
330 total += 128; }
332 for edge in self.edges.values() {
334 total += edge.edge_id.len() + edge.edge_type.len();
335 total += edge.source_id.len() + edge.target_id.len();
336 for (k, v) in &edge.properties {
337 total += k.len() + std::mem::size_of_val(v) + 48;
338 }
339 total += 128;
340 }
341 for (k, set) in &self.outgoing {
343 total += k.len() + set.len() * 32;
344 }
345 for (k, set) in &self.incoming {
346 total += k.len() + set.len() * 32;
347 }
348 for (k, set) in &self.by_type {
350 total += k.len() + set.len() * 32;
351 }
352 total += self.quarantined.len() * 48;
354 total
355 }
356
357 pub fn all_nodes(&self) -> Vec<&Node> {
359 self.nodes.values().filter(|n| !n.tombstoned).collect()
360 }
361
362 pub fn all_edges(&self) -> Vec<&Edge> {
364 self.edges
365 .values()
366 .filter(|e| {
367 !e.tombstoned && self.is_node_live(&e.source_id) && self.is_node_live(&e.target_id)
368 })
369 .collect()
370 }
371
372 pub fn neighbors(&self, node_id: &str) -> Vec<&str> {
374 self.outgoing_edges(node_id)
375 .iter()
376 .map(|e| e.target_id.as_str())
377 .collect()
378 }
379
380 pub fn reverse_neighbors(&self, node_id: &str) -> Vec<&str> {
382 self.incoming_edges(node_id)
383 .iter()
384 .map(|e| e.source_id.as_str())
385 .collect()
386 }
387
388 fn apply_add_node(
391 &mut self,
392 node_id: &str,
393 node_type: &str,
394 subtype: Option<&str>,
395 label: &str,
396 properties: &BTreeMap<String, Value>,
397 clock: &LamportClock,
398 ) {
399 if let Some(existing) = self.nodes.get_mut(node_id) {
400 existing.tombstoned = false;
402 if clock_wins(clock, &existing.last_add_clock) {
404 existing.last_add_clock = clock.clone();
405 }
406 if clock_wins(clock, &existing.last_clock) {
408 existing.label = label.to_string();
409 existing.subtype = subtype.map(|s| s.to_string());
410 existing.last_clock = clock.clone();
411 }
412 merge_properties_lww(
413 &mut existing.properties,
414 &mut existing.property_clocks,
415 properties,
416 clock,
417 );
418 } else {
419 let property_clocks: HashMap<String, LamportClock> = properties
420 .keys()
421 .map(|k| (k.clone(), clock.clone()))
422 .collect();
423 let node = Node {
424 node_id: node_id.to_string(),
425 node_type: node_type.to_string(),
426 subtype: subtype.map(|s| s.to_string()),
427 label: label.to_string(),
428 properties: properties.clone(),
429 property_clocks,
430 last_clock: clock.clone(),
431 last_add_clock: clock.clone(),
432 tombstoned: false,
433 };
434 self.by_type
435 .entry(node_type.to_string())
436 .or_default()
437 .insert(node_id.to_string());
438 self.nodes.insert(node_id.to_string(), node);
439 }
440 }
441
442 fn apply_add_edge(
443 &mut self,
444 edge_id: &str,
445 edge_type: &str,
446 source_id: &str,
447 target_id: &str,
448 properties: &BTreeMap<String, Value>,
449 clock: &LamportClock,
450 ) {
451 if let Some(existing) = self.edges.get_mut(edge_id) {
452 existing.tombstoned = false;
454 if clock_wins(clock, &existing.last_add_clock) {
455 existing.last_add_clock = clock.clone();
456 }
457 if clock_wins(clock, &existing.last_clock) {
458 existing.last_clock = clock.clone();
459 }
460 merge_properties_lww(
461 &mut existing.properties,
462 &mut existing.property_clocks,
463 properties,
464 clock,
465 );
466 } else {
467 let property_clocks: HashMap<String, LamportClock> = properties
468 .keys()
469 .map(|k| (k.clone(), clock.clone()))
470 .collect();
471 let edge = Edge {
472 edge_id: edge_id.to_string(),
473 edge_type: edge_type.to_string(),
474 source_id: source_id.to_string(),
475 target_id: target_id.to_string(),
476 properties: properties.clone(),
477 property_clocks,
478 last_clock: clock.clone(),
479 last_add_clock: clock.clone(),
480 tombstoned: false,
481 };
482 self.outgoing
483 .entry(source_id.to_string())
484 .or_default()
485 .insert(edge_id.to_string());
486 self.incoming
487 .entry(target_id.to_string())
488 .or_default()
489 .insert(edge_id.to_string());
490 self.edges.insert(edge_id.to_string(), edge);
491 }
492 }
493
494 fn apply_update_property(
495 &mut self,
496 entity_id: &str,
497 key: &str,
498 value: &Value,
499 clock: &LamportClock,
500 ) {
501 if let Some(node) = self.nodes.get_mut(entity_id) {
504 let dominated = node
505 .property_clocks
506 .get(key)
507 .map(|c| clock_wins(clock, c))
508 .unwrap_or(true);
509 if dominated {
510 node.properties.insert(key.to_string(), value.clone());
511 node.property_clocks.insert(key.to_string(), clock.clone());
512 }
513 if clock_wins(clock, &node.last_clock) {
515 node.last_clock = clock.clone();
516 }
517 } else if let Some(edge) = self.edges.get_mut(entity_id) {
518 let dominated = edge
519 .property_clocks
520 .get(key)
521 .map(|c| clock_wins(clock, c))
522 .unwrap_or(true);
523 if dominated {
524 edge.properties.insert(key.to_string(), value.clone());
525 edge.property_clocks.insert(key.to_string(), clock.clone());
526 }
527 if clock_wins(clock, &edge.last_clock) {
528 edge.last_clock = clock.clone();
529 }
530 }
531 }
533
534 fn apply_remove_node(&mut self, node_id: &str, clock: &LamportClock) {
535 if let Some(node) = self.nodes.get_mut(node_id) {
536 if clock_wins(clock, &node.last_add_clock) {
540 node.tombstoned = true;
541 node.last_clock = clock.clone();
542 }
543 }
544 }
547
548 fn apply_remove_edge(&mut self, edge_id: &str, clock: &LamportClock) {
549 if let Some(edge) = self.edges.get_mut(edge_id) {
550 if clock_wins(clock, &edge.last_add_clock) {
552 edge.tombstoned = true;
553 edge.last_clock = clock.clone();
554 }
555 }
556 }
557
558 fn is_node_live(&self, node_id: &str) -> bool {
559 self.nodes
560 .get(node_id)
561 .map(|n| !n.tombstoned)
562 .unwrap_or(false)
563 }
564}
565
566fn merge_properties_lww(
569 existing_props: &mut BTreeMap<String, Value>,
570 existing_clocks: &mut HashMap<String, LamportClock>,
571 new_props: &BTreeMap<String, Value>,
572 clock: &LamportClock,
573) {
574 for (k, v) in new_props {
575 let dominated = existing_clocks
576 .get(k)
577 .map(|c| clock_wins(clock, c))
578 .unwrap_or(true);
579 if dominated {
580 existing_props.insert(k.clone(), v.clone());
581 existing_clocks.insert(k.clone(), clock.clone());
582 }
583 }
584}
585
586fn clock_wins(new_clock: &LamportClock, existing_clock: &LamportClock) -> bool {
589 new_clock.cmp_order(existing_clock) == std::cmp::Ordering::Greater
590}
591
592#[cfg(test)]
593mod tests {
594 use super::*;
595 use crate::entry::Entry;
596 use crate::ontology::{EdgeTypeDef, NodeTypeDef};
597
598 fn test_ontology() -> Ontology {
599 Ontology {
600 node_types: BTreeMap::from([
601 (
602 "entity".into(),
603 NodeTypeDef {
604 description: None,
605 properties: BTreeMap::new(),
606 subtypes: None,
607 parent_type: None,
608 },
609 ),
610 (
611 "signal".into(),
612 NodeTypeDef {
613 description: None,
614 properties: BTreeMap::new(),
615 subtypes: None,
616 parent_type: None,
617 },
618 ),
619 ]),
620 edge_types: BTreeMap::from([
621 (
622 "RUNS_ON".into(),
623 EdgeTypeDef {
624 description: None,
625 source_types: vec!["entity".into()],
626 target_types: vec!["entity".into()],
627 properties: BTreeMap::new(),
628 },
629 ),
630 (
631 "OBSERVES".into(),
632 EdgeTypeDef {
633 description: None,
634 source_types: vec!["signal".into()],
635 target_types: vec!["entity".into()],
636 properties: BTreeMap::new(),
637 },
638 ),
639 ]),
640 }
641 }
642
643 fn make_entry(op: GraphOp, clock_time: u64, author: &str) -> Entry {
644 Entry::new(
645 op,
646 vec![],
647 vec![],
648 LamportClock::with_values(author, clock_time, 0),
649 author,
650 )
651 }
652
653 #[test]
656 fn add_node_appears_in_query() {
657 let mut g = MaterializedGraph::new(test_ontology());
658 let entry = make_entry(
659 GraphOp::AddNode {
660 node_id: "server-1".into(),
661 node_type: "entity".into(),
662 label: "Server 1".into(),
663 properties: BTreeMap::from([("ip".into(), Value::String("10.0.0.1".into()))]),
664 subtype: None,
665 },
666 1,
667 "inst-a",
668 );
669 g.apply(&entry);
670
671 let node = g.get_node("server-1").unwrap();
672 assert_eq!(node.node_type, "entity");
673 assert_eq!(node.label, "Server 1");
674 assert_eq!(
675 node.properties.get("ip"),
676 Some(&Value::String("10.0.0.1".into()))
677 );
678 }
679
680 #[test]
681 fn add_edge_creates_adjacency() {
682 let mut g = MaterializedGraph::new(test_ontology());
683 g.apply(&make_entry(
684 GraphOp::AddNode {
685 node_id: "svc".into(),
686 node_type: "entity".into(),
687 label: "svc".into(),
688 properties: BTreeMap::new(),
689 subtype: None,
690 },
691 1,
692 "inst-a",
693 ));
694 g.apply(&make_entry(
695 GraphOp::AddNode {
696 node_id: "srv".into(),
697 node_type: "entity".into(),
698 label: "srv".into(),
699 properties: BTreeMap::new(),
700 subtype: None,
701 },
702 2,
703 "inst-a",
704 ));
705 g.apply(&make_entry(
706 GraphOp::AddEdge {
707 edge_id: "e1".into(),
708 edge_type: "RUNS_ON".into(),
709 source_id: "svc".into(),
710 target_id: "srv".into(),
711 properties: BTreeMap::new(),
712 },
713 3,
714 "inst-a",
715 ));
716
717 let out = g.outgoing_edges("svc");
719 assert_eq!(out.len(), 1);
720 assert_eq!(out[0].target_id, "srv");
721
722 let inc = g.incoming_edges("srv");
723 assert_eq!(inc.len(), 1);
724 assert_eq!(inc[0].source_id, "svc");
725
726 assert_eq!(g.neighbors("svc"), vec!["srv"]);
727 }
728
729 #[test]
730 fn update_property_reflected() {
731 let mut g = MaterializedGraph::new(test_ontology());
732 g.apply(&make_entry(
733 GraphOp::AddNode {
734 node_id: "s1".into(),
735 node_type: "entity".into(),
736 label: "s1".into(),
737 properties: BTreeMap::new(),
738 subtype: None,
739 },
740 1,
741 "inst-a",
742 ));
743 g.apply(&make_entry(
744 GraphOp::UpdateProperty {
745 entity_id: "s1".into(),
746 key: "cpu".into(),
747 value: Value::Float(85.5),
748 },
749 2,
750 "inst-a",
751 ));
752
753 let node = g.get_node("s1").unwrap();
754 assert_eq!(node.properties.get("cpu"), Some(&Value::Float(85.5)));
755 }
756
757 #[test]
758 fn remove_node_cascades_edges() {
759 let mut g = MaterializedGraph::new(test_ontology());
760 g.apply(&make_entry(
761 GraphOp::AddNode {
762 node_id: "a".into(),
763 node_type: "entity".into(),
764 label: "a".into(),
765 properties: BTreeMap::new(),
766 subtype: None,
767 },
768 1,
769 "inst-a",
770 ));
771 g.apply(&make_entry(
772 GraphOp::AddNode {
773 node_id: "b".into(),
774 node_type: "entity".into(),
775 label: "b".into(),
776 properties: BTreeMap::new(),
777 subtype: None,
778 },
779 2,
780 "inst-a",
781 ));
782 g.apply(&make_entry(
783 GraphOp::AddEdge {
784 edge_id: "e1".into(),
785 edge_type: "RUNS_ON".into(),
786 source_id: "a".into(),
787 target_id: "b".into(),
788 properties: BTreeMap::new(),
789 },
790 3,
791 "inst-a",
792 ));
793 assert_eq!(g.all_edges().len(), 1);
794
795 g.apply(&make_entry(
797 GraphOp::RemoveNode {
798 node_id: "b".into(),
799 },
800 4,
801 "inst-a",
802 ));
803 assert!(g.get_node("b").is_none());
804 assert_eq!(g.all_edges().len(), 0);
806 assert_eq!(g.outgoing_edges("a").len(), 0);
808 }
809
810 #[test]
811 fn remove_edge_preserves_nodes() {
812 let mut g = MaterializedGraph::new(test_ontology());
813 g.apply(&make_entry(
814 GraphOp::AddNode {
815 node_id: "a".into(),
816 node_type: "entity".into(),
817 label: "a".into(),
818 properties: BTreeMap::new(),
819 subtype: None,
820 },
821 1,
822 "inst-a",
823 ));
824 g.apply(&make_entry(
825 GraphOp::AddNode {
826 node_id: "b".into(),
827 node_type: "entity".into(),
828 label: "b".into(),
829 properties: BTreeMap::new(),
830 subtype: None,
831 },
832 2,
833 "inst-a",
834 ));
835 g.apply(&make_entry(
836 GraphOp::AddEdge {
837 edge_id: "e1".into(),
838 edge_type: "RUNS_ON".into(),
839 source_id: "a".into(),
840 target_id: "b".into(),
841 properties: BTreeMap::new(),
842 },
843 3,
844 "inst-a",
845 ));
846 g.apply(&make_entry(
847 GraphOp::RemoveEdge {
848 edge_id: "e1".into(),
849 },
850 4,
851 "inst-a",
852 ));
853
854 assert!(g.get_node("a").is_some());
856 assert!(g.get_node("b").is_some());
857 assert!(g.get_edge("e1").is_none());
859 assert_eq!(g.all_edges().len(), 0);
860 }
861
862 #[test]
863 fn query_by_type_filters() {
864 let mut g = MaterializedGraph::new(test_ontology());
865 g.apply(&make_entry(
866 GraphOp::AddNode {
867 node_id: "s1".into(),
868 node_type: "entity".into(),
869 label: "s1".into(),
870 properties: BTreeMap::new(),
871 subtype: None,
872 },
873 1,
874 "inst-a",
875 ));
876 g.apply(&make_entry(
877 GraphOp::AddNode {
878 node_id: "s2".into(),
879 node_type: "entity".into(),
880 label: "s2".into(),
881 properties: BTreeMap::new(),
882 subtype: None,
883 },
884 2,
885 "inst-a",
886 ));
887 g.apply(&make_entry(
888 GraphOp::AddNode {
889 node_id: "alert".into(),
890 node_type: "signal".into(),
891 label: "alert".into(),
892 properties: BTreeMap::new(),
893 subtype: None,
894 },
895 3,
896 "inst-a",
897 ));
898
899 let entities = g.nodes_by_type("entity");
900 assert_eq!(entities.len(), 2);
901 let signals = g.nodes_by_type("signal");
902 assert_eq!(signals.len(), 1);
903 assert_eq!(signals[0].node_id, "alert");
904 }
905
906 #[test]
907 fn query_by_property_filters() {
908 let mut g = MaterializedGraph::new(test_ontology());
909 g.apply(&make_entry(
910 GraphOp::AddNode {
911 node_id: "s1".into(),
912 node_type: "entity".into(),
913 label: "s1".into(),
914 properties: BTreeMap::from([("status".into(), Value::String("alive".into()))]),
915 subtype: None,
916 },
917 1,
918 "inst-a",
919 ));
920 g.apply(&make_entry(
921 GraphOp::AddNode {
922 node_id: "s2".into(),
923 node_type: "entity".into(),
924 label: "s2".into(),
925 properties: BTreeMap::from([("status".into(), Value::String("dead".into()))]),
926 subtype: None,
927 },
928 2,
929 "inst-a",
930 ));
931
932 let alive = g.nodes_by_property("status", &Value::String("alive".into()));
933 assert_eq!(alive.len(), 1);
934 assert_eq!(alive[0].node_id, "s1");
935 }
936
937 #[test]
938 fn materialization_from_empty() {
939 let mut g1 = MaterializedGraph::new(test_ontology());
941 let entries = vec![
942 make_entry(
943 GraphOp::DefineOntology {
944 ontology: test_ontology(),
945 },
946 0,
947 "inst-a",
948 ),
949 make_entry(
950 GraphOp::AddNode {
951 node_id: "a".into(),
952 node_type: "entity".into(),
953 label: "a".into(),
954 properties: BTreeMap::new(),
955 subtype: None,
956 },
957 1,
958 "inst-a",
959 ),
960 make_entry(
961 GraphOp::AddNode {
962 node_id: "b".into(),
963 node_type: "entity".into(),
964 label: "b".into(),
965 properties: BTreeMap::new(),
966 subtype: None,
967 },
968 2,
969 "inst-a",
970 ),
971 make_entry(
972 GraphOp::AddEdge {
973 edge_id: "e1".into(),
974 edge_type: "RUNS_ON".into(),
975 source_id: "a".into(),
976 target_id: "b".into(),
977 properties: BTreeMap::new(),
978 },
979 3,
980 "inst-a",
981 ),
982 ];
983 for e in &entries {
984 g1.apply(e);
985 }
986
987 let mut g2 = MaterializedGraph::new(test_ontology());
989 let refs: Vec<&Entry> = entries.iter().collect();
990 g2.rebuild(&refs);
991
992 assert_eq!(g1.all_nodes().len(), g2.all_nodes().len());
994 assert_eq!(g1.all_edges().len(), g2.all_edges().len());
995 for node in g1.all_nodes() {
996 let n2 = g2.get_node(&node.node_id).unwrap();
997 assert_eq!(node.node_type, n2.node_type);
998 assert_eq!(node.properties, n2.properties);
999 }
1000 }
1001
1002 #[test]
1003 fn incremental_equals_full() {
1004 let entries = vec![
1005 make_entry(
1006 GraphOp::DefineOntology {
1007 ontology: test_ontology(),
1008 },
1009 0,
1010 "inst-a",
1011 ),
1012 make_entry(
1013 GraphOp::AddNode {
1014 node_id: "a".into(),
1015 node_type: "entity".into(),
1016 label: "a".into(),
1017 properties: BTreeMap::from([("x".into(), Value::Int(1))]),
1018 subtype: None,
1019 },
1020 1,
1021 "inst-a",
1022 ),
1023 make_entry(
1024 GraphOp::UpdateProperty {
1025 entity_id: "a".into(),
1026 key: "x".into(),
1027 value: Value::Int(2),
1028 },
1029 2,
1030 "inst-a",
1031 ),
1032 make_entry(
1033 GraphOp::AddNode {
1034 node_id: "b".into(),
1035 node_type: "entity".into(),
1036 label: "b".into(),
1037 properties: BTreeMap::new(),
1038 subtype: None,
1039 },
1040 3,
1041 "inst-a",
1042 ),
1043 make_entry(
1044 GraphOp::AddEdge {
1045 edge_id: "e1".into(),
1046 edge_type: "RUNS_ON".into(),
1047 source_id: "a".into(),
1048 target_id: "b".into(),
1049 properties: BTreeMap::new(),
1050 },
1051 4,
1052 "inst-a",
1053 ),
1054 make_entry(
1055 GraphOp::RemoveEdge {
1056 edge_id: "e1".into(),
1057 },
1058 5,
1059 "inst-a",
1060 ),
1061 ];
1062
1063 let mut g_inc = MaterializedGraph::new(test_ontology());
1065 for e in &entries {
1066 g_inc.apply(e);
1067 }
1068
1069 let mut g_full = MaterializedGraph::new(test_ontology());
1071 let refs: Vec<&Entry> = entries.iter().collect();
1072 g_full.rebuild(&refs);
1073
1074 assert_eq!(
1076 g_inc.get_node("a").unwrap().properties.get("x"),
1077 Some(&Value::Int(2))
1078 );
1079 assert_eq!(
1080 g_full.get_node("a").unwrap().properties.get("x"),
1081 Some(&Value::Int(2))
1082 );
1083 assert_eq!(g_inc.all_edges().len(), 0);
1085 assert_eq!(g_full.all_edges().len(), 0);
1086 }
1087
1088 #[test]
1089 fn lww_concurrent_property_update() {
1090 let mut g = MaterializedGraph::new(test_ontology());
1092 g.apply(&make_entry(
1093 GraphOp::AddNode {
1094 node_id: "s1".into(),
1095 node_type: "entity".into(),
1096 label: "s1".into(),
1097 properties: BTreeMap::new(),
1098 subtype: None,
1099 },
1100 1,
1101 "inst-a",
1102 ));
1103 g.apply(&make_entry(
1105 GraphOp::UpdateProperty {
1106 entity_id: "s1".into(),
1107 key: "status".into(),
1108 value: Value::String("alive".into()),
1109 },
1110 2,
1111 "inst-a",
1112 ));
1113 g.apply(&make_entry(
1115 GraphOp::UpdateProperty {
1116 entity_id: "s1".into(),
1117 key: "status".into(),
1118 value: Value::String("dead".into()),
1119 },
1120 3,
1121 "inst-b",
1122 ));
1123 assert_eq!(
1124 g.get_node("s1").unwrap().properties.get("status"),
1125 Some(&Value::String("dead".into()))
1126 );
1127 }
1128
1129 #[test]
1130 fn lww_tiebreak_by_instance_id() {
1131 let mut g = MaterializedGraph::new(test_ontology());
1133 g.apply(&make_entry(
1134 GraphOp::AddNode {
1135 node_id: "s1".into(),
1136 node_type: "entity".into(),
1137 label: "s1".into(),
1138 properties: BTreeMap::new(),
1139 subtype: None,
1140 },
1141 1,
1142 "inst-a",
1143 ));
1144 g.apply(&make_entry(
1146 GraphOp::UpdateProperty {
1147 entity_id: "s1".into(),
1148 key: "x".into(),
1149 value: Value::Int(1),
1150 },
1151 5,
1152 "inst-a",
1153 ));
1154 g.apply(&make_entry(
1155 GraphOp::UpdateProperty {
1156 entity_id: "s1".into(),
1157 key: "x".into(),
1158 value: Value::Int(2),
1159 },
1160 5,
1161 "inst-b",
1162 ));
1163 assert_eq!(
1165 g.get_node("s1").unwrap().properties.get("x"),
1166 Some(&Value::Int(1))
1167 );
1168 }
1169
1170 #[test]
1171 fn lww_per_property_concurrent_different_keys() {
1172 let mut g = MaterializedGraph::new(test_ontology());
1176 g.apply(&make_entry(
1177 GraphOp::AddNode {
1178 node_id: "s1".into(),
1179 node_type: "entity".into(),
1180 label: "s1".into(),
1181 properties: BTreeMap::from([
1182 ("x".into(), Value::Int(0)),
1183 ("y".into(), Value::Int(0)),
1184 ]),
1185 subtype: None,
1186 },
1187 1,
1188 "inst-a",
1189 ));
1190 g.apply(&make_entry(
1192 GraphOp::UpdateProperty {
1193 entity_id: "s1".into(),
1194 key: "x".into(),
1195 value: Value::Int(42),
1196 },
1197 3,
1198 "inst-a",
1199 ));
1200 g.apply(&make_entry(
1202 GraphOp::UpdateProperty {
1203 entity_id: "s1".into(),
1204 key: "y".into(),
1205 value: Value::Int(99),
1206 },
1207 3,
1208 "inst-b",
1209 ));
1210
1211 let node = g.get_node("s1").unwrap();
1212 assert_eq!(
1214 node.properties.get("x"),
1215 Some(&Value::Int(42)),
1216 "update to 'x' must not be rejected by concurrent update to 'y'"
1217 );
1218 assert_eq!(
1219 node.properties.get("y"),
1220 Some(&Value::Int(99)),
1221 "update to 'y' must not be rejected by concurrent update to 'x'"
1222 );
1223 }
1224
1225 #[test]
1226 fn lww_per_property_order_independent() {
1227 let mut g = MaterializedGraph::new(test_ontology());
1229 g.apply(&make_entry(
1230 GraphOp::AddNode {
1231 node_id: "s1".into(),
1232 node_type: "entity".into(),
1233 label: "s1".into(),
1234 properties: BTreeMap::from([
1235 ("x".into(), Value::Int(0)),
1236 ("y".into(), Value::Int(0)),
1237 ]),
1238 subtype: None,
1239 },
1240 1,
1241 "inst-a",
1242 ));
1243 g.apply(&make_entry(
1245 GraphOp::UpdateProperty {
1246 entity_id: "s1".into(),
1247 key: "y".into(),
1248 value: Value::Int(99),
1249 },
1250 3,
1251 "inst-b",
1252 ));
1253 g.apply(&make_entry(
1254 GraphOp::UpdateProperty {
1255 entity_id: "s1".into(),
1256 key: "x".into(),
1257 value: Value::Int(42),
1258 },
1259 3,
1260 "inst-a",
1261 ));
1262
1263 let node = g.get_node("s1").unwrap();
1264 assert_eq!(node.properties.get("x"), Some(&Value::Int(42)));
1265 assert_eq!(node.properties.get("y"), Some(&Value::Int(99)));
1266 }
1267
1268 #[test]
1269 fn add_wins_over_remove() {
1270 let mut g = MaterializedGraph::new(test_ontology());
1272 g.apply(&make_entry(
1273 GraphOp::AddNode {
1274 node_id: "s1".into(),
1275 node_type: "entity".into(),
1276 label: "s1".into(),
1277 properties: BTreeMap::new(),
1278 subtype: None,
1279 },
1280 1,
1281 "inst-a",
1282 ));
1283 g.apply(&make_entry(
1285 GraphOp::RemoveNode {
1286 node_id: "s1".into(),
1287 },
1288 2,
1289 "inst-a",
1290 ));
1291 assert!(g.get_node("s1").is_none());
1292
1293 g.apply(&make_entry(
1295 GraphOp::AddNode {
1296 node_id: "s1".into(),
1297 node_type: "entity".into(),
1298 label: "s1 v2".into(),
1299 properties: BTreeMap::new(),
1300 subtype: None,
1301 },
1302 3,
1303 "inst-b",
1304 ));
1305 let node = g.get_node("s1").unwrap();
1306 assert_eq!(node.label, "s1 v2");
1307 assert!(!node.tombstoned);
1308 }
1309
1310 #[test]
1314 fn checkpoint_replay_applies_inner_define_ontology() {
1315 let mut base = test_ontology();
1317 base.node_types.remove("signal");
1318 base.edge_types.remove("OBSERVES");
1319 let mut g = MaterializedGraph::new(base);
1320
1321 let checkpoint = make_entry(
1322 GraphOp::Checkpoint {
1323 ops: vec![
1324 GraphOp::DefineOntology {
1325 ontology: test_ontology(), },
1327 GraphOp::AddNode {
1328 node_id: "n1".into(),
1329 node_type: "entity".into(),
1330 label: "base-typed".into(),
1331 properties: BTreeMap::new(),
1332 subtype: None,
1333 },
1334 GraphOp::AddNode {
1335 node_id: "s1".into(),
1336 node_type: "signal".into(),
1337 label: "extension-typed".into(),
1338 properties: BTreeMap::new(),
1339 subtype: None,
1340 },
1341 ],
1342 op_clocks: vec![(1, 0), (1, 1), (1, 2)],
1343 compacted_at_physical_ms: 1,
1344 compacted_at_logical: 2,
1345 },
1346 1,
1347 "inst-a",
1348 );
1349 g.apply(&checkpoint);
1350
1351 assert!(g.get_node("n1").is_some());
1352 assert!(g.get_node("s1").is_some(), "extension-typed node lost");
1353 assert!(g.ontology.node_types.contains_key("signal"));
1354 assert!(g.quarantined.is_empty());
1355 }
1356
1357 #[test]
1362 fn checkpoint_replay_trusts_inner_ops() {
1363 use crate::ontology::{PropertyDef, ValueType};
1364
1365 let mut ont = test_ontology();
1366 ont.node_types.get_mut("entity").unwrap().properties.insert(
1367 "name".into(),
1368 PropertyDef {
1369 value_type: ValueType::String,
1370 required: true,
1371 description: None,
1372 constraints: None,
1373 },
1374 );
1375 let mut g = MaterializedGraph::new(ont.clone());
1376
1377 let checkpoint = make_entry(
1378 GraphOp::Checkpoint {
1379 ops: vec![
1380 GraphOp::DefineOntology { ontology: ont },
1381 GraphOp::AddNode {
1382 node_id: "n1".into(),
1383 node_type: "entity".into(),
1384 label: "req".into(),
1385 properties: BTreeMap::new(), subtype: None,
1387 },
1388 GraphOp::UpdateProperty {
1389 entity_id: "n1".into(),
1390 key: "name".into(),
1391 value: Value::String("x".into()),
1392 },
1393 ],
1394 op_clocks: vec![(1, 0), (1, 1), (1, 2)],
1395 compacted_at_physical_ms: 1,
1396 compacted_at_logical: 2,
1397 },
1398 1,
1399 "inst-a",
1400 );
1401 g.apply(&checkpoint);
1402
1403 let node = g.get_node("n1").expect("required-property node lost");
1404 assert_eq!(
1405 node.properties.get("name"),
1406 Some(&Value::String("x".into()))
1407 );
1408 assert!(g.quarantined.is_empty());
1409 }
1410}