1use std::collections::{BTreeMap, HashMap, HashSet};
2
3use crate::clock::LamportClock;
4use crate::entry::{Entry, GraphOp, Hash, Value};
5use crate::ontology::{Ontology, OntologyExtension, ValidationMode};
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
46#[derive(Debug, Clone, PartialEq)]
53pub struct QuarantineRecord {
54 pub op_kind: String,
56 pub reason: String,
58 pub ontology_hash: String,
61}
62
63#[derive(Clone)]
74pub struct MaterializedGraph {
75 pub nodes: HashMap<String, Node>,
77 pub edges: HashMap<String, Edge>,
79 pub outgoing: HashMap<String, HashSet<String>>,
81 pub incoming: HashMap<String, HashSet<String>>,
83 pub by_type: HashMap<String, HashSet<String>>,
85 pub ontology: Ontology,
87 base_ontology: Ontology,
96 pub quarantined: HashMap<Hash, QuarantineRecord>,
103 pending_edges: HashMap<Hash, Entry>,
109}
110
111impl MaterializedGraph {
112 pub fn new(ontology: Ontology) -> Self {
114 Self {
115 nodes: HashMap::new(),
116 edges: HashMap::new(),
117 outgoing: HashMap::new(),
118 incoming: HashMap::new(),
119 by_type: HashMap::new(),
120 base_ontology: ontology.clone(),
121 ontology,
122 quarantined: HashMap::new(),
123 pending_edges: HashMap::new(),
124 }
125 }
126
127 pub fn apply(&mut self, entry: &Entry) {
134 self.apply_entry(entry, ValidationMode::Full)
135 }
136
137 fn apply_entry(&mut self, entry: &Entry, mode: ValidationMode) {
138 self.apply_inner(entry, mode, true)
139 }
140
141 fn apply_inner(
159 &mut self,
160 entry: &Entry,
161 mode: ValidationMode,
162 adopt_checkpoint_ontology: bool,
163 ) {
164 macro_rules! quarantine {
165 ($kind:expr, $reason:expr) => {{
166 let record = QuarantineRecord {
167 op_kind: $kind.to_string(),
168 reason: $reason.to_string(),
169 ontology_hash: hex::encode(self.ontology.content_hash()),
170 };
171 self.quarantined.insert(entry.hash, record);
172 }};
173 }
174 match &entry.payload {
175 GraphOp::Checkpoint { ops, op_clocks, .. } => {
176 for (i, op) in ops.iter().enumerate() {
179 if let GraphOp::DefineOntology { ontology } = op {
186 if adopt_checkpoint_ontology {
187 self.ontology = ontology.clone();
188 }
189 continue;
190 }
191 let clock = if i < op_clocks.len() {
192 LamportClock::with_values(&entry.author, op_clocks[i].0, op_clocks[i].1)
193 } else {
194 entry.clock.clone() };
196 let synthetic = Entry::new(op.clone(), vec![], vec![], clock, &entry.author);
197 self.apply_entry(&synthetic, ValidationMode::SkipRequired);
198 }
199 self.quarantined.remove(&entry.hash);
201 }
202 GraphOp::DefineOntology { .. } => {
203 }
205 GraphOp::ExtendOntology { extension } => {
206 if let Err(e) = self.ontology.merge_extension(extension) {
207 quarantine!("extend_ontology", e);
208 } else {
209 self.quarantined.remove(&entry.hash);
210 }
211 }
212 GraphOp::AddNode {
213 node_id,
214 node_type,
215 subtype,
216 label,
217 properties,
218 } => {
219 if let Err(e) = self.ontology.validate_node_mode(
221 node_type,
222 subtype.as_deref(),
223 properties,
224 mode,
225 ) {
226 quarantine!("add_node", e);
227 return;
228 }
229 self.quarantined.remove(&entry.hash);
233 self.apply_add_node(
234 node_id,
235 node_type,
236 subtype.as_deref(),
237 label,
238 properties,
239 &entry.clock,
240 );
241 self.retry_pending_edges();
243 }
244 GraphOp::AddEdge {
245 edge_id,
246 edge_type,
247 source_id,
248 target_id,
249 properties,
250 } => {
251 if !self.ontology.edge_types.contains_key(edge_type.as_str()) {
253 quarantine!(
254 "add_edge",
255 crate::ontology::ValidationError::UnknownEdgeType(edge_type.clone())
256 );
257 return;
258 }
259 match (
265 self.nodes.get(source_id.as_str()),
266 self.nodes.get(target_id.as_str()),
267 ) {
268 (Some(src), Some(tgt)) => {
269 if let Err(e) = self.ontology.validate_edge_mode(
270 edge_type,
271 &src.node_type,
272 &tgt.node_type,
273 properties,
274 mode,
275 ) {
276 quarantine!("add_edge", e);
277 return;
278 }
279 }
280 _ => {
281 self.pending_edges.insert(entry.hash, entry.clone());
282 quarantine!(
283 "add_edge",
284 format!(
285 "endpoint not yet materialized (source '{source_id}', \
286 target '{target_id}'); held pending until both arrive"
287 )
288 );
289 return;
290 }
291 }
292 self.quarantined.remove(&entry.hash);
293 self.apply_add_edge(
294 edge_id,
295 edge_type,
296 source_id,
297 target_id,
298 properties,
299 &entry.clock,
300 );
301 }
302 GraphOp::UpdateProperty {
303 entity_id,
304 key,
305 value,
306 } => {
307 let verdict = if let Some(node) = self.nodes.get(entity_id.as_str()) {
315 self.ontology.validate_property_update(
316 &node.node_type,
317 node.subtype.as_deref(),
318 key,
319 value,
320 )
321 } else if let Some(edge) = self.edges.get(entity_id.as_str()) {
322 self.ontology
323 .validate_edge_property_update(&edge.edge_type, key, value)
324 } else {
325 Ok(())
326 };
327 if let Err(e) = verdict {
328 quarantine!("update_property", e);
329 return;
330 }
331 self.quarantined.remove(&entry.hash);
332 self.apply_update_property(entity_id, key, value, &entry.clock);
333 }
334 GraphOp::RemoveNode { node_id } => {
335 self.quarantined.remove(&entry.hash);
336 self.apply_remove_node(node_id, &entry.clock);
337 }
338 GraphOp::RemoveEdge { edge_id } => {
339 self.quarantined.remove(&entry.hash);
340 self.apply_remove_edge(edge_id, &entry.clock);
341 }
342 GraphOp::DefineLens { .. } => {
343 }
345 }
346 }
347
348 pub fn apply_all(&mut self, entries: &[&Entry]) {
350 for entry in entries {
351 self.apply(entry);
352 }
353 }
354
355 pub fn rebuild(&mut self, entries: &[&Entry]) {
357 self.rebuild_inner(entries, None)
358 }
359
360 pub fn rebuild_with_pending_extension(
368 &mut self,
369 entries: &[&Entry],
370 extension: &OntologyExtension,
371 ) {
372 self.rebuild_inner(entries, Some(extension))
373 }
374
375 fn rebuild_inner(&mut self, entries: &[&Entry], pending: Option<&OntologyExtension>) {
376 self.nodes.clear();
377 self.edges.clear();
378 self.outgoing.clear();
379 self.incoming.clear();
380 self.by_type.clear();
381 self.quarantined.clear();
382 self.pending_edges.clear();
383
384 self.ontology = self.base_ontology.clone();
396 for entry in entries {
397 match &entry.payload {
398 GraphOp::DefineOntology { .. } => {}
399 GraphOp::ExtendOntology { extension } => {
400 if let Err(e) = self.ontology.merge_extension(extension) {
401 self.quarantined.insert(
402 entry.hash,
403 QuarantineRecord {
404 op_kind: "extend_ontology".to_string(),
405 reason: e.to_string(),
406 ontology_hash: hex::encode(self.ontology.content_hash()),
407 },
408 );
409 }
410 }
411 GraphOp::Checkpoint { ops, .. } => {
412 for op in ops {
413 if let GraphOp::DefineOntology { ontology } = op {
414 self.ontology = ontology.clone();
415 }
416 }
417 }
418 _ => {}
419 }
420 }
421
422 if let Some(extension) = pending {
425 self.ontology
426 .merge_extension(extension)
427 .expect("caller validates the extension before previewing it");
428 }
429
430 for entry in entries {
433 if matches!(entry.payload, GraphOp::ExtendOntology { .. }) {
434 continue;
435 }
436 self.apply_inner(entry, ValidationMode::Full, false);
437 }
438 }
439
440 fn retry_pending_edges(&mut self) {
445 if self.pending_edges.is_empty() {
446 return;
447 }
448 let ready: Vec<Entry> = self
449 .pending_edges
450 .values()
451 .filter(|entry| match &entry.payload {
452 GraphOp::AddEdge {
453 source_id,
454 target_id,
455 ..
456 } => {
457 self.nodes.contains_key(source_id.as_str())
458 && self.nodes.contains_key(target_id.as_str())
459 }
460 _ => false,
461 })
462 .cloned()
463 .collect();
464 for entry in ready {
465 self.pending_edges.remove(&entry.hash);
466 self.apply_entry(&entry, ValidationMode::Full);
467 }
468 }
469
470 pub fn get_node(&self, node_id: &str) -> Option<&Node> {
474 self.nodes.get(node_id).filter(|n| !n.tombstoned)
475 }
476
477 pub fn get_edge(&self, edge_id: &str) -> Option<&Edge> {
479 self.edges.get(edge_id).filter(|e| !e.tombstoned)
480 }
481
482 pub fn nodes_by_type(&self, node_type: &str) -> Vec<&Node> {
485 let mut types = vec![node_type.to_string()];
486 types.extend(
487 self.ontology
488 .descendants(node_type)
489 .into_iter()
490 .map(|s| s.to_string()),
491 );
492 types
493 .iter()
494 .flat_map(|t| self.by_type.get(t.as_str()))
495 .flatten()
496 .filter_map(|id| self.get_node(id))
497 .collect()
498 }
499
500 pub fn nodes_by_subtype(&self, subtype: &str) -> Vec<&Node> {
502 self.nodes
503 .values()
504 .filter(|n| !n.tombstoned && n.subtype.as_deref() == Some(subtype))
505 .collect()
506 }
507
508 pub fn nodes_by_property(&self, key: &str, value: &Value) -> Vec<&Node> {
510 self.nodes
511 .values()
512 .filter(|n| !n.tombstoned && n.properties.get(key) == Some(value))
513 .collect()
514 }
515
516 pub fn outgoing_edges(&self, node_id: &str) -> Vec<&Edge> {
518 match self.outgoing.get(node_id) {
519 Some(edge_ids) => edge_ids
520 .iter()
521 .filter_map(|eid| self.get_edge(eid))
522 .filter(|e| self.is_node_live(&e.target_id))
523 .collect(),
524 None => vec![],
525 }
526 }
527
528 pub fn incoming_edges(&self, node_id: &str) -> Vec<&Edge> {
530 match self.incoming.get(node_id) {
531 Some(edge_ids) => edge_ids
532 .iter()
533 .filter_map(|eid| self.get_edge(eid))
534 .filter(|e| self.is_node_live(&e.source_id))
535 .collect(),
536 None => vec![],
537 }
538 }
539
540 pub fn estimated_memory_bytes(&self) -> usize {
545 let mut total = 0;
546 for node in self.nodes.values() {
548 total += node.node_id.len() + node.node_type.len() + node.label.len();
549 total += node.subtype.as_ref().map_or(0, |s| s.len());
550 for (k, v) in &node.properties {
552 total += k.len() + std::mem::size_of_val(v) + 48; }
554 total += 128; }
556 for edge in self.edges.values() {
558 total += edge.edge_id.len() + edge.edge_type.len();
559 total += edge.source_id.len() + edge.target_id.len();
560 for (k, v) in &edge.properties {
561 total += k.len() + std::mem::size_of_val(v) + 48;
562 }
563 total += 128;
564 }
565 for (k, set) in &self.outgoing {
567 total += k.len() + set.len() * 32;
568 }
569 for (k, set) in &self.incoming {
570 total += k.len() + set.len() * 32;
571 }
572 for (k, set) in &self.by_type {
574 total += k.len() + set.len() * 32;
575 }
576 total += self.quarantined.len() * 48;
578 total
579 }
580
581 pub fn all_nodes(&self) -> Vec<&Node> {
583 self.nodes.values().filter(|n| !n.tombstoned).collect()
584 }
585
586 pub fn all_edges(&self) -> Vec<&Edge> {
588 self.edges
589 .values()
590 .filter(|e| {
591 !e.tombstoned && self.is_node_live(&e.source_id) && self.is_node_live(&e.target_id)
592 })
593 .collect()
594 }
595
596 pub fn neighbors(&self, node_id: &str) -> Vec<&str> {
598 self.outgoing_edges(node_id)
599 .iter()
600 .map(|e| e.target_id.as_str())
601 .collect()
602 }
603
604 pub fn reverse_neighbors(&self, node_id: &str) -> Vec<&str> {
606 self.incoming_edges(node_id)
607 .iter()
608 .map(|e| e.source_id.as_str())
609 .collect()
610 }
611
612 fn apply_add_node(
615 &mut self,
616 node_id: &str,
617 node_type: &str,
618 subtype: Option<&str>,
619 label: &str,
620 properties: &BTreeMap<String, Value>,
621 clock: &LamportClock,
622 ) {
623 if let Some(existing) = self.nodes.get_mut(node_id) {
624 existing.tombstoned = false;
626 if clock_wins(clock, &existing.last_add_clock) {
628 existing.last_add_clock = clock.clone();
629 }
630 if clock_wins(clock, &existing.last_clock) {
632 existing.label = label.to_string();
633 existing.subtype = subtype.map(|s| s.to_string());
634 existing.last_clock = clock.clone();
635 }
636 merge_properties_lww(
637 &mut existing.properties,
638 &mut existing.property_clocks,
639 properties,
640 clock,
641 );
642 } else {
643 let property_clocks: HashMap<String, LamportClock> = properties
644 .keys()
645 .map(|k| (k.clone(), clock.clone()))
646 .collect();
647 let node = Node {
648 node_id: node_id.to_string(),
649 node_type: node_type.to_string(),
650 subtype: subtype.map(|s| s.to_string()),
651 label: label.to_string(),
652 properties: properties.clone(),
653 property_clocks,
654 last_clock: clock.clone(),
655 last_add_clock: clock.clone(),
656 tombstoned: false,
657 };
658 self.by_type
659 .entry(node_type.to_string())
660 .or_default()
661 .insert(node_id.to_string());
662 self.nodes.insert(node_id.to_string(), node);
663 }
664 }
665
666 fn apply_add_edge(
667 &mut self,
668 edge_id: &str,
669 edge_type: &str,
670 source_id: &str,
671 target_id: &str,
672 properties: &BTreeMap<String, Value>,
673 clock: &LamportClock,
674 ) {
675 if let Some(existing) = self.edges.get_mut(edge_id) {
676 existing.tombstoned = false;
678 if clock_wins(clock, &existing.last_add_clock) {
679 existing.last_add_clock = clock.clone();
680 }
681 if clock_wins(clock, &existing.last_clock) {
682 existing.last_clock = clock.clone();
683 }
684 merge_properties_lww(
685 &mut existing.properties,
686 &mut existing.property_clocks,
687 properties,
688 clock,
689 );
690 } else {
691 let property_clocks: HashMap<String, LamportClock> = properties
692 .keys()
693 .map(|k| (k.clone(), clock.clone()))
694 .collect();
695 let edge = Edge {
696 edge_id: edge_id.to_string(),
697 edge_type: edge_type.to_string(),
698 source_id: source_id.to_string(),
699 target_id: target_id.to_string(),
700 properties: properties.clone(),
701 property_clocks,
702 last_clock: clock.clone(),
703 last_add_clock: clock.clone(),
704 tombstoned: false,
705 };
706 self.outgoing
707 .entry(source_id.to_string())
708 .or_default()
709 .insert(edge_id.to_string());
710 self.incoming
711 .entry(target_id.to_string())
712 .or_default()
713 .insert(edge_id.to_string());
714 self.edges.insert(edge_id.to_string(), edge);
715 }
716 }
717
718 fn apply_update_property(
719 &mut self,
720 entity_id: &str,
721 key: &str,
722 value: &Value,
723 clock: &LamportClock,
724 ) {
725 if let Some(node) = self.nodes.get_mut(entity_id) {
728 let dominated = node
729 .property_clocks
730 .get(key)
731 .map(|c| clock_wins(clock, c))
732 .unwrap_or(true);
733 if dominated {
734 node.properties.insert(key.to_string(), value.clone());
735 node.property_clocks.insert(key.to_string(), clock.clone());
736 }
737 if clock_wins(clock, &node.last_clock) {
739 node.last_clock = clock.clone();
740 }
741 } else if let Some(edge) = self.edges.get_mut(entity_id) {
742 let dominated = edge
743 .property_clocks
744 .get(key)
745 .map(|c| clock_wins(clock, c))
746 .unwrap_or(true);
747 if dominated {
748 edge.properties.insert(key.to_string(), value.clone());
749 edge.property_clocks.insert(key.to_string(), clock.clone());
750 }
751 if clock_wins(clock, &edge.last_clock) {
752 edge.last_clock = clock.clone();
753 }
754 }
755 }
757
758 fn apply_remove_node(&mut self, node_id: &str, clock: &LamportClock) {
759 if let Some(node) = self.nodes.get_mut(node_id) {
760 if clock_wins(clock, &node.last_add_clock) {
764 node.tombstoned = true;
765 node.last_clock = clock.clone();
766 }
767 }
768 }
771
772 fn apply_remove_edge(&mut self, edge_id: &str, clock: &LamportClock) {
773 if let Some(edge) = self.edges.get_mut(edge_id) {
774 if clock_wins(clock, &edge.last_add_clock) {
776 edge.tombstoned = true;
777 edge.last_clock = clock.clone();
778 }
779 }
780 }
781
782 fn is_node_live(&self, node_id: &str) -> bool {
783 self.nodes
784 .get(node_id)
785 .map(|n| !n.tombstoned)
786 .unwrap_or(false)
787 }
788}
789
790fn merge_properties_lww(
793 existing_props: &mut BTreeMap<String, Value>,
794 existing_clocks: &mut HashMap<String, LamportClock>,
795 new_props: &BTreeMap<String, Value>,
796 clock: &LamportClock,
797) {
798 for (k, v) in new_props {
799 let dominated = existing_clocks
800 .get(k)
801 .map(|c| clock_wins(clock, c))
802 .unwrap_or(true);
803 if dominated {
804 existing_props.insert(k.clone(), v.clone());
805 existing_clocks.insert(k.clone(), clock.clone());
806 }
807 }
808}
809
810fn clock_wins(new_clock: &LamportClock, existing_clock: &LamportClock) -> bool {
813 new_clock.cmp_order(existing_clock) == std::cmp::Ordering::Greater
814}
815
816#[cfg(test)]
817mod tests {
818 use super::*;
819 use crate::entry::Entry;
820 use crate::ontology::{EdgeTypeDef, NodeTypeDef};
821
822 fn test_ontology() -> Ontology {
823 Ontology {
824 node_types: BTreeMap::from([
825 (
826 "entity".into(),
827 NodeTypeDef {
828 description: None,
829 properties: BTreeMap::new(),
830 subtypes: None,
831 parent_type: None,
832 },
833 ),
834 (
835 "signal".into(),
836 NodeTypeDef {
837 description: None,
838 properties: BTreeMap::new(),
839 subtypes: None,
840 parent_type: None,
841 },
842 ),
843 ]),
844 edge_types: BTreeMap::from([
845 (
846 "RUNS_ON".into(),
847 EdgeTypeDef {
848 description: None,
849 source_types: vec!["entity".into()],
850 target_types: vec!["entity".into()],
851 properties: BTreeMap::new(),
852 },
853 ),
854 (
855 "OBSERVES".into(),
856 EdgeTypeDef {
857 description: None,
858 source_types: vec!["signal".into()],
859 target_types: vec!["entity".into()],
860 properties: BTreeMap::new(),
861 },
862 ),
863 ]),
864 }
865 }
866
867 fn make_entry(op: GraphOp, clock_time: u64, author: &str) -> Entry {
868 Entry::new(
869 op,
870 vec![],
871 vec![],
872 LamportClock::with_values(author, clock_time, 0),
873 author,
874 )
875 }
876
877 #[test]
880 fn add_node_appears_in_query() {
881 let mut g = MaterializedGraph::new(test_ontology());
882 let entry = make_entry(
883 GraphOp::AddNode {
884 node_id: "server-1".into(),
885 node_type: "entity".into(),
886 label: "Server 1".into(),
887 properties: BTreeMap::from([("ip".into(), Value::String("10.0.0.1".into()))]),
888 subtype: None,
889 },
890 1,
891 "inst-a",
892 );
893 g.apply(&entry);
894
895 let node = g.get_node("server-1").unwrap();
896 assert_eq!(node.node_type, "entity");
897 assert_eq!(node.label, "Server 1");
898 assert_eq!(
899 node.properties.get("ip"),
900 Some(&Value::String("10.0.0.1".into()))
901 );
902 }
903
904 #[test]
905 fn add_edge_creates_adjacency() {
906 let mut g = MaterializedGraph::new(test_ontology());
907 g.apply(&make_entry(
908 GraphOp::AddNode {
909 node_id: "svc".into(),
910 node_type: "entity".into(),
911 label: "svc".into(),
912 properties: BTreeMap::new(),
913 subtype: None,
914 },
915 1,
916 "inst-a",
917 ));
918 g.apply(&make_entry(
919 GraphOp::AddNode {
920 node_id: "srv".into(),
921 node_type: "entity".into(),
922 label: "srv".into(),
923 properties: BTreeMap::new(),
924 subtype: None,
925 },
926 2,
927 "inst-a",
928 ));
929 g.apply(&make_entry(
930 GraphOp::AddEdge {
931 edge_id: "e1".into(),
932 edge_type: "RUNS_ON".into(),
933 source_id: "svc".into(),
934 target_id: "srv".into(),
935 properties: BTreeMap::new(),
936 },
937 3,
938 "inst-a",
939 ));
940
941 let out = g.outgoing_edges("svc");
943 assert_eq!(out.len(), 1);
944 assert_eq!(out[0].target_id, "srv");
945
946 let inc = g.incoming_edges("srv");
947 assert_eq!(inc.len(), 1);
948 assert_eq!(inc[0].source_id, "svc");
949
950 assert_eq!(g.neighbors("svc"), vec!["srv"]);
951 }
952
953 #[test]
954 fn update_property_reflected() {
955 let mut g = MaterializedGraph::new(test_ontology());
956 g.apply(&make_entry(
957 GraphOp::AddNode {
958 node_id: "s1".into(),
959 node_type: "entity".into(),
960 label: "s1".into(),
961 properties: BTreeMap::new(),
962 subtype: None,
963 },
964 1,
965 "inst-a",
966 ));
967 g.apply(&make_entry(
968 GraphOp::UpdateProperty {
969 entity_id: "s1".into(),
970 key: "cpu".into(),
971 value: Value::Float(85.5),
972 },
973 2,
974 "inst-a",
975 ));
976
977 let node = g.get_node("s1").unwrap();
978 assert_eq!(node.properties.get("cpu"), Some(&Value::Float(85.5)));
979 }
980
981 #[test]
982 fn remove_node_cascades_edges() {
983 let mut g = MaterializedGraph::new(test_ontology());
984 g.apply(&make_entry(
985 GraphOp::AddNode {
986 node_id: "a".into(),
987 node_type: "entity".into(),
988 label: "a".into(),
989 properties: BTreeMap::new(),
990 subtype: None,
991 },
992 1,
993 "inst-a",
994 ));
995 g.apply(&make_entry(
996 GraphOp::AddNode {
997 node_id: "b".into(),
998 node_type: "entity".into(),
999 label: "b".into(),
1000 properties: BTreeMap::new(),
1001 subtype: None,
1002 },
1003 2,
1004 "inst-a",
1005 ));
1006 g.apply(&make_entry(
1007 GraphOp::AddEdge {
1008 edge_id: "e1".into(),
1009 edge_type: "RUNS_ON".into(),
1010 source_id: "a".into(),
1011 target_id: "b".into(),
1012 properties: BTreeMap::new(),
1013 },
1014 3,
1015 "inst-a",
1016 ));
1017 assert_eq!(g.all_edges().len(), 1);
1018
1019 g.apply(&make_entry(
1021 GraphOp::RemoveNode {
1022 node_id: "b".into(),
1023 },
1024 4,
1025 "inst-a",
1026 ));
1027 assert!(g.get_node("b").is_none());
1028 assert_eq!(g.all_edges().len(), 0);
1030 assert_eq!(g.outgoing_edges("a").len(), 0);
1032 }
1033
1034 #[test]
1035 fn remove_edge_preserves_nodes() {
1036 let mut g = MaterializedGraph::new(test_ontology());
1037 g.apply(&make_entry(
1038 GraphOp::AddNode {
1039 node_id: "a".into(),
1040 node_type: "entity".into(),
1041 label: "a".into(),
1042 properties: BTreeMap::new(),
1043 subtype: None,
1044 },
1045 1,
1046 "inst-a",
1047 ));
1048 g.apply(&make_entry(
1049 GraphOp::AddNode {
1050 node_id: "b".into(),
1051 node_type: "entity".into(),
1052 label: "b".into(),
1053 properties: BTreeMap::new(),
1054 subtype: None,
1055 },
1056 2,
1057 "inst-a",
1058 ));
1059 g.apply(&make_entry(
1060 GraphOp::AddEdge {
1061 edge_id: "e1".into(),
1062 edge_type: "RUNS_ON".into(),
1063 source_id: "a".into(),
1064 target_id: "b".into(),
1065 properties: BTreeMap::new(),
1066 },
1067 3,
1068 "inst-a",
1069 ));
1070 g.apply(&make_entry(
1071 GraphOp::RemoveEdge {
1072 edge_id: "e1".into(),
1073 },
1074 4,
1075 "inst-a",
1076 ));
1077
1078 assert!(g.get_node("a").is_some());
1080 assert!(g.get_node("b").is_some());
1081 assert!(g.get_edge("e1").is_none());
1083 assert_eq!(g.all_edges().len(), 0);
1084 }
1085
1086 #[test]
1087 fn query_by_type_filters() {
1088 let mut g = MaterializedGraph::new(test_ontology());
1089 g.apply(&make_entry(
1090 GraphOp::AddNode {
1091 node_id: "s1".into(),
1092 node_type: "entity".into(),
1093 label: "s1".into(),
1094 properties: BTreeMap::new(),
1095 subtype: None,
1096 },
1097 1,
1098 "inst-a",
1099 ));
1100 g.apply(&make_entry(
1101 GraphOp::AddNode {
1102 node_id: "s2".into(),
1103 node_type: "entity".into(),
1104 label: "s2".into(),
1105 properties: BTreeMap::new(),
1106 subtype: None,
1107 },
1108 2,
1109 "inst-a",
1110 ));
1111 g.apply(&make_entry(
1112 GraphOp::AddNode {
1113 node_id: "alert".into(),
1114 node_type: "signal".into(),
1115 label: "alert".into(),
1116 properties: BTreeMap::new(),
1117 subtype: None,
1118 },
1119 3,
1120 "inst-a",
1121 ));
1122
1123 let entities = g.nodes_by_type("entity");
1124 assert_eq!(entities.len(), 2);
1125 let signals = g.nodes_by_type("signal");
1126 assert_eq!(signals.len(), 1);
1127 assert_eq!(signals[0].node_id, "alert");
1128 }
1129
1130 #[test]
1131 fn query_by_property_filters() {
1132 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::from([("status".into(), Value::String("alive".into()))]),
1139 subtype: None,
1140 },
1141 1,
1142 "inst-a",
1143 ));
1144 g.apply(&make_entry(
1145 GraphOp::AddNode {
1146 node_id: "s2".into(),
1147 node_type: "entity".into(),
1148 label: "s2".into(),
1149 properties: BTreeMap::from([("status".into(), Value::String("dead".into()))]),
1150 subtype: None,
1151 },
1152 2,
1153 "inst-a",
1154 ));
1155
1156 let alive = g.nodes_by_property("status", &Value::String("alive".into()));
1157 assert_eq!(alive.len(), 1);
1158 assert_eq!(alive[0].node_id, "s1");
1159 }
1160
1161 #[test]
1162 fn materialization_from_empty() {
1163 let mut g1 = MaterializedGraph::new(test_ontology());
1165 let entries = vec![
1166 make_entry(
1167 GraphOp::DefineOntology {
1168 ontology: test_ontology(),
1169 },
1170 0,
1171 "inst-a",
1172 ),
1173 make_entry(
1174 GraphOp::AddNode {
1175 node_id: "a".into(),
1176 node_type: "entity".into(),
1177 label: "a".into(),
1178 properties: BTreeMap::new(),
1179 subtype: None,
1180 },
1181 1,
1182 "inst-a",
1183 ),
1184 make_entry(
1185 GraphOp::AddNode {
1186 node_id: "b".into(),
1187 node_type: "entity".into(),
1188 label: "b".into(),
1189 properties: BTreeMap::new(),
1190 subtype: None,
1191 },
1192 2,
1193 "inst-a",
1194 ),
1195 make_entry(
1196 GraphOp::AddEdge {
1197 edge_id: "e1".into(),
1198 edge_type: "RUNS_ON".into(),
1199 source_id: "a".into(),
1200 target_id: "b".into(),
1201 properties: BTreeMap::new(),
1202 },
1203 3,
1204 "inst-a",
1205 ),
1206 ];
1207 for e in &entries {
1208 g1.apply(e);
1209 }
1210
1211 let mut g2 = MaterializedGraph::new(test_ontology());
1213 let refs: Vec<&Entry> = entries.iter().collect();
1214 g2.rebuild(&refs);
1215
1216 assert_eq!(g1.all_nodes().len(), g2.all_nodes().len());
1218 assert_eq!(g1.all_edges().len(), g2.all_edges().len());
1219 for node in g1.all_nodes() {
1220 let n2 = g2.get_node(&node.node_id).unwrap();
1221 assert_eq!(node.node_type, n2.node_type);
1222 assert_eq!(node.properties, n2.properties);
1223 }
1224 }
1225
1226 #[test]
1227 fn incremental_equals_full() {
1228 let entries = vec![
1229 make_entry(
1230 GraphOp::DefineOntology {
1231 ontology: test_ontology(),
1232 },
1233 0,
1234 "inst-a",
1235 ),
1236 make_entry(
1237 GraphOp::AddNode {
1238 node_id: "a".into(),
1239 node_type: "entity".into(),
1240 label: "a".into(),
1241 properties: BTreeMap::from([("x".into(), Value::Int(1))]),
1242 subtype: None,
1243 },
1244 1,
1245 "inst-a",
1246 ),
1247 make_entry(
1248 GraphOp::UpdateProperty {
1249 entity_id: "a".into(),
1250 key: "x".into(),
1251 value: Value::Int(2),
1252 },
1253 2,
1254 "inst-a",
1255 ),
1256 make_entry(
1257 GraphOp::AddNode {
1258 node_id: "b".into(),
1259 node_type: "entity".into(),
1260 label: "b".into(),
1261 properties: BTreeMap::new(),
1262 subtype: None,
1263 },
1264 3,
1265 "inst-a",
1266 ),
1267 make_entry(
1268 GraphOp::AddEdge {
1269 edge_id: "e1".into(),
1270 edge_type: "RUNS_ON".into(),
1271 source_id: "a".into(),
1272 target_id: "b".into(),
1273 properties: BTreeMap::new(),
1274 },
1275 4,
1276 "inst-a",
1277 ),
1278 make_entry(
1279 GraphOp::RemoveEdge {
1280 edge_id: "e1".into(),
1281 },
1282 5,
1283 "inst-a",
1284 ),
1285 ];
1286
1287 let mut g_inc = MaterializedGraph::new(test_ontology());
1289 for e in &entries {
1290 g_inc.apply(e);
1291 }
1292
1293 let mut g_full = MaterializedGraph::new(test_ontology());
1295 let refs: Vec<&Entry> = entries.iter().collect();
1296 g_full.rebuild(&refs);
1297
1298 assert_eq!(
1300 g_inc.get_node("a").unwrap().properties.get("x"),
1301 Some(&Value::Int(2))
1302 );
1303 assert_eq!(
1304 g_full.get_node("a").unwrap().properties.get("x"),
1305 Some(&Value::Int(2))
1306 );
1307 assert_eq!(g_inc.all_edges().len(), 0);
1309 assert_eq!(g_full.all_edges().len(), 0);
1310 }
1311
1312 #[test]
1313 fn lww_concurrent_property_update() {
1314 let mut g = MaterializedGraph::new(test_ontology());
1316 g.apply(&make_entry(
1317 GraphOp::AddNode {
1318 node_id: "s1".into(),
1319 node_type: "entity".into(),
1320 label: "s1".into(),
1321 properties: BTreeMap::new(),
1322 subtype: None,
1323 },
1324 1,
1325 "inst-a",
1326 ));
1327 g.apply(&make_entry(
1329 GraphOp::UpdateProperty {
1330 entity_id: "s1".into(),
1331 key: "status".into(),
1332 value: Value::String("alive".into()),
1333 },
1334 2,
1335 "inst-a",
1336 ));
1337 g.apply(&make_entry(
1339 GraphOp::UpdateProperty {
1340 entity_id: "s1".into(),
1341 key: "status".into(),
1342 value: Value::String("dead".into()),
1343 },
1344 3,
1345 "inst-b",
1346 ));
1347 assert_eq!(
1348 g.get_node("s1").unwrap().properties.get("status"),
1349 Some(&Value::String("dead".into()))
1350 );
1351 }
1352
1353 #[test]
1354 fn lww_tiebreak_by_instance_id() {
1355 let mut g = MaterializedGraph::new(test_ontology());
1357 g.apply(&make_entry(
1358 GraphOp::AddNode {
1359 node_id: "s1".into(),
1360 node_type: "entity".into(),
1361 label: "s1".into(),
1362 properties: BTreeMap::new(),
1363 subtype: None,
1364 },
1365 1,
1366 "inst-a",
1367 ));
1368 g.apply(&make_entry(
1370 GraphOp::UpdateProperty {
1371 entity_id: "s1".into(),
1372 key: "x".into(),
1373 value: Value::Int(1),
1374 },
1375 5,
1376 "inst-a",
1377 ));
1378 g.apply(&make_entry(
1379 GraphOp::UpdateProperty {
1380 entity_id: "s1".into(),
1381 key: "x".into(),
1382 value: Value::Int(2),
1383 },
1384 5,
1385 "inst-b",
1386 ));
1387 assert_eq!(
1389 g.get_node("s1").unwrap().properties.get("x"),
1390 Some(&Value::Int(1))
1391 );
1392 }
1393
1394 #[test]
1395 fn lww_per_property_concurrent_different_keys() {
1396 let mut g = MaterializedGraph::new(test_ontology());
1400 g.apply(&make_entry(
1401 GraphOp::AddNode {
1402 node_id: "s1".into(),
1403 node_type: "entity".into(),
1404 label: "s1".into(),
1405 properties: BTreeMap::from([
1406 ("x".into(), Value::Int(0)),
1407 ("y".into(), Value::Int(0)),
1408 ]),
1409 subtype: None,
1410 },
1411 1,
1412 "inst-a",
1413 ));
1414 g.apply(&make_entry(
1416 GraphOp::UpdateProperty {
1417 entity_id: "s1".into(),
1418 key: "x".into(),
1419 value: Value::Int(42),
1420 },
1421 3,
1422 "inst-a",
1423 ));
1424 g.apply(&make_entry(
1426 GraphOp::UpdateProperty {
1427 entity_id: "s1".into(),
1428 key: "y".into(),
1429 value: Value::Int(99),
1430 },
1431 3,
1432 "inst-b",
1433 ));
1434
1435 let node = g.get_node("s1").unwrap();
1436 assert_eq!(
1438 node.properties.get("x"),
1439 Some(&Value::Int(42)),
1440 "update to 'x' must not be rejected by concurrent update to 'y'"
1441 );
1442 assert_eq!(
1443 node.properties.get("y"),
1444 Some(&Value::Int(99)),
1445 "update to 'y' must not be rejected by concurrent update to 'x'"
1446 );
1447 }
1448
1449 #[test]
1450 fn lww_per_property_order_independent() {
1451 let mut g = MaterializedGraph::new(test_ontology());
1453 g.apply(&make_entry(
1454 GraphOp::AddNode {
1455 node_id: "s1".into(),
1456 node_type: "entity".into(),
1457 label: "s1".into(),
1458 properties: BTreeMap::from([
1459 ("x".into(), Value::Int(0)),
1460 ("y".into(), Value::Int(0)),
1461 ]),
1462 subtype: None,
1463 },
1464 1,
1465 "inst-a",
1466 ));
1467 g.apply(&make_entry(
1469 GraphOp::UpdateProperty {
1470 entity_id: "s1".into(),
1471 key: "y".into(),
1472 value: Value::Int(99),
1473 },
1474 3,
1475 "inst-b",
1476 ));
1477 g.apply(&make_entry(
1478 GraphOp::UpdateProperty {
1479 entity_id: "s1".into(),
1480 key: "x".into(),
1481 value: Value::Int(42),
1482 },
1483 3,
1484 "inst-a",
1485 ));
1486
1487 let node = g.get_node("s1").unwrap();
1488 assert_eq!(node.properties.get("x"), Some(&Value::Int(42)));
1489 assert_eq!(node.properties.get("y"), Some(&Value::Int(99)));
1490 }
1491
1492 #[test]
1493 fn add_wins_over_remove() {
1494 let mut g = MaterializedGraph::new(test_ontology());
1496 g.apply(&make_entry(
1497 GraphOp::AddNode {
1498 node_id: "s1".into(),
1499 node_type: "entity".into(),
1500 label: "s1".into(),
1501 properties: BTreeMap::new(),
1502 subtype: None,
1503 },
1504 1,
1505 "inst-a",
1506 ));
1507 g.apply(&make_entry(
1509 GraphOp::RemoveNode {
1510 node_id: "s1".into(),
1511 },
1512 2,
1513 "inst-a",
1514 ));
1515 assert!(g.get_node("s1").is_none());
1516
1517 g.apply(&make_entry(
1519 GraphOp::AddNode {
1520 node_id: "s1".into(),
1521 node_type: "entity".into(),
1522 label: "s1 v2".into(),
1523 properties: BTreeMap::new(),
1524 subtype: None,
1525 },
1526 3,
1527 "inst-b",
1528 ));
1529 let node = g.get_node("s1").unwrap();
1530 assert_eq!(node.label, "s1 v2");
1531 assert!(!node.tombstoned);
1532 }
1533
1534 #[test]
1538 fn checkpoint_replay_applies_inner_define_ontology() {
1539 let mut base = test_ontology();
1541 base.node_types.remove("signal");
1542 base.edge_types.remove("OBSERVES");
1543 let mut g = MaterializedGraph::new(base);
1544
1545 let checkpoint = make_entry(
1546 GraphOp::Checkpoint {
1547 ops: vec![
1548 GraphOp::DefineOntology {
1549 ontology: test_ontology(), },
1551 GraphOp::AddNode {
1552 node_id: "n1".into(),
1553 node_type: "entity".into(),
1554 label: "base-typed".into(),
1555 properties: BTreeMap::new(),
1556 subtype: None,
1557 },
1558 GraphOp::AddNode {
1559 node_id: "s1".into(),
1560 node_type: "signal".into(),
1561 label: "extension-typed".into(),
1562 properties: BTreeMap::new(),
1563 subtype: None,
1564 },
1565 ],
1566 op_clocks: vec![(1, 0), (1, 1), (1, 2)],
1567 compacted_at_physical_ms: 1,
1568 compacted_at_logical: 2,
1569 },
1570 1,
1571 "inst-a",
1572 );
1573 g.apply(&checkpoint);
1574
1575 assert!(g.get_node("n1").is_some());
1576 assert!(g.get_node("s1").is_some(), "extension-typed node lost");
1577 assert!(g.ontology.node_types.contains_key("signal"));
1578 assert!(g.quarantined.is_empty());
1579 }
1580
1581 #[test]
1586 fn checkpoint_replay_trusts_inner_ops() {
1587 use crate::ontology::{PropertyDef, ValueType};
1588
1589 let mut ont = test_ontology();
1590 ont.node_types.get_mut("entity").unwrap().properties.insert(
1591 "name".into(),
1592 PropertyDef {
1593 value_type: ValueType::String,
1594 required: true,
1595 description: None,
1596 constraints: None,
1597 },
1598 );
1599 let mut g = MaterializedGraph::new(ont.clone());
1600
1601 let checkpoint = make_entry(
1602 GraphOp::Checkpoint {
1603 ops: vec![
1604 GraphOp::DefineOntology { ontology: ont },
1605 GraphOp::AddNode {
1606 node_id: "n1".into(),
1607 node_type: "entity".into(),
1608 label: "req".into(),
1609 properties: BTreeMap::new(), subtype: None,
1611 },
1612 GraphOp::UpdateProperty {
1613 entity_id: "n1".into(),
1614 key: "name".into(),
1615 value: Value::String("x".into()),
1616 },
1617 ],
1618 op_clocks: vec![(1, 0), (1, 1), (1, 2)],
1619 compacted_at_physical_ms: 1,
1620 compacted_at_logical: 2,
1621 },
1622 1,
1623 "inst-a",
1624 );
1625 g.apply(&checkpoint);
1626
1627 let node = g.get_node("n1").expect("required-property node lost");
1628 assert_eq!(
1629 node.properties.get("name"),
1630 Some(&Value::String("x".into()))
1631 );
1632 assert!(g.quarantined.is_empty());
1633 }
1634}