From f43688dd82791bf6356769c70b092ea0ff4c2e9b Mon Sep 17 00:00:00 2001 From: Steve Brown Date: Fri, 23 Oct 2026 21:42:05 -0400 Subject: [PATCH 01/13] fix(changes): carry valid time and byte bodies so a feed replica equals its source A replica built only from /changes records diverged from its source in two ways (local-sync-mode.md F3, F4): - Records carried no valid time, so a fact valid since 2025 replayed as valid from its commit time. Records now carry valid_from and valid_to. A later retract closes the asserted row in place, so a row a later transaction closed reports valid_to null and the closing arrives as the retract's own record. Otherwise a served record would change after the fact, and a replica would assert an already-closed fact its retract could not find. - Byte values were reduced to bytes_len. They now also carry the body as base64. Both are additive fields. New replay tests rebuild a store from the feed and compare a canonical live-fact digest: a plain control, valid time (open and bounded), bytes, and an assert record that reads the same before and after a later retract. Removing each change turns its test red. Refs aegis-xdvg.1. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_0185RxTpLsPCQFB9ni72Nzac --- docs/book/src/architecture/change-feed.md | 12 +- src/store/changes.rs | 46 +++- src/store/changes_replay_tests.rs | 297 ++++++++++++++++++++++ 3 files changed, 343 insertions(+), 12 deletions(-) create mode 100644 src/store/changes_replay_tests.rs diff --git a/docs/book/src/architecture/change-feed.md b/docs/book/src/architecture/change-feed.md index ba596ab3a..09859c7f5 100644 --- a/docs/book/src/architecture/change-feed.md +++ b/docs/book/src/architecture/change-feed.md @@ -17,6 +17,11 @@ mirror exactly which facts did. - **Records** are `(tx, sequence, op, graph, entity, attribute, value)`, derived directly from fact rows — never a second log that can drift from the first. `op` is `assert`, `retract`, or `tombstone`. +- **Valid time rides each record** as `valid_from` and `valid_to`, so a + replica keeps the interval a fact was written with rather than its commit + time. An assert reports the bound it was written with; when a later + transaction closes the fact, that closing arrives as its own `retract` + record, and the assert's record reads the same before and after. - **The cursor is a transaction id, and pages end on transaction boundaries.** Every cursor is therefore a consistent prefix of commit history: a reader never observes half a transaction. @@ -64,7 +69,8 @@ Both return one page: "tx": 43, "sequence": 0, "timestamp": "2026-04-03", "actor": "ingest", "source": "episode:...", "op": "assert", "graph": "ROOT", "entity": "http://example.org/korimako", - "attribute": "http://example.org/cpuCores", "value": 8 + "attribute": "http://example.org/cpuCores", "value": 8, + "valid_from": "2026-04-03", "valid_to": null } ], "next_tx": 43, @@ -77,8 +83,8 @@ 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 `{"ref": ""}` so a consumer can tell an edge from a string that happens -to look like one; bytes report a length, not a body — the feed is a -notification surface, and blob bodies belong on the entity read path. +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. ## First consumers diff --git a/src/store/changes.rs b/src/store/changes.rs index 47065e6b0..01ceee950 100644 --- a/src/store/changes.rs +++ b/src/store/changes.rs @@ -7,7 +7,14 @@ //! - **Records** are `(tx, sequence, op, graph, entity, attribute, value)`, //! derived from rows — never a second log to drift from the first. A //! retract's op=0 row carries the retracted value, so the old value rides -//! the record without a lookup. +//! the record without a lookup. Each record also carries the row's valid +//! time (`valid_from`, `valid_to`): without it a replica can only guess the +//! commit time, and a fact valid since last year replays as valid since today. +//! An assert's `valid_to` is the bound it was written with. A later retract +//! closes the row in place, so a row a later transaction closed reports +//! `valid_to: null` here and the closing arrives as that retract's record — +//! otherwise a record would change after the fact and a replica would assert +//! an already-closed fact that its retract then cannot find. //! - **Cursor = transaction id, whole transactions only.** A page always ends //! on a tx boundary, so every cursor is a consistent prefix of commit //! history — a reader never observes half a transaction, which is the same @@ -30,6 +37,7 @@ //! - **The log is permanent.** Unlike Spanner's 1–30 day retention, the fact //! log never garbage-collects, so a cursor never expires. +use base64::{Engine, engine::general_purpose::STANDARD}; use rusqlite::params; use serde_json::{Value as Json, json}; @@ -191,7 +199,11 @@ impl Store { ) .unwrap_or((None, None, None)); - let mut sql = String::from("SELECT e, a, v, g, op FROM facts WHERE tx = ?1"); + let mut sql = String::from( + "SELECT e, a, v, g, op, valid_from, \ + CASE WHEN retracted_tx IS NULL THEN valid_to END \ + FROM facts WHERE tx = ?1", + ); if graph.is_some() { sql.push_str(" AND g = ?2"); } @@ -199,9 +211,19 @@ impl Store { // ordering promise rides on it. sql.push_str(" ORDER BY rowid"); let mut stmt = self.conn.prepare(&sql)?; - let rows: Vec<(i64, i64, Vec, i64, i32)> = { - let map = - |r: &rusqlite::Row<'_>| Ok((r.get(0)?, r.get(1)?, r.get(2)?, r.get(3)?, r.get(4)?)); + type Row = (i64, i64, Vec, i64, i32, String, Option); + let rows: Vec = { + let map = |r: &rusqlite::Row<'_>| { + Ok(( + r.get(0)?, + r.get(1)?, + r.get(2)?, + r.get(3)?, + r.get(4)?, + r.get(5)?, + r.get(6)?, + )) + }; match graph { Some(g) => stmt .query_map(params![tx, g], map)? @@ -212,7 +234,7 @@ impl Store { } }; - for (sequence, (e, a, v, g, op)) in rows.into_iter().enumerate() { + for (sequence, (e, a, v, g, op, valid_from, valid_to)) in rows.into_iter().enumerate() { let value = Value::from_bytes(&v)?; let op_name = match op { 0 => "retract", @@ -229,6 +251,8 @@ impl Store { "graph": if g == 0 { "ROOT".to_string() } else { self.resolve(g)? }, "entity": self.resolve(e)?, "attribute": self.resolve(a)?, + "valid_from": valid_from, + "valid_to": valid_to, }); match (op_name, capture) { // A retract under new_values identifies the fact but not the @@ -290,9 +314,9 @@ impl Store { Value::Int(i) => json!(i), Value::Float(f) => json!(f), Value::Bool(b) => json!(b), - // Length, not bytes: the feed is a notification surface, and blob - // bodies belong on the entity read path. - Value::Bytes(b) => json!({"bytes_len": b.len()}), + // The body rides along so a replica built from the feed can hold + // the value; a length alone replays as a different fact. + Value::Bytes(b) => json!({"bytes_len": b.len(), "base64": STANDARD.encode(b)}), Value::Lang { lexical, lang } => json!({"value": lexical, "lang": lang}), Value::Typed { lexical, datatype } => json!({"value": lexical, "datatype": datatype}), }) @@ -302,3 +326,7 @@ impl Store { #[cfg(test)] #[path = "changes_tests.rs"] mod tests; + +#[cfg(test)] +#[path = "changes_replay_tests.rs"] +mod replay_tests; diff --git a/src/store/changes_replay_tests.rs b/src/store/changes_replay_tests.rs new file mode 100644 index 000000000..a1a45252b --- /dev/null +++ b/src/store/changes_replay_tests.rs @@ -0,0 +1,297 @@ +//! Feed replay conformance (local-sync-mode.md §2, P0; aegis-xdvg.1). +//! +//! A replica built only from `/changes` records must equal its source. Each +//! test replays every page into an EMPTY store and compares a canonical digest +//! 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::store::{Datum, Store}; +use crate::types::Op; + +fn datum(e: i64, a: i64, value: Value, valid_from: &str, op: Op) -> Datum { + Datum { + entity: e, + attribute: a, + value, + valid_from: valid_from.into(), + valid_to: None, + op, + } +} + +/// `(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. +fn replay(source: &Store) -> Store { + let mut replica = Store::open_in_memory().unwrap(); + let page = source + .changes_after(0, usize::MAX >> 1, Capture::OldAndNewValues, None) + .unwrap(); + let mut by_tx: Vec<(i64, Vec<&Json>)> = Vec::new(); + for record in &page.records { + let tx = record["tx"].as_i64().unwrap(); + match by_tx.last_mut() { + Some((last, list)) if *last == tx => list.push(record), + _ => by_tx.push((tx, vec![record])), + } + } + 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(); + 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 + }) + .collect(); + if graph_name == "ROOT" { + replica.transact(&datums, ×tamp, None, None).unwrap(); + } else { + let g = replica.graph_create(&graph_name).unwrap(); + replica + .transact_to_graph(&datums, ×tamp, None, None, g) + .unwrap(); + } + } + replica +} + +fn assert_replica_equals_source(source: &Store) { + let replica = replay(source); + let (want, got) = (live_digest(source), live_digest(&replica)); + assert!(!want.is_empty(), "fixture wrote nothing live"); + let missing: Vec<_> = want.difference(&got).collect(); + let extra: Vec<_> = got.difference(&want).collect(); + assert!( + missing.is_empty() && extra.is_empty(), + "replica diverges from source\n missing: {missing:#?}\n extra: {extra:#?}" + ); +} + +const TS1: &str = "2026-04-01T00:00:00Z"; +const TS2: &str = "2026-04-02T00:00:00Z"; + +/// Positive control: asserts, an explicit retract and a graph replay exactly. +#[test] +fn control_plain_asserts_and_retracts_replay_exactly() { + let mut source = Store::open_in_memory().unwrap(); + let e = source.intern("http://example.org/korimako").unwrap(); + let a = source.intern("http://example.org/cpuCores").unwrap(); + let n = source.intern("http://example.org/name").unwrap(); + source + .transact( + &[ + datum(e, a, Value::Int(4), TS1, Op::Assert), + datum(e, n, Value::Str("korimako".into()), TS1, Op::Assert), + ], + TS1, + None, + None, + ) + .unwrap(); + source + .transact( + &[datum(e, a, Value::Int(4), TS2, Op::Retract)], + TS2, + None, + None, + ) + .unwrap(); + assert_replica_equals_source(&source); +} + +/// F3: change records carry no valid time, so a fact asserted with a valid +/// time other than its commit time replays at the wrong time. +#[test] +fn f3_valid_time_survives_replay() { + let mut source = Store::open_in_memory().unwrap(); + let e = source.intern("http://example.org/lease").unwrap(); + let a = source.intern("http://example.org/holder").unwrap(); + let mut bounded = datum( + e, + a, + Value::Str("globex".into()), + "2024-01-01T00:00:00Z", + Op::Assert, + ); + bounded.valid_to = Some("2024-12-31T00:00:00Z".into()); + source + .transact( + &[ + datum( + e, + a, + Value::Str("acme".into()), + "2025-01-01T00:00:00Z", + Op::Assert, + ), + bounded, + ], + TS1, + None, + None, + ) + .unwrap(); + assert_replica_equals_source(&source); +} + +/// F4: a byte value is reduced to its length, so the replica cannot hold it. +#[test] +fn f4_byte_values_replay_losslessly() { + let mut source = Store::open_in_memory().unwrap(); + let e = source.intern("http://example.org/blob").unwrap(); + let a = source.intern("http://example.org/body").unwrap(); + source + .transact( + &[datum( + e, + a, + Value::Bytes(vec![0, 159, 146, 150]), + TS1, + Op::Assert, + )], + TS1, + None, + None, + ) + .unwrap(); + assert_replica_equals_source(&source); +} + +/// A record must not change after it is served. A later retract closes the +/// asserted row in place; the assert's record still reads as it did at its tx. +#[test] +fn an_assert_record_reads_the_same_before_and_after_a_later_retract() { + let mut source = Store::open_in_memory().unwrap(); + let e = source.intern("http://example.org/korimako").unwrap(); + let a = source.intern("http://example.org/cpuCores").unwrap(); + let tx1 = source + .transact( + &[datum(e, a, Value::Int(4), TS1, Op::Assert)], + TS1, + None, + None, + ) + .unwrap(); + let before = source + .changes_after(tx1 - 1, 1, Capture::OldAndNewValues, None) + .unwrap() + .records; + source + .transact( + &[datum(e, a, Value::Int(4), TS2, Op::Retract)], + TS2, + None, + None, + ) + .unwrap(); + let after = source + .changes_after(tx1 - 1, 1, Capture::OldAndNewValues, None) + .unwrap() + .records; + assert_eq!(before, after); + assert!(after[0]["valid_to"].is_null(), "{after:#?}"); +} From 759527145cd642b3cff876910744ac0cdc562c8b Mon Sep 17 00:00:00 2001 From: Steve Brown Date: Fri, 23 Oct 2026 21:53:06 -0400 Subject: [PATCH 02/13] fix(owl): write a retract row when a functional value is superseded A functional-property supersede closed the prior value with an UPDATE and wrote no op=0 row, unlike an explicit retract. The change feed derives its records from rows, so it showed the new value and never the end of the old one, and a replica built from the feed kept both (local-sync-mode.md F1). The supersede now writes the same immutable op=0 row an explicit retract writes, at the superseding transaction. The resident read model reads op=1 rows and their retracted_tx stamp, so it is unaffected. The replay test asserts the replica holds exactly one current value; removing the new row turns it red. Refs aegis-xdvg.1. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_0185RxTpLsPCQFB9ni72Nzac --- src/store/changes_replay_tests.rs | 33 +++++++++++++++++++++++++++++++ src/store/gate.rs | 22 ++++++++++++++++++++- 2 files changed, 54 insertions(+), 1 deletion(-) diff --git a/src/store/changes_replay_tests.rs b/src/store/changes_replay_tests.rs index a1a45252b..8218f18cc 100644 --- a/src/store/changes_replay_tests.rs +++ b/src/store/changes_replay_tests.rs @@ -295,3 +295,36 @@ fn an_assert_record_reads_the_same_before_and_after_a_later_retract() { assert_eq!(before, after); assert!(after[0]["valid_to"].is_null(), "{after:#?}"); } + +/// F1: a functional-property supersede closes the old value with an UPDATE and +/// writes no retract record, so the replica keeps both values. +#[cfg(feature = "owl")] +#[test] +fn f1_functional_supersede_replays_as_one_current_value() { + const ONT: &str = "@prefix owl: .\n\ + @prefix ex: .\n\ + ex:contentHash a owl:DatatypeProperty, owl:FunctionalProperty .\n"; + let mut source = Store::open_in_memory().unwrap(); + source.owl_config_mut().validate_on_write = true; + source.load_ontology("t", ONT, TS1).unwrap(); + source.invalidate_owl_cache(); + let e = source.intern("http://example.org/doc1").unwrap(); + let a = source.intern("http://example.org/contentHash").unwrap(); + source + .transact( + &[datum(e, a, Value::Str("aaa".into()), TS1, Op::Assert)], + TS1, + None, + None, + ) + .unwrap(); + source + .transact( + &[datum(e, a, Value::Str("bbb".into()), TS2, Op::Assert)], + TS2, + None, + None, + ) + .unwrap(); + assert_replica_equals_source(&source); +} diff --git a/src/store/gate.rs b/src/store/gate.rs index 948539fbb..f03da1bd2 100644 --- a/src/store/gate.rs +++ b/src/store/gate.rs @@ -145,6 +145,14 @@ impl Store { "UPDATE facts SET valid_to = ?1, retracted_tx = ?6 \ WHERE e = ?2 AND a = ?3 AND v = ?4 AND g = ?5 AND op = 1 AND valid_to IS NULL", )?; + // The close needs the same immutable op=0 row an explicit retract + // writes. Without it the change feed shows the new value and never the + // end of the old one, so a replica built from the feed keeps both + // (local-sync-mode.md F1). + let mut retract_row = self.conn.prepare( + "INSERT INTO facts (e, a, v, g, tx, valid_from, valid_to, op) \ + VALUES (?1, ?2, ?3, ?4, ?5, ?6, NULL, ?7) ON CONFLICT(e, a, v, tx) DO NOTHING", + )?; for ((entity, attribute), values) in &proposed { // AMBIGUOUS BATCH: two distinct values for one functional property in // a single write. Superseding here would silently pick whichever the @@ -162,8 +170,20 @@ impl Store { )?.query_map(params![entity, attribute, graph], |row| row.get(0))?.collect::>()?; for bytes in existing { if crate::Value::from_bytes(&bytes)?.term_key() != *new_value { - closed += close_other + let count = close_other .execute(params![timestamp, entity, attribute, bytes, graph, tx_id])?; + if count > 0 { + retract_row.execute(params![ + entity, + attribute, + bytes, + graph, + tx_id, + timestamp, + Op::Retract as i32 + ])?; + } + closed += count; } } } From 15c812822886a1befe15efe412d5802ccebafbfe Mon Sep 17 00:00:00 2001 From: Steve Brown Date: Fri, 23 Oct 2026 22:08:47 -0400 Subject: [PATCH 03/13] fix(stream): send an oversized transaction as parts instead of ending the stream MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A /changes/stream page larger than the 1 MiB frame budget sent quipu.error without advancing the cursor and ended the stream. A consumer reconnecting at that cursor hit the same transaction and the same dead end forever (local-sync-mode.md F6). Such a transaction now arrives as quipu.changes.begin, numbered quipu.changes.part frames and a quipu.changes.commit barrier (design §6). Only the barrier carries the SSE id, so Last-Event-ID never lands inside a transaction and a reconnect replays it from the first part. The barrier digest is SHA-256 over the parts' data exactly as sent; each part carries the hash of the parts before it. A single record larger than a frame is refused with quipu.changes.tx_too_large and no id, leaving the cursor unchanged. Tests stream a 2500-record transaction: every part fits, the digests chain, the reassembled records equal the page, the next transaction follows, and a reconnect replays from part 1. Disabling the split turns the test red. Refs aegis-xdvg.1. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_0185RxTpLsPCQFB9ni72Nzac --- docs/book/src/reference/rest-api.md | 24 ++++- src/server/event_cursor.rs | 13 ++- src/server/feed_stream.rs | 124 ++++++++++++++++++++++- src/server/feed_stream_tests.rs | 151 ++++++++++++++++++++++++++++ 4 files changed, 307 insertions(+), 5 deletions(-) diff --git a/docs/book/src/reference/rest-api.md b/docs/book/src/reference/rest-api.md index f8bc01f79..cdfbf700f 100644 --- a/docs/book/src/reference/rest-api.md +++ b/docs/book/src/reference/rest-api.md @@ -2222,11 +2222,29 @@ ID is not an event offset and must never be sent to `/events/commit`. Fifteen-second heartbeat comments do not read the store; commit hints are checked against the durable feed after crossing the writer barrier. An hourly backstop covers writes from other connections. A `quipu.error` event has no ID -and terminates a failed or oversized page; keep the prior applied cursor. +and terminates a failed page; keep the prior applied cursor. + +A change transaction too large for one frame arrives as a sequence: + +1. `quipu.changes.begin`: `{source_tx, parts, records, bytes, digest}`. +2. `parts` frames of type `quipu.changes.part`: `{source_tx, index, + digest_before, records}`, in index order. +3. `quipu.changes.commit`: `{source_tx, parts, records, digest, next_tx, + watermark_tx, watermark_timestamp, capture}`. This is the only frame of the + sequence with an ID, the transaction ID. + +`digest` is SHA-256 (hex) over the concatenated `data` of every part, exactly +as received; `digest_before` is the same hash over the parts before that one. +Stage the parts and apply the transaction only after the commit frame's digest +verifies. Because only the commit frame carries an ID, a reconnect after a +partial sequence resumes from the previous transaction and receives this one +again from its first part. A single record larger than a frame cannot be split: +the stream sends `quipu.changes.tx_too_large` with no ID and ends, and the +cursor is unchanged. The process admits up to 64 streams, reads one transaction/event per page and -limits output frames to 1 MiB. One large transaction may require more memory -before the output check; this is not a total heap bound. See the +limits output frames to 1 MiB. A large transaction is still read whole before +it is split into parts, so this is not a total heap bound. See the [committed-feed protocol](../architecture/event-stream.md). ### `POST /events/commit` diff --git a/src/server/event_cursor.rs b/src/server/event_cursor.rs index dd1cbcd7e..64123f270 100644 --- a/src/server/event_cursor.rs +++ b/src/server/event_cursor.rs @@ -37,12 +37,23 @@ pub(super) fn validate_frame(previous: i64, offset: i64, json: &str) -> Result<( if json.contains(['\r', '\n']) { return Err("event data must be single-line JSON"); } - if json.len() > MAX_FRAME_BYTES.saturating_sub(64) { + if !fits_frame(json) { return Err("event exceeds transport frame budget"); } Ok(()) } +/// Whether one serialized event body fits the transport frame budget, with +/// room for the SSE `id:`/`event:` lines around it. +pub(super) fn fits_frame(json: &str) -> bool { + fits_frame_len(json.len()) +} + +/// [`fits_frame`] for a body whose length is already known. +pub(super) fn fits_frame_len(len: usize) -> bool { + len <= MAX_FRAME_BYTES.saturating_sub(64) +} + #[cfg(test)] mod tests { use super::*; diff --git a/src/server/feed_stream.rs b/src/server/feed_stream.rs index 73259585e..f574d7213 100644 --- a/src/server/feed_stream.rs +++ b/src/server/feed_stream.rs @@ -1,6 +1,7 @@ //! Commit-driven transport over existing logs; no shadow log or implicit ACK. use std::{ + collections::VecDeque, convert::Infallible, sync::{Arc, OnceLock}, time::Duration, @@ -16,12 +17,13 @@ use axum::{ }; use futures::stream::unfold; use serde_json::{Value, json}; +use sha2::{Digest, Sha256}; use tokio::sync::{OwnedSemaphorePermit, Semaphore, watch}; use super::{ SharedStore, base::blocking, - event_cursor::{resume_offset, validate_frame}, + event_cursor::{fits_frame, fits_frame_len, resume_offset, validate_frame}, feed, }; @@ -42,6 +44,8 @@ struct Cursor { changes: bool, wait: bool, terminal: bool, + /// Frames of one chunked transaction still to send, in order. + pending: VecDeque, _slot: OwnedSemaphorePermit, } @@ -105,9 +109,13 @@ fn stream(store: SharedStore, headers: &HeaderMap, params: Params, changes: bool changes, wait: false, terminal: false, + pending: VecDeque::new(), _slot: slot, }; let output = unfold(cursor, |mut cursor| async move { + if let Some(event) = cursor.pending.pop_front() { + return Some((Ok::<_, Infallible>(event), cursor)); + } if cursor.terminal { return None; } @@ -200,6 +208,21 @@ fn stream(store: SharedStore, headers: &HeaderMap, params: Params, changes: bool continue; } let data = body.to_string(); + if cursor.changes && !fits_frame(&data) { + match chunk_transaction(&body, next) { + Ok(frames) => { + cursor.offset = next; + cursor.wait = false; + cursor.pending = frames; + } + Err(refusal) => { + cursor.terminal = true; + cursor.pending.push_back(refusal); + } + } + let first = cursor.pending.pop_front(); + return first.map(|event| (Ok::<_, Infallible>(event), cursor)); + } if validate_frame(cursor.offset, next, &data).is_err() { cursor.terminal = true; return Some(( @@ -231,6 +254,105 @@ fn stream(store: SharedStore, headers: &HeaderMap, params: Params, changes: bool .into_response() } +/// Split one change page (a single transaction) that exceeds the frame budget +/// into `quipu.changes.begin`, numbered `quipu.changes.part` frames and a +/// `quipu.changes.commit` barrier (local-sync-mode.md §6, F6). +/// +/// Only the barrier carries the SSE id, so `Last-Event-ID` never lands inside a +/// transaction: a consumer that disconnects mid-transaction resumes from the +/// previous one and replays this transaction from its first part. `digest` is +/// SHA-256 over the concatenated `data` of every part, in order, exactly as +/// sent. Each part's `digest_before` is the same hash over the parts before it, +/// so a consumer can check the stream as it arrives without a part hashing +/// itself. +/// Before this, an oversized transaction ended the stream without advancing +/// the cursor, and every reconnect hit the same dead end. +/// +/// A single record larger than a frame cannot be split; it is refused with +/// `quipu.changes.tx_too_large` and the cursor is not advanced. +fn chunk_transaction(page: &Value, next: i64) -> Result, Event> { + let records = page + .get("records") + .and_then(Value::as_array) + .cloned() + .unwrap_or_default(); + // Sized with a placeholder digest of the final length, so the real part + // can never be larger than the one that was measured. + let placeholder = "0".repeat(64); + let part_data = |index: usize, chunk: &[Value], before: &str| { + json!({"source_tx": next, "index": index, "digest_before": before, "records": chunk}) + .to_string() + }; + // Sizes add exactly: an array serializes as `[`, the items joined by `,`, + // and `]`, so a part is its empty frame plus each record plus a comma. + let mut chunks: Vec> = Vec::new(); + let mut chunk: Vec = Vec::new(); + let mut size = part_data(0, &[], &placeholder).len(); + for record in &records { + let text = record.to_string().len(); + if !chunk.is_empty() && !fits_frame_len(size + text + 1) { + chunks.push(std::mem::take(&mut chunk)); + size = part_data(chunks.len(), &[], &placeholder).len(); + } + let len = text + usize::from(!chunk.is_empty()); + if !fits_frame_len(size + len) { + return Err(Event::default().event("quipu.changes.tx_too_large").data( + json!({ + "source_tx": next, + "reason": "one record exceeds the transport frame budget; cursor not advanced", + }) + .to_string(), + )); + } + size += len; + chunk.push(record.clone()); + } + if !chunk.is_empty() { + chunks.push(chunk); + } + let mut hasher = Sha256::new(); + let mut parts = Vec::with_capacity(chunks.len()); + for (index, chunk) in chunks.iter().enumerate() { + let before = hex::encode(hasher.clone().finalize()); + let data = part_data(index, chunk, &before); + hasher.update(data.as_bytes()); + parts.push(data); + } + let digest = hex::encode(hasher.finalize()); + let mut frames = VecDeque::new(); + let mut commit = json!({ + "source_tx": next, + "parts": parts.len(), + "records": records.len(), + "digest": digest, + }); + for key in ["next_tx", "watermark_tx", "watermark_timestamp", "capture"] { + commit[key] = page.get(key).cloned().unwrap_or(Value::Null); + } + frames.push_back( + Event::default().event("quipu.changes.begin").data( + json!({ + "source_tx": next, + "parts": parts.len(), + "records": records.len(), + "bytes": parts.iter().map(String::len).sum::(), + "digest": digest, + }) + .to_string(), + ), + ); + for part in parts { + frames.push_back(Event::default().event("quipu.changes.part").data(part)); + } + frames.push_back( + Event::default() + .event("quipu.changes.commit") + .id(next.to_string()) + .data(commit.to_string()), + ); + Ok(frames) +} + fn headers_for_feed() -> HeaderMap { let mut headers = HeaderMap::new(); headers.insert("x-quipu-client", "event-stream".parse().unwrap()); diff --git a/src/server/feed_stream_tests.rs b/src/server/feed_stream_tests.rs index 4fc48045d..f7cf83151 100644 --- a/src/server/feed_stream_tests.rs +++ b/src/server/feed_stream_tests.rs @@ -176,3 +176,154 @@ async fn inner_savepoint_hint_cannot_publish_speculative_rows() { 0 ); } + +/// One parsed SSE frame: `(event, id, data)`. +type Frame = (String, Option, String); + +/// Read frames until `stop` matches one (inclusive), or fail after 10 s. +async fn read_frames(response: Response, stop: impl Fn(&Frame) -> bool) -> Vec { + let mut bytes = response.into_body().into_data_stream(); + let (mut buffer, mut frames) = (String::new(), Vec::new()); + loop { + while let Some(end) = buffer.find("\n\n") { + let block: String = buffer.drain(..end + 2).collect(); + let (mut event, mut id, mut data) = (String::new(), None, String::new()); + for line in block.lines() { + if let Some(v) = line.strip_prefix("event: ") { + event = v.into(); + } else if let Some(v) = line.strip_prefix("id: ") { + id = Some(v.into()); + } else if let Some(v) = line.strip_prefix("data: ") { + data = v.into(); + } + } + if event.is_empty() { + continue; // keep-alive comment + } + let frame = (event, id, data); + let done = stop(&frame); + frames.push(frame); + if done { + return frames; + } + } + let chunk = tokio::time::timeout(Duration::from_secs(10), bytes.next()) + .await + .expect("stream stalled") + .expect("stream ended") + .unwrap(); + buffer.push_str(std::str::from_utf8(&chunk).unwrap()); + } +} + +fn wide_datum(store: &mut Store, name: &str, width: usize) -> Datum { + Datum { + value: FactValue::Str("x".repeat(width)), + ..datum(store, name) + } +} + +/// F6 (local-sync-mode.md §6): a transaction larger than one frame used to end +/// the stream with the cursor unmoved, so every reconnect hit the same dead +/// end. It now arrives as begin, parts and a commit barrier that alone carries +/// the id, and the stream continues with the next transaction. +#[tokio::test] +async fn oversized_transaction_streams_as_parts_then_continues() { + let mut store = Store::open_in_memory().unwrap(); + let big: Vec = (0..2500) + .map(|i| wide_datum(&mut store, &format!("urn:test:wide-{i}"), 600)) + .collect(); + let tx1 = store.transact(&big, "2026-01-01", None, None).unwrap(); + let small = datum(&mut store, "urn:test:after"); + let tx2 = store.transact(&[small], "2026-01-02", None, None).unwrap(); + let expected = store + .changes_after(0, 1, quipu::store::changes::Capture::OldAndNewValues, None) + .unwrap() + .records; + let shared = Arc::new(super::super::handle::StoreHandle::writer_only(store)); + + let open = |since: i64| { + changes_stream( + State(shared.clone()), + HeaderMap::new(), + params(json!({"since": since})), + ) + }; + let frames = read_frames(open(0).await, |f| f.0 == "quipu.changes").await; + let names: Vec<&str> = frames.iter().map(|f| f.0.as_str()).collect(); + let parts: Vec<&Frame> = frames + .iter() + .filter(|f| f.0 == "quipu.changes.part") + .collect(); + assert_eq!(names.first(), Some(&"quipu.changes.begin"), "{names:?}"); + assert!(parts.len() >= 2, "expected a split, got {names:?}"); + assert_eq!(names[names.len() - 2], "quipu.changes.commit", "{names:?}"); + // Only the barrier and the next whole transaction carry ids. + let ids: Vec> = frames.iter().map(|f| f.1.as_deref()).collect(); + assert!(ids[..ids.len() - 2].iter().all(Option::is_none), "{ids:?}"); + assert_eq!(ids[ids.len() - 2], Some(tx1.to_string().as_str())); + assert_eq!(ids[ids.len() - 1], Some(tx2.to_string().as_str())); + + // Every part fits; the digests chain over the parts exactly as sent. + let mut hasher = Sha256::new(); + let mut records = Vec::new(); + for (index, part) in parts.iter().enumerate() { + assert!( + fits_frame(&part.2), + "part {index} is {} bytes", + part.2.len() + ); + let body: Value = serde_json::from_str(&part.2).unwrap(); + assert_eq!(body["index"], index); + assert_eq!(body["source_tx"], tx1); + assert_eq!( + body["digest_before"], + hex::encode(hasher.clone().finalize()) + ); + hasher.update(part.2.as_bytes()); + records.extend(body["records"].as_array().unwrap().iter().cloned()); + } + let digest = hex::encode(hasher.finalize()); + let begin: Value = serde_json::from_str(&frames[0].2).unwrap(); + let commit: Value = serde_json::from_str(&frames[frames.len() - 2].2).unwrap(); + assert_eq!(begin["digest"], digest); + assert_eq!(commit["digest"], digest); + assert_eq!(commit["parts"], parts.len()); + assert_eq!(records, expected, "reassembled parts differ from the page"); + + // A reconnect before the barrier resumes from the previous transaction and + // replays this one from its first part; one after it skips straight on. + let again = read_frames(open(0).await, |f| f.0 == "quipu.changes.begin").await; + assert_eq!(again.len(), 1); + let mut headers = HeaderMap::new(); + headers.insert("last-event-id", tx1.to_string().parse().unwrap()); + let resumed = changes_stream(State(shared.clone()), headers, params(json!({}))).await; + let next = read_frames(resumed, |_| true).await; + assert_eq!(next[0].0, "quipu.changes"); + assert_eq!(next[0].1.as_deref(), Some(tx2.to_string().as_str())); +} + +/// One record larger than a frame cannot be split. It is refused with no id, +/// so the consumer's cursor stays put, and the stream ends. +#[tokio::test] +async fn a_record_larger_than_a_frame_is_refused_without_advancing() { + let mut store = Store::open_in_memory().unwrap(); + let huge = wide_datum(&mut store, "urn:test:huge", 1_200_000); + store.transact(&[huge], "2026-01-01", None, None).unwrap(); + let shared = Arc::new(super::super::handle::StoreHandle::writer_only(store)); + let response = + changes_stream(State(shared), HeaderMap::new(), params(json!({"since": 0}))).await; + let mut bytes = response.into_body().into_data_stream(); + let first = tokio::time::timeout(Duration::from_secs(10), bytes.next()) + .await + .unwrap() + .unwrap() + .unwrap(); + let text = String::from_utf8(first.to_vec()).unwrap(); + assert!(text.contains("event: quipu.changes.tx_too_large"), "{text}"); + assert!(!text.contains("id:"), "{text}"); + let end = tokio::time::timeout(Duration::from_secs(5), bytes.next()) + .await + .unwrap(); + assert!(end.is_none(), "stream should end after the refusal"); +} From 7529d93b5a959c5254b7f2ee7875ed0adf464106 Mon Sep 17 00:00:00 2001 From: Steve Brown Date: Fri, 23 Oct 2026 23:18:18 -0400 Subject: [PATCH 04/13] refactor(stream): build oversized-transaction frames independent of SSE chunk_transaction now returns transport-neutral frames (event name, optional cursor id, data string), and the SSE stream turns them into events only at the edge. Another transport can send the same frames and the same digests unchanged. A new test checks that the frames SSE sends equal the ones chunk_transaction returns, byte for byte. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01BzgSY2LpdGhBhb7D1nHbiF --- src/server/feed_stream.rs | 81 ++++++++++++++++++++++----------- src/server/feed_stream_tests.rs | 40 ++++++++++++++++ 2 files changed, 94 insertions(+), 27 deletions(-) diff --git a/src/server/feed_stream.rs b/src/server/feed_stream.rs index f574d7213..3c41b34e0 100644 --- a/src/server/feed_stream.rs +++ b/src/server/feed_stream.rs @@ -45,7 +45,7 @@ struct Cursor { wait: bool, terminal: bool, /// Frames of one chunked transaction still to send, in order. - pending: VecDeque, + pending: VecDeque, _slot: OwnedSemaphorePermit, } @@ -113,8 +113,8 @@ fn stream(store: SharedStore, headers: &HeaderMap, params: Params, changes: bool _slot: slot, }; let output = unfold(cursor, |mut cursor| async move { - if let Some(event) = cursor.pending.pop_front() { - return Some((Ok::<_, Infallible>(event), cursor)); + if let Some(frame) = cursor.pending.pop_front() { + return Some((Ok::<_, Infallible>(frame.into_sse()), cursor)); } if cursor.terminal { return None; @@ -221,7 +221,7 @@ fn stream(store: SharedStore, headers: &HeaderMap, params: Params, changes: bool } } let first = cursor.pending.pop_front(); - return first.map(|event| (Ok::<_, Infallible>(event), cursor)); + return first.map(|frame| (Ok::<_, Infallible>(frame.into_sse()), cursor)); } if validate_frame(cursor.offset, next, &data).is_err() { cursor.terminal = true; @@ -254,6 +254,28 @@ fn stream(store: SharedStore, headers: &HeaderMap, params: Params, changes: bool .into_response() } +/// One feed frame, independent of transport: the event name, the cursor id +/// it acknowledges (if any) and its data, byte-for-byte as sent. SSE builds an +/// `Event` from it only at the edge, so another transport can send the same +/// frames and digests unchanged. +#[derive(Debug, Clone, PartialEq, Eq)] +pub(crate) struct FeedFrame { + pub(crate) event: &'static str, + pub(crate) id: Option, + pub(crate) data: String, +} + +impl FeedFrame { + fn into_sse(self) -> Event { + let event = Event::default().event(self.event); + let event = match self.id { + Some(id) => event.id(id.to_string()), + None => event, + }; + event.data(self.data) + } +} + /// Split one change page (a single transaction) that exceeds the frame budget /// into `quipu.changes.begin`, numbered `quipu.changes.part` frames and a /// `quipu.changes.commit` barrier (local-sync-mode.md §6, F6). @@ -270,7 +292,7 @@ fn stream(store: SharedStore, headers: &HeaderMap, params: Params, changes: bool /// /// A single record larger than a frame cannot be split; it is refused with /// `quipu.changes.tx_too_large` and the cursor is not advanced. -fn chunk_transaction(page: &Value, next: i64) -> Result, Event> { +pub(crate) fn chunk_transaction(page: &Value, next: i64) -> Result, FeedFrame> { let records = page .get("records") .and_then(Value::as_array) @@ -296,13 +318,15 @@ fn chunk_transaction(page: &Value, next: i64) -> Result, Event> } let len = text + usize::from(!chunk.is_empty()); if !fits_frame_len(size + len) { - return Err(Event::default().event("quipu.changes.tx_too_large").data( - json!({ + return Err(FeedFrame { + event: "quipu.changes.tx_too_large", + id: None, + data: json!({ "source_tx": next, "reason": "one record exceeds the transport frame budget; cursor not advanced", }) .to_string(), - )); + }); } size += len; chunk.push(record.clone()); @@ -329,27 +353,30 @@ fn chunk_transaction(page: &Value, next: i64) -> Result, Event> for key in ["next_tx", "watermark_tx", "watermark_timestamp", "capture"] { commit[key] = page.get(key).cloned().unwrap_or(Value::Null); } - frames.push_back( - Event::default().event("quipu.changes.begin").data( - json!({ - "source_tx": next, - "parts": parts.len(), - "records": records.len(), - "bytes": parts.iter().map(String::len).sum::(), - "digest": digest, - }) - .to_string(), - ), - ); + frames.push_back(FeedFrame { + event: "quipu.changes.begin", + id: None, + data: json!({ + "source_tx": next, + "parts": parts.len(), + "records": records.len(), + "bytes": parts.iter().map(String::len).sum::(), + "digest": digest, + }) + .to_string(), + }); for part in parts { - frames.push_back(Event::default().event("quipu.changes.part").data(part)); + frames.push_back(FeedFrame { + event: "quipu.changes.part", + id: None, + data: part, + }); } - frames.push_back( - Event::default() - .event("quipu.changes.commit") - .id(next.to_string()) - .data(commit.to_string()), - ); + frames.push_back(FeedFrame { + event: "quipu.changes.commit", + id: Some(next), + data: commit.to_string(), + }); Ok(frames) } diff --git a/src/server/feed_stream_tests.rs b/src/server/feed_stream_tests.rs index f7cf83151..ba7f1503a 100644 --- a/src/server/feed_stream_tests.rs +++ b/src/server/feed_stream_tests.rs @@ -303,6 +303,46 @@ async fn oversized_transaction_streams_as_parts_then_continues() { assert_eq!(next[0].1.as_deref(), Some(tx2.to_string().as_str())); } +/// The split is transport-neutral: `chunk_transaction` returns the frames +/// and SSE only renders them, so another transport (the E2 hub) can send the +/// same names, ids and data bytes, and so the same digests. +#[tokio::test] +async fn sse_sends_the_transport_neutral_frames_unchanged() { + let mut store = Store::open_in_memory().unwrap(); + let big: Vec = (0..2500) + .map(|i| wide_datum(&mut store, &format!("urn:test:wide-{i}"), 600)) + .collect(); + let tx = store.transact(&big, "2026-01-01", None, None).unwrap(); + let shared = Arc::new(super::super::handle::StoreHandle::writer_only(store)); + let page = feed::changes_get( + State(shared.clone()), + headers_for_feed(), + Query( + serde_json::from_value( + json!({"since": 0, "limit": 1, "capture": "old_and_new_values"}), + ) + .unwrap(), + ), + ) + .await + .unwrap() + .0; + let neutral: Vec = chunk_transaction(&page, tx) + .unwrap() + .into_iter() + .map(|f| (f.event.to_string(), f.id.map(|id| id.to_string()), f.data)) + .collect(); + let response = + changes_stream(State(shared), HeaderMap::new(), params(json!({"since": 0}))).await; + let sent = read_frames(response, |f| f.0 == "quipu.changes.commit").await; + assert!( + neutral.len() >= 4, + "expected a split, got {}", + neutral.len() + ); + assert_eq!(sent, neutral); +} + /// One record larger than a frame cannot be split. It is refused with no id, /// so the consumer's cursor stays put, and the stream ends. #[tokio::test] From e6a6dd5ee4e1669eb3be0d8926b2a7b7066375ef Mon Sep 17 00:00:00 2001 From: Steve Brown Date: Fri, 23 Oct 2026 22:29:53 -0400 Subject: [PATCH 05/13] feat(changes): name the source store and log epoch on every page MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A change page did not say which store produced it, and nothing marked a history boundary, so a follower could not tell a restored or cloned primary, or one that had removed served records, from the one it was following (local-sync-mode.md §5, F5). Every page now carries source.store_id and source.log_epoch. The epoch lives in a new one-row log_epoch table, minted when the store is created (or first opened by this version) and minted again by Store::mint_log_epoch for an operation that removes served history. Share reconstruction excludes the table, so a restored copy mints its own epoch and its followers re-bootstrap. The explicit restore/adopt ownership step is deferred to aegis-s24.2, per the xdvg.1 ruling. Tests cover the page source, an epoch stable across reopen, a new epoch after minting, and a store that predates the table getting one on first open. Removing the field turns both tests red. Refs aegis-xdvg.1. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_0185RxTpLsPCQFB9ni72Nzac --- docs/book/src/architecture/change-feed.md | 9 +++- src/schema.rs | 10 +++++ src/share_completeness.rs | 5 +++ src/store/changes.rs | 8 ++++ src/store/changes_replay_tests.rs | 53 +++++++++++++++++++++++ src/store/open.rs | 48 +++++++++++++------- src/store/respace_map.rs | 4 ++ src/store/settings.rs | 22 ++++++++++ 8 files changed, 142 insertions(+), 17 deletions(-) diff --git a/docs/book/src/architecture/change-feed.md b/docs/book/src/architecture/change-feed.md index 09859c7f5..2e29257c4 100644 --- a/docs/book/src/architecture/change-feed.md +++ b/docs/book/src/architecture/change-feed.md @@ -27,6 +27,12 @@ mirror exactly which facts did. history: a reader never observes half a transaction. - **Ordering**: per entity, records arrive in commit order (transaction, then write order within it). Across entities there is no ordering promise. +- **Every page names its source**: `source.store_id` is the store, and + `source.log_epoch` is the epoch of its change log, minted when the store is + created and minted again by any operation that removes served history. A + follower that sees either value change is reading a different history and + must re-bootstrap rather than keep its cursor. A store reconstructed from a + share mints its own epoch. - **Watermark instead of heartbeat**: every page carries `watermark_tx` and `watermark_timestamp` — the newest committed transaction. An empty page with an advancing watermark means the store is idle (or your graph scope @@ -76,7 +82,8 @@ Both return one page: "next_tx": 43, "watermark_tx": 43, "watermark_timestamp": "2026-04-03", - "capture": "new_values" + "capture": "new_values", + "source": {"store_id": "urn:uuid:…", "log_epoch": "urn:uuid:…"} } ``` diff --git a/src/schema.rs b/src/schema.rs index 4d0885053..f20f2256a 100644 --- a/src/schema.rs +++ b/src/schema.rs @@ -37,6 +37,16 @@ CREATE TABLE IF NOT EXISTS store_identity ( store_id TEXT NOT NULL UNIQUE ); +-- The change log's epoch (local-sync-mode.md §5, F5). Minted when the store +-- is created and re-minted by any operation that removes served history, so a +-- follower holding a cursor from another epoch knows to re-bootstrap. +CREATE TABLE IF NOT EXISTS log_epoch ( + id INTEGER PRIMARY KEY CHECK (id = 1), + epoch TEXT NOT NULL, + minted_at TEXT NOT NULL, + reason TEXT NOT NULL +); + CREATE TABLE IF NOT EXISTS facts ( e INTEGER NOT NULL, a INTEGER NOT NULL, diff --git a/src/share_completeness.rs b/src/share_completeness.rs index d2913c4d4..29311306d 100644 --- a/src/share_completeness.rs +++ b/src/share_completeness.rs @@ -126,6 +126,11 @@ pub const DECLARED: &[(&str, Disposition)] = &[ // replay the origin had already spent is accepted on the copy. Excluding it // visibly is the only honest option. ("attestation_nonces", Disposition::Excluded), + // The change log's epoch names ONE copy's history. A reconstruction is a + // different copy, so it must not inherit the epoch: excluded, the restored + // 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), // 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 01ceee950..812e71884 100644 --- a/src/store/changes.rs +++ b/src/store/changes.rs @@ -90,6 +90,11 @@ pub struct ChangePage { pub watermark_tx: i64, pub watermark_timestamp: Option, pub capture: Capture, + /// The store these records came from, and its log epoch. A follower that + /// sees either change is reading a different history (a restore, a clone + /// or removed records) and must re-bootstrap rather than keep its cursor. + pub store_id: String, + pub log_epoch: String, } impl ChangePage { @@ -101,6 +106,7 @@ impl ChangePage { "watermark_tx": self.watermark_tx, "watermark_timestamp": self.watermark_timestamp, "capture": self.capture.as_str(), + "source": {"store_id": self.store_id, "log_epoch": self.log_epoch}, }) } } @@ -162,6 +168,8 @@ impl Store { watermark_tx, watermark_timestamp, capture, + store_id: self.store_id()?, + log_epoch: self.log_epoch()?, }) } diff --git a/src/store/changes_replay_tests.rs b/src/store/changes_replay_tests.rs index 8218f18cc..58f4899f0 100644 --- a/src/store/changes_replay_tests.rs +++ b/src/store/changes_replay_tests.rs @@ -328,3 +328,56 @@ fn f1_functional_supersede_replays_as_one_current_value() { .unwrap(); assert_replica_equals_source(&source); } + +/// F5: a page must name its source store and log epoch, so a follower can +/// detect a restored or cloned primary. +#[test] +fn f5_pages_carry_source_identity() { + let source = Store::open_in_memory().unwrap(); + let page = source + .changes_after(0, 10, Capture::NewValues, None) + .unwrap() + .to_json(); + assert!(page["source"]["store_id"].is_string(), "{page}"); + assert!(page["source"]["log_epoch"].is_string(), "{page}"); +} + +/// The epoch is minted once and kept across reopen; a store that predates it +/// gets one on its first open; minting a new one changes what pages report. +#[test] +fn f5_log_epoch_is_stable_until_minted_again() { + let dir = tempfile::tempdir().unwrap(); + let path = dir.path().join("store.db"); + let path = path.to_str().unwrap(); + let (store_id, first) = { + let store = Store::open(path).unwrap(); + (store.store_id().unwrap(), store.log_epoch().unwrap()) + }; + let store = Store::open(path).unwrap(); + assert_eq!(store.log_epoch().unwrap(), first, "epoch changed on reopen"); + let page = store + .changes_after(0, 1, Capture::NewValues, None) + .unwrap() + .to_json(); + assert_eq!(page["source"]["store_id"], store_id); + assert_eq!(page["source"]["log_epoch"], first); + + let minted = store.mint_log_epoch("test").unwrap(); + assert_ne!(minted, first); + let page = store + .changes_after(0, 1, Capture::NewValues, None) + .unwrap() + .to_json(); + assert_eq!(page["source"]["log_epoch"], minted); + assert_eq!( + page["source"]["store_id"], store_id, + "store identity is not the epoch" + ); + + store.conn.execute("DELETE FROM log_epoch", []).unwrap(); + drop(store); + let upgraded = Store::open(path).unwrap(); + let late = upgraded.log_epoch().unwrap(); + assert!(late.starts_with("urn:uuid:"), "{late}"); + assert_ne!(late, minted); +} diff --git a/src/store/open.rs b/src/store/open.rs index c1fe370d1..4902cadd1 100644 --- a/src/store/open.rs +++ b/src/store/open.rs @@ -250,6 +250,7 @@ impl Store { )?; super::search_scope_policy::initialize_schema(&conn)?; Self::ensure_store_identity(&conn)?; + Self::ensure_log_epoch(&conn)?; conn.execute_batch(VECTORS_SQL)?; // Frozen packs re-attach on every open (deep freeze): read the // registry BEFORE attach_all so the caller's attachments and the @@ -339,29 +340,25 @@ impl Store { |row| row.get(0), )?; if !exists { - let mut bytes = [0_u8; 16]; - SystemRandom::new() - .fill(&mut bytes) - .map_err(|_| Error::Store("cannot generate store identity".into()))?; - bytes[6] = (bytes[6] & 0x0f) | 0x40; - bytes[8] = (bytes[8] & 0x3f) | 0x80; - let hex = hex::encode(bytes); - let store_id = format!( - "urn:uuid:{}-{}-{}-{}-{}", - &hex[0..8], - &hex[8..12], - &hex[12..16], - &hex[16..20], - &hex[20..32] - ); conn.execute( "INSERT INTO store_identity (id, store_id) VALUES (1, ?1)", - params![store_id], + params![new_urn_uuid("store identity")?], )?; } Ok(()) } + /// A store opened for the first time, or first opened by a version that + /// predates the epoch, gets one now; its earlier history had none. + fn ensure_log_epoch(conn: &Connection) -> Result<()> { + conn.execute( + "INSERT OR IGNORE INTO log_epoch (id, epoch, minted_at, reason) \ + VALUES (1, ?1, strftime('%Y-%m-%dT%H:%M:%SZ', 'now'), 'created')", + params![new_urn_uuid("log epoch")?], + )?; + Ok(()) + } + /// The SQL [`Self::resolve`] runs. Exposed for the test that pins it /// byte-identical on an unattached store. #[cfg(test)] @@ -623,3 +620,22 @@ pub(crate) fn open_file_immutable(path: &str) -> Result { #[cfg(test)] #[path = "open_memory_tests.rs"] mod memory_tests; + +/// A random version-4 UUID as a `urn:uuid:` IRI. `what` names the failure. +pub(super) fn new_urn_uuid(what: &str) -> Result { + let mut bytes = [0_u8; 16]; + SystemRandom::new() + .fill(&mut bytes) + .map_err(|_| Error::Store(format!("cannot generate {what}")))?; + bytes[6] = (bytes[6] & 0x0f) | 0x40; + bytes[8] = (bytes[8] & 0x3f) | 0x80; + let hex = hex::encode(bytes); + Ok(format!( + "urn:uuid:{}-{}-{}-{}-{}", + &hex[0..8], + &hex[8..12], + &hex[12..16], + &hex[16..20], + &hex[20..32] + )) +} diff --git a/src/store/respace_map.rs b/src/store/respace_map.rs index 59817420a..f04d5429a 100644 --- a/src/store/respace_map.rs +++ b/src/store/respace_map.rs @@ -50,6 +50,10 @@ pub const COLUMN_CLASSIFICATION: &[(&str, &str, TermIdKind)] = &[ ("transaction_auth", "auth_class", TermIdKind::None), ("store_identity", "id", TermIdKind::None), ("store_identity", "store_id", TermIdKind::None), + ("log_epoch", "id", TermIdKind::None), + ("log_epoch", "epoch", TermIdKind::None), + ("log_epoch", "minted_at", TermIdKind::None), + ("log_epoch", "reason", TermIdKind::None), // Local search-default intent contains no interned term identifiers. ("search_default_policy", "id", TermIdKind::None), ("search_default_policy", "policy", TermIdKind::None), diff --git a/src/store/settings.rs b/src/store/settings.rs index e78fb8a72..716978714 100644 --- a/src/store/settings.rs +++ b/src/store/settings.rs @@ -30,6 +30,28 @@ impl Store { .map_err(Into::into) } + /// The change log's current epoch (local-sync-mode.md §5). A follower that + /// sees it change must re-bootstrap: history it already read may be gone. + pub fn log_epoch(&self) -> Result { + self.conn + .query_row("SELECT epoch FROM log_epoch WHERE id = 1", [], |row| { + row.get(0) + }) + .map_err(Into::into) + } + + /// Mint a new epoch, for an operation that removes served history or makes + /// this a different copy of it. Returns the new epoch. + pub fn mint_log_epoch(&self, reason: &str) -> Result { + let epoch = super::open::new_urn_uuid("log epoch")?; + self.conn.execute( + "UPDATE log_epoch SET epoch = ?1, reason = ?2, \ + minted_at = strftime('%Y-%m-%dT%H:%M:%SZ', 'now') WHERE id = 1", + rusqlite::params![epoch, reason], + )?; + Ok(epoch) + } + /// Defer auto-embedding: writes collect embed work instead of running the /// ONNX embed under the caller's lock. The caller MUST drain with /// [`Self::take_deferred_embed`] after each write and finish via From 614ce73d585123571c4a26d76a5d639c99d8e89f Mon Sep 17 00:00:00 2001 From: Steve Brown Date: Sat, 24 Oct 2026 01:22:45 -0400 Subject: [PATCH 06/13] fix(stream): name the source on a chunked transaction's begin and commit An oversized transaction's begin and commit frames carried no source, so a follower could not epoch-check a chunked transaction the way it checks a whole page, although a changed epoch must mean re-bootstrap, never splice. Both frames now carry the page's {store_id, log_epoch}. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01BzgSY2LpdGhBhb7D1nHbiF --- src/server/feed_stream.rs | 11 ++++++++++- src/server/feed_stream_tests.rs | 26 ++++++++++++++++++++++++++ 2 files changed, 36 insertions(+), 1 deletion(-) diff --git a/src/server/feed_stream.rs b/src/server/feed_stream.rs index 3c41b34e0..d126ab79d 100644 --- a/src/server/feed_stream.rs +++ b/src/server/feed_stream.rs @@ -350,7 +350,15 @@ pub(crate) fn chunk_transaction(page: &Value, next: i64) -> Result Result(), "digest": digest, + "source": page.get("source").cloned().unwrap_or(Value::Null), }) .to_string(), }); diff --git a/src/server/feed_stream_tests.rs b/src/server/feed_stream_tests.rs index ba7f1503a..9efac44e2 100644 --- a/src/server/feed_stream_tests.rs +++ b/src/server/feed_stream_tests.rs @@ -343,6 +343,32 @@ async fn sse_sends_the_transport_neutral_frames_unchanged() { assert_eq!(sent, neutral); } +/// A chunked transaction names its source on the frames a follower checks +/// before and after the parts: begin and the commit barrier. Without it a +/// follower cannot tell an epoch change inside a chunked transaction, and the +/// contract is "a changed epoch means re-bootstrap, never splice". +#[tokio::test] +async fn a_chunked_transaction_names_its_source_on_begin_and_commit() { + let mut store = Store::open_in_memory().unwrap(); + let big: Vec = (0..2500) + .map(|i| wide_datum(&mut store, &format!("urn:test:wide-{i}"), 600)) + .collect(); + store.transact(&big, "2026-01-01", None, None).unwrap(); + let expected = json!({ + "store_id": store.store_id().unwrap(), + "log_epoch": store.log_epoch().unwrap(), + }); + let shared = Arc::new(super::super::handle::StoreHandle::writer_only(store)); + let response = + changes_stream(State(shared), HeaderMap::new(), params(json!({"since": 0}))).await; + let frames = read_frames(response, |f| f.0 == "quipu.changes.commit").await; + for name in ["quipu.changes.begin", "quipu.changes.commit"] { + let frame = frames.iter().find(|f| f.0 == name).unwrap(); + let body: Value = serde_json::from_str(&frame.2).unwrap(); + assert_eq!(body["source"], expected, "{name}: {body}"); + } +} + /// One record larger than a frame cannot be split. It is refused with no id, /// so the consumer's cursor stays put, and the stream ends. #[tokio::test] From ff89873caf0779357afbb17e5c9b6873b5e6acb5 Mon Sep 17 00:00:00 2001 From: Steve Brown Date: Sat, 24 Oct 2026 01:28:01 -0400 Subject: [PATCH 07/13] feat(applier): the shared applier, slice 1: a replica from the shipped feed 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 (docs/design/shared-applier-api.md). This slice applies the frames that ship today, given as (event, data) text exactly as SSE or a WebSocket delivers them: /changes pages and the begin/part/commit frames of an oversized transaction. It does no I/O, so it builds for wasm32 too. - The first frame binds the replica to its source (store_id, log_epoch); a frame from another source returns Rebootstrap, never a splice. - Each source transaction applies in one savepoint with the persisted cursor; a chunked one is buffered, its running digest checked part by part, and nothing is visible until the commit verifies (or the caller is told to resume from the cursor). - A record the replica cannot reproduce exactly (a byte value sent as its length only, an unknown value shape) refuses the whole transaction. - Freshness is (last applied tx, time since the last frame), never bounded. The feed-replay conformance tests now use the same record->datum conversion (record_to_datum) and share live_digest, so there is one mapping. The cursor table is excluded from shares, like the log epoch it is bound to. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01BkdpDkAkbTvXvNbNacVpJb --- src/applier.rs | 562 ++++++++++++++++++++++++++++++ src/applier_tests.rs | 308 ++++++++++++++++ src/lib.rs | 1 + src/share_completeness.rs | 5 + src/store/changes_replay_tests.rs | 107 +----- 5 files changed, 882 insertions(+), 101 deletions(-) create mode 100644 src/applier.rs create mode 100644 src/applier_tests.rs diff --git a/src/applier.rs b/src/applier.rs new file mode 100644 index 000000000..b14e44096 --- /dev/null +++ b/src/applier.rs @@ -0,0 +1,562 @@ +//! 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.** A transaction at or below the cursor is a duplicate. +//! - **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}; + +/// 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. +#[derive(Debug, Clone, PartialEq, Eq)] +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 chunk sequence broke: 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 was applied. + Refused { + /// Why the frame cannot be applied exactly. + reason: String, + }, + /// An event this slice does not act on (heartbeats, newer frame kinds). + Ignored, +} + +/// What a reader may say about how current the replica is. +#[derive(Debug, Clone, PartialEq, Eq)] +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, +} + +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)?; + 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, + }) + } + + /// 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); + 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 { + 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); + } + 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"); + }; + 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); + } + 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'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) { + 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)?; + } + } + 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], + )?; + 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 = op == "retract"; + 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_tests.rs b/src/applier_tests.rs new file mode 100644 index 000000000..990528b71 --- /dev/null +++ b/src/applier_tests.rs @@ -0,0 +1,308 @@ +//! 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. +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. +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. +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}) + .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}); + if with_source { + commit["source"] = page["source"].clone(); + } + frames.push(("quipu.changes.commit", commit.to_string())); + frames +} + +#[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 + )); +} diff --git a/src/lib.rs b/src/lib.rs index b77737755..3538fd0a6 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/share_completeness.rs b/src/share_completeness.rs index 29311306d..ae0edc3a3 100644 --- a/src/share_completeness.rs +++ b/src/share_completeness.rs @@ -131,6 +131,11 @@ 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), // 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_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" { From 18781af820ddbd7ef61c4bd920e888acfe7c197c Mon Sep 17 00:00:00 2001 From: Steve Brown Date: Sat, 24 Oct 2026 01:34:18 -0400 Subject: [PATCH 08/13] test(applier): a digest-sensitivity control and the bridge's JSON contract The replica equality oracle now sits beside the code it judges, so prove it sees valid time, byte content and graph (review ask). Outcome and Freshness serialize tagged by kind, the contract a browser bridge returns. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01BkdpDkAkbTvXvNbNacVpJb --- src/applier.rs | 9 +++-- src/applier_tests.rs | 78 ++++++++++++++++++++++++++++++++++++++++++++ 2 files changed, 85 insertions(+), 2 deletions(-) diff --git a/src/applier.rs b/src/applier.rs index b14e44096..aa0eca7dc 100644 --- a/src/applier.rs +++ b/src/applier.rs @@ -58,7 +58,12 @@ pub struct Progress { } /// What one frame did, and what the caller must do next. -#[derive(Debug, Clone, PartialEq, Eq)] +/// +/// 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 { @@ -94,7 +99,7 @@ pub enum Outcome { } /// What a reader may say about how current the replica is. -#[derive(Debug, Clone, PartialEq, Eq)] +#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize)] pub struct Freshness { /// The last source transaction applied, if any. pub source_tx: Option, diff --git a/src/applier_tests.rs b/src/applier_tests.rs index 990528b71..cd650e60b 100644 --- a/src/applier_tests.rs +++ b/src/applier_tests.rs @@ -306,3 +306,81 @@ fn the_cursor_survives_reopening_and_freshness_is_never_bounded() { 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}) + ); +} From 569130d1229776312773cacb7de01f730e182e27 Mon Sep 17 00:00:00 2001 From: Steve Brown Date: Sat, 24 Oct 2026 02:55:39 -0400 Subject: [PATCH 09/13] fix(applier): refuse a gap; a page names the cursor it answers A /changes page covers every transaction in its scope after the cursor it was asked for, but did not say which cursor that was. Transaction ids are global and a per-graph page skips other graphs' ids, so the applier could not tell a lost page from an ordinary skip and would splice past it. - Every page now carries `since`, and the chunked `begin` and `commit` frames repeat it. - The applier resumes (nothing applied) when a page or a chunked transaction answers a cursor above its own. A new replica starts at 0 until a bootstrap exists. A page that names no `since` is refused. - A page arriving while a chunked transaction is open discards the chunk and resumes. Before, the page could apply the chunk's transaction, move the cursor past it, and the commit was then dropped as a duplicate. Seven tests: - a dropped page; - a first page not at 0; - a page without `since`; - a page inside an open chunk; - an interrupted chunk restarted from a new begin; - a re-opened applier holding no staged parts; - a chunk that skips a transaction. With the two checks disabled, four of them fail. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01BkdpDkAkbTvXvNbNacVpJb --- docs/book/src/architecture/change-feed.md | 8 +- src/applier.rs | 42 ++++++- src/applier_tests.rs | 147 +++++++++++++++++++++- src/server/feed_stream.rs | 2 + src/store/changes.rs | 6 + 5 files changed, 200 insertions(+), 5 deletions(-) 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/src/applier.rs b/src/applier.rs index aa0eca7dc..5419b5cd8 100644 --- a/src/applier.rs +++ b/src/applier.rs @@ -20,6 +20,12 @@ //! buffered, its digest chain checked part by part, and nothing is visible //! until the commit verifies. //! - **Idempotent.** A transaction at or below the cursor is a duplicate. +//! - **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 @@ -79,7 +85,8 @@ pub enum Outcome { Duplicate, /// A page with no records: the feed has nothing new. Idle, - /// The chunk sequence broke: nothing was applied; reconnect from `from`. + /// 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, @@ -89,7 +96,9 @@ pub enum Outcome { /// Why the replica cannot continue from its cursor. reason: String, }, - /// The frame cannot be applied exactly; nothing was applied. + /// 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, @@ -203,6 +212,13 @@ impl Applier { } 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")); }; @@ -212,6 +228,9 @@ impl Applier { 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); @@ -231,6 +250,12 @@ impl Applier { ) 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), @@ -296,6 +321,9 @@ impl Applier { 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) } @@ -303,6 +331,16 @@ impl Applier { 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 { diff --git a/src/applier_tests.rs b/src/applier_tests.rs index cd650e60b..96728d292 100644 --- a/src/applier_tests.rs +++ b/src/applier_tests.rs @@ -198,11 +198,11 @@ fn chunk_frames(page: &str, per_part: usize, with_source: bool) -> Vec<(&'static 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}) + 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}); + 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(); } @@ -210,6 +210,149 @@ fn chunk_frames(page: &str, per_part: usize, with_source: bool) -> Vec<(&'static 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(); diff --git a/src/server/feed_stream.rs b/src/server/feed_stream.rs index d126ab79d..4a90b8355 100644 --- a/src/server/feed_stream.rs +++ b/src/server/feed_stream.rs @@ -353,6 +353,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/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, From c34da2bea5593d01cdc3beebbddbedf297f89535 Mon Sep 17 00:00:00 2001 From: Steve Brown Date: Sat, 24 Oct 2026 05:55:19 -0400 Subject: [PATCH 10/13] fix(knowledge): refresh the src/lib.rs source witness after adding the applier The applier commit added `pub mod applier` to src/lib.rs, so its repository-knowledge witness hash went stale (the pendant source controls job). The record's summary still holds; only the hash changes. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01BkdpDkAkbTvXvNbNacVpJb --- docs/knowledge/repository-knowledge.json | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/docs/knowledge/repository-knowledge.json b/docs/knowledge/repository-knowledge.json index 093369cf6..bfaa5143a 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": "14927c98b7edc319ad3ee40d4829b2752ade3d593266d8acaf1663ee7831214c", + "source_sha256": "132902ff275bb8c777571b07aaee46aa6a3523afee047fab7255dba60771f2d0", "properties": { "interfaceKind": "library", "publicName": "quipu" From 0572529cc0307c3b96eec2585bc5c78f2d0a3e45 Mon Sep 17 00:00:00 2001 From: Steve Brown Date: Sat, 24 Oct 2026 07:54:57 -0400 Subject: [PATCH 11/13] fix(knowledge): refresh the rest-api.md source witness 15c81282 changed docs/book/src/reference/rest-api.md without moving its witness, so the repository-knowledge check refused the registry. Only the source_sha256 moves. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01BkdpDkAkbTvXvNbNacVpJb --- docs/knowledge/repository-knowledge.json | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/docs/knowledge/repository-knowledge.json b/docs/knowledge/repository-knowledge.json index bfaa5143a..14e7b38f9 100644 --- a/docs/knowledge/repository-knowledge.json +++ b/docs/knowledge/repository-knowledge.json @@ -117,7 +117,7 @@ "source": "docs/book/src/reference/rest-api.md", "subject": "http://aegis.gastown.local/ontology/doc/quipu/docs%2Fbook%2Fsrc%2Freference%2Frest-api.md#rest-api", "summary": "The native server exposes graph operations over HTTP; authentication, request budgets and provenance headers have separate roles.", - "source_sha256": "c3791da1d6b8bd46eed228ad9a603cfce4570bbeada550e54c7a5017594fa420", + "source_sha256": "f4df26ee18f2a67748746ce776fca3805fd5dfc32643a97acdc9cd0df8eebc48", "properties": { "interfaceKind": "REST", "publicName": "quipu-server HTTP" From 3442c9de35f7ec16ab34b1ef41ec36fc9684c719 Mon Sep 17 00:00:00 2001 From: Steve Brown Date: Sat, 24 Oct 2026 07:54:57 -0400 Subject: [PATCH 12/13] feat(applier): a duplicate is checked against the retained transaction digest Rule 6 of the shared applier API (docs/design/shared-applier-api.md): a transaction at or below the cursor was skipped as a duplicate without looking at it, so a redelivered tx 5 with changed records answered exactly like an exact replay. The applier now computes each transaction's content digest itself from the decoded records (domain tag, store id, log epoch, source tx, then each record's length-prefixed exact encoding, sorted), independent of framing, and retains it in replica_applied_tx in the same savepoint as the records and the cursor. A redelivery is Duplicate only when its digest equals the retained one; a different digest is Stopped { DigestMismatch }, no retained row is Stopped { DigestUnverifiable }, nothing is written, and the stop is sticky. A frame from a new source still answers Rebootstrap. replica_applied_tx is excluded from shares with the cursor it belongs to. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01BkdpDkAkbTvXvNbNacVpJb --- src/applier.rs | 39 ++++++- src/applier_digest.rs | 217 ++++++++++++++++++++++++++++++++++++ src/applier_digest_tests.rs | 178 +++++++++++++++++++++++++++++ src/applier_tests.rs | 10 +- src/share_completeness.rs | 4 + 5 files changed, 444 insertions(+), 4 deletions(-) create mode 100644 src/applier_digest.rs create mode 100644 src/applier_digest_tests.rs diff --git a/src/applier.rs b/src/applier.rs index 5419b5cd8..b1d93d6fb 100644 --- a/src/applier.rs +++ b/src/applier.rs @@ -19,7 +19,10 @@ //! 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.** A transaction at or below the cursor is a duplicate. +//! - **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. @@ -45,6 +48,10 @@ 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 { @@ -105,6 +112,14 @@ pub enum Outcome { }, /// 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. @@ -134,6 +149,7 @@ pub struct Applier { progress: Option, chunk: Option, last_frame_ms: Option, + stopped: Option, } const PROGRESS_DDL: &str = "CREATE TABLE IF NOT EXISTS replica_progress (\ @@ -147,6 +163,7 @@ impl Applier { /// [`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)?; let progress = store .conn .query_row( @@ -167,6 +184,7 @@ impl Applier { progress, chunk: None, last_frame_ms: None, + stopped: None, }) } @@ -199,6 +217,15 @@ impl Applier { 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)), @@ -372,6 +399,10 @@ impl Applier { 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)? { @@ -444,11 +475,17 @@ fn apply_one_tx( 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 { diff --git a/src/applier_digest.rs b/src/applier_digest.rs new file mode 100644 index 000000000..1c5020ae3 --- /dev/null +++ b/src/applier_digest.rs @@ -0,0 +1,217 @@ +//! 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 == "retract"; + 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) +} + +/// 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(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 }, + }; + 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..a83f3651b --- /dev/null +++ b/src/applier_digest_tests.rs @@ -0,0 +1,178 @@ +//! 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}}) + ); +} diff --git a/src/applier_tests.rs b/src/applier_tests.rs index 96728d292..78322d2ca 100644 --- a/src/applier_tests.rs +++ b/src/applier_tests.rs @@ -21,7 +21,7 @@ fn d(store: &Store, e: &str, a: &str, v: Value, op: Op) -> Datum { } /// A source with ROOT and named-graph writes, a retraction and bytes. -fn source() -> Store { +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( @@ -63,7 +63,7 @@ fn source() -> Store { } /// The feed one transaction per page, as `/changes/stream` sends it. -fn pages(source: &Store) -> Vec { +pub(super) fn pages(source: &Store) -> Vec { let mut out = Vec::new(); let mut cursor = 0; loop { @@ -181,7 +181,11 @@ fn a_length_only_byte_value_is_refused_and_nothing_is_written() { /// (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. -fn chunk_frames(page: &str, per_part: usize, with_source: bool) -> Vec<(&'static str, String)> { +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(); diff --git a/src/share_completeness.rs b/src/share_completeness.rs index ae0edc3a3..6d2624d61 100644 --- a/src/share_completeness.rs +++ b/src/share_completeness.rs @@ -136,6 +136,10 @@ pub const DECLARED: &[(&str, Disposition)] = &[ // 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), // 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. From 258cba7a53a2268eb5ab92bd8e64c7ceb5c6fda9 Mon Sep 17 00:00:00 2001 From: Steve Brown Date: Sat, 24 Oct 2026 12:17:29 -0400 Subject: [PATCH 13/13] fix(applier): a stop persists across reopening, and an unknown op is refused Review findings on the shared applier: - P1: the sticky Stopped outcome lived only in memory, so a fresh Applier on the same replica tailed on past a recorded digest mismatch. The stop is now kept in replica_stop (first stop wins) and restored by open; a replica whose history disagreed with its source stays stopped until it is rebuilt. - P2: any op other than "retract" was applied as an assert. Only "assert" and "retract" are accepted, in the record conversion and the transaction digest alike; anything else refuses the transaction with nothing written. replica_stop is excluded from shares with the cursor it belongs to. Controls: a stop survives reopening (fails with open ignoring the stop), and an unknown op is refused (fails with unknown ops read as assert). Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01BkdpDkAkbTvXvNbNacVpJb --- src/applier.rs | 8 +++++-- src/applier_digest.rs | 48 +++++++++++++++++++++++++++++++++++-- src/applier_digest_tests.rs | 41 +++++++++++++++++++++++++++++++ src/share_completeness.rs | 3 +++ 4 files changed, 96 insertions(+), 4 deletions(-) diff --git a/src/applier.rs b/src/applier.rs index b1d93d6fb..f3e30f80b 100644 --- a/src/applier.rs +++ b/src/applier.rs @@ -164,6 +164,7 @@ impl Applier { 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( @@ -184,7 +185,7 @@ impl Applier { progress, chunk: None, last_frame_ms: None, - stopped: None, + stopped: digest::load_stop(store)?, }) } @@ -519,7 +520,10 @@ pub fn record_to_datum( ) else { return Ok(Err("record lacks entity, attribute or op".into())); }; - let retract = op == "retract"; + 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, diff --git a/src/applier_digest.rs b/src/applier_digest.rs index 1c5020ae3..dfc3cf3e8 100644 --- a/src/applier_digest.rs +++ b/src/applier_digest.rs @@ -66,7 +66,7 @@ fn encode_record(record: &Json, tx_timestamp: &str) -> std::result::Result std::result::Result, String> { 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( @@ -171,7 +214,7 @@ pub(super) fn retained(store: &Store, source: &Source, tx: i64) -> Result