diff --git a/src/applier.rs b/src/applier.rs index f3e30f80b..6fd54feac 100644 --- a/src/applier.rs +++ b/src/applier.rs @@ -45,12 +45,20 @@ use serde_json::Value as Json; use sha2::{Digest, Sha256}; use crate::error::Result; -use crate::store::{Datum, Store}; +#[cfg(test)] +use crate::store::Datum; +use crate::store::Store; +#[cfg(test)] use crate::types::{Op, Value}; #[path = "applier_digest.rs"] mod digest; +#[path = "applier_record.rs"] +mod record; +#[path = "applier_start.rs"] +mod start; pub use digest::Stop; +pub use record::record_to_datum; /// The source log a replica follows. #[derive(Debug, Clone, PartialEq, Eq)] @@ -397,6 +405,11 @@ impl Applier { }; by_tx.entry(tx).or_default().push(record); } + if self.progress.is_none() + && let Some(reason) = Self::new_replica_refusal(store)? + { + return Ok(Outcome::Refused { reason }); + } let mut applied = Vec::new(); for (tx, records) in by_tx { if self.progress.as_ref().is_some_and(|p| tx <= p.source_tx) { @@ -503,94 +516,6 @@ fn apply_one_tx( } } -/// One feed record as the datum that reproduces it, or why it cannot be -/// reproduced exactly. The outer `Result` is a store failure. -/// -/// # Errors -/// [`Error::Sqlite`] if interning a term fails. -pub fn record_to_datum( - store: &Store, - record: &Json, - tx_timestamp: &str, -) -> Result> { - let (Some(entity), Some(attribute), Some(op)) = ( - record["entity"].as_str(), - record["attribute"].as_str(), - record["op"].as_str(), - ) else { - return Ok(Err("record lacks entity, attribute or op".into())); - }; - let retract = match digest::op_is_retract(op) { - Ok(r) => r, - Err(reason) => return Ok(Err(reason)), - }; - let value_field = if retract { "old_value" } else { "value" }; - let value = match value_of(store, &record[value_field])? { - Ok(v) => v, - Err(reason) => return Ok(Err(reason)), - }; - let valid_from = record["valid_from"] - .as_str() - .unwrap_or(tx_timestamp) - .to_string(); - Ok(Ok(Datum { - entity: store.intern(entity)?, - attribute: store.intern(attribute)?, - value, - valid_from, - // A retract names the fact it closes; its own row's valid time - // describes the closing, not an interval to replay. - valid_to: if retract { - None - } else { - record["valid_to"].as_str().map(String::from) - }, - op: if retract { Op::Retract } else { Op::Assert }, - })) -} - -fn value_of(store: &Store, json: &Json) -> Result> { - use base64::Engine as _; - Ok(Ok(match json { - Json::String(s) => Value::Str(s.clone()), - Json::Bool(b) => Value::Bool(*b), - Json::Number(n) => match n.as_i64() { - Some(i) => Value::Int(i), - None => match n.as_f64() { - Some(f) => Value::Float(f), - None => return Ok(Err(format!("unrepresentable number {n}"))), - }, - }, - Json::Object(o) if o.get("ref").and_then(Json::as_str).is_some() => { - Value::Ref(store.intern(o["ref"].as_str().unwrap_or_default())?) - } - Json::Object(o) if o.contains_key("lang") => Value::Lang { - lexical: o["value"].as_str().unwrap_or_default().into(), - lang: o["lang"].as_str().unwrap_or_default().into(), - }, - Json::Object(o) if o.contains_key("datatype") => Value::Typed { - lexical: o["value"].as_str().unwrap_or_default().into(), - datatype: o["datatype"].as_str().unwrap_or_default().into(), - }, - Json::Object(o) if o.get("base64").and_then(Json::as_str).is_some() => { - match base64::engine::general_purpose::STANDARD - .decode(o["base64"].as_str().unwrap_or_default()) - { - Ok(bytes) => Value::Bytes(bytes), - Err(_) => return Ok(Err("byte value is not valid base64".into())), - } - } - // A source from before byte bodies were carried sends only the length. - // Writing empty bytes would be a silent loss: refuse instead. - Json::Object(o) if o.contains_key("bytes_len") => { - return Ok(Err( - "byte value sent as its length only; the source predates lossless bytes".into(), - )); - } - other => return Ok(Err(format!("unreplayable value {other}"))), - })) -} - /// Live facts of a store as resolved strings, so two stores with different /// term ids compare by meaning; valid time is part of a fact's identity here. /// Shared by the applier and feed-replay conformance tests. diff --git a/src/applier_record.rs b/src/applier_record.rs new file mode 100644 index 000000000..36293562b --- /dev/null +++ b/src/applier_record.rs @@ -0,0 +1,97 @@ +//! Feed records decoded into the datums that reproduce them exactly, or +//! refused when they cannot be. + +use serde_json::Value as Json; + +use super::digest; +use crate::error::Result; +use crate::store::{Datum, Store}; +use crate::types::{Op, Value}; + +/// One feed record as the datum that reproduces it, or why it cannot be +/// reproduced exactly. The outer `Result` is a store failure. +/// +/// # Errors +/// [`Error::Sqlite`] if interning a term fails. +pub fn record_to_datum( + store: &Store, + record: &Json, + tx_timestamp: &str, +) -> Result> { + let (Some(entity), Some(attribute), Some(op)) = ( + record["entity"].as_str(), + record["attribute"].as_str(), + record["op"].as_str(), + ) else { + return Ok(Err("record lacks entity, attribute or op".into())); + }; + let retract = match digest::op_is_retract(op) { + Ok(r) => r, + Err(reason) => return Ok(Err(reason)), + }; + let value_field = if retract { "old_value" } else { "value" }; + let value = match value_of(store, &record[value_field])? { + Ok(v) => v, + Err(reason) => return Ok(Err(reason)), + }; + let valid_from = record["valid_from"] + .as_str() + .unwrap_or(tx_timestamp) + .to_string(); + Ok(Ok(Datum { + entity: store.intern(entity)?, + attribute: store.intern(attribute)?, + value, + valid_from, + // A retract names the fact it closes; its own row's valid time + // describes the closing, not an interval to replay. + valid_to: if retract { + None + } else { + record["valid_to"].as_str().map(String::from) + }, + op: if retract { Op::Retract } else { Op::Assert }, + })) +} + +fn value_of(store: &Store, json: &Json) -> Result> { + use base64::Engine as _; + Ok(Ok(match json { + Json::String(s) => Value::Str(s.clone()), + Json::Bool(b) => Value::Bool(*b), + Json::Number(n) => match n.as_i64() { + Some(i) => Value::Int(i), + None => match n.as_f64() { + Some(f) => Value::Float(f), + None => return Ok(Err(format!("unrepresentable number {n}"))), + }, + }, + Json::Object(o) if o.get("ref").and_then(Json::as_str).is_some() => { + Value::Ref(store.intern(o["ref"].as_str().unwrap_or_default())?) + } + Json::Object(o) if o.contains_key("lang") => Value::Lang { + lexical: o["value"].as_str().unwrap_or_default().into(), + lang: o["lang"].as_str().unwrap_or_default().into(), + }, + Json::Object(o) if o.contains_key("datatype") => Value::Typed { + lexical: o["value"].as_str().unwrap_or_default().into(), + datatype: o["datatype"].as_str().unwrap_or_default().into(), + }, + Json::Object(o) if o.get("base64").and_then(Json::as_str).is_some() => { + match base64::engine::general_purpose::STANDARD + .decode(o["base64"].as_str().unwrap_or_default()) + { + Ok(bytes) => Value::Bytes(bytes), + Err(_) => return Ok(Err("byte value is not valid base64".into())), + } + } + // A source from before byte bodies were carried sends only the length. + // Writing empty bytes would be a silent loss: refuse instead. + Json::Object(o) if o.contains_key("bytes_len") => { + return Ok(Err( + "byte value sent as its length only; the source predates lossless bytes".into(), + )); + } + other => return Ok(Err(format!("unreplayable value {other}"))), + })) +} diff --git a/src/applier_start.rs b/src/applier_start.rs new file mode 100644 index 000000000..691ab8a4b --- /dev/null +++ b/src/applier_start.rs @@ -0,0 +1,41 @@ +//! Which stores may become a replica. A replica must equal its source at its +//! cursor, so it starts with no facts of its own (or from a verified +//! bootstrap, which refuses a store holding facts on its own). + +use super::Applier; +use crate::error::Result; +use crate::store::Store; + +impl Applier { + /// Why `store` cannot become a new replica, or `None` when it can: it + /// already follows a source, or it holds no facts (shapes and settings are + /// not facts). Local facts with no cursor would be mixed into the replica, + /// which would then equal no source while claiming to. + /// + /// # Errors + /// [`Error::Sqlite`] if the store cannot be read. + pub fn new_replica_refusal(store: &Store) -> Result> { + let follows: bool = store.conn.query_row( + "SELECT EXISTS (SELECT 1 FROM sqlite_master \ + WHERE type = 'table' AND name = 'replica_progress')", + [], + |r| r.get(0), + )?; + if follows + && store + .conn + .query_row("SELECT EXISTS (SELECT 1 FROM replica_progress)", [], |r| { + r.get::<_, bool>(0) + })? + { + return Ok(None); + } + let holds_facts: bool = + store + .conn + .query_row("SELECT EXISTS (SELECT 1 FROM facts)", [], |r| r.get(0))?; + Ok(holds_facts.then(|| { + "a new replica must start with no facts; this store holds local facts".to_string() + })) + } +} diff --git a/src/applier_tests.rs b/src/applier_tests.rs index 78322d2ca..a266deeb4 100644 --- a/src/applier_tests.rs +++ b/src/applier_tests.rs @@ -531,3 +531,31 @@ fn outcomes_serialize_tagged_by_kind() { json!({"source_tx": 9, "last_frame_age_ms": null, "bounded": false}) ); } + +#[test] +fn a_store_holding_local_facts_cannot_start_a_replica() { + let src = source(); + let feed = pages(&src); + let mut replica = Store::open_in_memory().unwrap(); + let local = d( + &replica, + "urn:local", + "urn:p:title", + Value::Str("mine".into()), + Op::Assert, + ); + replica.transact(&[local], T, None, None).unwrap(); + let before = live_digest(&replica); + let mut applier = Applier::open(&replica).unwrap(); + let o = applier + .apply_frame(&mut replica, "quipu.changes", &feed[0], 1) + .unwrap(); + assert!(matches!(o, Outcome::Refused { .. }), "{o:?}"); + assert!(applier.progress().is_none()); + assert_eq!(live_digest(&replica), before); + // Once a replica follows a source, its own facts are the source's. + let mut empty = Store::open_in_memory().unwrap(); + let mut follower = Applier::open(&empty).unwrap(); + replay(&mut empty, &mut follower, &feed[..1]); + assert_eq!(Applier::new_replica_refusal(&empty).unwrap(), None); +} diff --git a/src/bindings.rs b/src/bindings.rs index 784120bd8..427c9ae59 100644 --- a/src/bindings.rs +++ b/src/bindings.rs @@ -142,6 +142,10 @@ impl BindingError { /// A local engine owns its store; adapters cannot mutate it around validation. pub struct Engine { store: Store, + /// Set once this store becomes a replica (the first `replica_*` call). + /// From then on it is read-only to local callers: only the applier + /// writes, and it writes what the source already accepted. + replica: Option, } impl Engine { @@ -154,7 +158,10 @@ impl Engine { { store.shacl_config_mut().validate_on_write = true; } - Ok(Self { store }) + Ok(Self { + store, + replica: None, + }) } /// Report this build and actual dispatch availability, not global parity. @@ -201,6 +208,14 @@ impl Engine { "operation is not available in this binding".into(), ) })?; + if self.replica.is_some() && writes(operation, input) { + return Err(BindingError::new( + "read_only_replica", + operation, + "not_started", + "this store is a replica: only its source's feed writes to it".into(), + )); + } if op.requires_shacl && !cfg!(feature = "shacl") { return Err(BindingError::new( "unsupported_capability", @@ -226,6 +241,85 @@ impl Engine { }) } + /// Apply one feed frame through the shared applier, making this store a + /// read-only replica on first use. Only a store with no facts (or one that + /// already follows a source) is promoted; any other is refused unchanged. Returns the `Outcome` JSON (tagged by + /// `kind`); a protocol refusal is a returned outcome, never an error. + /// + /// The source already validated what it accepted, so the replica applies + /// it verbatim: local write validation is suspended for the applier's own + /// writes and restored afterwards. Re-judging source data against local + /// shapes could refuse it and silently fork the replica. + /// + /// # Errors + /// [`BindingError`] if the local store fails; the frame is not applied. + pub fn replica_apply_frame( + &mut self, + event: &str, + data: &str, + now_ms: u64, + ) -> std::result::Result { + const OP: &str = "replica_apply_frame"; + if self.replica.is_none() { + // A store holding local facts is not promoted: it stays writable, + // and the frame is refused with nothing applied. + if let Some(reason) = crate::applier::Applier::new_replica_refusal(&self.store) + .map_err(|error| BindingError::native(OP, error))? + { + return serde_json::to_value(crate::applier::Outcome::Refused { reason }).map_err( + |error| BindingError::new("internal", OP, "unknown", error.to_string()), + ); + } + let applier = crate::applier::Applier::open(&self.store) + .map_err(|error| BindingError::native(OP, error))?; + self.replica = Some(applier); + } + let applier = self.replica.as_mut().expect("opened above"); + #[cfg(feature = "shacl")] + let validate = + std::mem::replace(&mut self.store.shacl_config_mut().validate_on_write, false); + let outcome = applier.apply_frame(&mut self.store, event, data, now_ms); + #[cfg(feature = "shacl")] + { + self.store.shacl_config_mut().validate_on_write = validate; + } + let outcome = outcome.map_err(|error| BindingError::native(OP, error))?; + serde_json::to_value(outcome) + .map_err(|error| BindingError::new("internal", OP, "unknown", error.to_string())) + } + + /// The replica cursor `{store_id, log_epoch, source_tx}`, or `null` when + /// this store is not (yet) a replica. + #[must_use] + pub fn replica_progress(&self) -> Value { + match self + .replica + .as_ref() + .and_then(crate::applier::Applier::progress) + { + Some(p) => json!({ + "store_id": p.source.store_id, + "log_epoch": p.source.log_epoch, + "source_tx": p.source_tx, + }), + None => Value::Null, + } + } + + /// `{source_tx, last_frame_age_ms, bounded}`; `bounded` is always false. + #[must_use] + pub fn replica_freshness(&self, now_ms: u64) -> Value { + let freshness = match &self.replica { + Some(applier) => applier.freshness(now_ms), + None => crate::applier::Freshness { + source_tx: None, + last_frame_age_ms: None, + bounded: false, + }, + }; + serde_json::to_value(freshness).unwrap_or(Value::Null) + } + /// JSON-text transport preserves native integer lexemes. JS adapters must /// not parse this into lossy Numbers without an explicit conversion policy. pub fn invoke_json( @@ -248,5 +342,18 @@ impl Engine { } } +/// Whether `operation` with `input` writes to the store. A replica refuses +/// these: its only writer is the applier. +fn writes(operation: &str, input: &Value) -> bool { + match operation { + "quipu_knot" | "quipu_episode" | "quipu_set" | "quipu_retract" => true, + "quipu_shapes" => matches!( + input.get("action").and_then(Value::as_str), + Some("load" | "remove") + ), + _ => false, + } +} + #[cfg(test)] mod tests; diff --git a/src/bindings/tests.rs b/src/bindings/tests.rs index d74b52da8..f55e6879b 100644 --- a/src/bindings/tests.rs +++ b/src/bindings/tests.rs @@ -1,6 +1,5 @@ use super::*; -#[cfg(feature = "shacl")] const TS: &str = "2026-10-20T00:00:00Z"; #[test] @@ -209,3 +208,151 @@ fn binding_and_direct_native_operations_return_the_same_results() { crate::tool_query(&direct.store, &input).unwrap() ); } + +/// The source's feed, one transaction per page, as frame text. +fn feed_of(source: &Store) -> Vec { + let mut pages = Vec::new(); + let mut cursor = 0; + loop { + let page = source + .changes_after( + cursor, + 1, + crate::store::changes::Capture::OldAndNewValues, + None, + ) + .unwrap(); + if page.records.is_empty() { + return pages; + } + cursor = page.next_tx; + pages.push(page.to_json().to_string()); + } +} + +fn record_source() -> Store { + let mut source = Store::open_in_memory().unwrap(); + let typ = source + .intern("http://www.w3.org/1999/02/22-rdf-syntax-ns#type") + .unwrap(); + let r = source.intern("urn:r1").unwrap(); + let class = source.intern("urn:Record").unwrap(); + let datum = crate::store::Datum { + entity: r, + attribute: typ, + value: crate::types::Value::Ref(class), + valid_from: TS.into(), + valid_to: None, + op: crate::types::Op::Assert, + }; + source.transact(&[datum], TS, None, None).unwrap(); + source +} + +#[test] +fn a_replica_applies_its_source_feed_and_refuses_local_writes() { + let source = record_source(); + let mut engine = Engine::open_in_memory().unwrap(); + assert_eq!(engine.replica_progress(), Value::Null); + for page in feed_of(&source) { + let outcome = engine + .replica_apply_frame("quipu.changes", &page, 10) + .unwrap(); + assert_eq!(outcome["kind"], "applied", "{outcome}"); + } + assert_eq!( + engine.replica_progress()["source_tx"], + json!(source.latest_tx_id().unwrap()) + ); + assert_eq!(engine.replica_freshness(15)["last_frame_age_ms"], json!(5)); + assert_eq!(engine.replica_freshness(15)["bounded"], json!(false)); + // Reads still work. + let read = engine + .invoke( + "quipu_query", + &json!({"query": "SELECT ?s WHERE { ?s a }"}), + ) + .unwrap(); + assert!(read.to_string().contains("urn:r1"), "{read}"); + // Local writes are refused before dispatch: only the feed writes here. + let refused = engine + .invoke( + "quipu_set", + &json!({"entity": "urn:r1", "attribute": "urn:title", "value": "x"}), + ) + .unwrap_err(); + assert_eq!(refused.code, "read_only_replica"); + assert_eq!(refused.commit_state, "not_started"); +} + +/// A store with local facts is never promoted: the frame is refused, nothing is +/// applied, and the store stays writable. +#[test] +fn a_store_holding_local_facts_is_not_promoted_to_a_replica() { + let source = record_source(); + let mut engine = Engine::open_in_memory().unwrap(); + let local = crate::store::Datum { + entity: engine.store.intern("urn:local").unwrap(), + attribute: engine.store.intern("urn:title").unwrap(), + value: crate::types::Value::Str("mine".into()), + valid_from: TS.into(), + valid_to: None, + op: crate::types::Op::Assert, + }; + engine.store.transact(&[local], TS, None, None).unwrap(); + for page in feed_of(&source) { + let outcome = engine + .replica_apply_frame("quipu.changes", &page, 1) + .unwrap(); + assert_eq!(outcome["kind"], "refused", "{outcome}"); + } + assert_eq!(engine.replica_progress(), Value::Null); + assert!(engine.store.lookup("urn:r1").unwrap().is_none()); + // Still an ordinary store: a local write is not refused as a replica's. + let write = engine.invoke( + "quipu_set", + &json!({"entity": "urn:local", "attribute": "urn:title", "value": "again"}), + ); + assert!( + write + .as_ref() + .err() + .is_none_or(|e| e.code != "read_only_replica"), + "{write:?}" + ); +} + +/// The source already judged what it accepted: a replica that re-judged it +/// against its own shapes could refuse it and silently fork. +#[cfg(feature = "shacl")] +#[test] +fn a_replica_applies_source_data_its_local_shapes_would_refuse() { + // Control: the same data written locally IS refused by the adopted shape. + let mut local = Engine::open_in_memory().unwrap(); + load_shape(&mut local); + let control = local.invoke( + "quipu_knot", + &json!({"turtle": " a .", "timestamp": TS}), + ); + let refused = match control { + Err(_) => true, + Ok(v) => v.get("conforms") == Some(&json!(false)), + }; + assert!( + refused, + "control: the shape must refuse an untitled record locally" + ); + + let source = record_source(); + let mut replica = Engine::open_in_memory().unwrap(); + load_shape(&mut replica); + for page in feed_of(&source) { + let outcome = replica + .replica_apply_frame("quipu.changes", &page, 1) + .unwrap(); + assert_eq!(outcome["kind"], "applied", "{outcome}"); + } + assert!(replica.store.lookup("urn:r1").unwrap().is_some()); + // Validation is restored for anything that is not the applier. + assert!(replica.store.shacl_config_mut().validate_on_write); +} diff --git a/wasm/explorer/src/engine.rs b/wasm/explorer/src/engine.rs index 0917dd0ef..853c13c58 100644 --- a/wasm/explorer/src/engine.rs +++ b/wasm/explorer/src/engine.rs @@ -26,6 +26,44 @@ impl Database { serde_json::to_string(&self.engine.capabilities()).map_err(super::err_js) } + /// Apply one feed frame (the SSE or WebSocket `event` and `data` text) + /// through the shared applier. The first call makes this store a + /// read-only replica: afterwards only its source's feed writes to it. + /// Returns the outcome as JSON text tagged by `kind`; a protocol refusal + /// is a returned outcome, and only a local store failure throws. + #[wasm_bindgen(js_name = replicaApplyFrame)] + pub fn replica_apply_frame( + &mut self, + event: &str, + data: &str, + now_ms: f64, + ) -> Result { + let outcome = self + .engine + .replica_apply_frame(event, data, millis(now_ms)) + .map_err(|error| { + JsValue::from_str( + &serde_json::to_string(&error).unwrap_or_else(|_| error.to_string()), + ) + })?; + serde_json::to_string(&outcome).map_err(super::err_js) + } + + /// The replica cursor as JSON text (`{store_id, log_epoch, source_tx}`), + /// or `null` while this store is not a replica. + #[wasm_bindgen(js_name = replicaProgress)] + pub fn replica_progress(&self) -> Result { + serde_json::to_string(&self.engine.replica_progress()).map_err(super::err_js) + } + + /// `{source_tx, last_frame_age_ms, bounded}` as JSON text. `bounded` stays + /// false until a freshness challenge exists. + #[wasm_bindgen(js_name = replicaFreshness)] + pub fn replica_freshness(&self, now_ms: f64) -> Result { + serde_json::to_string(&self.engine.replica_freshness(millis(now_ms))) + .map_err(super::err_js) + } + /// Invoke a supported native handler with lossless JSON text in and out. /// Errors are JSON text including commit state and native validation data. pub fn invoke(&mut self, operation: &str, input: &str) -> Result { @@ -37,3 +75,15 @@ impl Database { }) } } + +/// `Date.now()` arrives as an f64; clamp it into whole non-negative milliseconds. +fn millis(now_ms: f64) -> u64 { + if now_ms.is_finite() && now_ms > 0.0 { + // Truncation is intended: sub-millisecond precision is meaningless here. + #[allow(clippy::cast_possible_truncation, clippy::cast_sign_loss)] + let ms = now_ms as u64; + ms + } else { + 0 + } +}