uqa_graph/
versioned_store.rs

1//
2// Unified Query Algebra
3//
4// Copyright (c) 2023-2026 Cognica, Inc.
5//
6
7//! Version-tracked graph store (Section 9.3, Paper 2).
8//!
9//! Wraps a [`GraphStore`] and applies [`GraphDelta`]s with a monotonic
10//! version counter. Each apply records an inverse delta so the store
11//! can rewind to an earlier version. Invalidation callbacks fire on
12//! affected edge labels so dependent path indexes can refresh.
13
14use std::collections::BTreeSet;
15
16use crate::delta::{DeltaOp, GraphDelta};
17use crate::store::{GraphStore, GraphStoreError, GraphStoreResult};
18
19type InvalidationCallback = Box<dyn Fn(&BTreeSet<String>) + Send + Sync>;
20type InverseDelta = Vec<(String, DeltaOp)>;
21
22fn record_vertex_removal<G: GraphStore>(
23    candidate: &mut G,
24    graph: &str,
25    vertex_id: u64,
26    inverse: &mut InverseDelta,
27    affected_labels: &mut BTreeSet<String>,
28) -> GraphStoreResult<()> {
29    if !candidate
30        .vertex_graphs(vertex_id)?
31        .iter()
32        .any(|owner| owner == graph)
33    {
34        return Ok(());
35    }
36    let existing = candidate
37        .get_vertex(vertex_id)?
38        .ok_or_else(|| GraphStoreError::CorruptGraph(format!("missing vertex {vertex_id}")))?;
39    let mut incident_edges = Vec::new();
40    let mut ids = candidate.out_edge_ids(vertex_id, graph)?;
41    ids.extend(candidate.in_edge_ids(vertex_id, graph)?);
42    for id in ids {
43        let edge = candidate
44            .get_edge(id)?
45            .ok_or_else(|| GraphStoreError::CorruptGraph(format!("missing incident edge {id}")))?;
46        affected_labels.insert(edge.label.clone());
47        incident_edges.push(edge);
48    }
49    candidate.remove_vertex(vertex_id, graph)?;
50    for edge in incident_edges {
51        inverse.push((graph.to_string(), DeltaOp::AddEdge(edge)));
52    }
53    inverse.push((graph.to_string(), DeltaOp::AddVertex(existing)));
54    Ok(())
55}
56
57pub struct VersionedGraphStore<'a, G: GraphStore> {
58    base: &'a mut G,
59    graph: String,
60    version: u64,
61    deltas: Vec<GraphDelta>,
62    inverse_deltas: Vec<InverseDelta>,
63    on_invalidate: Vec<InvalidationCallback>,
64}
65
66impl<'a, G: GraphStore> VersionedGraphStore<'a, G> {
67    pub fn new(base: &'a mut G, graph: impl Into<String>) -> Self {
68        Self {
69            base,
70            graph: graph.into(),
71            version: 0,
72            deltas: Vec::new(),
73            inverse_deltas: Vec::new(),
74            on_invalidate: Vec::new(),
75        }
76    }
77
78    pub fn version(&self) -> u64 {
79        self.version
80    }
81
82    pub fn base(&self) -> &G {
83        &*self.base
84    }
85
86    pub fn base_mut(&mut self) -> &mut G {
87        self.base
88    }
89
90    /// Apply a delta to the base store, accumulating an inverse delta
91    /// for rollback. Returns the new version number.
92    pub fn apply(&mut self, delta: GraphDelta) -> GraphStoreResult<u64> {
93        if !self.base.has_graph(&self.graph)? {
94            return Err(GraphStoreError::UnknownGraph(self.graph.clone()));
95        }
96        let next_version = self.version.checked_add(1).ok_or_else(|| {
97            GraphStoreError::IdExhausted("graph version counter overflow".to_string())
98        })?;
99        let mut inverse = InverseDelta::new();
100        let mut affected_labels = delta.affected_edge_labels();
101        self.base.transaction(|candidate| {
102            for op in delta.ops() {
103                match op {
104                    DeltaOp::AddVertex(vertex) => {
105                        if let Some(previous) = candidate.get_vertex(vertex.vertex_id)? {
106                            let owners = candidate.vertex_graphs(vertex.vertex_id)?;
107                            let owner = owners.first().ok_or_else(|| {
108                                GraphStoreError::CorruptGraph(format!(
109                                    "vertex {} has no owning graph",
110                                    vertex.vertex_id
111                                ))
112                            })?;
113                            inverse.push((owner.clone(), DeltaOp::AddVertex(previous)));
114                            if !owners.contains(&self.graph) {
115                                inverse.push((
116                                    self.graph.clone(),
117                                    DeltaOp::RemoveVertex(vertex.vertex_id),
118                                ));
119                            }
120                        } else {
121                            inverse.push((
122                                self.graph.clone(),
123                                DeltaOp::RemoveVertex(vertex.vertex_id),
124                            ));
125                        }
126                        candidate.add_vertex(vertex.clone(), &self.graph)?;
127                    }
128                    DeltaOp::RemoveVertex(vertex_id) => {
129                        record_vertex_removal(
130                            candidate,
131                            &self.graph,
132                            *vertex_id,
133                            &mut inverse,
134                            &mut affected_labels,
135                        )?;
136                    }
137                    DeltaOp::AddEdge(edge) => {
138                        if let Some(previous) = candidate.get_edge(edge.edge_id)? {
139                            affected_labels.insert(previous.label.clone());
140                            let owners = candidate.edge_graphs(edge.edge_id)?;
141                            let owner = owners.first().ok_or_else(|| {
142                                GraphStoreError::CorruptGraph(format!(
143                                    "edge {} has no owning graph",
144                                    edge.edge_id
145                                ))
146                            })?;
147                            inverse.push((owner.clone(), DeltaOp::AddEdge(previous)));
148                            if !owners.contains(&self.graph) {
149                                inverse
150                                    .push((self.graph.clone(), DeltaOp::RemoveEdge(edge.edge_id)));
151                            }
152                        } else {
153                            inverse.push((self.graph.clone(), DeltaOp::RemoveEdge(edge.edge_id)));
154                        }
155                        candidate.add_edge(edge.clone(), &self.graph)?;
156                    }
157                    DeltaOp::RemoveEdge(edge_id) => {
158                        if !candidate.edge_graphs(*edge_id)?.contains(&self.graph) {
159                            continue;
160                        }
161                        let existing = candidate.get_edge(*edge_id)?.ok_or_else(|| {
162                            GraphStoreError::CorruptGraph(format!("missing edge {edge_id}"))
163                        })?;
164                        affected_labels.insert(existing.label.clone());
165                        candidate.remove_edge(*edge_id, &self.graph)?;
166                        inverse.push((self.graph.clone(), DeltaOp::AddEdge(existing)));
167                    }
168                }
169            }
170            Ok(())
171        })?;
172        self.version = next_version;
173        self.deltas.push(delta);
174        self.inverse_deltas.push(inverse);
175        if !affected_labels.is_empty() {
176            for callback in &self.on_invalidate {
177                callback(&affected_labels);
178            }
179        }
180        Ok(self.version)
181    }
182
183    /// Rewind to the given version by replaying inverse deltas. Errors
184    /// when the target version is in the future or below zero.
185    pub fn rollback(&mut self, to_version: u64) -> GraphStoreResult<()> {
186        if to_version > self.version {
187            return Err(GraphStoreError::InvalidMutation(format!(
188                "cannot rollback to version {to_version} (current: {})",
189                self.version
190            )));
191        }
192        let mut remaining_version = self.version;
193        let mut inverse_count = 0usize;
194        self.base.transaction(|candidate| {
195            while remaining_version > to_version {
196                let offset = inverse_count.checked_add(1).ok_or_else(|| {
197                    GraphStoreError::CorruptGraph("version history index overflow".to_string())
198                })?;
199                let inverse = self
200                    .inverse_deltas
201                    .get(
202                        self.inverse_deltas
203                            .len()
204                            .checked_sub(offset)
205                            .ok_or_else(|| {
206                                GraphStoreError::CorruptGraph(
207                                    "version history is shorter than the current graph version"
208                                        .to_string(),
209                                )
210                            })?,
211                    )
212                    .ok_or_else(|| {
213                        GraphStoreError::CorruptGraph(
214                            "version history is shorter than the current graph version".to_string(),
215                        )
216                    })?;
217                for (graph, op) in inverse.iter().rev() {
218                    match op {
219                        DeltaOp::AddVertex(vertex) => {
220                            candidate.add_vertex(vertex.clone(), graph)?;
221                        }
222                        DeltaOp::RemoveVertex(vertex_id) => {
223                            candidate.remove_vertex(*vertex_id, graph)?;
224                        }
225                        DeltaOp::AddEdge(edge) => {
226                            candidate.add_edge(edge.clone(), graph)?;
227                        }
228                        DeltaOp::RemoveEdge(edge_id) => {
229                            candidate.remove_edge(*edge_id, graph)?;
230                        }
231                    }
232                }
233                remaining_version = remaining_version.checked_sub(1).ok_or_else(|| {
234                    GraphStoreError::CorruptGraph("graph version underflow".to_string())
235                })?;
236                inverse_count = inverse_count.checked_add(1).ok_or_else(|| {
237                    GraphStoreError::CorruptGraph("version history index overflow".to_string())
238                })?;
239            }
240            Ok(())
241        })?;
242        self.version = remaining_version;
243        let new_len = self
244            .inverse_deltas
245            .len()
246            .checked_sub(inverse_count)
247            .ok_or_else(|| {
248                GraphStoreError::CorruptGraph("version history truncation underflow".to_string())
249            })?;
250        self.inverse_deltas.truncate(new_len);
251        self.deltas.truncate(new_len);
252        Ok(())
253    }
254
255    /// Register a callback fired with the set of affected edge labels
256    /// every time `apply` lands a delta that touches at least one edge.
257    pub fn on_invalidate<F>(&mut self, callback: F)
258    where
259        F: Fn(&BTreeSet<String>) + Send + Sync + 'static,
260    {
261        self.on_invalidate.push(Box::new(callback));
262    }
263}
uqa_graph/versioned_store.rs

uqa_graph/
versioned_store.rs
