vertep_runtime/
storage.rs

1//! Where a node keeps journals and graphs.
2//!
3//! A [`Storage`] pairs one [`JournalStore`] with the [`GraphStore`] that holds each actor's
4//! current-state graph (on stores that commit state with appends, the same store), its
5//! snapshots, outbox marks, dead letters and store-backed mailbox.
6
7use oxrdf::{NamedNode, Quad, Term};
8use vertep_store::{GraphStore, JournalStore, MemoryStore, Solutions, StoreError};
9
10/// The journal store and graph stores of a node.
11pub trait Storage: Send + Sync {
12    /// The journal store.
13    fn journal(&self) -> &dyn JournalStore;
14    /// The graph store holding `actor`'s current state, snapshots and records.
15    fn graph(&self, actor: &NamedNode) -> &dyn GraphStore;
16    /// Every graph store, for scans across actors (pending timers, mailboxes).
17    fn graphs(&self) -> Vec<&dyn GraphStore>;
18}
19
20impl<S: Storage + ?Sized> Storage for std::sync::Arc<S> {
21    fn journal(&self) -> &dyn JournalStore {
22        (**self).journal()
23    }
24    fn graph(&self, actor: &NamedNode) -> &dyn GraphStore {
25        (**self).graph(actor)
26    }
27    fn graphs(&self) -> Vec<&dyn GraphStore> {
28        (**self).graphs()
29    }
30}
31
32impl Storage for MemoryStore {
33    fn journal(&self) -> &dyn JournalStore {
34        self
35    }
36    fn graph(&self, _actor: &NamedNode) -> &dyn GraphStore {
37        self
38    }
39    fn graphs(&self) -> Vec<&dyn GraphStore> {
40        vec![self]
41    }
42}
43
44#[cfg(feature = "tiramemsu")]
45impl Storage for vertep_tiramemsu::TiramemsuShard {
46    fn journal(&self) -> &dyn JournalStore {
47        self
48    }
49    fn graph(&self, _actor: &NamedNode) -> &dyn GraphStore {
50        self
51    }
52    fn graphs(&self) -> Vec<&dyn GraphStore> {
53        vec![self]
54    }
55}
56
57#[cfg(feature = "tiramemsu")]
58impl Storage for vertep_tiramemsu::TiramemsuBackend {
59    fn journal(&self) -> &dyn JournalStore {
60        self
61    }
62    fn graph(&self, actor: &NamedNode) -> &dyn GraphStore {
63        self.shard_for(actor)
64    }
65    fn graphs(&self) -> Vec<&dyn GraphStore> {
66        self.shards().iter().map(|s| s as &dyn GraphStore).collect()
67    }
68}
69
70/// A journal store and a separate graph store (for example a journal store without state in
71/// appends and an Oxigraph graph store). Current state is then written after each append.
72pub struct SplitStorage<J, G> {
73    /// The journal store.
74    pub journal: J,
75    /// The graph store.
76    pub graph: G,
77}
78
79impl<J, G> std::fmt::Debug for SplitStorage<J, G> {
80    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
81        f.debug_struct("SplitStorage").finish_non_exhaustive()
82    }
83}
84
85impl<J: JournalStore, G: GraphStore> Storage for SplitStorage<J, G> {
86    fn journal(&self) -> &dyn JournalStore {
87        &self.journal
88    }
89    fn graph(&self, _actor: &NamedNode) -> &dyn GraphStore {
90        &self.graph
91    }
92    fn graphs(&self) -> Vec<&dyn GraphStore> {
93        vec![&self.graph]
94    }
95}
96
97/// Runs a SELECT query.
98pub(crate) fn select(g: &dyn GraphStore, sparql: &str) -> Result<Solutions, StoreError> {
99    g.query(sparql)?.into_solutions()
100}
101
102/// The lexical form of a literal or the IRI of a named node.
103pub(crate) fn text(t: &Term) -> Option<String> {
104    match t {
105        Term::Literal(l) => Some(l.value().to_owned()),
106        Term::NamedNode(n) => Some(n.as_str().to_owned()),
107        _ => None,
108    }
109}
110
111/// Every quad of graph `graph` in `g`.
112pub(crate) fn graph_quads(g: &dyn GraphStore, graph: &NamedNode) -> Result<Vec<Quad>, StoreError> {
113    let q = format!("SELECT ?s ?p ?o WHERE {{ GRAPH {graph} {{ ?s ?p ?o }} }}");
114    let sol = select(g, &q)?;
115    let mut out = Vec::new();
116    for i in 0..sol.len() {
117        let (Some(s), Some(p), Some(o)) = (sol.get(i, "s"), sol.get(i, "p"), sol.get(i, "o"))
118        else {
119            continue;
120        };
121        let s = match s {
122            Term::NamedNode(n) => oxrdf::NamedOrBlankNode::from(n.clone()),
123            Term::BlankNode(b) => oxrdf::NamedOrBlankNode::from(b.clone()),
124            _ => continue,
125        };
126        let Term::NamedNode(p) = p else { continue };
127        out.push(Quad::new(s, p.clone(), o.clone(), graph.clone()));
128    }
129    Ok(out)
130}
vertep_runtime/storage.rs

vertep_runtime/
storage.rs
