vertep_store/
projector.rs

1//! Eventual projection of journals into an external graph store.
2
3use std::collections::HashSet;
4
5use oxrdf::vocab::xsd;
6use oxrdf::{GraphName, Literal, NamedNode, Quad, Term, Triple};
7
8use crate::error::StoreError;
9use crate::graph::GraphStore;
10use crate::journal::{JournalStore, Position};
11use crate::vocab;
12
13/// What one [`Projector::project_actor`] run did.
14#[derive(Clone, Debug, PartialEq, Eq)]
15pub struct ProjectionReport {
16    /// The actor.
17    pub actor: NamedNode,
18    /// The projected position before the run.
19    pub from: Position,
20    /// The projected position after the run.
21    pub to: Position,
22    /// The number of entries written.
23    pub entries: usize,
24}
25
26/// Copies journal entries into a separate graph store in journal order, idempotently, and
27/// records per actor the last projected position in the target store.
28///
29/// Entry graphs are written with one insert per entry, in position order. When a state source
30/// is configured (the graph side of a journal store that keeps current state, such as a
31/// tiramemsu shard), the actor's current-state graph is then brought to the source's content by
32/// a difference (removals and insertions, never a clear), and finally the position is recorded
33/// as `<actor> vt:projectedPosition n` in the graph [`vocab::PROJECTION_GRAPH`].
34///
35/// Every write is idempotent for entries without blank nodes, so a run interrupted at any point
36/// (the target unreachable, the process stopped) is completed by running again: entries after
37/// the recorded position are written again and change nothing they had already written.
38// @tg: implements:: [[openspec:storage#Eventual Projection To External Graph Stores]]
39pub struct Projector<'a> {
40    journal: &'a dyn JournalStore,
41    target: &'a dyn GraphStore,
42    state_source: Option<&'a dyn GraphStore>,
43}
44
45impl std::fmt::Debug for Projector<'_> {
46    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
47        f.debug_struct("Projector")
48            .field("state_source", &self.state_source.is_some())
49            .finish_non_exhaustive()
50    }
51}
52
53impl<'a> Projector<'a> {
54    /// A projector from `journal` into `target`.
55    pub fn new(journal: &'a dyn JournalStore, target: &'a dyn GraphStore) -> Self {
56        Projector {
57            journal,
58            target,
59            state_source: None,
60        }
61    }
62
63    /// This projector also copying each actor's current-state graph from `source`.
64    pub fn with_state_source(mut self, source: &'a dyn GraphStore) -> Self {
65        self.state_source = Some(source);
66        self
67    }
68
69    /// The last position of `actor` recorded in the target, 0 when none.
70    ///
71    /// # Errors
72    ///
73    /// The errors of [`GraphStore::query`].
74    pub fn projected_position(&self, actor: &NamedNode) -> Result<Position, StoreError> {
75        let q = format!(
76            "SELECT ?n WHERE {{ GRAPH <{g}> {{ {actor} <{p}> ?n }} }}",
77            g = vocab::PROJECTION_GRAPH,
78            p = vocab::PROJECTED_POSITION,
79        );
80        let solutions = self.target.query(&q)?.into_solutions()?;
81        Ok(Position::new(
82            solutions
83                .values("n")
84                .iter()
85                .filter_map(|t| match t {
86                    Term::Literal(l) => l.value().parse::<u64>().ok(),
87                    _ => None,
88                })
89                .max()
90                .unwrap_or(0),
91        ))
92    }
93
94    /// Projects the entries of `actor` after its recorded position, then its state, then
95    /// records the new position.
96    ///
97    /// # Errors
98    ///
99    /// The errors of the journal read and of the graph store writes. Nothing already written
100    /// needs undoing: running again completes the projection.
101    pub fn project_actor(&self, actor: &NamedNode) -> Result<ProjectionReport, StoreError> {
102        let from = self.projected_position(actor)?;
103        let records = self.journal.read(actor, from.next())?;
104        let mut to = from;
105        for record in &records {
106            debug_assert_eq!(record.position, to.next(), "journal reads are gap-free");
107            self.target.insert_quads(&record.entry.quads())?;
108            to = record.position;
109        }
110        if records.is_empty() {
111            return Ok(ProjectionReport {
112                actor: actor.clone(),
113                from,
114                to,
115                entries: 0,
116            });
117        }
118        if let Some(source) = self.state_source {
119            self.copy_state(source, actor)?;
120        }
121        self.record_position(actor, from, to)?;
122        Ok(ProjectionReport {
123            actor: actor.clone(),
124            from,
125            to,
126            entries: records.len(),
127        })
128    }
129
130    /// Projects every actor of the journal store.
131    ///
132    /// # Errors
133    ///
134    /// The first error of [`Projector::project_actor`]; actors before it stay projected.
135    pub fn project_all(&self) -> Result<Vec<ProjectionReport>, StoreError> {
136        self.journal
137            .actors()?
138            .iter()
139            .map(|a| self.project_actor(a))
140            .collect()
141    }
142
143    fn graph_triples(store: &dyn GraphStore, graph: &NamedNode) -> Result<Vec<Triple>, StoreError> {
144        store
145            .query(&format!(
146                "CONSTRUCT {{ ?s ?p ?o }} WHERE {{ GRAPH {graph} {{ ?s ?p ?o }} }}"
147            ))?
148            .into_graph()
149    }
150
151    fn copy_state(&self, source: &dyn GraphStore, actor: &NamedNode) -> Result<(), StoreError> {
152        let want: HashSet<Triple> = Self::graph_triples(source, actor)?.into_iter().collect();
153        let have: HashSet<Triple> = Self::graph_triples(self.target, actor)?
154            .into_iter()
155            .collect();
156        let g = GraphName::NamedNode(actor.clone());
157        let add: Vec<Quad> = want
158            .difference(&have)
159            .map(|t| t.clone().in_graph(g.clone()))
160            .collect();
161        let remove: Vec<Quad> = have
162            .difference(&want)
163            .map(|t| t.clone().in_graph(g.clone()))
164            .collect();
165        if !add.is_empty() {
166            self.target.insert_quads(&add)?;
167        }
168        if !remove.is_empty() {
169            self.target.remove_quads(&remove)?;
170        }
171        Ok(())
172    }
173
174    fn record_position(
175        &self,
176        actor: &NamedNode,
177        old: Position,
178        new: Position,
179    ) -> Result<(), StoreError> {
180        let quad = |p: Position| {
181            Quad::new(
182                actor.clone(),
183                vocab::node(vocab::PROJECTED_POSITION),
184                Literal::new_typed_literal(p.get().to_string(), xsd::INTEGER),
185                vocab::node(vocab::PROJECTION_GRAPH),
186            )
187        };
188        // insert the new position before removing the old one: a crash in between leaves two
189        // values, and readers take the larger
190        self.target.insert_quads(&[quad(new)])?;
191        if old != Position::ZERO && old != new {
192            self.target.remove_quads(&[quad(old)])?;
193        }
194        Ok(())
195    }
196}
vertep_store/projector.rs

vertep_store/
projector.rs
