feat: public batch merge and snapshot install

benches/query_result_collection.rs

@@ -26,6 +26,8 @@ fn record_allocation(bytes: usize) {
     PEAK_LIVE_BYTES.fetch_max(live, Ordering::Relaxed);
 }
 
+// `try_update` needs Rust 1.95; the deprecated name stays until the MSRV moves.
+#[allow(deprecated)]
 fn record_deallocation(bytes: usize) {
     let _ = LIVE_BYTES.fetch_update(Ordering::Relaxed, Ordering::Relaxed, |live| {
         Some(live.saturating_sub(bytes))

src/internal/core.rs

@@ -439,6 +439,99 @@ pub struct Batch {
     pub timestamp: DateTime<Utc>,
 }
 
+/// A change carried a graph other than the batch's own graph.
+#[derive(Debug, Clone, PartialEq, Eq, thiserror::Error)]
+#[error("batch graph `{expected}` contained change for `{actual}`")]
+pub struct CrossGraphChange {
+    pub expected: GraphId,
+    pub actual: GraphId,
+}
+
+impl CrossGraphChange {
+    pub fn kind(&self) -> crate::CraqleErrorKind {
+        crate::CraqleErrorKind::InvalidInput
+    }
+}
+
+impl Batch {
+    /// Build a replication batch from a materialized change set, without
+    /// touching any store.
+    ///
+    /// Every insert shares the single dot `(actor, counter)`; every delete
+    /// becomes an OR-Set remove that witnesses `base_clock`, so it drops
+    /// exactly the dots the author had seen. Op order is the change order,
+    /// which a delete-then-add pair on the same quad depends on.
+    ///
+    /// `counter` must not repeat for `actor` in `graph`: a merge that already
+    /// saw `(actor, counter)` treats the batch as applied and skips it.
+    ///
+    /// Fails when a change targets a graph other than `graph`.
+    pub fn from_changes(
+        graph: GraphId,
+        actor: ActorId,
+        counter: u64,
+        base_clock: VectorClock,
+        changes: impl IntoIterator<Item = MaterializedQuadChange>,
+        timestamp: DateTime<Utc>,
+    ) -> Result<Self, CrossGraphChange> {
+        let dot = Dot { actor, counter };
+        let changes = changes.into_iter();
+        let mut ops = Vec::with_capacity(changes.size_hint().0);
+        for change in changes {
+            match change {
+                MaterializedQuadChange::Insert {
+                    graph: change_graph,
+                    subject,
+                    predicate,
+                    object,
+                } => {
+                    ensure_change_graph(&graph, change_graph)?;
+                    ops.push(QuadOp::Add {
+                        subject,
+                        predicate,
+                        object,
+                        dot,
+                    });
+                }
+                MaterializedQuadChange::Delete {
+                    graph: change_graph,
+                    subject,
+                    predicate,
+                    object,
+                } => {
+                    ensure_change_graph(&graph, change_graph)?;
+                    ops.push(QuadOp::Remove {
+                        subject,
+                        predicate,
+                        object,
+                        witnessed: base_clock.clone(),
+                    });
+                }
+            }
+        }
+
+        Ok(Self {
+            graph,
+            actor,
+            counter,
+            base_clock,
+            ops,
+            timestamp,
+        })
+    }
+}
+
+fn ensure_change_graph(expected: &GraphId, actual: GraphId) -> Result<(), CrossGraphChange> {
+    if *expected == actual {
+        Ok(())
+    } else {
+        Err(CrossGraphChange {
+            expected: expected.clone(),
+            actual,
+        })
+    }
+}
+
 /// Read-only state of a live quad and its OR-Set dot set.
 #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
 pub struct SnapshotQuadState {

src/internal/replication.rs

@@ -79,8 +79,12 @@ impl MergeError {
     }
 }
 
+/// Outcome of merging replicated state into local state.
 #[derive(Debug)]
-pub(crate) struct MergeResult {
+pub struct MergeResult {
+    /// `true` when the merge changed local state. `false` when there was
+    /// nothing left to do: the batch was already applied, the snapshot added
+    /// no dot and no clock entry, or the graph is tombstoned.
     pub applied: bool,
 }
 
@@ -596,6 +600,19 @@ impl ReplicationEngine {
         })
     }
 
+    /// Run the checks a local apply of `changes` would run, without writing.
+    ///
+    /// The verdict describes the state visible now; a concurrent write to
+    /// `graph` can still invalidate the planned change set.
+    pub(crate) fn check_planned_changes(
+        &self,
+        graph: &GraphId,
+        changes: &[MaterializedQuadChange],
+    ) -> Result<(), UpdateError> {
+        self.ensure_change_set_targets(graph, changes)?;
+        self.validate(graph, changes)
+    }
+
     fn validate(
         &self,
         graph: &GraphId,
@@ -1778,6 +1795,17 @@ impl ReplicationEngine {
         self.apply_irokle_batch_with_plan(&incoming, None, DiagnosticsMode::Immediate)
     }
 
+    /// Merge a batch that reached this node outside irokle.
+    /// **Call with the graph's write lock held.**
+    ///
+    /// The ops are term-checked first, exactly as a replicated record is, so a
+    /// foreign transport cannot hand the store content it could only fail on.
+    pub(crate) fn merge_batch(&self, incoming: &Batch) -> Result<MergeResult, MergeError> {
+        crate::sync::check_ops(&incoming.ops)
+            .map_err(|error| MergeError::InputRejected(error.to_string()))?;
+        self.apply_irokle_batch_with_plan(incoming, None, DiagnosticsMode::Immediate)
+    }
+
     /// **Call with the graph's write lock held.** Every caller does, and so
     /// does every writer of a graph tombstone, which is what makes the check
     /// below atomic against a concurrent delete.
@@ -1917,6 +1945,127 @@ impl ReplicationEngine {
             .transpose()
     }
 
+    /// Join a snapshot taken on another replica into local state.
+    /// **Call with the graph's write lock held.**
+    ///
+    /// State-based OR-Set join: a snapshot dot joins a quad's local dot set
+    /// only when the local graph clock does not already cover it, because a
+    /// covered dot the local quad lacks is a removal this node has already
+    /// seen. The graph clock then becomes the element-wise maximum.
+    #[tracing::instrument(level = "debug", skip_all, fields(graph = %snapshot.graph.as_str(), quad_count = snapshot.quads.len()))]
+    pub(crate) fn install_snapshot(
+        &self,
+        snapshot: &GraphReplicaSnapshot,
+    ) -> Result<MergeResult, MergeError> {
+        let graph = &snapshot.graph;
+        crate::sync::check_snapshot(snapshot)
+            .map_err(|error| MergeError::InputRejected(error.to_string()))?;
+        if self.store.graph_tombstoned(graph)? {
+            return Ok(MergeResult { applied: false });
+        }
+        // Self-guarding, so it must run before the commit guard is taken.
+        if !self.store.contains_graph(graph)? {
+            self.store.create_graph(graph)?;
+        }
+
+        #[cfg(feature = "shacl-core")]
+        let pending_graphs;
+        let applied = {
+            let _commit_guard = self.store.graph_commit_guard(graph);
+            #[cfg(feature = "shacl-core")]
+            {
+                pending_graphs = self.store.affected_shacl_graphs(graph)?;
+            }
+            self.join_snapshot(snapshot)?
+        };
+
+        #[cfg(feature = "shacl-core")]
+        if applied {
+            let settled = self.settle_current_post_commit(graph);
+            self.settle_shacl_graphs(&pending_graphs, (!settled).then_some(graph));
+        }
+        Ok(MergeResult { applied })
+    }
+
+    /// Write the OR-Set join of local state and `snapshot`, reporting whether
+    /// anything changed. **Call with the graph commit guard held.**
+    fn join_snapshot(&self, snapshot: &GraphReplicaSnapshot) -> Result<bool, MergeError> {
+        let graph = &snapshot.graph;
+        // The causal context of this node's state, read before the join: it
+        // decides which snapshot dots are new and which are already-seen
+        // removals.
+        let seen = self.store.get_vector_clock(graph)?;
+        let mut changed = false;
+        for (actor, counter) in &snapshot.clock.0 {
+            changed |= seen.0.get(actor).is_none_or(|local| local < counter);
+        }
+        let mut clock = seen.clone();
+        clock.merge(&snapshot.clock);
+
+        let mut batch = self.store.new_batch();
+        let mut affected_subjects = HashSet::new();
+        let mut term_cache = HashMap::new();
+        let mut cx = BatchTermCtx {
+            batch: &mut batch,
+            cache: &mut term_cache,
+        };
+        self.store.seed_term_cache(
+            &mut cx,
+            snapshot
+                .quads
+                .iter()
+                .flat_map(|quad| [&quad.subject, &quad.predicate, &quad.object]),
+        )?;
+        let graph_id = self
+            .store
+            .resolve_term(&EncodedTerm::from_named_node(&graph.0))?;
+
+        for state in &snapshot.quads {
+            let quad = self.resolve_quad(
+                &mut cx,
+                QuadTerms {
+                    graph_id,
+                    subject: &state.subject,
+                    predicate: &state.predicate,
+                    object: &state.object,
+                },
+            )?;
+            for dot in state.dots.iter().filter(|dot| !seen.contains(dot)) {
+                changed |= self
+                    .store
+                    .insert_quad(cx.batch, QuadAdd { quad, dot: *dot })?;
+            }
+            affected_subjects.insert(quad.subject);
+        }
+
+        // Nothing to commit: the staged term interning is content-addressed, so
+        // discarding the batch loses no state.
+        if !changed {
+            return Ok(false);
+        }
+
+        self.store.set_vector_clock(
+            &mut batch,
+            ClockUpdate {
+                graph_id,
+                clock: &clock,
+            },
+        )?;
+        self.store.enqueue_fts_subjects(
+            &mut batch,
+            FtsEnqueue {
+                graph_id,
+                subjects: &affected_subjects,
+            },
+        )?;
+        #[cfg(feature = "shacl-core")]
+        self.store
+            .stage_pending_bindings(&mut batch, graph, clock_digest(&clock)?)?;
+        self.store.commit(batch)?;
+        self.recompute_graph_diagnostics(graph)?;
+        Ok(true)
+    }
+
     #[tracing::instrument(level = "debug", skip_all, fields(graph = %incoming.graph.as_str(), op_count = incoming.ops.len()))]
     fn apply_single_batch(
         &self,

src/internal/rocrate.rs

@@ -695,6 +695,28 @@ impl RoCrateManager {
         additional_triples: Vec<(NamedNode, oxrdf::Term)>,
         replaced_predicates: &[NamedNode],
     ) -> Result<Batch, RoCrateError> {
+        let changes = self.plan_patch_data(
+            graph_id,
+            entity_id,
+            entity_type,
+            name,
+            &additional_triples,
+            replaced_predicates,
+        )?;
+        Ok(self.engine.local_apply_changes(graph_id, changes)?)
+    }
+
+    /// Change set [`Self::patch_data_entity`] would commit against the state
+    /// visible now. Mutates nothing.
+    pub(crate) fn plan_patch_data(
+        &self,
+        graph_id: &GraphId,
+        entity_id: &str,
+        entity_type: &str,
+        name: &str,
+        additional_triples: &[(NamedNode, oxrdf::Term)],
+        replaced_predicates: &[NamedNode],
+    ) -> Result<Vec<MaterializedQuadChange>, RoCrateError> {
         let entity_id = normalize_entity_id(entity_id);
         let cx = self.crate_ctx(graph_id)?;
         self.require_rocrate_initialized(&cx)?;
@@ -710,7 +732,7 @@ impl RoCrateManager {
                     entity_id: &entity_id,
                     entity_type,
                     name,
-                    additional_triples: &additional_triples,
+                    additional_triples,
                 })?,
                 replaced_predicates,
             },
@@ -729,7 +751,7 @@ impl RoCrateManager {
                 encoded_subject(&entity_id),
             ));
         }
-        Ok(self.engine.local_apply_changes(graph_id, changes)?)
+        Ok(changes)
     }
 
     pub(crate) fn append_new_root_data_entities(
@@ -833,23 +855,44 @@ impl RoCrateManager {
         additional_triples: Vec<(NamedNode, oxrdf::Term)>,
         replaced_predicates: &[NamedNode],
     ) -> Result<Batch, RoCrateError> {
+        let changes = self.plan_patch_contextual(
+            graph_id,
+            entity_id,
+            entity_type,
+            name,
+            &additional_triples,
+            replaced_predicates,
+        )?;
+        Ok(self.engine.local_apply_changes(graph_id, changes)?)
+    }
+
+    /// Change set [`Self::patch_contextual_entity`] would commit against the
+    /// state visible now. Mutates nothing.
+    pub(crate) fn plan_patch_contextual(
+        &self,
+        graph_id: &GraphId,
+        entity_id: &str,
+        entity_type: &str,
+        name: &str,
+        additional_triples: &[(NamedNode, oxrdf::Term)],
+        replaced_predicates: &[NamedNode],
+    ) -> Result<Vec<MaterializedQuadChange>, RoCrateError> {
         let cx = self.crate_ctx(graph_id)?;
         self.require_rocrate_initialized(&cx)?;
         let entity_id = normalize_entity_id(entity_id);
-        let changes = self.patch_subject_changes(
+        self.patch_subject_changes(
             &cx,
             SubjectPatch {
                 subject_id: &entity_id,
                 desired: entity_subject_triples(&EntitySpec {
                     entity_id: &entity_id,
                     entity_type,
                     name,
-                    additional_triples: &additional_triples,
+                    additional_triples,
                 })?,
                 replaced_predicates,
             },
-        )?;
-        Ok(self.engine.local_apply_changes(graph_id, changes)?)
+        )
     }
 
     /// Export a graph to RO-Crate JSON-LD.
@@ -3483,7 +3526,9 @@ fn decode_hex(encoded: &str) -> Result<Vec<u8>, RoCrateError> {
     }
     encoded
         .as_bytes()
-        .chunks_exact(2)
+        .as_chunks::<2>()
+        .0
+        .iter()
         .map(|pair| {
             let high = decode_hex_digit(pair[0])?;
             let low = decode_hex_digit(pair[1])?;