diff --git a/benches/query_result_collection.rs b/benches/query_result_collection.rs index e3d4304..f608157 100644 --- a/benches/query_result_collection.rs +++ b/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)) diff --git a/src/internal/core.rs b/src/internal/core.rs index 34cf49a..54bab0d 100644 --- a/src/internal/core.rs +++ b/src/internal/core.rs @@ -439,6 +439,99 @@ pub struct Batch { pub timestamp: DateTime, } +/// 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, + timestamp: DateTime, + ) -> Result { + 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 { diff --git a/src/internal/replication.rs b/src/internal/replication.rs index d954ce8..0e207b4 100644 --- a/src/internal/replication.rs +++ b/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 { + 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 { + 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 { + 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, diff --git a/src/internal/rocrate.rs b/src/internal/rocrate.rs index a0dce61..ebbea3a 100644 --- a/src/internal/rocrate.rs +++ b/src/internal/rocrate.rs @@ -695,6 +695,28 @@ impl RoCrateManager { additional_triples: Vec<(NamedNode, oxrdf::Term)>, replaced_predicates: &[NamedNode], ) -> Result { + 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, 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,10 +855,32 @@ impl RoCrateManager { additional_triples: Vec<(NamedNode, oxrdf::Term)>, replaced_predicates: &[NamedNode], ) -> Result { + 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, 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, @@ -844,12 +888,11 @@ impl RoCrateManager { 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, 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])?; diff --git a/src/internal/store.rs b/src/internal/store.rs index c8342d4..3ada548 100644 --- a/src/internal/store.rs +++ b/src/internal/store.rs @@ -6618,13 +6618,25 @@ impl GraphStore { continue; } let quad = Self::decode_quad_key(key.as_ref())?; + let mut dots = decode_dots(value.as_ref())?; + // Term ids and dot arrival order are node-local, so both are sorted + // by value: two replicas holding the same state must produce equal + // snapshots. + dots.sort_unstable_by_key(|dot| (dot.actor, dot.counter)); quads.push(SnapshotQuadState { subject: self.decode_term_arc(quad.subject)?.as_ref().clone(), predicate: self.decode_term_arc(quad.predicate)?.as_ref().clone(), object: self.decode_term_arc(quad.object)?.as_ref().clone(), - dots: decode_dots(value.as_ref())?, + dots, }); } + quads.sort_unstable_by(|left, right| { + (&left.subject, &left.predicate, &left.object).cmp(&( + &right.subject, + &right.predicate, + &right.object, + )) + }); Ok(GraphReplicaSnapshot { graph: graph.clone(), diff --git a/src/lib.rs b/src/lib.rs index 8ce1417..7fe4004 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -99,14 +99,16 @@ use chrono::Utc; use oxrdf::{NamedNode, Term}; pub use crate::core::{ - ActorId, Batch, CrateViolation, EncodedTerm, EventId, GraphDiagnostics, GraphId, GraphPolicy, - GraphTombstone, MaterializedQuadChange, PolicyTag, PredicateFilter, TaggedGraphPolicy, - UnsupportedRdfStarTerm, VectorClock, vocab, + ActorId, Batch, CrateViolation, CrossGraphChange, EncodedTerm, EventId, GraphDiagnostics, + GraphId, GraphPolicy, GraphTombstone, MaterializedQuadChange, PolicyTag, PredicateFilter, + TaggedGraphPolicy, UnsupportedRdfStarTerm, VectorClock, vocab, }; pub use crate::core::{Dot, GraphReplicaSnapshot, QuadOp, SnapshotQuadState}; pub use crate::planner::{JoinKind, JoinMode, PlannedJoin}; pub use crate::query_context::{QueryCancellation, QueryReadMode, ReadAccessPath, ReadStatistics}; -pub use crate::replication::{CheckMode, DiagnosticsMode, MergeError, UpdateError, WriteChecks}; +pub use crate::replication::{ + CheckMode, DiagnosticsMode, MergeError, MergeResult, UpdateError, WriteChecks, +}; pub use crate::rocrate::{ AppendDataEntitiesReport, CanonicalJsonLd, NewDataEntity, PrepareRoCrateOptions, PreparedGraphBase, PreparedRoCrateDocument, PreparedRoCrateStatistics, RoCrateError, @@ -2694,6 +2696,73 @@ impl CraqleNode { Ok(self.manager().plan_import_jsonld_checked(&graph, jsonld)?) } + /// Change set the strict RO-Crate replacement would commit for `jsonld`, + /// without applying it. + /// + /// Runs the same complete-RO-Crate validation as + /// [`CraqleNode::apply_rocrate_document_checked_with_policy`] and requires + /// the same write authorization as + /// [`CraqleNode::apply_rocrate_document`]. Mutates nothing: no quads, no + /// policy, no search queue, no replication record. + /// + /// The change set describes the state visible now, so a concurrent write + /// to `graph` can still invalidate it. + pub fn plan_rocrate_document_checked( + &self, + auth: &dyn Authorizer, + graph: &GraphId, + jsonld: &str, + ) -> Result> { + self.ensure_graph_action(graph, auth, Action::Write)?; + Ok(self.manager().plan_import_jsonld_checked(graph, jsonld)?) + } + + /// Change set [`CraqleNode::patch_data_with`] would commit for `request`, + /// without applying it. + /// + /// Structurally validated exactly as the applying variant is, against the + /// state visible now, which a concurrent write to the graph can still + /// invalidate. Mutates nothing. + pub fn plan_patch_data( + &self, + auth: &dyn Authorizer, + request: &PatchEntityRequest, + ) -> Result> { + let graph = &request.entity.graph; + self.ensure_graph_action(graph, auth, Action::Write)?; + let changes = self.manager().plan_patch_data( + graph, + &request.entity.entity_id, + &request.entity.entity_type, + &request.entity.name, + &request.entity.additional_triples, + &request.replaced_predicates, + )?; + self.replication.check_planned_changes(graph, &changes)?; + Ok(changes) + } + + /// Change set [`CraqleNode::patch_contextual_with`] would commit for + /// `request`, without applying it. See [`CraqleNode::plan_patch_data`]. + pub fn plan_patch_contextual( + &self, + auth: &dyn Authorizer, + request: &PatchEntityRequest, + ) -> Result> { + let graph = &request.entity.graph; + self.ensure_graph_action(graph, auth, Action::Write)?; + let changes = self.manager().plan_patch_contextual( + graph, + &request.entity.entity_id, + &request.entity.entity_type, + &request.entity.name, + &request.entity.additional_triples, + &request.replaced_predicates, + )?; + self.replication.check_planned_changes(graph, &changes)?; + Ok(changes) + } + /// Preview the canonical RDF changes implied by a JSON-LD document. pub fn preview_rocrate_update( &self, @@ -3386,12 +3455,77 @@ impl CraqleNode { Ok(self.store.graph_fingerprint(graph)?) } - /// Read-only dump of one graph's quad and dot state, for diagnostics and - /// test assertions. Not a sync mechanism. + /// Read-only dump of one graph's quad and dot state. + /// + /// Quads and each quad's dots are sorted by value, so two replicas holding + /// the same state produce equal snapshots regardless of local term ids or + /// arrival order. That makes this both a diagnostic and the state an + /// application replicates with [`CraqleNode::install_graph_snapshot`]. + /// + /// A graph this node does not hold reports an empty clock and no quads. pub fn graph_snapshot(&self, graph: &GraphId) -> Result { Ok(self.store.graph_snapshot(graph)?) } + /// Merge a batch authored on another replica that reached this node + /// through the application's own transport instead of irokle. + /// + /// Applies the batch's ops in order under the graph's write lock with the + /// OR-Set semantics of replicated records: adds carry the batch dot, + /// removes drop exactly the dots they witnessed, and the graph is created + /// when it is missing. Nothing is published back to irokle. + /// + /// Idempotent by the batch dot: merging a batch whose `(actor, counter)` + /// this graph's clock already contains reports `applied: false`, as does + /// merging into a graph this node has tombstoned. + /// + /// Fails when an op carries a term the store cannot hold: an RDF-star + /// term, a term over four megabytes, or one that is not an encoded IRI, + /// literal or blank node. + pub fn merge_batch(&self, batch: &Batch) -> Result { + // Orders this merge against every other write to the same graph; see + // `replication::GRAPH_WRITE_LOCKS`. + let _write_guard = replication::graph_write_guard(&batch.graph); + let merged = self.replication.merge_batch(batch)?; + if merged.applied { + self.schedule_search_update_for_graph(&batch.graph)?; + } + self.persist_fjall()?; + Ok(merged) + } + + /// Install a snapshot taken on another replica, seeding or repairing a + /// local copy of that graph. + /// + /// The result is the state-based OR-Set join of the two states: a snapshot + /// dot joins a quad's local dot set only when this graph's clock does not + /// already cover it, the graph clock then becomes the element-wise maximum + /// of the two clocks, and a graph this node does not hold is created. + /// + /// A dot the local clock covers but the local quad no longer carries is a + /// removal this node has already seen, so it stays removed: installing a + /// lagging replica's snapshot never resurrects a quad, and a device may be + /// seeded from several holders in any order. + /// + /// Later [`CraqleNode::merge_batch`] calls compose with the join, so a + /// remove that witnessed the snapshot's dots still removes them. + /// + /// Installing the same snapshot twice reports `applied: false`, as does + /// installing a snapshot this node's clock already covers or installing + /// into a graph this node has tombstoned. + /// + /// Fails when a quad carries a term the store cannot hold, as for + /// [`CraqleNode::merge_batch`]. + pub fn install_graph_snapshot(&self, snapshot: &GraphReplicaSnapshot) -> Result { + let _write_guard = replication::graph_write_guard(&snapshot.graph); + let merged = self.replication.install_snapshot(snapshot)?; + if merged.applied { + self.schedule_search_update_for_graph(&snapshot.graph)?; + } + self.persist_fjall()?; + Ok(merged) + } + /// Build the per-graph state `describe_in_ctx` needs. fn describe_ctx(&self, graph: &GraphId) -> Result { let graph_term = EncodedTerm::from_named_node(&graph.0); diff --git a/src/sync.rs b/src/sync.rs index 9ab8237..6c6b9fc 100644 --- a/src/sync.rs +++ b/src/sync.rs @@ -2,7 +2,7 @@ use std::collections::{BTreeSet, HashMap}; use std::sync::{Arc, PoisonError, RwLock}; use crate::core::{ - ActorId, Batch, ContextTag, Dot, EncodedTerm, GraphId, GraphTombstone, MaterializedQuadChange, + ActorId, Batch, ContextTag, EncodedTerm, GraphId, GraphTombstone, MaterializedQuadChange, QuadOp, RoCrateRenderHints, TaggedGraphPolicy, TaggedRoCrateRenderHints, VectorClock, }; use crate::store::GraphStore; @@ -922,54 +922,15 @@ where I: IntoIterator, { let EventBatchCtx { graph, meta } = cx; - let actor = actor_from_irokle(meta.actor_id); - let counter = meta.actor_seq; - let base_clock = clock_from_irokle(&meta.observed_clock); - 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(Batch { - graph: graph.clone(), - actor, - counter, - base_clock, - ops, - timestamp: Utc::now(), - }) + Batch::from_changes( + graph.clone(), + actor_from_irokle(meta.actor_id), + meta.actor_seq, + clock_from_irokle(&meta.observed_clock), + changes, + Utc::now(), + ) + .map_err(|error| CraqleSyncError::InvalidEvent(error.to_string())) } /// Largest term craqle accepts from a topic. Well past any real IRI or literal, @@ -1029,6 +990,41 @@ fn check_changes(changes: &[MaterializedQuadChange]) -> SyncResult<()> { Ok(()) } +/// Same guard for a batch that reached craqle outside irokle: no op may carry +/// content the store could only fail on. +pub(crate) fn check_ops(ops: &[QuadOp]) -> SyncResult<()> { + for op in ops { + let terms = match op { + QuadOp::Add { + subject, + predicate, + object, + .. + } + | QuadOp::Remove { + subject, + predicate, + object, + .. + } => [subject, predicate, object], + }; + for term in terms { + check_term(term)?; + } + } + Ok(()) +} + +/// Same guard for a replica snapshot handed to craqle by an application. +pub(crate) fn check_snapshot(snapshot: &crate::GraphReplicaSnapshot) -> SyncResult<()> { + for quad in &snapshot.quads { + for term in [&quad.subject, &quad.predicate, &quad.object] { + check_term(term)?; + } + } + Ok(()) +} + /// Borrowing variant, for callers that only hold a reference to the record /// (catch-up and reconcile replay both re-read their records afterwards). pub(crate) fn batch_from_record( @@ -1108,18 +1104,6 @@ pub(crate) fn batch_from_owned( })) } -fn ensure_change_graph(expected: &GraphId, actual: &GraphId) -> SyncResult<()> { - if expected == actual { - Ok(()) - } else { - Err(CraqleSyncError::InvalidEvent(format!( - "event graph `{}` contained change for `{}`", - expected.as_str(), - actual.as_str() - ))) - } -} - fn actor_from_irokle(actor: irokle::ActorId) -> ActorId { ActorId::from_bytes(*actor.as_bytes()) } diff --git a/tests/merge_api.rs b/tests/merge_api.rs new file mode 100644 index 0000000..e7520f1 --- /dev/null +++ b/tests/merge_api.rs @@ -0,0 +1,376 @@ +use chrono::Utc; +use craqle::*; + +fn policy() -> GraphPolicy { + GraphPolicy { + public: true, + permission_paths: vec!["/tests/merge".to_string()], + } +} + +fn actor(seed: u8) -> ActorId { + ActorId::from_bytes([seed; 32]) +} + +fn node(root: &std::path::Path, name: &str) -> CraqleNode { + CraqleNode::open_with_actor(root.join(name), actor(name.as_bytes()[0])).unwrap() +} + +fn doc(graph: &GraphId, file_count: usize) -> String { + let root = graph.as_str(); + let mut entries = vec![ + serde_json::json!({ + "@id": "ro-crate-metadata.json", + "@type": "CreativeWork", + "conformsTo": {"@id": "https://w3id.org/ro/crate/1.2"}, + "about": {"@id": root} + }), + serde_json::json!({ + "@id": root, + "@type": "Dataset", + "name": "Merge Dataset", + "description": "Merge API test", + "datePublished": "2026-06-10", + "license": {"@id": "https://creativecommons.org/licenses/by/4.0/"}, + "hasPart": (0..file_count) + .map(|idx| serde_json::json!({"@id": format!("./data/file-{idx}.raw")})) + .collect::>() + }), + ]; + for idx in 0..file_count { + entries.push(serde_json::json!({ + "@id": format!("./data/file-{idx}.raw"), + "@type": "File", + "name": format!("file-{idx}.raw") + })); + } + serde_json::json!({ + "@context": "https://w3id.org/ro/crate/1.2/context", + "@graph": entries + }) + .to_string() +} + +fn apply_doc(node: &CraqleNode, graph: &GraphId, jsonld: &str) -> Batch { + node.apply_rocrate_document_checked_with_policy( + &AllowAllAuthorizer, + graph.clone(), + jsonld, + policy(), + ) + .unwrap() +} + +fn keyword_change(graph: &GraphId, keyword: &str, insert: bool) -> MaterializedQuadChange { + let subject = EncodedTerm::from_named_node(&graph.0); + let predicate = EncodedTerm::from_named_node(&vocab::schema_keywords()); + let object = EncodedTerm(format!("\"{keyword}\"")); + if insert { + MaterializedQuadChange::Insert { + graph: graph.clone(), + subject, + predicate, + object, + } + } else { + MaterializedQuadChange::Delete { + graph: graph.clone(), + subject, + predicate, + object, + } + } +} + +fn objects_for(node: &CraqleNode, graph: &GraphId, predicate: &str) -> Vec { + node.graph_snapshot(graph) + .unwrap() + .quads + .into_iter() + .filter(|quad| quad.predicate.0 == format!("<{predicate}>")) + .map(|quad| quad.object.0) + .collect() +} + +#[test] +fn merge_batch_converges() { + let tmp = tempfile::tempdir().unwrap(); + let author = node(tmp.path(), "author"); + let holder = node(tmp.path(), "holder"); + let graph = GraphId::new("https://w3id.org/aruna/merge-batch"); + let batch = apply_doc(&author, &graph, &doc(&graph, 2)); + + let merged = holder.merge_batch(&batch).unwrap(); + assert!(merged.applied); + assert_eq!( + author.graph_snapshot(&graph).unwrap(), + holder.graph_snapshot(&graph).unwrap() + ); + assert!(!holder.merge_batch(&batch).unwrap().applied); + assert_eq!( + author.graph_snapshot(&graph).unwrap(), + holder.graph_snapshot(&graph).unwrap() + ); +} + +#[test] +fn concurrent_batches_converge() { + // Both replicas merge both batches; add-wins, and the remove only takes + // the dots it witnessed. + let tmp = tempfile::tempdir().unwrap(); + let left = node(tmp.path(), "left"); + let right = node(tmp.path(), "right"); + let graph = GraphId::new("https://w3id.org/aruna/merge-concurrent"); + apply_doc(&left, &graph, &doc(&graph, 1)); + left.apply_changes( + &AllowAllAuthorizer, + &graph, + vec![keyword_change(&graph, "x", true)], + ) + .unwrap(); + assert!( + right + .install_graph_snapshot(&left.graph_snapshot(&graph).unwrap()) + .unwrap() + .applied + ); + + let left_batch = Batch::from_changes( + graph.clone(), + actor(200), + 1, + left.vector_clock(&graph).unwrap(), + vec![ + keyword_change(&graph, "x", false), + keyword_change(&graph, "y", true), + ], + Utc::now(), + ) + .unwrap(); + let right_batch = Batch::from_changes( + graph.clone(), + actor(201), + 1, + right.vector_clock(&graph).unwrap(), + vec![keyword_change(&graph, "z", true)], + Utc::now(), + ) + .unwrap(); + + for batch in [&left_batch, &right_batch] { + assert!(left.merge_batch(batch).unwrap().applied); + } + for batch in [&right_batch, &left_batch] { + assert!(right.merge_batch(batch).unwrap().applied); + } + + let keywords = objects_for(&left, &graph, vocab::schema_keywords().as_str()); + assert_eq!( + vec!["\"y\"".to_string(), "\"z\"".to_string()], + { + let mut sorted = keywords.clone(); + sorted.sort(); + sorted + }, + "expected the concurrent adds and not the removed quad" + ); + assert_eq!( + left.graph_snapshot(&graph).unwrap(), + right.graph_snapshot(&graph).unwrap() + ); +} + +#[test] +fn install_then_remove() { + // Installing twice changes nothing, and a batch witnessing the installed + // dots still removes them. + let tmp = tempfile::tempdir().unwrap(); + let author = node(tmp.path(), "author"); + let replica = node(tmp.path(), "replica"); + let graph = GraphId::new("https://w3id.org/aruna/merge-snapshot"); + apply_doc(&author, &graph, &doc(&graph, 1)); + author + .apply_changes( + &AllowAllAuthorizer, + &graph, + vec![keyword_change(&graph, "x", true)], + ) + .unwrap(); + let snapshot = author.graph_snapshot(&graph).unwrap(); + + assert!(replica.install_graph_snapshot(&snapshot).unwrap().applied); + assert!(!replica.install_graph_snapshot(&snapshot).unwrap().applied); + assert_eq!(snapshot, replica.graph_snapshot(&graph).unwrap()); + + let removal = Batch::from_changes( + graph.clone(), + actor(202), + 1, + snapshot.clock.clone(), + vec![keyword_change(&graph, "x", false)], + Utc::now(), + ) + .unwrap(); + assert!(replica.merge_batch(&removal).unwrap().applied); + assert!(objects_for(&replica, &graph, vocab::schema_keywords().as_str()).is_empty()); +} + +#[test] +fn plan_matches_apply() { + let tmp = tempfile::tempdir().unwrap(); + let planner = node(tmp.path(), "planner"); + let graph = GraphId::new("https://w3id.org/aruna/merge-plan"); + apply_doc(&planner, &graph, &doc(&graph, 1)); + let before = planner.graph_snapshot(&graph).unwrap(); + + let jsonld = doc(&graph, 3); + let planned = planner + .plan_rocrate_document_checked(&AllowAllAuthorizer, &graph, &jsonld) + .unwrap(); + assert!(!planned.is_empty()); + assert_eq!(before, planner.graph_snapshot(&graph).unwrap()); + + let batch = apply_doc(&planner, &graph, &jsonld); + let applied = batch + .ops + .iter() + .map(|op| match op { + QuadOp::Add { + subject, + predicate, + object, + .. + } => MaterializedQuadChange::Insert { + graph: graph.clone(), + subject: subject.clone(), + predicate: predicate.clone(), + object: object.clone(), + }, + QuadOp::Remove { + subject, + predicate, + object, + .. + } => MaterializedQuadChange::Delete { + graph: graph.clone(), + subject: subject.clone(), + predicate: predicate.clone(), + object: object.clone(), + }, + }) + .collect::>(); + assert_eq!(planned, applied); +} + +#[test] +fn plan_patch_merges() { + let tmp = tempfile::tempdir().unwrap(); + let planner = node(tmp.path(), "planner"); + let graph = GraphId::new("https://w3id.org/aruna/merge-plan-patch"); + apply_doc(&planner, &graph, &doc(&graph, 1)); + let before = planner.graph_snapshot(&graph).unwrap(); + + let request = PatchEntityRequest { + entity: CreateEntityRequest { + graph: graph.clone(), + entity_id: "./data/file-0.raw".to_string(), + entity_type: "File".to_string(), + name: "renamed.raw".to_string(), + additional_triples: Vec::new(), + }, + replaced_predicates: vec![vocab::schema_name()], + }; + let planned = planner + .plan_patch_data(&AllowAllAuthorizer, &request) + .unwrap(); + assert!(!planned.is_empty()); + assert_eq!(before, planner.graph_snapshot(&graph).unwrap()); + + let holder = node(tmp.path(), "holder"); + assert!(holder.install_graph_snapshot(&before).unwrap().applied); + let batch = Batch::from_changes( + graph.clone(), + actor(203), + 1, + planner.vector_clock(&graph).unwrap(), + planned, + Utc::now(), + ) + .unwrap(); + assert!(planner.merge_batch(&batch).unwrap().applied); + assert!(holder.merge_batch(&batch).unwrap().applied); + assert_eq!( + planner.graph_snapshot(&graph).unwrap(), + holder.graph_snapshot(&graph).unwrap() + ); + assert!( + objects_for(&planner, &graph, vocab::schema_name().as_str()) + .contains(&"\"renamed.raw\"".to_string()) + ); +} + +#[test] +fn rejects_foreign_graph() { + let graph = GraphId::new("https://w3id.org/aruna/merge-builder"); + let other = GraphId::new("https://w3id.org/aruna/merge-builder-other"); + let error = Batch::from_changes( + graph.clone(), + actor(204), + 1, + VectorClock::new(), + vec![keyword_change(&other, "x", true)], + Utc::now(), + ) + .unwrap_err(); + assert_eq!(graph, error.expected); + assert_eq!(other, error.actual); +} + +#[test] +fn stale_snapshot_ignored() { + // A device seeded from a current holder must not get a removed quad back + // from a lagging holder's snapshot. + let tmp = tempfile::tempdir().unwrap(); + let alpha = node(tmp.path(), "alpha"); + let beta = node(tmp.path(), "beta"); + let device = node(tmp.path(), "device"); + let graph = GraphId::new("https://w3id.org/aruna/merge-stale"); + apply_doc(&alpha, &graph, &doc(&graph, 1)); + alpha + .apply_changes( + &AllowAllAuthorizer, + &graph, + vec![keyword_change(&graph, "x", true)], + ) + .unwrap(); + assert!( + beta.install_graph_snapshot(&alpha.graph_snapshot(&graph).unwrap()) + .unwrap() + .applied + ); + + let removal = Batch::from_changes( + graph.clone(), + actor(210), + 1, + alpha.vector_clock(&graph).unwrap(), + vec![keyword_change(&graph, "x", false)], + Utc::now(), + ) + .unwrap(); + assert!(alpha.merge_batch(&removal).unwrap().applied); + assert!(objects_for(&alpha, &graph, vocab::schema_keywords().as_str()).is_empty()); + + let current = alpha.graph_snapshot(&graph).unwrap(); + assert!(device.install_graph_snapshot(¤t).unwrap().applied); + let stale = beta.graph_snapshot(&graph).unwrap(); + assert!( + stale.quads.iter().any(|quad| quad.object.0 == "\"x\""), + "the lagging holder must still carry the removed quad" + ); + assert!(!device.install_graph_snapshot(&stale).unwrap().applied); + assert_eq!(current, device.graph_snapshot(&graph).unwrap()); + + assert!(beta.merge_batch(&removal).unwrap().applied); + assert_eq!(current, beta.graph_snapshot(&graph).unwrap()); +}