diff --git a/docs/book/src/architecture/change-feed.md b/docs/book/src/architecture/change-feed.md index 2e29257c4..ebc87ec93 100644 --- a/docs/book/src/architecture/change-feed.md +++ b/docs/book/src/architecture/change-feed.md @@ -79,6 +79,7 @@ Both return one page: "valid_from": "2026-04-03", "valid_to": null } ], + "since": 42, "next_tx": 43, "watermark_tx": 43, "watermark_timestamp": "2026-04-03", @@ -88,7 +89,12 @@ Both return one page: ``` Pass `next_tx` back as the cursor (`--from` / `since`). On an empty page it -stays put, so polling is a fixpoint, not a rewind. `Ref` values resolve to +stays put, so polling is a fixpoint, not a rewind. `since` echoes the cursor +the page answers: the page holds every transaction in scope in +`(since, next_tx]`. A follower whose own cursor is below `since` has missed +the transactions between them and must resume from its cursor rather than +apply the page. The chunked `begin` and `commit` frames carry the same +`since`. `Ref` values resolve to `{"ref": ""}` so a consumer can tell an edge from a string that happens to look like one. Bytes arrive as `{"bytes_len": n, "base64": "..."}` — the whole body, so a replica built from the feed holds the same value. diff --git a/docs/knowledge/repository-knowledge.json b/docs/knowledge/repository-knowledge.json index 881ff8d47..5f782aa72 100644 --- a/docs/knowledge/repository-knowledge.json +++ b/docs/knowledge/repository-knowledge.json @@ -35,7 +35,7 @@ "source": "src/lib.rs", "subject": "http://aegis.gastown.local/ontology/code/quipu/src%2Flib.rs", "summary": "The Rust library exposes the embedded graph engine used by native clients and repository integrations.", - "source_sha256": "824b93dc8e56c4199a7ded61398ea444524ed106bf8bc9dec5e58ef879bb0784", + "source_sha256": "827def19e784624513b3c4e0d67ae936e31fefb12bf67f464e9239ec3a3fee53", "properties": { "interfaceKind": "library", "publicName": "quipu" diff --git a/src/applier.rs b/src/applier.rs new file mode 100644 index 000000000..f3e30f80b --- /dev/null +++ b/src/applier.rs @@ -0,0 +1,646 @@ +//! The shared applier, slice 1 (`docs/design/shared-applier-api.md`). +//! +//! ONE component turns the committed feed into a read-only replica, so the +//! HA follower, a scoped follower and the in-page browser store never write +//! their own projection logic. This slice consumes the feed frames that ship +//! today (`/changes` pages and the chunked `begin`/`part`/`commit` frames of an +//! oversized transaction) as `(event, data)` text pairs, exactly as SSE or a +//! WebSocket delivers them. It does no I/O of its own, so it builds for the +//! native server and for `wasm32` alike. +//! +//! The rules are the contract's, in the order [`Applier::apply_frame`] checks +//! them: +//! +//! - **One source, one epoch.** The first frame binds the replica to its source +//! `(store_id, log_epoch)`. A frame from any other source means a restore, a +//! clone or removed history: re-bootstrap, never splice. +//! - **Whole transactions only.** Each source transaction is applied in ONE +//! local savepoint together with the progress record, so the replica is +//! always at a source transaction boundary. A chunked transaction is +//! buffered, its digest chain checked part by part, and nothing is visible +//! until the commit verifies. +//! - **Idempotent, and checked.** A transaction at or below the cursor is a +//! duplicate only when its content digest equals the one retained when it +//! was applied. A different digest, or none retained, stops the replica: +//! its history and the source's disagree, and nothing more is applied. +//! - **No gaps.** A page covers every transaction in scope after the `since` +//! it names, so a page (or a chunked transaction) whose `since` is above +//! the cursor would skip whatever lies between: resume from the cursor. +//! Until a bootstrap exists, a new replica starts at transaction 0. A page +//! arriving while a chunked transaction is still open breaks the sequence +//! the same way: the chunk is discarded and the caller resumes. +//! - **Refuse rather than guess.** A record the replica cannot reproduce +//! exactly (a byte value sent as its length only, an unknown value shape) +//! refuses the whole transaction. Writing an approximation would make the +//! replica disagree with its source without saying so. +//! +//! Freshness is reported as (last applied transaction, time since the last +//! frame) and never claims a bound: a bounded claim needs the source's +//! freshness challenge, which a later slice adds. + +use std::collections::BTreeMap; + +use rusqlite::{OptionalExtension, params}; +use serde_json::Value as Json; +use sha2::{Digest, Sha256}; + +use crate::error::Result; +use crate::store::{Datum, Store}; +use crate::types::{Op, Value}; + +#[path = "applier_digest.rs"] +mod digest; +pub use digest::Stop; + +/// The source log a replica follows. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct Source { + /// The source store's identity, minted at its creation. + pub store_id: String, + /// The source log's epoch; re-minted by any history-removing operation. + pub log_epoch: String, +} + +/// The replica's cursor, persisted with every applied transaction. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct Progress { + /// The source this replica follows. + pub source: Source, + /// The last source transaction applied in full. + pub source_tx: i64, +} + +/// What one frame did, and what the caller must do next. +/// +/// Serialized (for the browser bridge) as an object tagged by `kind`, in +/// snake case: `{"kind": "applied", "source_txs": [..]}`, `{"kind": "resume", +/// "from": N}`, `{"kind": "duplicate"}`, and so on. +#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize)] +#[serde(tag = "kind", rename_all = "snake_case")] +pub enum Outcome { + /// These source transactions are now visible, with the cursor moved past them. + Applied { + /// The source transactions applied, in commit order. + source_txs: Vec, + }, + /// A part of a chunked transaction is held; nothing is visible yet. + Buffered { + /// The chunked source transaction being assembled. + source_tx: i64, + }, + /// Everything in the frame was at or below the cursor. + Duplicate, + /// A page with no records: the feed has nothing new. + Idle, + /// The frame does not continue from the cursor (a dropped page, or a + /// broken or interrupted chunk): nothing was applied; reconnect from `from`. + Resume { + /// The cursor to reconnect from (the last applied source transaction). + from: i64, + }, + /// The source changed: build a new replica from a fresh bootstrap. + Rebootstrap { + /// Why the replica cannot continue from its cursor. + reason: String, + }, + /// The frame cannot be applied exactly. Nothing from the refused source + /// transaction was written; earlier whole transactions in the same page + /// may have been, and [`Applier::progress`] says where the cursor is. + Refused { + /// Why the frame cannot be applied exactly. + reason: String, + }, + /// An event this slice does not act on (heartbeats, newer frame kinds). + Ignored, + /// A redelivered transaction does not match the one applied, or cannot + /// be checked. Nothing from it was written, and every later frame answers + /// the same until the replica is rebuilt (a new source still answers + /// [`Outcome::Rebootstrap`]). + Stopped { + /// Which check failed. + reason: Stop, + }, +} + +/// What a reader may say about how current the replica is. +#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize)] +pub struct Freshness { + /// The last source transaction applied, if any. + pub source_tx: Option, + /// Milliseconds since the last frame of any kind; `None` before the first. + pub last_frame_age_ms: Option, + /// Always `false` in this slice: no freshness challenge has been answered. + pub bounded: bool, +} + +struct Chunk { + source_tx: i64, + parts: usize, + /// Parts accepted so far; the next part's index must equal it. + seen: usize, + records: usize, + digest: String, + hasher: Sha256, + buffered: Vec, +} + +/// The applier for one replica store. +pub struct Applier { + progress: Option, + chunk: Option, + last_frame_ms: Option, + stopped: Option, +} + +const PROGRESS_DDL: &str = "CREATE TABLE IF NOT EXISTS replica_progress (\ + id INTEGER PRIMARY KEY CHECK (id = 1), \ + store_id TEXT NOT NULL, log_epoch TEXT NOT NULL, source_tx INTEGER NOT NULL)"; + +impl Applier { + /// Open the applier for `store`, resuming its persisted cursor if any. + /// + /// # Errors + /// [`Error::Sqlite`] if the progress record cannot be read. + pub fn open(store: &Store) -> Result { + store.conn.execute_batch(PROGRESS_DDL)?; + store.conn.execute_batch(digest::APPLIED_TX_DDL)?; + store.conn.execute_batch(digest::STOP_DDL)?; + let progress = store + .conn + .query_row( + "SELECT store_id, log_epoch, source_tx FROM replica_progress WHERE id = 1", + [], + |r| { + Ok(Progress { + source: Source { + store_id: r.get(0)?, + log_epoch: r.get(1)?, + }, + source_tx: r.get(2)?, + }) + }, + ) + .optional()?; + Ok(Self { + progress, + chunk: None, + last_frame_ms: None, + stopped: digest::load_stop(store)?, + }) + } + + /// The persisted cursor, or `None` before the first applied transaction. + #[must_use] + pub fn progress(&self) -> Option<&Progress> { + self.progress.as_ref() + } + + /// What may be said about freshness at `now_ms` (never a bound here). + #[must_use] + pub fn freshness(&self, now_ms: u64) -> Freshness { + Freshness { + source_tx: self.progress.as_ref().map(|p| p.source_tx), + last_frame_age_ms: self.last_frame_ms.map(|t| now_ms.saturating_sub(t)), + bounded: false, + } + } + + /// Apply one feed frame, given as the transport delivered it. + /// + /// # Errors + /// [`Error::Sqlite`] or a write error if the local store fails; the frame + /// is then not applied and the cursor has not moved. + pub fn apply_frame( + &mut self, + store: &mut Store, + event: &str, + data: &str, + now_ms: u64, + ) -> Result { + self.last_frame_ms = Some(now_ms); + if let Some(stopped) = self.stopped.clone() { + // A new source still comes first: it means a new replica. + let changed = serde_json::from_str::(data) + .ok() + .as_ref() + .and_then(source_of) + .and_then(|source| self.check_source(&source)); + return Ok(changed.unwrap_or(stopped)); + } + match event { + "quipu.changes" => self.apply_page(store, data), + "quipu.changes.begin" => Ok(self.begin_chunk(data)), + "quipu.changes.part" => Ok(self.chunk_part(data)), + "quipu.changes.commit" => self.commit_chunk(store, data), + "quipu.changes.tx_too_large" => Ok(Outcome::Refused { + reason: "the source cannot send this transaction in frames".into(), + }), + _ => Ok(Outcome::Ignored), + } + } + + fn apply_page(&mut self, store: &mut Store, data: &str) -> Result { + // A page never arrives inside a chunked transaction; if one does, the + // chunk can no longer complete in order. + if self.chunk.take().is_some() { + return Ok(Outcome::Resume { + from: self.resume_point(), + }); + } + let Ok(page) = serde_json::from_str::(data) else { + return Ok(refused("page is not JSON")); + }; + let Some(source) = source_of(&page) else { + return Ok(refused("page does not name its source store and log epoch")); + }; + if let Some(outcome) = self.check_source(&source) { + return Ok(outcome); + } + if let Some(outcome) = self.check_since(&page) { + return Ok(outcome); + } + let records = page["records"].as_array().cloned().unwrap_or_default(); + if records.is_empty() { + return Ok(Outcome::Idle); + } + self.apply_records(store, &source, &records) + } + + fn begin_chunk(&mut self, data: &str) -> Outcome { + let Ok(begin) = serde_json::from_str::(data) else { + return refused("begin frame is not JSON"); + }; + let (Some(source_tx), Some(parts), Some(records), Some(digest)) = ( + begin["source_tx"].as_i64(), + begin["parts"].as_u64(), + begin["records"].as_u64(), + begin["digest"].as_str(), + ) else { + return refused("begin frame is incomplete"); + }; + if let Some(outcome) = self.check_since(&begin) { + self.chunk = None; + return outcome; + } + // A new begin replaces any chunk in progress: a reconnect replays an + // interrupted transaction from its first part. + self.chunk = Some(Chunk { + source_tx, + parts: usize::try_from(parts).unwrap_or(usize::MAX), + records: usize::try_from(records).unwrap_or(usize::MAX), + seen: 0, + digest: digest.to_string(), + hasher: Sha256::new(), + buffered: Vec::new(), + }); + Outcome::Buffered { source_tx } + } + + fn chunk_part(&mut self, data: &str) -> Outcome { + let resume = self.resume_point(); + let Some(chunk) = self.chunk.as_mut() else { + return Outcome::Resume { from: resume }; + }; + let Ok(part) = serde_json::from_str::(data) else { + self.chunk = None; + return Outcome::Resume { from: resume }; + }; + let expected_before = hex::encode(chunk.hasher.clone().finalize()); + let in_order = part["source_tx"].as_i64() == Some(chunk.source_tx) + && part["index"].as_u64() == u64::try_from(chunk.seen).ok() + && part["digest_before"].as_str() == Some(expected_before.as_str()); + if !in_order { + self.chunk = None; + return Outcome::Resume { from: resume }; + } + // The digest covers the part EXACTLY as sent, so it is hashed as text. + chunk.hasher.update(data.as_bytes()); + chunk.seen += 1; + chunk + .buffered + .extend(part["records"].as_array().cloned().unwrap_or_default()); + Outcome::Buffered { + source_tx: chunk.source_tx, + } + } + + fn commit_chunk(&mut self, store: &mut Store, data: &str) -> Result { + let resume = self.resume_point(); + let Some(chunk) = self.chunk.take() else { + return Ok(Outcome::Resume { from: resume }); + }; + let Ok(commit) = serde_json::from_str::(data) else { + return Ok(Outcome::Resume { from: resume }); + }; + let digest = hex::encode(chunk.hasher.clone().finalize()); + let complete = commit["source_tx"].as_i64() == Some(chunk.source_tx) + && commit["digest"].as_str() == Some(chunk.digest.as_str()) + && digest == chunk.digest + && chunk.seen == chunk.parts + && chunk.buffered.len() == chunk.records; + if !complete { + return Ok(Outcome::Resume { from: resume }); + } + let Some(source) = source_of(&commit) else { + return Ok(refused( + "chunked transaction does not name its source store and log epoch", + )); + }; + if let Some(outcome) = self.check_source(&source) { + return Ok(outcome); + } + if let Some(outcome) = self.check_since(&commit) { + return Ok(outcome); + } + self.apply_records(store, &source, &chunk.buffered) + } + + fn resume_point(&self) -> i64 { + self.progress.as_ref().map_or(0, |p| p.source_tx) + } + + /// `None` when the frame continues from the cursor: its `since` is at or + /// below it (the overlap is skipped as duplicates). + fn check_since(&self, frame: &Json) -> Option { + let Some(since) = frame["since"].as_i64() else { + return Some(refused("frame does not name the cursor it answers (since)")); + }; + let from = self.resume_point(); + (since > from).then_some(Outcome::Resume { from }) + } + + /// `None` when the frame's source is this replica's (or the first one). + fn check_source(&self, source: &Source) -> Option { + match &self.progress { + Some(p) if p.source != *source => Some(Outcome::Rebootstrap { + reason: format!( + "source changed from {}/{} to {}/{}", + p.source.store_id, p.source.log_epoch, source.store_id, source.log_epoch + ), + }), + _ => None, + } + } + + /// Apply whole source transactions in commit order, each in one local + /// savepoint with its progress record. + fn apply_records( + &mut self, + store: &mut Store, + source: &Source, + records: &[Json], + ) -> Result { + let mut by_tx: BTreeMap> = BTreeMap::new(); + for record in records { + let Some(tx) = record["tx"].as_i64() else { + return Ok(refused("record without a source transaction")); + }; + by_tx.entry(tx).or_default().push(record); + } + let mut applied = Vec::new(); + for (tx, records) in by_tx { + if self.progress.as_ref().is_some_and(|p| tx <= p.source_tx) { + if let Some(stop) = digest::check_duplicate(store, source, tx, &records)? { + self.stopped = Some(stop.clone()); + return Ok(stop); + } + continue; + } + match apply_one_tx(store, source, tx, &records)? { + None => { + self.progress = Some(Progress { + source: source.clone(), + source_tx: tx, + }); + applied.push(tx); + } + Some(reason) => return Ok(Outcome::Refused { reason }), + } + } + Ok(if applied.is_empty() { + Outcome::Duplicate + } else { + Outcome::Applied { + source_txs: applied, + } + }) + } +} + +fn refused(reason: &str) -> Outcome { + Outcome::Refused { + reason: reason.to_string(), + } +} + +fn source_of(frame: &Json) -> Option { + Some(Source { + store_id: frame["source"]["store_id"].as_str()?.to_string(), + log_epoch: frame["source"]["log_epoch"].as_str()?.to_string(), + }) +} + +/// One source transaction, all-or-nothing. `Ok(Some(reason))` refuses it with +/// nothing written; `Ok(None)` committed it and the cursor. +fn apply_one_tx( + store: &mut Store, + source: &Source, + tx: i64, + records: &[&Json], +) -> Result> { + let timestamp = records[0]["timestamp"] + .as_str() + .unwrap_or_default() + .to_string(); + // A transaction may touch several graphs: one local write per graph, all + // inside the same savepoint. + let mut by_graph: BTreeMap> = BTreeMap::new(); + for record in records { + let graph = record["graph"].as_str().unwrap_or("ROOT").to_string(); + by_graph.entry(graph).or_default().push(record); + } + store.conn.execute_batch("SAVEPOINT replica_apply")?; + let result = (|| -> Result> { + for (graph, records) in &by_graph { + let mut datums = Vec::with_capacity(records.len()); + for record in records { + match record_to_datum(store, record, ×tamp)? { + Ok(d) => datums.push(d), + Err(reason) => return Ok(Some(format!("source tx {tx}: {reason}"))), + } + } + if graph == "ROOT" { + store.transact(&datums, ×tamp, None, None)?; + } else { + let g = store.graph_create(graph)?; + store.transact_to_graph(&datums, ×tamp, None, None, g)?; + } + } + // Every record decoded above, so its exact encoding exists. + let tx_digest = match digest::tx_digest(source, tx, records) { + Ok(d) => d, + Err(reason) => return Ok(Some(format!("source tx {tx}: {reason}"))), + }; + store.conn.execute( + "INSERT INTO replica_progress (id, store_id, log_epoch, source_tx) VALUES (1, ?1, ?2, ?3) \ + ON CONFLICT(id) DO UPDATE SET store_id = ?1, log_epoch = ?2, source_tx = ?3", + params![source.store_id, source.log_epoch, tx], + )?; + digest::retain(store, source, tx, &tx_digest)?; + Ok(None) + })(); + match result { + Ok(None) => { + store.conn.execute_batch("RELEASE replica_apply")?; + Ok(None) + } + other => { + store + .conn + .execute_batch("ROLLBACK TO replica_apply; RELEASE replica_apply")?; + other + } + } +} + +/// 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. +#[cfg(test)] +type FactRow = (i64, i64, Vec, i64, String, Option); + +#[cfg(test)] +pub(crate) fn live_digest(store: &Store) -> std::collections::BTreeSet { + let mut stmt = store + .conn + .prepare( + "SELECT e, a, v, g, valid_from, valid_to FROM facts \ + WHERE op = 1 AND retracted_tx IS NULL", + ) + .unwrap(); + let rows: Vec = stmt + .query_map([], |r| { + Ok(( + r.get(0)?, + r.get(1)?, + r.get(2)?, + r.get(3)?, + r.get(4)?, + r.get(5)?, + )) + }) + .unwrap() + .collect::>() + .unwrap(); + rows.into_iter() + .map(|(e, a, v, g, from, to)| { + let value = match Value::from_bytes(&v).unwrap() { + Value::Ref(id) => format!("ref:{}", store.resolve(id).unwrap()), + Value::Bytes(b) => format!("bytes:{b:?}"), + other => format!("{other:?}"), + }; + let graph = if g == 0 { + "ROOT".to_string() + } else { + store.resolve(g).unwrap() + }; + format!( + "{graph} {} {} {value} [{from}, {to:?})", + store.resolve(e).unwrap(), + store.resolve(a).unwrap() + ) + }) + .collect() +} + +#[cfg(test)] +#[path = "applier_tests.rs"] +mod tests; diff --git a/src/applier_digest.rs b/src/applier_digest.rs new file mode 100644 index 000000000..dfc3cf3e8 --- /dev/null +++ b/src/applier_digest.rs @@ -0,0 +1,261 @@ +//! The applier's transaction digest and its retained `AppliedTx` rows +//! (`docs/design/shared-applier-api.md`: "Transaction digest", rule 6). +//! +//! A redelivered transaction is a duplicate only when its CONTENT equals the +//! one applied, so the applier needs the identity of what it applied. The +//! digest is computed from the decoded records, never from a digest a frame +//! declares, and it does not depend on framing: one page or a chunked +//! begin/part/commit of the same transaction hash the same. + +use rusqlite::{OptionalExtension, params}; +use serde_json::Value as Json; +use sha2::{Digest, Sha256}; + +use super::{Outcome, Source}; +use crate::error::Result; +use crate::store::Store; + +const DOMAIN: &[u8] = b"quipu-applier-tx/v1"; + +pub(super) const APPLIED_TX_DDL: &str = "CREATE TABLE IF NOT EXISTS replica_applied_tx (\ + store_id TEXT NOT NULL, log_epoch TEXT NOT NULL, source_tx INTEGER NOT NULL, \ + tx_digest TEXT NOT NULL, PRIMARY KEY (store_id, log_epoch, source_tx))"; + +/// Append one length-prefixed field, so no two field sequences encode alike. +fn put(out: &mut Vec, field: &[u8]) { + out.extend_from_slice(&u64::try_from(field.len()).unwrap_or(u64::MAX).to_be_bytes()); + out.extend_from_slice(field); +} + +/// The digest of one source transaction's records, or why a record cannot be +/// encoded exactly (the same records the applier refuses to write). +pub(super) fn tx_digest( + source: &Source, + tx: i64, + records: &[&Json], +) -> std::result::Result { + let timestamp = records + .first() + .and_then(|r| r["timestamp"].as_str()) + .unwrap_or_default(); + let mut encoded = Vec::with_capacity(records.len()); + for record in records { + encoded.push(encode_record(record, timestamp)?); + } + // Record order is not part of a transaction's identity. + encoded.sort(); + let mut out = Vec::new(); + put(&mut out, DOMAIN); + put(&mut out, source.store_id.as_bytes()); + put(&mut out, source.log_epoch.as_bytes()); + put(&mut out, &tx.to_be_bytes()); + for record in &encoded { + put(&mut out, record); + } + Ok(hex::encode(Sha256::digest(&out))) +} + +/// One record as the fields the applier writes, with the same defaults +/// (`valid_from` falls back to the transaction time; a retract carries no +/// `valid_to`). +fn encode_record(record: &Json, tx_timestamp: &str) -> std::result::Result, String> { + let (Some(entity), Some(attribute), Some(op)) = ( + record["entity"].as_str(), + record["attribute"].as_str(), + record["op"].as_str(), + ) else { + return Err("record lacks entity, attribute or op".into()); + }; + let retract = op_is_retract(op)?; + let value = encode_value(&record[if retract { "old_value" } else { "value" }])?; + let mut out = Vec::new(); + put( + &mut out, + record["graph"].as_str().unwrap_or("ROOT").as_bytes(), + ); + put(&mut out, entity.as_bytes()); + put(&mut out, attribute.as_bytes()); + put(&mut out, &value); + put( + &mut out, + record["valid_from"] + .as_str() + .unwrap_or(tx_timestamp) + .as_bytes(), + ); + match record["valid_to"].as_str().filter(|_| !retract) { + Some(to) => put(&mut out, &[b"\x01", to.as_bytes()].concat()), + None => put(&mut out, b"\x00"), + } + put(&mut out, if retract { b"retract" } else { b"assert" }); + Ok(out) +} + +/// A value's exact bytes and kind, decoded the way `value_of` decodes it. +fn encode_value(json: &Json) -> std::result::Result, String> { + use base64::Engine as _; + let str_of = |o: &serde_json::Map, k: &str| { + o.get(k) + .and_then(Json::as_str) + .unwrap_or_default() + .to_string() + }; + let mut out = Vec::new(); + match json { + Json::String(s) => { + put(&mut out, b"str"); + put(&mut out, s.as_bytes()); + } + Json::Bool(b) => { + put(&mut out, b"bool"); + put(&mut out, &[u8::from(*b)]); + } + Json::Number(n) => match (n.as_i64(), n.as_f64()) { + (Some(i), _) => { + put(&mut out, b"int"); + put(&mut out, &i.to_be_bytes()); + } + (None, Some(f)) => { + put(&mut out, b"float"); + put(&mut out, &f.to_bits().to_be_bytes()); + } + (None, None) => return Err(format!("unrepresentable number {n}")), + }, + Json::Object(o) if o.get("ref").and_then(Json::as_str).is_some() => { + put(&mut out, b"ref"); + put(&mut out, str_of(o, "ref").as_bytes()); + } + Json::Object(o) if o.contains_key("lang") => { + put(&mut out, b"lang"); + put(&mut out, str_of(o, "value").as_bytes()); + put(&mut out, str_of(o, "lang").as_bytes()); + } + Json::Object(o) if o.contains_key("datatype") => { + put(&mut out, b"typed"); + put(&mut out, str_of(o, "value").as_bytes()); + put(&mut out, str_of(o, "datatype").as_bytes()); + } + Json::Object(o) if o.get("base64").and_then(Json::as_str).is_some() => { + let bytes = base64::engine::general_purpose::STANDARD + .decode(str_of(o, "base64")) + .map_err(|_| "byte value is not valid base64".to_string())?; + put(&mut out, b"bytes"); + put(&mut out, &bytes); + } + other => return Err(format!("value {other} has no exact encoding")), + } + Ok(out) +} + +/// The feed's two ops. Anything else is refused, never read as an assert. +pub(super) fn op_is_retract(op: &str) -> std::result::Result { + match op { + "retract" => Ok(true), + "assert" => Ok(false), + other => Err(format!("unknown op {other:?}")), + } +} + +/// A replica's stop, persisted so a reopened applier stays stopped: a history +/// that disagreed with its source once is not trusted again by reopening. +pub(super) const STOP_DDL: &str = "CREATE TABLE IF NOT EXISTS replica_stop (\ + id INTEGER PRIMARY KEY CHECK (id = 1), reason TEXT NOT NULL)"; + +/// Record why the replica stopped (the first stop wins; it is never cleared +/// by the applier, only by rebuilding the replica). +pub(super) fn persist_stop(store: &Store, reason: &Stop) -> Result<()> { + let text = + serde_json::to_string(reason).map_err(|e| crate::Error::InvalidValue(e.to_string()))?; + store.conn.execute( + "INSERT OR IGNORE INTO replica_stop (id, reason) VALUES (1, ?1)", + params![text], + )?; + Ok(()) +} + +/// The persisted stop, if this replica stopped before. +pub(super) fn load_stop(store: &Store) -> Result> { + let text: Option = store + .conn + .query_row("SELECT reason FROM replica_stop WHERE id = 1", [], |r| { + r.get(0) + }) + .optional()?; + Ok(match text { + Some(t) => Some(Outcome::Stopped { + reason: serde_json::from_str(&t) + .map_err(|e| crate::Error::InvalidValue(e.to_string()))?, + }), + None => None, + }) +} + +/// Keep the digest of a transaction just applied (inside its savepoint). +pub(super) fn retain(store: &Store, source: &Source, tx: i64, digest: &str) -> Result<()> { + store.conn.execute( + "INSERT INTO replica_applied_tx (store_id, log_epoch, source_tx, tx_digest) \ + VALUES (?1, ?2, ?3, ?4)", + params![source.store_id, source.log_epoch, tx, digest], + )?; + Ok(()) +} + +/// The digest retained for an applied transaction, if one was kept. +pub(super) fn retained(store: &Store, source: &Source, tx: i64) -> Result> { + Ok(store + .conn + .query_row( + "SELECT tx_digest FROM replica_applied_tx \ + WHERE store_id = ?1 AND log_epoch = ?2 AND source_tx = ?3", + params![source.store_id, source.log_epoch, tx], + |r| r.get(0), + ) + .optional()?) +} + +/// Why a replica stopped. +#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)] +#[serde(tag = "kind", rename_all = "snake_case")] +pub enum Stop { + /// The redelivered transaction's content differs from the applied one. + DigestMismatch { + /// The source transaction. + source_tx: i64, + /// The digest retained when it was applied. + applied: String, + /// The digest of the redelivery (or why it has no exact encoding). + received: String, + }, + /// No digest was retained for the redelivered transaction. + DigestUnverifiable { + /// The source transaction. + source_tx: i64, + }, +} + +/// `None` when a redelivered transaction equals the one applied; otherwise +/// the [`Outcome::Stopped`] that halts the replica. +pub(super) fn check_duplicate( + store: &Store, + source: &Source, + tx: i64, + records: &[&Json], +) -> Result> { + let received = tx_digest(source, tx, records) + .unwrap_or_else(|reason| format!("no exact encoding: {reason}")); + let reason = match retained(store, source, tx)? { + Some(applied) if applied == received => return Ok(None), + Some(applied) => Stop::DigestMismatch { + source_tx: tx, + applied, + received, + }, + None => Stop::DigestUnverifiable { source_tx: tx }, + }; + persist_stop(store, &reason)?; + Ok(Some(Outcome::Stopped { reason })) +} + +#[cfg(test)] +#[path = "applier_digest_tests.rs"] +mod tests; diff --git a/src/applier_digest_tests.rs b/src/applier_digest_tests.rs new file mode 100644 index 000000000..6f4910b97 --- /dev/null +++ b/src/applier_digest_tests.rs @@ -0,0 +1,219 @@ +//! Rule 6 of the applier contract: a redelivered transaction is a duplicate +//! only when its content digest equals the retained one. An exact replay and +//! a changed replay of the same source transaction never answer alike. +use serde_json::{Value as Json, json}; + +use super::super::tests::{chunk_frames, pages, source}; +use super::super::{Applier, Outcome, Stop, live_digest}; +use super::*; + +fn replica_of(feed: &[String]) -> (Store, Applier) { + let mut replica = Store::open_in_memory().unwrap(); + let mut applier = Applier::open(&replica).unwrap(); + for page in feed { + let outcome = applier + .apply_frame(&mut replica, "quipu.changes", page, 1) + .unwrap(); + assert!(matches!(outcome, Outcome::Applied { .. }), "{outcome:?}"); + } + (replica, applier) +} + +/// The first page with its first record's value replaced. +fn tampered(page: &str) -> String { + let mut page: Json = serde_json::from_str(page).unwrap(); + page["records"][0]["value"] = json!("tampered"); + page.to_string() +} + +fn first_tx(page: &str) -> i64 { + serde_json::from_str::(page).unwrap()["records"][0]["tx"] + .as_i64() + .unwrap() +} + +#[test] +fn an_exact_redelivery_is_a_duplicate_as_a_page_or_as_a_chunk() { + let feed = pages(&source()); + let (mut replica, mut applier) = replica_of(&feed); + let before = live_digest(&replica); + let page = applier + .apply_frame(&mut replica, "quipu.changes", &feed[0], 2) + .unwrap(); + assert_eq!(page, Outcome::Duplicate); + // The same transaction framed differently has the same content digest. + let frames = chunk_frames(&feed[3], 1, true); + let last = frames + .iter() + .map(|(e, d)| applier.apply_frame(&mut replica, e, d, 3).unwrap()) + .last() + .unwrap(); + assert_eq!(last, Outcome::Duplicate); + assert_eq!(before, live_digest(&replica)); +} + +#[test] +fn a_changed_redelivery_stops_the_replica_and_writes_nothing() { + let src = source(); + let feed = pages(&src); + let (mut replica, mut applier) = replica_of(&feed); + let before = live_digest(&replica); + let outcome = applier + .apply_frame(&mut replica, "quipu.changes", &tampered(&feed[0]), 2) + .unwrap(); + let Outcome::Stopped { + reason: + Stop::DigestMismatch { + source_tx, + applied, + received, + }, + } = &outcome + else { + panic!("{outcome:?}"); + }; + assert_eq!(*source_tx, first_tx(&feed[0])); + assert_ne!(applied, received); + assert_eq!(before, live_digest(&replica)); + // Sticky: even an exact redelivery now answers the same stop. + let again = applier + .apply_frame(&mut replica, "quipu.changes", &feed[0], 3) + .unwrap(); + assert_eq!(again, outcome); + let begin = applier + .apply_frame(&mut replica, "quipu.changes.begin", "{}", 4) + .unwrap(); + assert_eq!(begin, outcome); +} + +#[test] +fn a_redelivery_with_no_retained_digest_is_unverifiable() { + let feed = pages(&source()); + let (mut replica, mut applier) = replica_of(&feed); + let tx = first_tx(&feed[0]); + replica + .conn + .execute("DELETE FROM replica_applied_tx WHERE source_tx = ?1", [tx]) + .unwrap(); + let outcome = applier + .apply_frame(&mut replica, "quipu.changes", &feed[0], 2) + .unwrap(); + assert_eq!( + outcome, + Outcome::Stopped { + reason: Stop::DigestUnverifiable { source_tx: tx } + } + ); +} + +#[test] +fn a_new_source_after_a_stop_still_means_rebootstrap() { + let feed = pages(&source()); + let (mut replica, mut applier) = replica_of(&feed); + applier + .apply_frame(&mut replica, "quipu.changes", &tampered(&feed[0]), 2) + .unwrap(); + let mut moved: Json = serde_json::from_str(&feed[0]).unwrap(); + moved["source"]["log_epoch"] = json!("another-epoch"); + let outcome = applier + .apply_frame(&mut replica, "quipu.changes", &moved.to_string(), 3) + .unwrap(); + assert!( + matches!(outcome, Outcome::Rebootstrap { .. }), + "{outcome:?}" + ); +} + +#[test] +fn retained_digests_survive_reopening() { + let feed = pages(&source()); + let (mut replica, _) = replica_of(&feed); + let mut reopened = Applier::open(&replica).unwrap(); + let outcome = reopened + .apply_frame(&mut replica, "quipu.changes", &tampered(&feed[0]), 2) + .unwrap(); + assert!( + matches!( + outcome, + Outcome::Stopped { + reason: Stop::DigestMismatch { .. } + } + ), + "{outcome:?}" + ); +} + +#[test] +fn the_digest_ignores_record_order_but_not_identity() { + let page: Json = serde_json::from_str(&pages(&source())[3]).unwrap(); + let records: Vec<&Json> = page["records"].as_array().unwrap().iter().collect(); + assert!( + records.len() > 1, + "the retract+assert transaction has two records" + ); + let source = Source { + store_id: "s".into(), + log_epoch: "e".into(), + }; + let forward = tx_digest(&source, 4, &records).unwrap(); + let reversed: Vec<&Json> = records.iter().rev().copied().collect(); + assert_eq!(forward, tx_digest(&source, 4, &reversed).unwrap()); + assert_ne!(forward, tx_digest(&source, 5, &records).unwrap()); + let other = Source { + store_id: "s".into(), + log_epoch: "f".into(), + }; + assert_ne!(forward, tx_digest(&other, 4, &records).unwrap()); +} + +#[test] +fn a_stop_serializes_tagged_by_kind() { + let stop = Outcome::Stopped { + reason: Stop::DigestUnverifiable { source_tx: 5 }, + }; + assert_eq!( + serde_json::to_value(stop).unwrap(), + json!({"kind": "stopped", "reason": {"kind": "digest_unverifiable", "source_tx": 5}}) + ); +} + +#[test] +fn a_stop_survives_reopening_so_a_diverged_replica_never_resumes() { + let feed = pages(&source()); + let (mut replica, mut applier) = replica_of(&feed[..2]); + let stop = applier + .apply_frame(&mut replica, "quipu.changes", &tampered(&feed[0]), 2) + .unwrap(); + assert!(matches!(stop, Outcome::Stopped { .. }), "{stop:?}"); + drop(applier); + // A fresh applier on the same replica must not tail on past the stop. + let mut reopened = Applier::open(&replica).unwrap(); + let before = live_digest(&replica); + let next = reopened + .apply_frame(&mut replica, "quipu.changes", &feed[2], 3) + .unwrap(); + assert_eq!(next, stop, "the persisted stop answers after reopening"); + assert_eq!( + before, + live_digest(&replica), + "nothing applied after the stop" + ); +} + +#[test] +fn an_unknown_op_is_refused_never_read_as_an_assert() { + let feed = pages(&source()); + let mut page: Json = serde_json::from_str(&feed[0]).unwrap(); + page["records"][0]["op"] = json!("Assert"); + let mut replica = Store::open_in_memory().unwrap(); + let mut applier = Applier::open(&replica).unwrap(); + let outcome = applier + .apply_frame(&mut replica, "quipu.changes", &page.to_string(), 1) + .unwrap(); + assert!( + matches!(&outcome, Outcome::Refused { reason } if reason.contains("unknown op")), + "{outcome:?}" + ); + assert!(live_digest(&replica).is_empty(), "nothing written"); + assert!(applier.progress().is_none()); +} diff --git a/src/applier_tests.rs b/src/applier_tests.rs new file mode 100644 index 000000000..78322d2ca --- /dev/null +++ b/src/applier_tests.rs @@ -0,0 +1,533 @@ +//! The shared applier against the shipped feed: equal replicas from whole +//! pages and from chunked transactions, and a refusal (with nothing written) +//! for every case the contract says must not splice. +use serde_json::{Value as Json, json}; +use sha2::{Digest, Sha256}; + +use super::*; +use crate::store::changes::Capture; + +const T: &str = "2026-10-24T00:00:00Z"; + +fn d(store: &Store, e: &str, a: &str, v: Value, op: Op) -> Datum { + Datum { + entity: store.intern(e).unwrap(), + attribute: store.intern(a).unwrap(), + value: v, + valid_from: T.into(), + valid_to: None, + op, + } +} + +/// A source with ROOT and named-graph writes, a retraction and bytes. +pub(super) fn source() -> Store { + let mut s = Store::open_in_memory().unwrap(); + let r = |s: &Store, iri: &str| Value::Ref(s.intern(iri).unwrap()); + let a = d( + &s, + "urn:item:1", + "urn:p:status", + Value::Str("open".into()), + Op::Assert, + ); + s.transact(&[a], T, None, None).unwrap(); + let b = d( + &s, + "urn:item:1", + "urn:p:blob", + Value::Bytes(vec![0, 1, 2, 255]), + Op::Assert, + ); + s.transact(&[b], T, None, None).unwrap(); + let g = s.graph_create("urn:graph:board").unwrap(); + let link = r(&s, "urn:item:2"); + let c = d(&s, "urn:item:1", "urn:p:blocks", link, Op::Assert); + s.transact_to_graph(&[c], T, None, None, g).unwrap(); + let retract = d( + &s, + "urn:item:1", + "urn:p:status", + Value::Str("open".into()), + Op::Retract, + ); + let closed = d( + &s, + "urn:item:1", + "urn:p:status", + Value::Str("closed".into()), + Op::Assert, + ); + s.transact(&[retract, closed], T, None, None).unwrap(); + s +} + +/// The feed one transaction per page, as `/changes/stream` sends it. +pub(super) fn pages(source: &Store) -> Vec { + let mut out = Vec::new(); + let mut cursor = 0; + loop { + let page = source + .changes_after(cursor, 1, Capture::OldAndNewValues, None) + .unwrap(); + if page.records.is_empty() { + return out; + } + cursor = page.next_tx; + out.push(page.to_json().to_string()); + } +} + +fn replay(replica: &mut Store, applier: &mut Applier, pages: &[String]) -> Vec { + pages + .iter() + .map(|p| applier.apply_frame(replica, "quipu.changes", p, 1).unwrap()) + .collect() +} + +#[test] +fn the_feed_builds_a_replica_equal_to_its_source() { + let src = source(); + let mut replica = Store::open_in_memory().unwrap(); + let mut applier = Applier::open(&replica).unwrap(); + let outcomes = replay(&mut replica, &mut applier, &pages(&src)); + assert!( + outcomes + .iter() + .all(|o| matches!(o, Outcome::Applied { .. })), + "{outcomes:?}" + ); + assert_eq!(live_digest(&src), live_digest(&replica)); + assert_eq!( + applier.progress().unwrap().source_tx, + src.latest_tx_id().unwrap() + ); +} + +#[test] +fn a_repeated_feed_is_a_duplicate_and_changes_nothing() { + let src = source(); + let mut replica = Store::open_in_memory().unwrap(); + let mut applier = Applier::open(&replica).unwrap(); + let feed = pages(&src); + replay(&mut replica, &mut applier, &feed); + let before = live_digest(&replica); + let again = replay(&mut replica, &mut applier, &feed); + assert!(again.iter().all(|o| *o == Outcome::Duplicate), "{again:?}"); + assert_eq!(before, live_digest(&replica)); +} + +#[test] +fn a_new_source_epoch_means_rebootstrap_never_splice() { + let src = source(); + let mut replica = Store::open_in_memory().unwrap(); + let mut applier = Applier::open(&replica).unwrap(); + let feed = pages(&src); + replay(&mut replica, &mut applier, &feed[..2]); + let before = live_digest(&replica); + let mut page: Json = serde_json::from_str(&feed[2]).unwrap(); + page["source"]["log_epoch"] = json!("another-epoch"); + let outcome = applier + .apply_frame(&mut replica, "quipu.changes", &page.to_string(), 2) + .unwrap(); + assert!( + matches!(outcome, Outcome::Rebootstrap { .. }), + "{outcome:?}" + ); + assert_eq!(before, live_digest(&replica), "nothing may be spliced in"); +} + +#[test] +fn a_page_without_its_source_is_refused() { + let src = source(); + let mut replica = Store::open_in_memory().unwrap(); + let mut applier = Applier::open(&replica).unwrap(); + let mut page: Json = serde_json::from_str(&pages(&src)[0]).unwrap(); + page.as_object_mut().unwrap().remove("source"); + let outcome = applier + .apply_frame(&mut replica, "quipu.changes", &page.to_string(), 1) + .unwrap(); + assert!(matches!(outcome, Outcome::Refused { .. }), "{outcome:?}"); + assert!(applier.progress().is_none()); +} + +#[test] +fn a_length_only_byte_value_is_refused_and_nothing_is_written() { + let src = source(); + let mut replica = Store::open_in_memory().unwrap(); + let mut applier = Applier::open(&replica).unwrap(); + let feed = pages(&src); + replay(&mut replica, &mut applier, &feed[..1]); + let before = (live_digest(&replica), applier.progress().cloned()); + // The second source transaction carries the bytes: strip them to the + // length-only form a pre-F4 source sends. + let mut page: Json = serde_json::from_str(&feed[1]).unwrap(); + for record in page["records"].as_array_mut().unwrap() { + if record["value"].get("base64").is_some() { + record["value"] = json!({"bytes_len": 4}); + } + } + let outcome = applier + .apply_frame(&mut replica, "quipu.changes", &page.to_string(), 2) + .unwrap(); + assert!( + matches!(&outcome, Outcome::Refused { reason } if reason.contains("length only")), + "{outcome:?}" + ); + assert_eq!(before, (live_digest(&replica), applier.progress().cloned())); +} + +/// The chunked frames for one page, in the shape the server sends them +/// (begin, parts with a running digest of the parts as sent, commit). The +/// server builds these in `feed_stream::chunk_transaction`; this mirrors its +/// output format for a library test. +pub(super) fn chunk_frames( + page: &str, + per_part: usize, + with_source: bool, +) -> Vec<(&'static str, String)> { + let page: Json = serde_json::from_str(page).unwrap(); + let records = page["records"].as_array().unwrap().clone(); + let tx = page["next_tx"].as_i64().unwrap(); + let mut hasher = Sha256::new(); + let mut parts = Vec::new(); + for (index, chunk) in records.chunks(per_part).enumerate() { + let before = hex::encode(hasher.clone().finalize()); + let data = + json!({"source_tx": tx, "index": index, "digest_before": before, "records": chunk}) + .to_string(); + hasher.update(data.as_bytes()); + parts.push(data); + } + let digest = hex::encode(hasher.finalize()); + let mut frames = vec![( + "quipu.changes.begin", + json!({"source_tx": tx, "parts": parts.len(), "records": records.len(), "digest": digest, "since": page["since"]}) + .to_string(), + )]; + frames.extend(parts.into_iter().map(|p| ("quipu.changes.part", p))); + let mut commit = json!({"source_tx": tx, "parts": frames.len() - 1, "records": records.len(), "digest": digest, "since": page["since"]}); + if with_source { + commit["source"] = page["source"].clone(); + } + frames.push(("quipu.changes.commit", commit.to_string())); + frames +} + +fn apply(replica: &mut Store, applier: &mut Applier, frames: &[(&str, String)]) -> Vec { + frames + .iter() + .map(|(e, data)| applier.apply_frame(replica, e, data, 1).unwrap()) + .collect() +} + +#[test] +fn a_dropped_page_resumes_with_nothing_applied() { + let src = source(); + let feed = pages(&src); + let mut replica = Store::open_in_memory().unwrap(); + let mut applier = Applier::open(&replica).unwrap(); + replay(&mut replica, &mut applier, &feed[..1]); + let cursor = applier.progress().unwrap().source_tx; + let before = live_digest(&replica); + // Page 1 is lost in transit; page 2 answers a cursor ahead of ours. + let o = applier + .apply_frame(&mut replica, "quipu.changes", &feed[2], 2) + .unwrap(); + assert_eq!(o, Outcome::Resume { from: cursor }); + assert_eq!(before, live_digest(&replica), "nothing may be spliced in"); + assert_eq!(applier.progress().unwrap().source_tx, cursor); + // Resuming from the cursor converges. + replay(&mut replica, &mut applier, &feed[1..]); + assert_eq!(live_digest(&src), live_digest(&replica)); +} + +#[test] +fn a_new_replica_starts_at_transaction_zero() { + let src = source(); + let feed = pages(&src); + let mut replica = Store::open_in_memory().unwrap(); + let mut applier = Applier::open(&replica).unwrap(); + let o = applier + .apply_frame(&mut replica, "quipu.changes", &feed[1], 1) + .unwrap(); + assert_eq!(o, Outcome::Resume { from: 0 }); + assert!(applier.progress().is_none()); +} + +#[test] +fn a_page_that_does_not_name_its_cursor_is_refused() { + let src = source(); + let mut page: Json = serde_json::from_str(&pages(&src)[0]).unwrap(); + page.as_object_mut().unwrap().remove("since"); + let mut replica = Store::open_in_memory().unwrap(); + let mut applier = Applier::open(&replica).unwrap(); + let o = applier + .apply_frame(&mut replica, "quipu.changes", &page.to_string(), 1) + .unwrap(); + assert!(matches!(o, Outcome::Refused { .. }), "{o:?}"); + assert!(applier.progress().is_none()); +} + +#[test] +fn a_page_inside_an_open_chunk_discards_it_and_resumes() { + let src = source(); + let feed = pages(&src); + let last = feed.last().unwrap(); + let frames = chunk_frames(last, 1, true); + let mut replica = Store::open_in_memory().unwrap(); + let mut applier = Applier::open(&replica).unwrap(); + replay(&mut replica, &mut applier, &feed[..feed.len() - 1]); + let cursor = applier.progress().unwrap().source_tx; + let before = live_digest(&replica); + apply(&mut replica, &mut applier, &frames[..2]); + // Without this rule the page would apply the chunk's transaction, move + // the cursor past it, and the commit would be dropped as a duplicate. + let o = applier + .apply_frame(&mut replica, "quipu.changes", last, 2) + .unwrap(); + assert_eq!(o, Outcome::Resume { from: cursor }); + assert_eq!(before, live_digest(&replica)); + let rest = apply(&mut replica, &mut applier, &frames[2..]); + assert!( + rest.iter().all(|o| *o == Outcome::Resume { from: cursor }), + "{rest:?}" + ); + assert_eq!(before, live_digest(&replica)); + replay(&mut replica, &mut applier, &feed[feed.len() - 1..]); + assert_eq!(live_digest(&src), live_digest(&replica)); +} + +#[test] +fn an_interrupted_chunk_restarts_from_a_new_begin() { + let src = source(); + let feed = pages(&src); + let frames = chunk_frames(feed.last().unwrap(), 1, true); + let mut replica = Store::open_in_memory().unwrap(); + let mut applier = Applier::open(&replica).unwrap(); + replay(&mut replica, &mut applier, &feed[..feed.len() - 1]); + // begin, part 0, then the reconnect replays from begin. + apply(&mut replica, &mut applier, &frames[..2]); + let outcomes = apply(&mut replica, &mut applier, &frames); + assert!( + matches!(outcomes.last(), Some(Outcome::Applied { .. })), + "{outcomes:?}" + ); + assert_eq!(live_digest(&src), live_digest(&replica)); +} + +#[test] +fn a_reopened_applier_holds_no_staged_parts() { + let src = source(); + let feed = pages(&src); + let frames = chunk_frames(feed.last().unwrap(), 1, true); + let mut replica = Store::open_in_memory().unwrap(); + let mut applier = Applier::open(&replica).unwrap(); + replay(&mut replica, &mut applier, &feed[..feed.len() - 1]); + let cursor = applier.progress().unwrap().source_tx; + let before = live_digest(&replica); + apply(&mut replica, &mut applier, &frames[..2]); + // The page restarts: the staged part was memory only. + let mut applier = Applier::open(&replica).unwrap(); + let rest = apply(&mut replica, &mut applier, &frames[2..]); + assert!( + rest.iter().all(|o| *o == Outcome::Resume { from: cursor }), + "{rest:?}" + ); + assert_eq!(before, live_digest(&replica)); +} + +#[test] +fn a_chunk_that_skips_a_transaction_resumes_at_its_begin() { + let src = source(); + let feed = pages(&src); + let frames = chunk_frames(feed.last().unwrap(), 1, true); + let mut replica = Store::open_in_memory().unwrap(); + let mut applier = Applier::open(&replica).unwrap(); + replay(&mut replica, &mut applier, &feed[..feed.len() - 2]); + let cursor = applier.progress().unwrap().source_tx; + let before = live_digest(&replica); + let outcomes = apply(&mut replica, &mut applier, &frames); + assert!( + outcomes + .iter() + .all(|o| *o == Outcome::Resume { from: cursor }), + "{outcomes:?}" + ); + assert_eq!(before, live_digest(&replica)); +} + +#[test] +fn a_chunked_transaction_applies_only_when_its_commit_verifies() { + let src = source(); + let feed = pages(&src); + // The last source transaction has two records: retract + assert. + let frames = chunk_frames(feed.last().unwrap(), 1, true); + assert_eq!(frames.len(), 4); + + let mut replica = Store::open_in_memory().unwrap(); + let mut applier = Applier::open(&replica).unwrap(); + replay(&mut replica, &mut applier, &feed[..feed.len() - 1]); + let before = live_digest(&replica); + for (event, data) in &frames[..3] { + let o = applier.apply_frame(&mut replica, event, data, 3).unwrap(); + assert!(matches!(o, Outcome::Buffered { .. }), "{o:?}"); + assert_eq!( + before, + live_digest(&replica), + "a buffered part is not visible" + ); + } + let (event, data) = &frames[3]; + let o = applier.apply_frame(&mut replica, event, data, 4).unwrap(); + assert!(matches!(o, Outcome::Applied { .. }), "{o:?}"); + assert_eq!(live_digest(&src), live_digest(&replica)); +} + +#[test] +fn a_part_out_of_order_resumes_with_nothing_applied() { + let src = source(); + let feed = pages(&src); + let frames = chunk_frames(feed.last().unwrap(), 1, true); + let mut replica = Store::open_in_memory().unwrap(); + let mut applier = Applier::open(&replica).unwrap(); + replay(&mut replica, &mut applier, &feed[..feed.len() - 1]); + let before = live_digest(&replica); + let cursor = applier.progress().unwrap().source_tx; + applier + .apply_frame(&mut replica, frames[0].0, &frames[0].1, 1) + .unwrap(); + // Skip part 0: part 1 arrives first. + let o = applier + .apply_frame(&mut replica, frames[2].0, &frames[2].1, 2) + .unwrap(); + assert_eq!(o, Outcome::Resume { from: cursor }); + let o = applier + .apply_frame(&mut replica, frames[3].0, &frames[3].1, 3) + .unwrap(); + assert_eq!( + o, + Outcome::Resume { from: cursor }, + "a commit without its chunk resumes" + ); + assert_eq!(before, live_digest(&replica)); +} + +#[test] +fn a_chunked_commit_that_does_not_name_its_source_is_refused() { + let src = source(); + let feed = pages(&src); + let frames = chunk_frames(feed.last().unwrap(), 1, false); + let mut replica = Store::open_in_memory().unwrap(); + let mut applier = Applier::open(&replica).unwrap(); + replay(&mut replica, &mut applier, &feed[..feed.len() - 1]); + let before = live_digest(&replica); + let outcomes: Vec = frames + .iter() + .map(|(e, data)| applier.apply_frame(&mut replica, e, data, 1).unwrap()) + .collect(); + assert!( + matches!(outcomes.last(), Some(Outcome::Refused { .. })), + "{outcomes:?}" + ); + assert_eq!(before, live_digest(&replica)); +} + +#[test] +fn the_cursor_survives_reopening_and_freshness_is_never_bounded() { + let src = source(); + let mut replica = Store::open_in_memory().unwrap(); + let mut applier = Applier::open(&replica).unwrap(); + assert_eq!(applier.freshness(10).last_frame_age_ms, None); + replay(&mut replica, &mut applier, &pages(&src)); + let progress = applier.progress().cloned(); + let reopened = Applier::open(&replica).unwrap(); + assert_eq!(reopened.progress().cloned(), progress); + let f = applier.freshness(101); + assert_eq!(f.last_frame_age_ms, Some(100)); + assert!(!f.bounded); + assert!(matches!( + applier + .apply_frame(&mut replica, "quipu.heartbeat", "{}", 5) + .unwrap(), + Outcome::Ignored + )); +} + +/// The oracle the replica tests rely on: stores that differ ONLY in valid +/// time, only in byte content, or only in graph must digest differently, and +/// identically built stores the same. Without this, an equality assertion +/// could pass because the digest is blind to the difference under test. +#[test] +fn the_digest_sees_valid_time_bytes_and_graph() { + let build = |valid_to: Option<&str>, bytes: Vec, graph: Option<&str>| { + let mut s = Store::open_in_memory().unwrap(); + let mut a = d( + &s, + "urn:item:1", + "urn:p:status", + Value::Str("open".into()), + Op::Assert, + ); + a.valid_to = valid_to.map(String::from); + let b = d( + &s, + "urn:item:1", + "urn:p:blob", + Value::Bytes(bytes), + Op::Assert, + ); + match graph { + None => s.transact(&[a, b], T, None, None).unwrap(), + Some(g) => { + let g = s.graph_create(g).unwrap(); + s.transact_to_graph(&[a, b], T, None, None, g).unwrap() + } + }; + live_digest(&s) + }; + let base = build(None, vec![1, 2, 3], None); + assert_eq!( + base, + build(None, vec![1, 2, 3], None), + "same content, same digest" + ); + assert_ne!( + base, + build(Some("2027-01-01T00:00:00Z"), vec![1, 2, 3], None), + "valid_to" + ); + assert_ne!(base, build(None, vec![1, 2, 4], None), "byte content"); + assert_ne!( + base, + build(None, vec![1, 2, 3], Some("urn:graph:other")), + "graph" + ); +} + +/// The browser bridge's contract: outcomes serialize tagged by `kind`. +#[test] +fn outcomes_serialize_tagged_by_kind() { + let applied = serde_json::to_value(Outcome::Applied { + source_txs: vec![3, 4], + }) + .unwrap(); + assert_eq!(applied, json!({"kind": "applied", "source_txs": [3, 4]})); + assert_eq!( + serde_json::to_value(Outcome::Duplicate).unwrap(), + json!({"kind": "duplicate"}) + ); + assert_eq!( + serde_json::to_value(Outcome::Resume { from: 7 }).unwrap(), + json!({"kind": "resume", "from": 7}) + ); + let fresh = Freshness { + source_tx: Some(9), + last_frame_age_ms: None, + bounded: false, + }; + assert_eq!( + serde_json::to_value(fresh).unwrap(), + json!({"source_tx": 9, "last_frame_age_ms": null, "bounded": false}) + ); +} diff --git a/src/lib.rs b/src/lib.rs index 3bec9be83..55ab0afe4 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -5,6 +5,7 @@ pub mod acquisition; pub mod align; +pub mod applier; pub mod bindings; pub mod compact; pub mod config; diff --git a/src/server/feed_stream.rs b/src/server/feed_stream.rs index a7b250a3c..a25d0bc24 100644 --- a/src/server/feed_stream.rs +++ b/src/server/feed_stream.rs @@ -361,6 +361,7 @@ pub(crate) fn chunk_transaction(page: &Value, next: i64) -> Result Result(), "digest": digest, + "since": page.get("since").cloned().unwrap_or(Value::Null), "source": page.get("source").cloned().unwrap_or(Value::Null), }) .to_string(), diff --git a/src/share_completeness.rs b/src/share_completeness.rs index 19a16ef06..503753421 100644 --- a/src/share_completeness.rs +++ b/src/share_completeness.rs @@ -133,6 +133,18 @@ pub const DECLARED: &[(&str, Disposition)] = &[ // store mints its own on first open, and a follower holding a cursor into // the original sees the epoch change and re-bootstraps (local-sync-mode §5). ("log_epoch", Disposition::Excluded), + // A replica's cursor into ONE source's history (the shared applier, + // aegis-s24.2). A share of the replica is a different copy with no cursor + // of its own: carrying it would let the copy splice a feed it never + // bootstrapped from. Excluded, like the epoch it is bound to. + ("replica_progress", Disposition::Excluded), + // The digests of the transactions that cursor applied, kept to tell an + // exact redelivery from a changed one. They belong to the same cursor and + // mean nothing without it: excluded with it. + ("replica_applied_tx", Disposition::Excluded), + // Why that cursor stopped (a history that disagreed with its source). It + // belongs to the same replica and cursor: excluded with them. + ("replica_stop", Disposition::Excluded), // Spent SEALED-DECISION nonces (aegis-kzt0ql.9.3). Same replay reasoning as // attestation_nonces: carried, a legitimate re-attestation on the copy is // refused; omitted silently, a spent one is accepted again. diff --git a/src/store/changes.rs b/src/store/changes.rs index 812e71884..e56043cc1 100644 --- a/src/store/changes.rs +++ b/src/store/changes.rs @@ -86,6 +86,10 @@ impl Capture { #[derive(Debug)] pub struct ChangePage { pub records: Vec, + /// The cursor this page answers: it covers every transaction in scope in + /// `(since, next_tx]`. A follower whose cursor is below `since` would skip + /// transactions by applying it, so it must resume instead. + pub since: i64, pub next_tx: i64, pub watermark_tx: i64, pub watermark_timestamp: Option, @@ -102,6 +106,7 @@ impl ChangePage { pub fn to_json(&self) -> Json { json!({ "records": self.records, + "since": self.since, "next_tx": self.next_tx, "watermark_tx": self.watermark_tx, "watermark_timestamp": self.watermark_timestamp, @@ -164,6 +169,7 @@ impl Store { } Ok(ChangePage { records, + since: since_tx, next_tx: tx_ids.last().copied().unwrap_or(since_tx), watermark_tx, watermark_timestamp, diff --git a/src/store/changes_replay_tests.rs b/src/store/changes_replay_tests.rs index 58f4899f0..099dfd7a1 100644 --- a/src/store/changes_replay_tests.rs +++ b/src/store/changes_replay_tests.rs @@ -5,11 +5,10 @@ //! of the live facts. The plain fixture is the positive control: it proves the //! instrument can say "equal". Each defect fixture isolates one feed gap. -use std::collections::BTreeSet; - use serde_json::Value as Json; use super::*; +use crate::applier::live_digest; use crate::store::{Datum, Store}; use crate::types::Op; @@ -24,86 +23,6 @@ fn datum(e: i64, a: i64, value: Value, valid_from: &str, op: Op) -> Datum { } } -/// `(e, a, v, g, valid_from, valid_to)` as stored. -type FactRow = (i64, i64, Vec, i64, String, Option); - -/// Live facts as resolved strings, so two stores with different term ids -/// compare by meaning. Valid time is part of a fact's identity here. -fn live_digest(store: &Store) -> BTreeSet { - let mut stmt = store - .conn - .prepare( - "SELECT e, a, v, g, valid_from, valid_to FROM facts \ - WHERE op = 1 AND retracted_tx IS NULL", - ) - .unwrap(); - let rows: Vec = stmt - .query_map([], |r| { - Ok(( - r.get(0)?, - r.get(1)?, - r.get(2)?, - r.get(3)?, - r.get(4)?, - r.get(5)?, - )) - }) - .unwrap() - .collect::>() - .unwrap(); - rows.into_iter() - .map(|(e, a, v, g, from, to)| { - let value = match Value::from_bytes(&v).unwrap() { - Value::Ref(id) => format!("ref:{}", store.resolve(id).unwrap()), - Value::Bytes(b) => format!("bytes:{b:?}"), - other => format!("{other:?}"), - }; - let graph = if g == 0 { - "ROOT".to_string() - } else { - store.resolve(g).unwrap() - }; - format!( - "{graph} {} {} {value} [{from}, {to:?})", - store.resolve(e).unwrap(), - store.resolve(a).unwrap() - ) - }) - .collect() -} - -/// Rebuild a stored value from its feed JSON. What the feed leaves out cannot -/// be invented here: a `bytes_len` stub replays as empty bytes. -fn value_of(replica: &Store, json: &Json) -> Value { - match json { - Json::String(s) => Value::Str(s.clone()), - Json::Bool(b) => Value::Bool(*b), - Json::Number(n) if n.is_i64() => Value::Int(n.as_i64().unwrap()), - Json::Number(n) => Value::Float(n.as_f64().unwrap()), - Json::Object(o) if o.contains_key("ref") => { - Value::Ref(replica.intern(o["ref"].as_str().unwrap()).unwrap()) - } - Json::Object(o) if o.contains_key("lang") => Value::Lang { - lexical: o["value"].as_str().unwrap().into(), - lang: o["lang"].as_str().unwrap().into(), - }, - Json::Object(o) if o.contains_key("datatype") => Value::Typed { - lexical: o["value"].as_str().unwrap().into(), - datatype: o["datatype"].as_str().unwrap().into(), - }, - Json::Object(o) if o.contains_key("base64") => { - use base64::Engine as _; - Value::Bytes( - base64::engine::general_purpose::STANDARD - .decode(o["base64"].as_str().unwrap()) - .unwrap(), - ) - } - Json::Object(o) if o.contains_key("bytes_len") => Value::Bytes(Vec::new()), - other => panic!("unreplayable value {other}"), - } -} - /// Replay the whole feed, one source transaction per replica transaction. /// Valid time comes from the record when the feed carries it, else from the /// record's commit timestamp, which is all a consumer has today. @@ -123,28 +42,14 @@ fn replay(source: &Store) -> Store { for (_, records) in by_tx { let timestamp = records[0]["timestamp"].as_str().unwrap().to_string(); let graph_name = records[0]["graph"].as_str().unwrap().to_string(); + // The applier's conversion, so the conformance tests and every + // replica share one record -> datum mapping (aegis-s24.2). let datums: Vec = records .iter() .map(|r| { - let e = replica.intern(r["entity"].as_str().unwrap()).unwrap(); - let a = replica.intern(r["attribute"].as_str().unwrap()).unwrap(); - let from = r["valid_from"].as_str().unwrap_or(×tamp).to_string(); - let mut d = match r["op"].as_str().unwrap() { - "retract" => datum( - e, - a, - value_of(&replica, &r["old_value"]), - &from, - Op::Retract, - ), - _ => datum(e, a, value_of(&replica, &r["value"]), &from, Op::Assert), - }; - // A retract names the fact it closes; its own row's valid time - // describes the closing, not an interval to replay. - if d.op == Op::Assert { - d.valid_to = r["valid_to"].as_str().map(String::from); - } - d + crate::applier::record_to_datum(&replica, r, ×tamp) + .unwrap() + .expect("every post-F4 record is reproducible") }) .collect(); if graph_name == "ROOT" {