swaits/wanderbooks
1
0 Fork 0
Code Issues Pull Requests Actions 10 Packages Projects Releases Wiki Activity

Store: say "newer wanderbooks", not "corrupt", for books from a newer version

Browse Source 
An unknown event type on a whole, chained line, or a journal.format_set
declaring a newer journal format, now raises StoreError::NewerJournal
(update wanderbooks) instead of JournalCorrupt. Corruption (bad hash,
seq, or JSON) is still corruption. The gate runs in the store, the
Snapshot and the GUI, after each line's parse, seq and hash checks, and
never skips a line it can't read. Adds the journal.format_set event
(format 1 today; nothing writes it yet) and a registry of every event
type this version reads, tested against the replay and projection arms.

Review fixes (P0 review, 2026-10-16):
- A line's own hash is checked only by the next line's prev, and the
  gate used to refuse a line before reading the next: a flipped bit in a
  middle line's type (or a format_set edited from 1 to 2) was reported
  as "newer, the books are not damaged". Now the reader settles every
  refusal against the next line (format::settle): a next line that
  doesn't chain makes it corrupt; one that chains makes it newer and
  verified; with no next line the message says "newer or damaged; update,
  then run verify", never "not damaged" (NewerJournal.verified).
- A newer format number is judged newer before its features are parsed:
  a newer format may shape them differently.
- Snapshot treats a blank complete line as corrupt, as the writer does.
- docs/spec/rules.md B-27, which the code cites, now exists.

Round-2 review fixes (2026-10-16):
- The envelope {seq, ts, type, payload, prev} and its hash (SHA-256 of
  the canonical JSON) are frozen across formats (rules.md B-27); new
  fields go in payload. settle reads only the next line's prev, leniently,
  so a newer format's next line (a float, a string seq, keys of its own)
  is never called corrupt: no readable prev means unverified newer.
- A key beside the envelope's was silently dropped (and the chain then
  broke as "corrupt"): it is now NewerJournal, hashed over the raw
  canonical JSON so the next line still checks it; JournalLine::parse
  refuses such a line.
- "the books are not damaged" overclaimed: a verified refusal says line N
  is intact and this version can't check the lines after it.
- The writer read the journal with BufRead::lines (invalid UTF-8 was an
  I/O error); it now splits bytes on newlines as the Snapshot does, so a
  non-UTF-8 line is JournalCorrupt in both.
- Store::open truncated a torn tail before the gate ran, altering newer
  books (perhaps mid-append) before refusing them. The gate now runs over
  the complete lines first; the tail is set aside (and journaled) only
  when they pass.
- Mutation testing (docs/testing.md, P0): the viewer and the writer say
  the same words about the same damaged line (the newline is no part of
  it); read_journal_and_format returns every line and the format in
  force; the journal page tells a format declaration.
Plan: plan/served.md P0 and 5.1.

Review round 3 (S6, S7). Store::open replays every complete line in
memory (the first line's currency, then the ledger) before a cut-off
last line is set aside, so a complete line the books can't apply
refuses the open with the journal byte for byte as it was and nothing
set aside (B-27 SCENARIO 3; test
a_bad_complete_line_before_a_torn_tail_leaves_the_journal_as_it_was);
catch_up now only projects. The viewer
applies the writer's first-line rule (created_currency, shared) instead
of guessing USD: any other first line, or a store.created without a
sound base_currency, is corrupt to both (test
the_first_line_is_store_created_for_the_viewer_and_the_writer). fuzz_store_replay varies the
first line (14,696 runs clean).

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01T3gXMVt7XJPi6ens1or9Dj
This commit is contained in:
swaitsandClaude Opus 5.5 committed 2026-10-17 00:32:42 -06:00
1 parent 4f8047689f
commit 9f9f00ed9a
16 files changed
+2029 -77
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

crates/ledger-store/src/error.rs
+50
View File
@@ -55,6 +55,33 @@ pub enum StoreError {
#[error("journal corrupt: {0}")]
JournalCorrupt(String),
/// A whole, correctly chained journal line this version can't read:
/// a newer wanderbooks wrote the books. When the line after it chains
/// onto it (`verified`), its own bytes are checked: it is intact,
/// though the lines after it are the newer version's to check. When
/// nothing checks it (it is the last line, or the next has no `prev`
/// this version can read), it is either newer or damaged, and the
/// message says so (rules.md B-27).
#[error("{}", newer_journal_text(*seq, event_type, need, *verified))]
NewerJournal {
/// The first line this version can't read.
seq: u64,
/// Its type.
event_type: String,
/// What reading it needs: a journal format, features, or an event
/// type this version doesn't know.
need: String,
/// Whether the line's own hash was checked (the next line's `prev`
/// matches it). False when nothing checks it: a damaged line would
/// read the same.
verified: bool,
},
/// A journal format this version can't declare: newer than it reads,
/// lower than the books already use, or with a feature it doesn't know.
#[error("{0}")]
FormatRefused(String),
/// Replay produced different ids than the journal recorded.
#[error("replay mismatch: {0}")]
ReplayMismatch(String),
@@ -109,3 +136,26 @@ pub enum StoreError {
#[error("base currency mismatch: store is {0}, requested {1}")]
CurrencyMismatch(String, String),
}
/// What [`StoreError::NewerJournal`] says. Only a line whose own hash was
/// checked lets it say that line is intact; it never vouches for the
/// lines after it, which this version can't read.
fn newer_journal_text(seq: u64, event_type: &str, need: &str, verified: bool) -> String {
let format = crate::format::SUPPORTED_JOURNAL_FORMAT;
if verified {
format!(
"these books were written by a newer version of wanderbooks: journal line {seq} \
({event_type}) needs {need}, and this version reads journal format {format}. \
Update wanderbooks to open them. Line {seq} is intact (the line after it checks \
it); this version can't check the lines after it"
)
} else {
format!(
"journal line {seq} ({event_type}) needs {need}, and this version reads journal \
format {format}. Nothing this version can read checks that line (it is the last \
line, or the line after it isn't one this version reads): either a newer version \
of wanderbooks wrote it, or the line is damaged. Update wanderbooks, then run \
verify to tell which"
)
}
}
crates/ledger-store/src/format.rs
+469
View File
@@ -0,0 +1,469 @@
//! The format gate: what this version of wanderbooks can read.
//!
//! Every journal line has a type, and this version knows a fixed set of
//! them ([`EVENT_TYPES`]). Books written by a newer wanderbooks can hold a
//! type this version has never seen, or declare a newer journal format
//! (`journal.format_set`) before the first event that needs it. Either way
//! the line is whole and chained: the books are fine, and this reader is
//! old. It says so ([`StoreError::NewerJournal`]) instead of calling the
//! books corrupt, and stops there: it never skips a line it can't read.
//!
//! The gate runs as each line is read, after its parse, seq and hash
//! checks ([`crate::journal::read_journal_and_format`], and
//! [`crate::Snapshot`] line by line).
//!
//! What every format shares is frozen (rules.md B-27): a line is the
//! envelope `{seq, ts, type, payload, prev}` and nothing beside it, and
//! its hash is the SHA-256 of its canonical JSON (keys sorted, no
//! whitespace). A newer format adds its fields inside `payload`. So this
//! version can always follow the chain one line past the first line it
//! can't read, and say whether that line is intact.
use crate::error::StoreError;
use crate::journal::{JournalLine, ReadLine};
use serde_json::{Value, json};
/// The newest journal format this version reads and writes.
pub const SUPPORTED_JOURNAL_FORMAT: u32 = 1;
/// Optional journal features this version reads (none yet).
pub const KNOWN_FEATURES: &[&str] = &[];
/// The line type that declares the journal format.
pub const FORMAT_SET: &str = "journal.format_set";
/// Every journal line type this version reads. Each has a replay arm and a
/// projection arm in `store.rs` (`tests/format.rs` checks both ways).
pub const EVENT_TYPES: &[&str] = &[
"account.added",
"account.renumbered",
"account.tax_set",
"asset.acquisition_linked",
"asset.corrected",
"asset.depreciation_posted",
"asset.described",
"asset.disposed",
"asset.registered",
"asset.serial_set",
"asset.tagged",
"asset.tax_basis_set",
"attachment.linked",
"attachment.unlinked",
"attachment.withdrawn",
"bill.created",
"bill.paid",
"class.added",
"customer.added",
"document.filed",
"entry.created",
"entry.line_added",
"entry.line_class_set",
"entry.no_party",
"entry.parties_set",
"entry.party_set",
"entry.posted",
"entry.reversed",
"entry.vendor_set",
"fx.rate_set",
"invoice.created",
"invoice.payment_received",
FORMAT_SET,
"journal.tail_repaired",
"line.cleared",
"party.added",
"party.role_set",
"party.tin_set",
"party.updated",
"party.w4_set",
"party.w9_set",
"payroll.run",
"payroll.taxes_remitted",
"period.closed",
"period.created",
"receipt.attached",
"receipt.linked",
"receipt.unlinked",
"reconciliation.completed",
"reconciliation.voided",
"recurring.generated",
"recurring.registered",
"role.assigned",
"schedule.posted",
"schedule.registered",
"statement.recorded",
"statement.voided",
"store.created",
"vendor.1099_set",
"vendor.added",
];
/// The journal format in force: the last `journal.format_set` read, or
/// format 1 with no features for books that never declared one.
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct JournalFormat {
/// Format number: 1 for every journal written before the gate.
pub format: u32,
/// Optional features in use, as declared.
pub features: Vec<String>,
}
impl Default for JournalFormat {
fn default() -> Self {
Self {
format: 1,
features: Vec::new(),
}
}
}
impl JournalFormat {
/// Pass one verified line through the gate, before it is replayed.
///
/// ```text
/// GOAL: an older wanderbooks never misreads newer books, and never
/// calls them corrupt.
/// SOURCE: rules.md B-27; plan/served.md 5.1 (format gate).
/// INVARIANTS:
/// 1. Only lines that parsed, chained and kept their seq reach the
/// gate: damage stays JournalCorrupt.
/// 2. NEVER admits a line this version can't replay.
/// SCENARIOS:
/// 1. IF the type is one this version doesn't know: SHOULD refuse
/// with NewerJournal
/// BECAUSE skipping it would show books that differ from the
/// record without saying so.
/// 2. IF it is a journal.format_set newer than
/// SUPPORTED_JOURNAL_FORMAT, or with a feature not in
/// KNOWN_FEATURES: SHOULD refuse with NewerJournal.
/// 3. IF it is a journal.format_set this version reads: SHOULD
/// record it as the format in force.
/// 4. IF a journal.format_set payload is malformed: SHOULD refuse
/// with JournalCorrupt.
/// ```
pub fn admit(&mut self, line: &JournalLine) -> Result<(), StoreError> {
if line.line_type == FORMAT_SET {
*self = declared(line)?;
Ok(())
} else if EVENT_TYPES.contains(&line.line_type.as_str()) {
Ok(())
} else {
Err(unknown_type(line))
}
}
/// Check a declaration this version is about to write: a format it
/// reads, not lower than the one in force, with features it knows.
/// Returns the journal payload.
pub(crate) fn declaration(&self, format: u32, features: &[&str]) -> Result<Value, StoreError> {
if format == 0 || format > SUPPORTED_JOURNAL_FORMAT {
return Err(StoreError::FormatRefused(format!(
"journal format {format}: this version writes 1 through {SUPPORTED_JOURNAL_FORMAT}"
)));
}
if format < self.format {
return Err(StoreError::FormatRefused(format!(
"journal format {format}: the books already use format {}, and a format is never lowered",
self.format
)));
}
if let Some(f) = features.iter().find(|f| !KNOWN_FEATURES.contains(f)) {
return Err(StoreError::FormatRefused(format!(
"journal feature {f:?}: this version doesn't know it"
)));
}
let mut features: Vec<&str> = features.to_vec();
features.sort_unstable();
features.dedup();
Ok(json!({ "format": format, "features": features }))
}
}
/// Read a `journal.format_set` line, refusing a format or feature this
/// version doesn't read.
pub(crate) fn declared(line: &JournalLine) -> Result<JournalFormat, StoreError> {
let p = &line.payload;
let format = p
.get("format")
.and_then(Value::as_u64)
.filter(|&f| f > 0)
.ok_or_else(|| {
StoreError::JournalCorrupt(format!("seq {}: {FORMAT_SET} format", line.seq))
})?;
// A newer format may shape the rest of its declaration differently:
// it is newer, whatever its features look like.
if format > u64::from(SUPPORTED_JOURNAL_FORMAT) {
return Err(newer(line, format!("journal format {format}")));
}
let format = u32::try_from(format).unwrap_or(SUPPORTED_JOURNAL_FORMAT);
let features = p
.get("features")
.and_then(Value::as_array)
.and_then(|a| {
a.iter()
.map(|f| f.as_str().map(str::to_string))
.collect::<Option<Vec<_>>>()
})
.ok_or_else(|| {
StoreError::JournalCorrupt(format!("seq {}: {FORMAT_SET} features", line.seq))
})?;
let unknown: Vec<&str> = features
.iter()
.map(String::as_str)
.filter(|f| !KNOWN_FEATURES.contains(f))
.collect();
if !unknown.is_empty() {
return Err(newer(
line,
format!("journal features {}", unknown.join(", ")),
));
}
Ok(JournalFormat { format, features })
}
/// The refusal for a line whose type this version doesn't know.
pub(crate) fn unknown_type(line: &JournalLine) -> StoreError {
newer(line, "an event type this version doesn't know".into())
}
/// The gate for one line as read: a line with a key beside the envelope's
/// is newer (rules.md B-27: this version never drops a key it can't
/// read), and any other goes to `admit` (JournalFormat::admit).
pub(crate) fn gate(
read: &ReadLine,
admit: &mut impl FnMut(&JournalLine) -> Result<(), StoreError>,
) -> Result<(), StoreError> {
if read.unknown_keys.is_empty() {
admit(&read.line)
} else {
Err(newer(
&read.line,
format!(
"journal line fields this version doesn't know ({})",
read.unknown_keys.join(", ")
),
))
}
}
/// The gate's refusal, unverified: only the reader knows whether a later
/// line checks this one's hash ([`settle`]).
fn newer(line: &JournalLine, need: String) -> StoreError {
StoreError::NewerJournal {
seq: line.seq,
event_type: line.line_type.clone(),
need,
verified: false,
}
}
/// Settle the gate's refusal of a line whose hash is `hash`, given the raw
/// journal line after it (`next`, None when the refused line is the last).
///
/// ```text
/// GOAL: damage is never reported as books from a newer wanderbooks, and
/// books from a newer wanderbooks never as damage.
/// SOURCE: rules.md B-27 (the envelope and its hash are frozen across
/// formats); docs/receipt-store.md "Line format" (a line's hash is
/// checked only by the next line's prev).
/// INVARIANTS:
/// 1. Only a NewerJournal refusal is settled; anything else passes
/// through unchanged.
/// 2. Of the next line only `prev` is read, leniently (any JSON object
/// with a string `prev`): it is written in a format this version
/// may not read, with floats, a string seq or keys of its own.
/// SCENARIOS:
/// 1. IF the next line has a prev that isn't `hash`: SHOULD be
/// JournalCorrupt
/// BECAUSE the refused line changed after it was written (a flipped
/// bit in its type reads like a type this version doesn't know).
/// 2. IF the next line's prev is `hash`: SHOULD be NewerJournal,
/// verified: the refused line is intact (the lines after it are the
/// newer version's to check).
/// 3. IF there is no next line, or it has no prev this version can
/// read: SHOULD be NewerJournal, unverified
/// BECAUSE nothing checks the refused line's own bytes, so the
/// message says "newer or damaged", never "intact".
/// ```
pub(crate) fn settle(refusal: StoreError, hash: &str, next: Option<&[u8]>) -> StoreError {
let StoreError::NewerJournal {
seq,
event_type,
need,
..
} = refusal
else {
return refusal;
};
let prev = next
.and_then(|raw| serde_json::from_slice::<Value>(raw).ok())
.and_then(|v| v.get("prev").and_then(Value::as_str).map(str::to_string));
match prev {
Some(prev) if prev != hash => {
StoreError::JournalCorrupt(format!("hash chain broken at seq {}", seq + 1))
}
found => StoreError::NewerJournal {
seq,
event_type,
need,
verified: found.is_some(),
},
}
}
#[cfg(test)]
mod tests {
use super::*;
use crate::store::{Store, apply_line_to_sqlite, replay_line_into_ledger};
use ledger_core::{Currency, Ledger};
fn line(line_type: &str, payload: Value) -> JournalLine {
JournalLine {
seq: 7,
ts: "2026-10-16T00:00:00Z".into(),
line_type: line_type.into(),
payload,
prev: String::new(),
}
}
fn is_newer(r: &Result<(), StoreError>) -> bool {
matches!(r, Err(StoreError::NewerJournal { .. }))
}
/// The list and the code agree: every type the gate admits has a
/// replay arm and a projection arm (a known type with an empty payload
/// fails as corrupt, never as newer).
#[test]
fn every_known_event_type_has_replay_and_projection_arms() {
let conn = rusqlite::Connection::open_in_memory().unwrap();
Store::migrate(&conn).unwrap();
for &t in EVENT_TYPES {
let l = line(t, json!({}));
let mut ledger = Ledger::new(Currency::USD);
let replayed = replay_line_into_ledger(&mut ledger, &l, Currency::USD);
assert!(!is_newer(&replayed), "{t}: no replay arm");
assert!(
!is_newer(&apply_line_to_sqlite(&conn, &l)),
"{t}: no projection arm"
);
}
let l = line("settings.set", json!({}));
let mut ledger = Ledger::new(Currency::USD);
assert!(is_newer(&replay_line_into_ledger(
&mut ledger,
&l,
Currency::USD
)));
assert!(is_newer(&apply_line_to_sqlite(&conn, &l)));
}
/// A declaration never lowers the format in force, and never names a
/// format or feature this version can't read back.
#[test]
fn a_declared_format_is_never_lowered_or_unreadable() {
let at1 = JournalFormat::default();
assert_eq!(
at1.declaration(1, &[]).unwrap(),
json!({"format": 1, "features": []})
);
for (format, features) in [(0, &[][..]), (2, &[][..]), (1, &["x"][..])] {
assert!(
matches!(
at1.declaration(format, features),
Err(StoreError::FormatRefused(_))
),
"{format} {features:?}"
);
}
let at2 = JournalFormat {
format: 2,
features: Vec::new(),
};
assert!(matches!(
at2.declaration(1, &[]),
Err(StoreError::FormatRefused(m)) if m.contains("never lowered")
));
}
/// What the gate records, and what it refuses, line by line.
#[test]
fn the_gate_records_a_readable_format_and_refuses_a_newer_one() {
let mut f = JournalFormat::default();
f.admit(&line(FORMAT_SET, json!({"format": 1, "features": []})))
.unwrap();
assert_eq!(f, JournalFormat::default());
f.admit(&line("entry.posted", json!({}))).unwrap();
match f.admit(&line(FORMAT_SET, json!({"format": 2, "features": []}))) {
Err(StoreError::NewerJournal {
seq,
event_type,
need,
verified,
}) => {
// The gate alone never vouches for the line's bytes.
assert!(!verified);
assert_eq!((seq, event_type.as_str()), (7, FORMAT_SET));
assert_eq!(need, "journal format 2");
}
other => panic!("expected NewerJournal, got {other:?}"),
}
match f.admit(&line(
FORMAT_SET,
json!({"format": 1, "features": ["a", "b"]}),
)) {
Err(StoreError::NewerJournal { need, .. }) => {
assert_eq!(need, "journal features a, b")
}
other => panic!("expected NewerJournal, got {other:?}"),
}
// A refused line leaves the format in force alone.
assert_eq!(f, JournalFormat::default());
for bad in [
json!({"features": []}),
json!({"format": 0, "features": []}),
json!({"format": "1", "features": []}),
json!({"format": 1}),
json!({"format": 1, "features": [1]}),
json!({"format": 1, "features": "a"}),
] {
assert!(
matches!(
f.admit(&line(FORMAT_SET, bad.clone())),
Err(StoreError::JournalCorrupt(_))
),
"{bad}"
);
}
}
/// A format newer than this version reads is newer whatever shape the
/// rest of its declaration takes, and however large its number: a
/// newer format may define its features differently, so they are
/// judged only for a format this version reads (rules.md B-27).
#[test]
fn a_newer_format_is_newer_whatever_its_features_look_like() {
let mut f = JournalFormat::default();
for newer in [
json!({"format": 2, "features": {"actors": true}}),
json!({"format": 2, "features": [1, 2]}),
json!({"format": 2}),
json!({"format": 4_294_967_296_u64, "features": []}),
] {
assert!(
is_newer(&f.admit(&line(FORMAT_SET, newer.clone()))),
"{newer}"
);
}
assert_eq!(f, JournalFormat::default());
// The same shapes under a format this version reads are damage.
for bad in [
json!({"format": 1, "features": {"actors": true}}),
json!({"format": 1, "features": [1, 2]}),
] {
assert!(matches!(
f.admit(&line(FORMAT_SET, bad)),
Err(StoreError::JournalCorrupt(_))
));
}
}
}
crates/ledger-store/src/journal.rs
+136 -17
View File
@@ -10,11 +10,12 @@
//! and the chain verifies.
use crate::error::StoreError;
use crate::format::JournalFormat;
use serde::{Deserialize, Serialize};
use serde_json::Value;
use sha2::{Digest, Sha256};
use std::fs::{File, OpenOptions};
use std::io::{BufRead, BufReader, Write};
use std::io::Write;
use std::path::{Path, PathBuf};
/// `prev` value of the first journal line: 64 zeros.
@@ -28,9 +29,11 @@ pub struct JournalLine {
/// UTC timestamp, RFC 3339. Recorded, never trusted for ordering.
pub ts: String,
/// Event type, e.g. `account.added`, `entry.posted`, `asset.tagged`:
/// every type the store writes has a replay arm in
/// `replay_line_into_ledger` (an unknown type is refused as corrupt),
/// and journal text in the viewer (`ledger-gui` tests check both).
/// every type the store writes is in [`crate::format::EVENT_TYPES`],
/// has a replay arm in `replay_line_into_ledger`, and has journal text
/// in the viewer (`ledger-store` and `ledger-gui` tests check). A type
/// this version doesn't know is refused as `NewerJournal`, not as
/// corrupt: a newer wanderbooks wrote it.
#[serde(rename = "type")]
pub line_type: String,
/// The full domain facts. Everything a replay needs.
@@ -67,6 +70,12 @@ fn sort_value(v: &Value) -> Value {
}
}
/// The keys of a journal line, the same in every journal format
/// (rules.md B-27): a newer format adds its fields inside `payload`, never
/// beside it, and a line's hash is always the SHA-256 of its canonical
/// JSON ([`canonical_json`]), whatever the format.
pub const ENVELOPE_KEYS: [&str; 5] = ["payload", "prev", "seq", "ts", "type"];
impl JournalLine {
/// Parse one journal line, refusing what the journal never holds.
///
@@ -74,22 +83,70 @@ impl JournalLine {
/// GOAL: read a journal line so that its canonical form (which the
/// hash chain covers) is a fixed point.
/// SOURCE: R-2 (no floats in money paths); found by fuzz_journal_line
/// (2026-10-13): a float payload re-canonicalized differently.
/// (2026-10-13): a float payload re-canonicalized differently;
/// rules.md B-27 (the envelope is frozen).
/// INVARIANTS:
/// 1. NEVER accepts a floating-point number anywhere in the line:
/// amounts are decimal strings and ids are integers.
/// 2. For an accepted line, parse(canonical) re-canonicalizes to the
/// 2. NEVER drops a key: a line with a key beside the five of
/// ENVELOPE_KEYS is refused (the store's reader calls it newer).
/// 3. For an accepted line, parse(canonical) re-canonicalizes to the
/// same bytes.
/// ```
pub fn parse(raw: &[u8]) -> Result<JournalLine, String> {
let value: Value = serde_json::from_slice(raw).map_err(|e| e.to_string())?;
if has_float(&value) {
return Err("a number with a fraction or exponent; the journal holds none".into());
let read = read_line(raw)?;
if read.unknown_keys.is_empty() {
Ok(read.line)
} else {
Err(format!(
"line fields this version doesn't know: {}",
read.unknown_keys.join(", ")
))
}
serde_json::from_value(value).map_err(|e| e.to_string())
}
}
/// One journal line as the store's readers take it in.
pub(crate) struct ReadLine {
/// The line, without any key this version doesn't know.
pub line: JournalLine,
/// SHA-256 of the line's canonical JSON, every key included: what the
/// next line's `prev` must be.
pub hash: String,
/// Keys beside [`ENVELOPE_KEYS`], sorted: none in any line this
/// version writes.
pub unknown_keys: Vec<String>,
}
/// Read one journal line (JournalLine::parse INVARIANT 1), keeping the
/// keys this version doesn't know out of the line but in its hash, so a
/// reader can still check the chain through it.
pub(crate) fn read_line(raw: &[u8]) -> Result<ReadLine, String> {
let mut value: Value = serde_json::from_slice(raw).map_err(|e| e.to_string())?;
if has_float(&value) {
return Err("a number with a fraction or exponent; the journal holds none".into());
}
let hash = sha256_hex(canonical_json(&value).as_bytes());
let mut unknown_keys = Vec::new();
if let Value::Object(map) = &mut value {
unknown_keys = map
.keys()
.filter(|k| !ENVELOPE_KEYS.contains(&k.as_str()))
.cloned()
.collect();
unknown_keys.sort();
for k in &unknown_keys {
map.remove(k);
}
}
let line = serde_json::from_value(value).map_err(|e| e.to_string())?;
Ok(ReadLine {
line,
hash,
unknown_keys,
})
}
fn has_float(v: &Value) -> bool {
match v {
Value::Number(n) => n.is_f64(),
@@ -193,16 +250,72 @@ impl JournalWriter {
/// Read and fully verify a journal: every line parses, seqs are gapless from
/// 1, and the hash chain is intact. Any failure names the offending seq.
/// This is the journal's own integrity, whatever its lines' types; opening
/// the books also passes every line through the format gate
/// ([`read_journal_and_format`]).
pub fn read_and_verify_journal(path: &Path) -> Result<Vec<JournalLine>, StoreError> {
let file = File::open(path)?;
let reader = BufReader::new(file);
verify_lines(&std::fs::read(path)?, |_| Ok(()))
}
/// [`read_and_verify_journal`] with the format gate, also returning the
/// journal format in force.
///
/// ```text
/// GOAL: books read whole and verified, or refused with the reason.
/// SOURCE: docs/receipt-store.md "Line format"; rules.md B-27.
/// INVARIANTS:
/// 1. Line by line, in order: parse, seq, chain, then the format gate.
/// 2. Damage before the first line this version can't read is
/// JournalCorrupt, that line included: its seq and prev are checked
/// before the gate, and its own hash by the next line's prev
/// (format::settle), so a flipped bit in its type is corrupt, not
/// newer.
/// 3. At the first whole, chained line this version can't read (an
/// unknown type, a newer journal.format_set, or a key beside the
/// envelope's): SHOULD stop with NewerJournal (verified when the next
/// line's prev matches it; unverified, saying "newer or damaged",
/// when there is no next line or it has no prev this version can
/// read)
/// BECAUSE a newer format may frame later lines in ways this version
/// can't check; judging them is the newer version's job.
/// 4. The journal is read as bytes, split on newlines, as the viewer
/// reads it (crate::Snapshot): a line that isn't UTF-8 JSON is
/// JournalCorrupt, never an I/O error.
/// ```
pub fn read_journal_and_format(
path: &Path,
) -> Result<(Vec<JournalLine>, JournalFormat), StoreError> {
verify_journal_and_format(&std::fs::read(path)?)
}
/// [`read_journal_and_format`] over a journal's bytes (`raw`).
pub(crate) fn verify_journal_and_format(
raw: &[u8],
) -> Result<(Vec<JournalLine>, JournalFormat), StoreError> {
let mut format = JournalFormat::default();
let lines = verify_lines(raw, |line| format.admit(line))?;
Ok((lines, format))
}
/// Check each line of `raw` (bytes, split on newlines; a final newline
/// ends the last line): its parse, seq and chain, then `admit` on it, in
/// order; the first failure stops the read.
fn verify_lines(
raw: &[u8],
mut admit: impl FnMut(&JournalLine) -> Result<(), StoreError>,
) -> Result<Vec<JournalLine>, StoreError> {
let mut lines = Vec::new();
if raw.is_empty() {
return Ok(lines);
}
let body = raw.strip_suffix(b"\n").unwrap_or(raw);
let mut expected_prev = GENESIS_PREV.to_string();
for (i, raw) in reader.lines().enumerate() {
let raw = raw?;
let line: JournalLine = JournalLine::parse(raw.as_bytes()).map_err(|e| {
let mut raws = body.split(|&b| b == b'\n').enumerate().peekable();
while let Some((i, raw)) = raws.next() {
let read = read_line(raw).map_err(|e| {
StoreError::JournalCorrupt(format!("seq {}: unparseable line: {e}", i + 1))
})?;
let line = &read.line;
let want_seq = i as u64 + 1;
if line.seq != want_seq {
return Err(StoreError::JournalCorrupt(format!(
@@ -216,8 +329,14 @@ pub fn read_and_verify_journal(path: &Path) -> Result<Vec<JournalLine>, StoreErr
line.seq
)));
}
expected_prev = sha256_hex(line.to_canonical().as_bytes());
lines.push(line);
if let Err(refused) = crate::format::gate(&read, &mut admit) {
// A line's own hash is checked only by the next line's prev:
// read it before saying whether the books are newer or damaged.
let next = raws.peek().map(|(_, next)| *next);
return Err(crate::format::settle(refused, &read.hash, next));
}
expected_prev = read.hash;
lines.push(read.line);
}
Ok(lines)
}
crates/ledger-store/src/lib.rs
+2
View File
@@ -20,6 +20,7 @@
pub mod blob;
pub mod error;
pub mod format;
pub mod journal;
pub mod reconcilable;
pub mod schema;
@@ -29,6 +30,7 @@ pub mod viewer;
pub use blob::BlobStore;
pub use error::StoreError;
pub use format::{JournalFormat, SUPPORTED_JOURNAL_FORMAT};
pub use journal::{GENESIS_PREV, JournalLine, JournalWriter};
pub use snapshot::{DiskCheck, Snapshot};
pub use store::{
crates/ledger-store/src/snapshot.rs
+61 -19
View File
@@ -16,7 +16,8 @@
use crate::blob::BlobStore;
use crate::error::StoreError;
use crate::journal::{GENESIS_PREV, JournalLine, sha256_hex};
use crate::format::JournalFormat;
use crate::journal::{GENESIS_PREV, JournalLine};
use crate::store::{
Attachment, BLOBS_DIR, DB_FILE, JOURNAL_FILE, Reconciliation, Statement, Store, Subject,
apply_line_to_sqlite, compare_projections, projection_rows, query_attachment_links,
@@ -71,6 +72,8 @@ pub struct Snapshot {
offset: u64,
/// Hash of the last line read: what the next line must chain onto.
head: String,
/// The journal format in force (the format gate's record).
format: JournalFormat,
}
impl Snapshot {
@@ -91,6 +94,7 @@ impl Snapshot {
lines: Vec::new(),
offset: 0,
head: GENESIS_PREV.to_string(),
format: JournalFormat::default(),
};
snap.read_new()?;
Ok(snap)
@@ -114,7 +118,16 @@ impl Snapshot {
/// 1. IF nothing was appended: SHOULD return 0 and change nothing.
/// 2. IF the file shrank or no longer chains (a restore, a repaired
/// tail): SHOULD reload from scratch.
/// 3. IF a complete new line fails its hash or parse: SHOULD refuse.
/// 3. IF a complete new line fails its hash or parse (a blank line
/// included, as the writer reads it): SHOULD refuse.
/// 4. IF a complete new line is whole and chained but one this
/// version can't read (an unknown type, a newer format, a key
/// beside the envelope's): SHOULD refuse with NewerJournal
/// (verified or not as format::settle finds it: a next line whose
/// prev doesn't match makes it corrupt), applying
/// none of the new lines (as the writer refuses the books whole),
/// and refuse the same way on every refresh until wanderbooks is
/// updated; what was read before stays.
/// ```
pub fn refresh(&mut self) -> Result<usize, StoreError> {
let len = std::fs::metadata(self.data_dir.join(JOURNAL_FILE))?.len();
@@ -145,40 +158,63 @@ impl Snapshot {
return Ok(0);
};
let complete = &buf[..=last_newline];
let mut added = 0;
for raw in complete.split(|&b| b == b'\n').filter(|l| !l.is_empty()) {
let line: JournalLine = JournalLine::parse(raw).map_err(|e| {
StoreError::JournalCorrupt(format!(
"seq {}: unparseable line: {e}",
self.lines.len() + 1
))
// First check every new line, as the writer does when it opens the
// books: parse, seq, chain, then the format gate, stopping at the
// first failure. A batch holding a line this version can't read is
// refused whole, so the viewer and the writer give the same answer
// for the same journal.
let mut format = self.format.clone();
let mut head = self.head.clone();
let mut checked: Vec<(JournalLine, String, u64)> = Vec::new();
let mut raws = complete
.split_inclusive(|&b| b == b'\n')
.map(|raw| (raw.len() as u64, &raw[..raw.len() - 1]))
.peekable();
while let Some((len, raw)) = raws.next() {
let seq = self.lines.len() + checked.len() + 1;
// A blank complete line is no line at all: corrupt, as the
// writer finds it (it never writes one).
let read = crate::journal::read_line(raw).map_err(|e| {
StoreError::JournalCorrupt(format!("seq {seq}: unparseable line: {e}"))
})?;
let want_seq = self.lines.len() as u64 + 1;
if line.seq != want_seq {
let line = &read.line;
if line.seq != seq as u64 {
return Err(StoreError::JournalCorrupt(format!(
"seq gap: expected {want_seq}, found {}",
"seq gap: expected {seq}, found {}",
line.seq
)));
}
if line.prev != self.head {
if line.prev != head {
return Err(StoreError::JournalCorrupt(format!(
"hash chain broken at seq {}",
line.seq
)));
}
if let Err(refused) = crate::format::gate(&read, &mut |l| format.admit(l)) {
// Its own hash is checked by the next line's prev, when
// there is one (format::settle).
let next = raws.peek().map(|(_, raw)| *raw);
return Err(crate::format::settle(refused, &read.hash, next));
}
head = read.hash;
checked.push((read.line, head.clone(), len));
}
// Then apply them. The offset advances line by line, so if replay
// refuses one, the snapshot holds exactly the lines before it.
let added = checked.len();
for (line, hash, len) in checked {
if line.seq == 1 {
let code = line.payload["base_currency"].as_str().unwrap_or("USD");
self.base_currency = Currency::new(code)
.map_err(|_| StoreError::JournalCorrupt(format!("bad base_currency {code}")))?;
// The writer's rule: the first line is a sound store.created.
self.base_currency = crate::store::created_currency(&line)?;
self.ledger = Ledger::new(self.base_currency);
}
self.format.admit(&line)?;
replay_line_into_ledger(&mut self.ledger, &line, self.base_currency)?;
apply_line_to_sqlite(&self.conn, &line)?;
self.head = sha256_hex(line.to_canonical().as_bytes());
self.head = hash;
self.lines.push(line);
added += 1;
self.offset += len;
}
self.offset += complete.len() as u64;
Ok(added)
}
@@ -192,6 +228,12 @@ impl Snapshot {
self.base_currency
}
/// The journal format in force: format 1 unless the books declared
/// another with `journal.format_set`.
pub fn journal_format(&self) -> &JournalFormat {
&self.format
}
/// The data directory.
pub fn data_dir(&self) -> &Path {
&self.data_dir
crates/ledger-store/src/store.rs
+100 -27
View File
@@ -32,6 +32,7 @@
use crate::blob::BlobStore;
use crate::error::StoreError;
use crate::format::JournalFormat;
use crate::journal::{
GENESIS_PREV, JournalLine, JournalWriter, read_and_verify_journal, sha256_hex,
};
@@ -202,6 +203,8 @@ pub struct Store {
blobs: BlobStore,
ledger: Ledger,
base_currency: Currency,
/// The journal format in force (the format gate's record).
format: JournalFormat,
}
/// One filing of bytes into the blob store and journal.
@@ -254,6 +257,7 @@ impl Store {
blobs: BlobStore::open(&data_dir.join(BLOBS_DIR))?,
ledger: Ledger::new(base_currency),
base_currency,
format: JournalFormat::default(),
};
let line = store.journal.append(
"store.created",
@@ -279,10 +283,19 @@ impl Store {
/// SCENARIOS:
/// 1. IF there is no journal: SHOULD refuse (NotFound).
/// 2. IF another writer has the books: SHOULD refuse (Locked).
/// 3. IF the last line is cut off (no newline): SHOULD set it aside,
/// truncate, and journal `journal.tail_repaired`.
/// 3. IF the last line is cut off (no newline) and every complete
/// line passes (its chain, the format gate, the first line's
/// store.created, and its replay into the ledger, all in memory):
/// SHOULD set it aside, truncate, and journal
/// `journal.tail_repaired`; IF a complete line fails: SHOULD
/// refuse with the journal byte for byte as it was.
/// 4. IF a complete line fails its hash, parse, or `seq`: SHOULD refuse
/// - NEVER repair or skip it (that is corruption, for a human).
/// 4a. IF a whole, chained line is one this version can't read (an
/// unknown type, a newer journal format, a key beside the
/// envelope's): SHOULD refuse with NewerJournal before touching
/// the database or the journal, a cut-off last line included
/// BECAUSE a newer wanderbooks wrote it, and may be appending.
/// 5. IF the database is missing: SHOULD rebuild it from the journal.
/// 6. IF the database fails SQLite's integrity check, or records a
/// base currency other than the journal's: SHOULD keep it aside
@@ -297,14 +310,31 @@ impl Store {
}
// The writer lock first: nothing is read until no one else writes.
let lock = lock_books(data_dir)?;
// An append cut off mid-line was never acknowledged: set it aside.
let torn = crate::journal::repair_torn_tail(&journal_path)?;
// Single verified scan; the writer reuses it.
let lines = read_and_verify_journal(&journal_path)?;
let journal = JournalWriter::from_lines(&journal_path, &lines)?;
// Single verified scan of the complete lines, through the format
// gate, before anything is changed: books from a newer wanderbooks
// are refused whole, never read in part, and never altered (a
// newer version may be mid-append). The writer reuses the scan.
let raw = std::fs::read(&journal_path)?;
let complete = raw.iter().rposition(|&b| b == b'\n').map_or(0, |i| i + 1);
let (lines, format) = crate::journal::verify_journal_and_format(&raw[..complete])?;
// The base currency comes from the journal, the source of truth; a
// journal without a sound store.created line is corrupt.
let base_currency = journal_currency(&lines)?;
// Every complete line replayed, in memory, before anything on disk
// changes (SCENARIO 3): a line the books can't apply refuses the
// open with the journal and the database as they were.
let mut ledger = Ledger::new(base_currency);
for line in &lines {
replay_line_into_ledger(&mut ledger, line, base_currency)?;
}
// Only then is an append cut off mid-line (never acknowledged) set
// aside, and journaled below.
let torn = if complete < raw.len() {
crate::journal::repair_torn_tail(&journal_path)?
} else {
None
};
let journal = JournalWriter::from_lines(&journal_path, &lines)?;
let db_path = data_dir.join(DB_FILE);
let mut fresh_db = !db_path.exists();
let damage = if fresh_db {
@@ -350,8 +380,9 @@ impl Store {
journal,
conn,
blobs: BlobStore::open(&data_dir.join(BLOBS_DIR))?,
ledger: Ledger::new(base_currency),
ledger,
base_currency,
format,
};
store.catch_up(&lines)?;
if let Some(t) = torn {
@@ -448,8 +479,9 @@ impl Store {
Ok(())
}
/// Replay the journal: hydrate the in-memory ledger fully, and apply to
/// SQLite every line past `last_applied_seq`.
/// Catch the projection up: apply to SQLite every line past
/// `last_applied_seq` (the ledger is already the replay of every line,
/// done in memory by [`Store::open`] before anything changed).
fn catch_up(&mut self, lines: &[JournalLine]) -> Result<(), StoreError> {
let last_applied: u64 = self
.conn
@@ -461,11 +493,10 @@ impl Store {
.optional()?
.and_then(|s| s.parse().ok())
.unwrap_or(0);
for line in lines {
replay_line_into_ledger(&mut self.ledger, line, self.base_currency)?;
if line.seq > last_applied {
apply_line_to_sqlite(&self.conn, line)?;
}
// The ledger is already their replay (Store::open): only the
// projection is caught up here.
for line in lines.iter().filter(|l| l.seq > last_applied) {
apply_line_to_sqlite(&self.conn, line)?;
}
Ok(())
}
@@ -484,6 +515,41 @@ impl Store {
self.base_currency
}
/// The journal format in force: format 1 unless the books declared
/// another with `journal.format_set`.
pub fn journal_format(&self) -> &JournalFormat {
&self.format
}
/// Declare the journal format: journal a `journal.format_set` line.
/// A writer declares the format before the first event that needs it,
/// so an older reader stops at the declaration and says it is too old.
///
/// ```text
/// GOAL: books that tell every reader which format they are in.
/// SOURCE: rules.md B-27; plan/served.md 5.1 (format gate).
/// SCENARIOS:
/// 1. IF the format is newer than SUPPORTED_JOURNAL_FORMAT, or 0:
/// SHOULD refuse (FormatRefused), journaling nothing
/// BECAUSE this version couldn't open the books it left.
/// 2. IF the format is lower than the one in force: SHOULD refuse
/// BECAUSE events written under it may need the higher one.
/// 3. IF a feature isn't in KNOWN_FEATURES: SHOULD refuse.
/// 4. Otherwise: SHOULD journal it (features sorted, once each) and
/// make it the format in force.
/// ```
pub fn declare_journal_format(
&mut self,
format: u32,
features: &[&str],
) -> Result<(), StoreError> {
let payload = self.format.declaration(format, features)?;
let line = self.append_journal(crate::format::FORMAT_SET, payload)?;
self.format = crate::format::declared(&line)?;
apply_line_to_sqlite(&self.conn, &line)?;
Ok(())
}
/// Every torn journal tail set aside on open, oldest first, as
/// (when, bytes, file it was saved to).
pub fn tail_repairs(&self) -> Result<Vec<(String, u64, String)>, StoreError> {
@@ -4416,11 +4482,14 @@ pub(crate) fn replay_line_into_ledger(
)?;
}
"journal.tail_repaired" => {}
other => {
return Err(StoreError::JournalCorrupt(format!(
"unknown line type: {other}"
)));
"journal.format_set" => {
// No ledger state; a format this version can't read is
// refused here too, for callers that skip the gate.
crate::format::declared(line)?;
}
// A whole, chained line of a type this version doesn't know: a
// newer wanderbooks wrote it (the format gate).
_ => return Err(crate::format::unknown_type(line)),
}
Ok(())
}
@@ -4643,7 +4712,7 @@ pub(crate) fn apply_line_to_sqlite(
fn apply_line_rows(conn: &Connection, line: &JournalLine) -> Result<(), StoreError> {
let p = &line.payload;
match line.line_type.as_str() {
"store.created" | "journal.tail_repaired" => {}
"store.created" | "journal.tail_repaired" | "journal.format_set" => {}
"account.added" => {
conn.execute(
"INSERT INTO accounts(id, code, name, type, parent, placeholder, currency) VALUES (?,?,?,?,?,?,?)",
@@ -5197,11 +5266,7 @@ fn apply_line_rows(conn: &Connection, line: &JournalLine) -> Result<(), StoreErr
],
)?;
}
other => {
return Err(StoreError::JournalCorrupt(format!(
"unknown line type: {other}"
)));
}
_ => return Err(crate::format::unknown_type(line)),
}
conn.execute(
"INSERT INTO events(type, payload, created_at) VALUES (?,?,?)",
@@ -5607,12 +5672,20 @@ fn row_to_attachment(r: &rusqlite::Row) -> rusqlite::Result<Attachment> {
})
}
/// The base currency of the books, from the journal's first line, which
/// must be `store.created`.
/// The base currency of a journal: its first line's, which must be a
/// sound `store.created` ([`created_currency`]).
fn journal_currency(lines: &[JournalLine]) -> Result<Currency, StoreError> {
let first = lines
.first()
.ok_or_else(|| StoreError::JournalCorrupt("journal has no store.created line".into()))?;
created_currency(first)
}
/// The base currency the journal's first line sets: it must be a
/// `store.created` line naming a currency, or the journal is corrupt. The
/// one rule for the writer ([`Store::open`]) and the viewer
/// ([`crate::Snapshot`]), so the two never read different books.
pub(crate) fn created_currency(first: &JournalLine) -> Result<Currency, StoreError> {
if first.line_type != "store.created" {
return Err(StoreError::JournalCorrupt(
"journal's first line is not store.created".into(),