silk/
graph.rs

1use std::collections::{BTreeMap, HashMap, HashSet};
2
3use crate::clock::LamportClock;
4use crate::entry::{Entry, GraphOp, Hash, Value};
5use crate::ontology::Ontology;
6
7/// A materialized node in the graph.
8#[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    /// Per-property clocks for LWW conflict resolution.
16    /// Each property key tracks the clock of its last write, so
17    /// concurrent updates to different properties don't conflict.
18    pub property_clocks: HashMap<String, LamportClock>,
19    /// Clock of the entry that last modified this node.
20    /// Used for add-wins semantics and label LWW.
21    pub last_clock: LamportClock,
22    /// Clock of the most recent AddNode for this node.
23    /// Used for add-wins semantics: remove only wins if its clock
24    /// is strictly greater than last_add_clock.
25    pub last_add_clock: LamportClock,
26    /// Whether this node has been tombstoned (removed).
27    pub tombstoned: bool,
28}
29
30/// A materialized edge in the graph.
31#[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    /// Per-property clocks for LWW conflict resolution.
39    pub property_clocks: HashMap<String, LamportClock>,
40    pub last_clock: LamportClock,
41    /// Clock of the most recent AddEdge for this edge.
42    pub last_add_clock: LamportClock,
43    pub tombstoned: bool,
44}
45
46/// Materialized graph — derived from the op log.
47///
48/// Provides fast queries without replaying the full log.
49/// Updated incrementally as new entries arrive, or rebuilt
50/// from scratch by replaying the entire op log.
51///
52/// CRDT semantics:
53/// - **Add-wins** for topology (concurrent add + remove → node/edge exists)
54/// - **LWW** (Last-Writer-Wins) per property key (highest Lamport clock wins)
55/// - **Tombstones** for deletes (mark as deleted, don't physically remove)
56pub struct MaterializedGraph {
57    /// node_id → Node
58    pub nodes: HashMap<String, Node>,
59    /// edge_id → Edge
60    pub edges: HashMap<String, Edge>,
61    /// node_id → set of outgoing edge_ids
62    pub outgoing: HashMap<String, HashSet<String>>,
63    /// node_id → set of incoming edge_ids
64    pub incoming: HashMap<String, HashSet<String>>,
65    /// node_type → set of node_ids (type index)
66    pub by_type: HashMap<String, HashSet<String>>,
67    /// The ontology (for validation during materialization)
68    pub ontology: Ontology,
69    /// R-02: entries that failed ontology validation during apply().
70    /// These entries exist in the oplog (for CRDT convergence) but are
71    /// invisible in the materialized graph. Grow-only within a single
72    /// materialization pass. Cleared and rebuilt on `rebuild()` — this
73    /// allows previously-quarantined entries to be re-evaluated when the
74    /// ontology evolves (e.g., after ExtendOntology arrives via sync).
75    pub quarantined: HashSet<Hash>,
76}
77
78impl MaterializedGraph {
79    /// Create an empty materialized graph with the given ontology.
80    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    /// Apply a single entry to the graph (incremental materialization).
93    ///
94    /// R-02: Validates AddNode/AddEdge payloads against the ontology.
95    /// Invalid entries are quarantined (added to `self.quarantined`) and
96    /// skipped for materialization. They remain in the oplog for CRDT
97    /// convergence — quarantine is a graph-layer concern, not an oplog concern.
98    pub fn apply(&mut self, entry: &Entry) {
99        match &entry.payload {
100            GraphOp::Checkpoint { ops, op_clocks, .. } => {
101                // R-08: Replay synthetic ops to restore graph state.
102                // Bug 6 fix: use per-op clocks (preserves LWW metadata).
103                for (i, op) in ops.iter().enumerate() {
104                    // Bug 14 fix: compaction folds every ExtendOntology into the
105                    // checkpoint's inner DefineOntology. Apply it — otherwise
106                    // extension-typed entities fail validation and quarantine on
107                    // replay, and a reopened store materializes without them.
108                    // The oplog is authoritative; a declared ontology only seeds
109                    // new stores.
110                    if let GraphOp::DefineOntology { ontology } = op {
111                        self.ontology = ontology.clone();
112                        continue;
113                    }
114                    let clock = if i < op_clocks.len() {
115                        LamportClock::with_values(&entry.author, op_clocks[i].0, op_clocks[i].1)
116                    } else {
117                        entry.clock.clone() // fallback for old checkpoints without op_clocks
118                    };
119                    let synthetic = Entry::new(op.clone(), vec![], vec![], clock, &entry.author);
120                    self.apply(&synthetic);
121                }
122            }
123            GraphOp::DefineOntology { .. } => {
124                // Genesis — nothing to materialize.
125            }
126            GraphOp::ExtendOntology { extension } => {
127                if let Err(_e) = self.ontology.merge_extension(extension) {
128                    self.quarantined.insert(entry.hash);
129                }
130            }
131            GraphOp::AddNode {
132                node_id,
133                node_type,
134                subtype,
135                label,
136                properties,
137            } => {
138                // R-02: validate against ontology, quarantine if invalid
139                if let Err(_e) =
140                    self.ontology
141                        .validate_node(node_type, subtype.as_deref(), properties)
142                {
143                    self.quarantined.insert(entry.hash);
144                    return;
145                }
146                self.apply_add_node(
147                    node_id,
148                    node_type,
149                    subtype.as_deref(),
150                    label,
151                    properties,
152                    &entry.clock,
153                );
154            }
155            GraphOp::AddEdge {
156                edge_id,
157                edge_type,
158                source_id,
159                target_id,
160                properties,
161            } => {
162                // R-02: validate edge type exists.
163                if !self.ontology.edge_types.contains_key(edge_type.as_str()) {
164                    self.quarantined.insert(entry.hash);
165                    return;
166                }
167                // Bug 13 fix: validate source/target type constraints when both nodes
168                // are materialized. If one is missing (out-of-order sync), skip —
169                // validation happens on rebuild.
170                if let (Some(src), Some(tgt)) = (
171                    self.nodes.get(source_id.as_str()),
172                    self.nodes.get(target_id.as_str()),
173                ) {
174                    if self
175                        .ontology
176                        .validate_edge(edge_type, &src.node_type, &tgt.node_type, properties)
177                        .is_err()
178                    {
179                        self.quarantined.insert(entry.hash);
180                        return;
181                    }
182                }
183                self.apply_add_edge(
184                    edge_id,
185                    edge_type,
186                    source_id,
187                    target_id,
188                    properties,
189                    &entry.clock,
190                );
191            }
192            GraphOp::UpdateProperty {
193                entity_id,
194                key,
195                value,
196            } => {
197                self.apply_update_property(entity_id, key, value, &entry.clock);
198            }
199            GraphOp::RemoveNode { node_id } => {
200                self.apply_remove_node(node_id, &entry.clock);
201            }
202            GraphOp::RemoveEdge { edge_id } => {
203                self.apply_remove_edge(edge_id, &entry.clock);
204            }
205            GraphOp::DefineLens { .. } => {
206                // Reserved. No materialization — lenses are metadata, not graph state.
207            }
208        }
209    }
210
211    /// Apply a sequence of entries (full rematerialization from op log).
212    pub fn apply_all(&mut self, entries: &[&Entry]) {
213        for entry in entries {
214            self.apply(entry);
215        }
216    }
217
218    /// Rebuild from scratch: clear everything and replay all entries.
219    pub fn rebuild(&mut self, entries: &[&Entry]) {
220        self.nodes.clear();
221        self.edges.clear();
222        self.outgoing.clear();
223        self.incoming.clear();
224        self.by_type.clear();
225        self.quarantined.clear();
226        self.apply_all(entries);
227    }
228
229    // -- Queries --
230
231    /// Get a node by ID (returns None if not found or tombstoned).
232    pub fn get_node(&self, node_id: &str) -> Option<&Node> {
233        self.nodes.get(node_id).filter(|n| !n.tombstoned)
234    }
235
236    /// Get an edge by ID (returns None if not found or tombstoned).
237    pub fn get_edge(&self, edge_id: &str) -> Option<&Edge> {
238        self.edges.get(edge_id).filter(|e| !e.tombstoned)
239    }
240
241    /// Query all live nodes of a given type, including descendants (RDFS rdfs9).
242    /// If "entity" has children "server" and "project", querying "entity" returns all three.
243    pub fn nodes_by_type(&self, node_type: &str) -> Vec<&Node> {
244        let mut types = vec![node_type.to_string()];
245        types.extend(
246            self.ontology
247                .descendants(node_type)
248                .into_iter()
249                .map(|s| s.to_string()),
250        );
251        types
252            .iter()
253            .flat_map(|t| self.by_type.get(t.as_str()))
254            .flatten()
255            .filter_map(|id| self.get_node(id))
256            .collect()
257    }
258
259    /// Query all live nodes of a given subtype.
260    pub fn nodes_by_subtype(&self, subtype: &str) -> Vec<&Node> {
261        self.nodes
262            .values()
263            .filter(|n| !n.tombstoned && n.subtype.as_deref() == Some(subtype))
264            .collect()
265    }
266
267    /// Query nodes by a property value.
268    pub fn nodes_by_property(&self, key: &str, value: &Value) -> Vec<&Node> {
269        self.nodes
270            .values()
271            .filter(|n| !n.tombstoned && n.properties.get(key) == Some(value))
272            .collect()
273    }
274
275    /// Get outgoing edges for a node (only live edges with live endpoints).
276    pub fn outgoing_edges(&self, node_id: &str) -> Vec<&Edge> {
277        match self.outgoing.get(node_id) {
278            Some(edge_ids) => edge_ids
279                .iter()
280                .filter_map(|eid| self.get_edge(eid))
281                .filter(|e| self.is_node_live(&e.target_id))
282                .collect(),
283            None => vec![],
284        }
285    }
286
287    /// Get incoming edges for a node (only live edges with live endpoints).
288    pub fn incoming_edges(&self, node_id: &str) -> Vec<&Edge> {
289        match self.incoming.get(node_id) {
290            Some(edge_ids) => edge_ids
291                .iter()
292                .filter_map(|eid| self.get_edge(eid))
293                .filter(|e| self.is_node_live(&e.source_id))
294                .collect(),
295            None => vec![],
296        }
297    }
298
299    /// Approximate heap memory used by the materialized graph (bytes).
300    /// Uses fixed overhead estimates per node/edge. Does not account for heap
301    /// allocations behind String/Vec in property values or allocator fragmentation.
302    /// Actual memory may be 2-3x higher for string-heavy graphs.
303    pub fn estimated_memory_bytes(&self) -> usize {
304        let mut total = 0;
305        // Nodes: id string + type string + label + properties + clocks + overhead
306        for node in self.nodes.values() {
307            total += node.node_id.len() + node.node_type.len() + node.label.len();
308            total += node.subtype.as_ref().map_or(0, |s| s.len());
309            // Properties: key + estimated value size + clock per property
310            for (k, v) in &node.properties {
311                total += k.len() + std::mem::size_of_val(v) + 48; // key + value + clock overhead
312            }
313            total += 128; // fixed struct overhead (clocks, bools, HashMap shells)
314        }
315        // Edges: similar structure
316        for edge in self.edges.values() {
317            total += edge.edge_id.len() + edge.edge_type.len();
318            total += edge.source_id.len() + edge.target_id.len();
319            for (k, v) in &edge.properties {
320                total += k.len() + std::mem::size_of_val(v) + 48;
321            }
322            total += 128;
323        }
324        // Adjacency indexes: outgoing + incoming (id strings + HashSet overhead)
325        for (k, set) in &self.outgoing {
326            total += k.len() + set.len() * 32;
327        }
328        for (k, set) in &self.incoming {
329            total += k.len() + set.len() * 32;
330        }
331        // Type index
332        for (k, set) in &self.by_type {
333            total += k.len() + set.len() * 32;
334        }
335        // Quarantine set
336        total += self.quarantined.len() * 48;
337        total
338    }
339
340    /// All live nodes.
341    pub fn all_nodes(&self) -> Vec<&Node> {
342        self.nodes.values().filter(|n| !n.tombstoned).collect()
343    }
344
345    /// All live edges (with live endpoints).
346    pub fn all_edges(&self) -> Vec<&Edge> {
347        self.edges
348            .values()
349            .filter(|e| {
350                !e.tombstoned && self.is_node_live(&e.source_id) && self.is_node_live(&e.target_id)
351            })
352            .collect()
353    }
354
355    /// Neighbors of a node (connected via outgoing edges).
356    pub fn neighbors(&self, node_id: &str) -> Vec<&str> {
357        self.outgoing_edges(node_id)
358            .iter()
359            .map(|e| e.target_id.as_str())
360            .collect()
361    }
362
363    /// Reverse neighbors (connected via incoming edges).
364    pub fn reverse_neighbors(&self, node_id: &str) -> Vec<&str> {
365        self.incoming_edges(node_id)
366            .iter()
367            .map(|e| e.source_id.as_str())
368            .collect()
369    }
370
371    // -- CRDT application helpers --
372
373    fn apply_add_node(
374        &mut self,
375        node_id: &str,
376        node_type: &str,
377        subtype: Option<&str>,
378        label: &str,
379        properties: &BTreeMap<String, Value>,
380        clock: &LamportClock,
381    ) {
382        if let Some(existing) = self.nodes.get_mut(node_id) {
383            // Add-wins: always resurrect from tombstone.
384            existing.tombstoned = false;
385            // Track the latest add clock for add-wins semantics.
386            if clock_wins(clock, &existing.last_add_clock) {
387                existing.last_add_clock = clock.clone();
388            }
389            // LWW merge for label, subtype, and properties.
390            if clock_wins(clock, &existing.last_clock) {
391                existing.label = label.to_string();
392                existing.subtype = subtype.map(|s| s.to_string());
393                existing.last_clock = clock.clone();
394            }
395            merge_properties_lww(
396                &mut existing.properties,
397                &mut existing.property_clocks,
398                properties,
399                clock,
400            );
401        } else {
402            let property_clocks: HashMap<String, LamportClock> = properties
403                .keys()
404                .map(|k| (k.clone(), clock.clone()))
405                .collect();
406            let node = Node {
407                node_id: node_id.to_string(),
408                node_type: node_type.to_string(),
409                subtype: subtype.map(|s| s.to_string()),
410                label: label.to_string(),
411                properties: properties.clone(),
412                property_clocks,
413                last_clock: clock.clone(),
414                last_add_clock: clock.clone(),
415                tombstoned: false,
416            };
417            self.by_type
418                .entry(node_type.to_string())
419                .or_default()
420                .insert(node_id.to_string());
421            self.nodes.insert(node_id.to_string(), node);
422        }
423    }
424
425    fn apply_add_edge(
426        &mut self,
427        edge_id: &str,
428        edge_type: &str,
429        source_id: &str,
430        target_id: &str,
431        properties: &BTreeMap<String, Value>,
432        clock: &LamportClock,
433    ) {
434        if let Some(existing) = self.edges.get_mut(edge_id) {
435            // Add-wins: always resurrect if tombstoned.
436            existing.tombstoned = false;
437            if clock_wins(clock, &existing.last_add_clock) {
438                existing.last_add_clock = clock.clone();
439            }
440            if clock_wins(clock, &existing.last_clock) {
441                existing.last_clock = clock.clone();
442            }
443            merge_properties_lww(
444                &mut existing.properties,
445                &mut existing.property_clocks,
446                properties,
447                clock,
448            );
449        } else {
450            let property_clocks: HashMap<String, LamportClock> = properties
451                .keys()
452                .map(|k| (k.clone(), clock.clone()))
453                .collect();
454            let edge = Edge {
455                edge_id: edge_id.to_string(),
456                edge_type: edge_type.to_string(),
457                source_id: source_id.to_string(),
458                target_id: target_id.to_string(),
459                properties: properties.clone(),
460                property_clocks,
461                last_clock: clock.clone(),
462                last_add_clock: clock.clone(),
463                tombstoned: false,
464            };
465            self.outgoing
466                .entry(source_id.to_string())
467                .or_default()
468                .insert(edge_id.to_string());
469            self.incoming
470                .entry(target_id.to_string())
471                .or_default()
472                .insert(edge_id.to_string());
473            self.edges.insert(edge_id.to_string(), edge);
474        }
475    }
476
477    fn apply_update_property(
478        &mut self,
479        entity_id: &str,
480        key: &str,
481        value: &Value,
482        clock: &LamportClock,
483    ) {
484        // Try node first, then edge. Per-property LWW: each key competes
485        // only with other writes to the same key, not the entire entity.
486        if let Some(node) = self.nodes.get_mut(entity_id) {
487            let dominated = node
488                .property_clocks
489                .get(key)
490                .map(|c| clock_wins(clock, c))
491                .unwrap_or(true);
492            if dominated {
493                node.properties.insert(key.to_string(), value.clone());
494                node.property_clocks.insert(key.to_string(), clock.clone());
495            }
496            // Update entity-level clock for add-wins tracking.
497            if clock_wins(clock, &node.last_clock) {
498                node.last_clock = clock.clone();
499            }
500        } else if let Some(edge) = self.edges.get_mut(entity_id) {
501            let dominated = edge
502                .property_clocks
503                .get(key)
504                .map(|c| clock_wins(clock, c))
505                .unwrap_or(true);
506            if dominated {
507                edge.properties.insert(key.to_string(), value.clone());
508                edge.property_clocks.insert(key.to_string(), clock.clone());
509            }
510            if clock_wins(clock, &edge.last_clock) {
511                edge.last_clock = clock.clone();
512            }
513        }
514        // If entity not found, silently ignore (may arrive out of order in sync).
515    }
516
517    fn apply_remove_node(&mut self, node_id: &str, clock: &LamportClock) {
518        if let Some(node) = self.nodes.get_mut(node_id) {
519            // Add-wins: only tombstone if the remove clock is strictly greater
520            // than the last add clock. If a concurrent (or later) add exists,
521            // the node stays alive.
522            if clock_wins(clock, &node.last_add_clock) {
523                node.tombstoned = true;
524                node.last_clock = clock.clone();
525            }
526        }
527        // Tombstoning a node doesn't physically remove edges — they just become
528        // invisible via is_node_live() checks in queries.
529    }
530
531    fn apply_remove_edge(&mut self, edge_id: &str, clock: &LamportClock) {
532        if let Some(edge) = self.edges.get_mut(edge_id) {
533            // Add-wins: only tombstone if remove clock > last add clock.
534            if clock_wins(clock, &edge.last_add_clock) {
535                edge.tombstoned = true;
536                edge.last_clock = clock.clone();
537            }
538        }
539    }
540
541    fn is_node_live(&self, node_id: &str) -> bool {
542        self.nodes
543            .get(node_id)
544            .map(|n| !n.tombstoned)
545            .unwrap_or(false)
546    }
547}
548
549/// Per-property LWW merge: each property from `new_props` competes with
550/// existing properties. Higher clock wins per key.
551fn merge_properties_lww(
552    existing_props: &mut BTreeMap<String, Value>,
553    existing_clocks: &mut HashMap<String, LamportClock>,
554    new_props: &BTreeMap<String, Value>,
555    clock: &LamportClock,
556) {
557    for (k, v) in new_props {
558        let dominated = existing_clocks
559            .get(k)
560            .map(|c| clock_wins(clock, c))
561            .unwrap_or(true);
562        if dominated {
563            existing_props.insert(k.clone(), v.clone());
564            existing_clocks.insert(k.clone(), clock.clone());
565        }
566    }
567}
568
569/// LWW comparison: returns true if `new_clock` wins over `existing_clock`.
570/// Uses HybridClock total ordering: (physical_ms, logical, id).
571fn clock_wins(new_clock: &LamportClock, existing_clock: &LamportClock) -> bool {
572    new_clock.cmp_order(existing_clock) == std::cmp::Ordering::Greater
573}
574
575#[cfg(test)]
576mod tests {
577    use super::*;
578    use crate::entry::Entry;
579    use crate::ontology::{EdgeTypeDef, NodeTypeDef};
580
581    fn test_ontology() -> Ontology {
582        Ontology {
583            node_types: BTreeMap::from([
584                (
585                    "entity".into(),
586                    NodeTypeDef {
587                        description: None,
588                        properties: BTreeMap::new(),
589                        subtypes: None,
590                        parent_type: None,
591                    },
592                ),
593                (
594                    "signal".into(),
595                    NodeTypeDef {
596                        description: None,
597                        properties: BTreeMap::new(),
598                        subtypes: None,
599                        parent_type: None,
600                    },
601                ),
602            ]),
603            edge_types: BTreeMap::from([
604                (
605                    "RUNS_ON".into(),
606                    EdgeTypeDef {
607                        description: None,
608                        source_types: vec!["entity".into()],
609                        target_types: vec!["entity".into()],
610                        properties: BTreeMap::new(),
611                    },
612                ),
613                (
614                    "OBSERVES".into(),
615                    EdgeTypeDef {
616                        description: None,
617                        source_types: vec!["signal".into()],
618                        target_types: vec!["entity".into()],
619                        properties: BTreeMap::new(),
620                    },
621                ),
622            ]),
623        }
624    }
625
626    fn make_entry(op: GraphOp, clock_time: u64, author: &str) -> Entry {
627        Entry::new(
628            op,
629            vec![],
630            vec![],
631            LamportClock::with_values(author, clock_time, 0),
632            author,
633        )
634    }
635
636    // -- test_graph.rs spec from docs/silk.md --
637
638    #[test]
639    fn add_node_appears_in_query() {
640        let mut g = MaterializedGraph::new(test_ontology());
641        let entry = make_entry(
642            GraphOp::AddNode {
643                node_id: "server-1".into(),
644                node_type: "entity".into(),
645                label: "Server 1".into(),
646                properties: BTreeMap::from([("ip".into(), Value::String("10.0.0.1".into()))]),
647                subtype: None,
648            },
649            1,
650            "inst-a",
651        );
652        g.apply(&entry);
653
654        let node = g.get_node("server-1").unwrap();
655        assert_eq!(node.node_type, "entity");
656        assert_eq!(node.label, "Server 1");
657        assert_eq!(
658            node.properties.get("ip"),
659            Some(&Value::String("10.0.0.1".into()))
660        );
661    }
662
663    #[test]
664    fn add_edge_creates_adjacency() {
665        let mut g = MaterializedGraph::new(test_ontology());
666        g.apply(&make_entry(
667            GraphOp::AddNode {
668                node_id: "svc".into(),
669                node_type: "entity".into(),
670                label: "svc".into(),
671                properties: BTreeMap::new(),
672                subtype: None,
673            },
674            1,
675            "inst-a",
676        ));
677        g.apply(&make_entry(
678            GraphOp::AddNode {
679                node_id: "srv".into(),
680                node_type: "entity".into(),
681                label: "srv".into(),
682                properties: BTreeMap::new(),
683                subtype: None,
684            },
685            2,
686            "inst-a",
687        ));
688        g.apply(&make_entry(
689            GraphOp::AddEdge {
690                edge_id: "e1".into(),
691                edge_type: "RUNS_ON".into(),
692                source_id: "svc".into(),
693                target_id: "srv".into(),
694                properties: BTreeMap::new(),
695            },
696            3,
697            "inst-a",
698        ));
699
700        // Both endpoints know about the edge.
701        let out = g.outgoing_edges("svc");
702        assert_eq!(out.len(), 1);
703        assert_eq!(out[0].target_id, "srv");
704
705        let inc = g.incoming_edges("srv");
706        assert_eq!(inc.len(), 1);
707        assert_eq!(inc[0].source_id, "svc");
708
709        assert_eq!(g.neighbors("svc"), vec!["srv"]);
710    }
711
712    #[test]
713    fn update_property_reflected() {
714        let mut g = MaterializedGraph::new(test_ontology());
715        g.apply(&make_entry(
716            GraphOp::AddNode {
717                node_id: "s1".into(),
718                node_type: "entity".into(),
719                label: "s1".into(),
720                properties: BTreeMap::new(),
721                subtype: None,
722            },
723            1,
724            "inst-a",
725        ));
726        g.apply(&make_entry(
727            GraphOp::UpdateProperty {
728                entity_id: "s1".into(),
729                key: "cpu".into(),
730                value: Value::Float(85.5),
731            },
732            2,
733            "inst-a",
734        ));
735
736        let node = g.get_node("s1").unwrap();
737        assert_eq!(node.properties.get("cpu"), Some(&Value::Float(85.5)));
738    }
739
740    #[test]
741    fn remove_node_cascades_edges() {
742        let mut g = MaterializedGraph::new(test_ontology());
743        g.apply(&make_entry(
744            GraphOp::AddNode {
745                node_id: "a".into(),
746                node_type: "entity".into(),
747                label: "a".into(),
748                properties: BTreeMap::new(),
749                subtype: None,
750            },
751            1,
752            "inst-a",
753        ));
754        g.apply(&make_entry(
755            GraphOp::AddNode {
756                node_id: "b".into(),
757                node_type: "entity".into(),
758                label: "b".into(),
759                properties: BTreeMap::new(),
760                subtype: None,
761            },
762            2,
763            "inst-a",
764        ));
765        g.apply(&make_entry(
766            GraphOp::AddEdge {
767                edge_id: "e1".into(),
768                edge_type: "RUNS_ON".into(),
769                source_id: "a".into(),
770                target_id: "b".into(),
771                properties: BTreeMap::new(),
772            },
773            3,
774            "inst-a",
775        ));
776        assert_eq!(g.all_edges().len(), 1);
777
778        // Remove node 'b' — edge becomes invisible (dangling target).
779        g.apply(&make_entry(
780            GraphOp::RemoveNode {
781                node_id: "b".into(),
782            },
783            4,
784            "inst-a",
785        ));
786        assert!(g.get_node("b").is_none());
787        // Edge still exists but not returned by all_edges (target tombstoned).
788        assert_eq!(g.all_edges().len(), 0);
789        // Outgoing from 'a' also filters out dangling edges.
790        assert_eq!(g.outgoing_edges("a").len(), 0);
791    }
792
793    #[test]
794    fn remove_edge_preserves_nodes() {
795        let mut g = MaterializedGraph::new(test_ontology());
796        g.apply(&make_entry(
797            GraphOp::AddNode {
798                node_id: "a".into(),
799                node_type: "entity".into(),
800                label: "a".into(),
801                properties: BTreeMap::new(),
802                subtype: None,
803            },
804            1,
805            "inst-a",
806        ));
807        g.apply(&make_entry(
808            GraphOp::AddNode {
809                node_id: "b".into(),
810                node_type: "entity".into(),
811                label: "b".into(),
812                properties: BTreeMap::new(),
813                subtype: None,
814            },
815            2,
816            "inst-a",
817        ));
818        g.apply(&make_entry(
819            GraphOp::AddEdge {
820                edge_id: "e1".into(),
821                edge_type: "RUNS_ON".into(),
822                source_id: "a".into(),
823                target_id: "b".into(),
824                properties: BTreeMap::new(),
825            },
826            3,
827            "inst-a",
828        ));
829        g.apply(&make_entry(
830            GraphOp::RemoveEdge {
831                edge_id: "e1".into(),
832            },
833            4,
834            "inst-a",
835        ));
836
837        // Nodes still exist.
838        assert!(g.get_node("a").is_some());
839        assert!(g.get_node("b").is_some());
840        // Edge is gone.
841        assert!(g.get_edge("e1").is_none());
842        assert_eq!(g.all_edges().len(), 0);
843    }
844
845    #[test]
846    fn query_by_type_filters() {
847        let mut g = MaterializedGraph::new(test_ontology());
848        g.apply(&make_entry(
849            GraphOp::AddNode {
850                node_id: "s1".into(),
851                node_type: "entity".into(),
852                label: "s1".into(),
853                properties: BTreeMap::new(),
854                subtype: None,
855            },
856            1,
857            "inst-a",
858        ));
859        g.apply(&make_entry(
860            GraphOp::AddNode {
861                node_id: "s2".into(),
862                node_type: "entity".into(),
863                label: "s2".into(),
864                properties: BTreeMap::new(),
865                subtype: None,
866            },
867            2,
868            "inst-a",
869        ));
870        g.apply(&make_entry(
871            GraphOp::AddNode {
872                node_id: "alert".into(),
873                node_type: "signal".into(),
874                label: "alert".into(),
875                properties: BTreeMap::new(),
876                subtype: None,
877            },
878            3,
879            "inst-a",
880        ));
881
882        let entities = g.nodes_by_type("entity");
883        assert_eq!(entities.len(), 2);
884        let signals = g.nodes_by_type("signal");
885        assert_eq!(signals.len(), 1);
886        assert_eq!(signals[0].node_id, "alert");
887    }
888
889    #[test]
890    fn query_by_property_filters() {
891        let mut g = MaterializedGraph::new(test_ontology());
892        g.apply(&make_entry(
893            GraphOp::AddNode {
894                node_id: "s1".into(),
895                node_type: "entity".into(),
896                label: "s1".into(),
897                properties: BTreeMap::from([("status".into(), Value::String("alive".into()))]),
898                subtype: None,
899            },
900            1,
901            "inst-a",
902        ));
903        g.apply(&make_entry(
904            GraphOp::AddNode {
905                node_id: "s2".into(),
906                node_type: "entity".into(),
907                label: "s2".into(),
908                properties: BTreeMap::from([("status".into(), Value::String("dead".into()))]),
909                subtype: None,
910            },
911            2,
912            "inst-a",
913        ));
914
915        let alive = g.nodes_by_property("status", &Value::String("alive".into()));
916        assert_eq!(alive.len(), 1);
917        assert_eq!(alive[0].node_id, "s1");
918    }
919
920    #[test]
921    fn materialization_from_empty() {
922        // Build graph incrementally.
923        let mut g1 = MaterializedGraph::new(test_ontology());
924        let entries = vec![
925            make_entry(
926                GraphOp::DefineOntology {
927                    ontology: test_ontology(),
928                },
929                0,
930                "inst-a",
931            ),
932            make_entry(
933                GraphOp::AddNode {
934                    node_id: "a".into(),
935                    node_type: "entity".into(),
936                    label: "a".into(),
937                    properties: BTreeMap::new(),
938                    subtype: None,
939                },
940                1,
941                "inst-a",
942            ),
943            make_entry(
944                GraphOp::AddNode {
945                    node_id: "b".into(),
946                    node_type: "entity".into(),
947                    label: "b".into(),
948                    properties: BTreeMap::new(),
949                    subtype: None,
950                },
951                2,
952                "inst-a",
953            ),
954            make_entry(
955                GraphOp::AddEdge {
956                    edge_id: "e1".into(),
957                    edge_type: "RUNS_ON".into(),
958                    source_id: "a".into(),
959                    target_id: "b".into(),
960                    properties: BTreeMap::new(),
961                },
962                3,
963                "inst-a",
964            ),
965        ];
966        for e in &entries {
967            g1.apply(e);
968        }
969
970        // Rebuild from scratch.
971        let mut g2 = MaterializedGraph::new(test_ontology());
972        let refs: Vec<&Entry> = entries.iter().collect();
973        g2.rebuild(&refs);
974
975        // Same result.
976        assert_eq!(g1.all_nodes().len(), g2.all_nodes().len());
977        assert_eq!(g1.all_edges().len(), g2.all_edges().len());
978        for node in g1.all_nodes() {
979            let n2 = g2.get_node(&node.node_id).unwrap();
980            assert_eq!(node.node_type, n2.node_type);
981            assert_eq!(node.properties, n2.properties);
982        }
983    }
984
985    #[test]
986    fn incremental_equals_full() {
987        let entries = vec![
988            make_entry(
989                GraphOp::DefineOntology {
990                    ontology: test_ontology(),
991                },
992                0,
993                "inst-a",
994            ),
995            make_entry(
996                GraphOp::AddNode {
997                    node_id: "a".into(),
998                    node_type: "entity".into(),
999                    label: "a".into(),
1000                    properties: BTreeMap::from([("x".into(), Value::Int(1))]),
1001                    subtype: None,
1002                },
1003                1,
1004                "inst-a",
1005            ),
1006            make_entry(
1007                GraphOp::UpdateProperty {
1008                    entity_id: "a".into(),
1009                    key: "x".into(),
1010                    value: Value::Int(2),
1011                },
1012                2,
1013                "inst-a",
1014            ),
1015            make_entry(
1016                GraphOp::AddNode {
1017                    node_id: "b".into(),
1018                    node_type: "entity".into(),
1019                    label: "b".into(),
1020                    properties: BTreeMap::new(),
1021                    subtype: None,
1022                },
1023                3,
1024                "inst-a",
1025            ),
1026            make_entry(
1027                GraphOp::AddEdge {
1028                    edge_id: "e1".into(),
1029                    edge_type: "RUNS_ON".into(),
1030                    source_id: "a".into(),
1031                    target_id: "b".into(),
1032                    properties: BTreeMap::new(),
1033                },
1034                4,
1035                "inst-a",
1036            ),
1037            make_entry(
1038                GraphOp::RemoveEdge {
1039                    edge_id: "e1".into(),
1040                },
1041                5,
1042                "inst-a",
1043            ),
1044        ];
1045
1046        // Incremental.
1047        let mut g_inc = MaterializedGraph::new(test_ontology());
1048        for e in &entries {
1049            g_inc.apply(e);
1050        }
1051
1052        // Full replay.
1053        let mut g_full = MaterializedGraph::new(test_ontology());
1054        let refs: Vec<&Entry> = entries.iter().collect();
1055        g_full.rebuild(&refs);
1056
1057        // Property should be 2 (updated).
1058        assert_eq!(
1059            g_inc.get_node("a").unwrap().properties.get("x"),
1060            Some(&Value::Int(2))
1061        );
1062        assert_eq!(
1063            g_full.get_node("a").unwrap().properties.get("x"),
1064            Some(&Value::Int(2))
1065        );
1066        // Edge should be removed.
1067        assert_eq!(g_inc.all_edges().len(), 0);
1068        assert_eq!(g_full.all_edges().len(), 0);
1069    }
1070
1071    #[test]
1072    fn lww_concurrent_property_update() {
1073        // Two instances update the same property — higher clock wins.
1074        let mut g = MaterializedGraph::new(test_ontology());
1075        g.apply(&make_entry(
1076            GraphOp::AddNode {
1077                node_id: "s1".into(),
1078                node_type: "entity".into(),
1079                label: "s1".into(),
1080                properties: BTreeMap::new(),
1081                subtype: None,
1082            },
1083            1,
1084            "inst-a",
1085        ));
1086        // inst-a sets status=alive at time 2
1087        g.apply(&make_entry(
1088            GraphOp::UpdateProperty {
1089                entity_id: "s1".into(),
1090                key: "status".into(),
1091                value: Value::String("alive".into()),
1092            },
1093            2,
1094            "inst-a",
1095        ));
1096        // inst-b sets status=dead at time 3 — wins (higher clock)
1097        g.apply(&make_entry(
1098            GraphOp::UpdateProperty {
1099                entity_id: "s1".into(),
1100                key: "status".into(),
1101                value: Value::String("dead".into()),
1102            },
1103            3,
1104            "inst-b",
1105        ));
1106        assert_eq!(
1107            g.get_node("s1").unwrap().properties.get("status"),
1108            Some(&Value::String("dead".into()))
1109        );
1110    }
1111
1112    #[test]
1113    fn lww_tiebreak_by_instance_id() {
1114        // Same clock time — higher instance ID wins.
1115        let mut g = MaterializedGraph::new(test_ontology());
1116        g.apply(&make_entry(
1117            GraphOp::AddNode {
1118                node_id: "s1".into(),
1119                node_type: "entity".into(),
1120                label: "s1".into(),
1121                properties: BTreeMap::new(),
1122                subtype: None,
1123            },
1124            1,
1125            "inst-a",
1126        ));
1127        // Both at physical_ms=5, logical=0. Lower id wins → "inst-a" wins.
1128        g.apply(&make_entry(
1129            GraphOp::UpdateProperty {
1130                entity_id: "s1".into(),
1131                key: "x".into(),
1132                value: Value::Int(1),
1133            },
1134            5,
1135            "inst-a",
1136        ));
1137        g.apply(&make_entry(
1138            GraphOp::UpdateProperty {
1139                entity_id: "s1".into(),
1140                key: "x".into(),
1141                value: Value::Int(2),
1142            },
1143            5,
1144            "inst-b",
1145        ));
1146        // inst-a has lower id → wins the tiebreak → value stays Int(1).
1147        assert_eq!(
1148            g.get_node("s1").unwrap().properties.get("x"),
1149            Some(&Value::Int(1))
1150        );
1151    }
1152
1153    #[test]
1154    fn lww_per_property_concurrent_different_keys() {
1155        // Two instances concurrently update DIFFERENT properties at the same
1156        // clock time. Both updates must be accepted — they don't conflict.
1157        // This requires per-property LWW, not node-level LWW.
1158        let mut g = MaterializedGraph::new(test_ontology());
1159        g.apply(&make_entry(
1160            GraphOp::AddNode {
1161                node_id: "s1".into(),
1162                node_type: "entity".into(),
1163                label: "s1".into(),
1164                properties: BTreeMap::from([
1165                    ("x".into(), Value::Int(0)),
1166                    ("y".into(), Value::Int(0)),
1167                ]),
1168                subtype: None,
1169            },
1170            1,
1171            "inst-a",
1172        ));
1173        // inst-a updates "x" at time 3
1174        g.apply(&make_entry(
1175            GraphOp::UpdateProperty {
1176                entity_id: "s1".into(),
1177                key: "x".into(),
1178                value: Value::Int(42),
1179            },
1180            3,
1181            "inst-a",
1182        ));
1183        // inst-b updates "y" at time 3 (concurrent, different property)
1184        g.apply(&make_entry(
1185            GraphOp::UpdateProperty {
1186                entity_id: "s1".into(),
1187                key: "y".into(),
1188                value: Value::Int(99),
1189            },
1190            3,
1191            "inst-b",
1192        ));
1193
1194        let node = g.get_node("s1").unwrap();
1195        // Both updates must be applied — no conflict.
1196        assert_eq!(
1197            node.properties.get("x"),
1198            Some(&Value::Int(42)),
1199            "update to 'x' must not be rejected by concurrent update to 'y'"
1200        );
1201        assert_eq!(
1202            node.properties.get("y"),
1203            Some(&Value::Int(99)),
1204            "update to 'y' must not be rejected by concurrent update to 'x'"
1205        );
1206    }
1207
1208    #[test]
1209    fn lww_per_property_order_independent() {
1210        // Same scenario but applied in reverse order — result must be identical.
1211        let mut g = MaterializedGraph::new(test_ontology());
1212        g.apply(&make_entry(
1213            GraphOp::AddNode {
1214                node_id: "s1".into(),
1215                node_type: "entity".into(),
1216                label: "s1".into(),
1217                properties: BTreeMap::from([
1218                    ("x".into(), Value::Int(0)),
1219                    ("y".into(), Value::Int(0)),
1220                ]),
1221                subtype: None,
1222            },
1223            1,
1224            "inst-a",
1225        ));
1226        // Apply inst-b first this time
1227        g.apply(&make_entry(
1228            GraphOp::UpdateProperty {
1229                entity_id: "s1".into(),
1230                key: "y".into(),
1231                value: Value::Int(99),
1232            },
1233            3,
1234            "inst-b",
1235        ));
1236        g.apply(&make_entry(
1237            GraphOp::UpdateProperty {
1238                entity_id: "s1".into(),
1239                key: "x".into(),
1240                value: Value::Int(42),
1241            },
1242            3,
1243            "inst-a",
1244        ));
1245
1246        let node = g.get_node("s1").unwrap();
1247        assert_eq!(node.properties.get("x"), Some(&Value::Int(42)));
1248        assert_eq!(node.properties.get("y"), Some(&Value::Int(99)));
1249    }
1250
1251    #[test]
1252    fn add_wins_over_remove() {
1253        // Concurrent add + remove → node should exist (add-wins).
1254        let mut g = MaterializedGraph::new(test_ontology());
1255        g.apply(&make_entry(
1256            GraphOp::AddNode {
1257                node_id: "s1".into(),
1258                node_type: "entity".into(),
1259                label: "s1".into(),
1260                properties: BTreeMap::new(),
1261                subtype: None,
1262            },
1263            1,
1264            "inst-a",
1265        ));
1266        // Remove at time 2.
1267        g.apply(&make_entry(
1268            GraphOp::RemoveNode {
1269                node_id: "s1".into(),
1270            },
1271            2,
1272            "inst-a",
1273        ));
1274        assert!(g.get_node("s1").is_none());
1275
1276        // Re-add at time 3 (add-wins — resurrects).
1277        g.apply(&make_entry(
1278            GraphOp::AddNode {
1279                node_id: "s1".into(),
1280                node_type: "entity".into(),
1281                label: "s1 v2".into(),
1282                properties: BTreeMap::new(),
1283                subtype: None,
1284            },
1285            3,
1286            "inst-b",
1287        ));
1288        let node = g.get_node("s1").unwrap();
1289        assert_eq!(node.label, "s1 v2");
1290        assert!(!node.tombstoned);
1291    }
1292
1293    /// Bug 14: a checkpoint's inner DefineOntology carries the merged ontology
1294    /// (compaction folds ExtendOntology entries into it). Replay must apply it,
1295    /// or entities typed by an extension quarantine and vanish from the graph.
1296    #[test]
1297    fn checkpoint_replay_applies_inner_define_ontology() {
1298        // Base ontology: "entity" only — no "signal".
1299        let mut base = test_ontology();
1300        base.node_types.remove("signal");
1301        base.edge_types.remove("OBSERVES");
1302        let mut g = MaterializedGraph::new(base);
1303
1304        let checkpoint = make_entry(
1305            GraphOp::Checkpoint {
1306                ops: vec![
1307                    GraphOp::DefineOntology {
1308                        ontology: test_ontology(), // merged: has "signal"
1309                    },
1310                    GraphOp::AddNode {
1311                        node_id: "n1".into(),
1312                        node_type: "entity".into(),
1313                        label: "base-typed".into(),
1314                        properties: BTreeMap::new(),
1315                        subtype: None,
1316                    },
1317                    GraphOp::AddNode {
1318                        node_id: "s1".into(),
1319                        node_type: "signal".into(),
1320                        label: "extension-typed".into(),
1321                        properties: BTreeMap::new(),
1322                        subtype: None,
1323                    },
1324                ],
1325                op_clocks: vec![(1, 0), (1, 1), (1, 2)],
1326                compacted_at_physical_ms: 1,
1327                compacted_at_logical: 2,
1328            },
1329            1,
1330            "inst-a",
1331        );
1332        g.apply(&checkpoint);
1333
1334        assert!(g.get_node("n1").is_some());
1335        assert!(g.get_node("s1").is_some(), "extension-typed node lost");
1336        assert!(g.ontology.node_types.contains_key("signal"));
1337        assert!(g.quarantined.is_empty());
1338    }
1339}
silk/graph.rs

silk/
graph.rs
