From d67e897f1bf8c3d723b815533da816ce1712086e Mon Sep 17 00:00:00 2001 From: Stephen Waits Date: Tue, 13 Oct 2026 19:37:20 -0600 Subject: [PATCH] store: a database with the wrong currency is rebuilt, not blamed on the journal Store::open refused books whose database recorded a base currency other than the journal's, with JournalCorrupt, though the journal was fine and the database is derived (a mutation-testing survivor: nothing tested the case). Now the currency comes from the journal, and a database that fails its integrity check or disagrees on currency goes through the same path: kept aside for forensics and rebuilt (open SCENARIO 6). Also tests: the viewer tells a database Behind the journal from one Ahead of what it has read; the table check counts missing and unexpected rows; a new journal creates its directory; a stored blob exists and an unknown one doesn't. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01GWVo7x3UHw5kwCUaWxiejX --- crates/ledger-store/src/store.rs | 108 ++++++++++++++------------- crates/ledger-store/tests/store.rs | 114 +++++++++++++++++++++++++++++ 2 files changed, 173 insertions(+), 49 deletions(-) diff --git a/crates/ledger-store/src/store.rs b/crates/ledger-store/src/store.rs index f8290ab..399ebea 100644 --- a/crates/ledger-store/src/store.rs +++ b/crates/ledger-store/src/store.rs @@ -284,8 +284,10 @@ impl Store { /// 4. IF a complete line fails its hash, parse, or `seq`: SHOULD refuse /// - NEVER repair or skip it (that is corruption, for a human). /// 5. IF the database is missing: SHOULD rebuild it from the journal. - /// 6. IF the database fails SQLite's integrity check: SHOULD keep it - /// aside for forensics and rebuild 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 + /// for forensics and rebuild from the journal + /// BECAUSE the database is derived; the journal is the record. /// 7. IF the database is behind the journal: SHOULD apply the rest. /// ``` pub fn open(data_dir: &Path) -> Result { @@ -300,11 +302,28 @@ impl Store { // Single verified scan; the writer reuses it. let lines = read_and_verify_journal(&journal_path)?; let journal = JournalWriter::from_lines(&journal_path, &lines)?; + // 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)?; let db_path = data_dir.join(DB_FILE); let mut fresh_db = !db_path.exists(); - if !fresh_db && !Self::db_file_healthy(&db_path) { - // The projection is corrupt. The journal is the source of truth, - // so quarantine the damaged file (for forensics) and rebuild. + let damage = if fresh_db { + None + } else if !Self::db_file_healthy(&db_path) { + Some("failed integrity_check".to_string()) + } else { + match Self::db_currency(&db_path) { + Some(c) if c == base_currency => None, + other => Some(format!( + "says base currency {}, the journal {}", + other.map_or("(none)".to_string(), |c| c.code().to_string()), + base_currency.code() + )), + } + }; + if let Some(why) = damage { + // The projection is wrong. The journal is the source of truth, + // so quarantine the database (for forensics) and rebuild. let stamp = chrono::Utc::now().format("%Y%m%dT%H%M%SZ").to_string(); let quarantine = db_path.with_extension(format!("corrupt.{stamp}")); std::fs::rename(&db_path, &quarantine).map_err(StoreError::Io)?; @@ -313,47 +332,17 @@ impl Store { let _ = std::fs::remove_file(db_path.with_extension("db-wal")); let _ = std::fs::remove_file(db_path.with_extension("db-shm")); eprintln!( - "wanderbooks: sqlite projection failed integrity_check; quarantined to {} and rebuilding from journal", + "wanderbooks: sqlite projection {why}; quarantined to {} and rebuilding from journal", quarantine.display() ); fresh_db = true; } let conn = Self::open_conn(&db_path)?; Self::migrate(&conn)?; - let base_currency = if fresh_db { - // Heal from the journal: currency comes from store.created. - let first = lines.first().ok_or_else(|| { - StoreError::JournalCorrupt("journal has no store.created line".into()) - })?; - if first.line_type != "store.created" { - return Err(StoreError::JournalCorrupt( - "journal's first line is not store.created".into(), - )); - } - let code = first.payload["base_currency"].as_str().ok_or_else(|| { - StoreError::JournalCorrupt("store.created without base_currency".into()) - })?; - let cur = Currency::new(code) - .map_err(|_| StoreError::JournalCorrupt(format!("bad currency {code}")))?; - Self::write_meta(&conn, "base_currency", cur.code())?; + if fresh_db { + Self::write_meta(&conn, "base_currency", base_currency.code())?; Self::write_meta(&conn, "last_applied_seq", "0")?; - cur - } else { - let cur = Self::read_currency(&conn)?; - // Cross-check the projection against the journal. - if let Some(first) = lines.first() - && first.line_type == "store.created" - { - let jc = first.payload["base_currency"].as_str().unwrap_or(""); - if jc != cur.code() { - return Err(StoreError::JournalCorrupt(format!( - "base currency mismatch: journal says {jc}, sqlite says {}", - cur.code() - ))); - } - } - cur - }; + } let mut store = Self { _lock: lock, data_dir: data_dir.to_path_buf(), @@ -385,6 +374,20 @@ impl Store { Ok(store) } + /// The base currency an existing database records, if it can say. + fn db_currency(db_path: &Path) -> Option { + let conn = Connection::open_with_flags(db_path, rusqlite::OpenFlags::SQLITE_OPEN_READ_ONLY) + .ok()?; + let code: String = conn + .query_row( + "SELECT value FROM meta WHERE key='base_currency'", + [], + |r| r.get(0), + ) + .ok()?; + Currency::new(&code).ok() + } + fn open_conn(db_path: &Path) -> Result { let conn = Connection::open(db_path)?; conn.execute_batch( @@ -445,16 +448,6 @@ impl Store { Ok(()) } - fn read_currency(conn: &Connection) -> Result { - let code: String = conn.query_row( - "SELECT value FROM meta WHERE key='base_currency'", - [], - |r| r.get(0), - )?; - Currency::new(&code) - .map_err(|_| StoreError::JournalCorrupt(format!("bad base_currency in meta: {code}"))) - } - /// Replay the journal: hydrate the in-memory ledger fully, and apply to /// SQLite every line past `last_applied_seq`. fn catch_up(&mut self, lines: &[JournalLine]) -> Result<(), StoreError> { @@ -4270,3 +4263,20 @@ fn row_to_attachment(r: &rusqlite::Row) -> rusqlite::Result { withdrawn_reason: r.get(10)?, }) } + +/// The base currency of the books, from the journal's first line, which +/// must be `store.created`. +fn journal_currency(lines: &[JournalLine]) -> Result { + let first = lines + .first() + .ok_or_else(|| StoreError::JournalCorrupt("journal has no store.created line".into()))?; + if first.line_type != "store.created" { + return Err(StoreError::JournalCorrupt( + "journal's first line is not store.created".into(), + )); + } + let code = first.payload["base_currency"] + .as_str() + .ok_or_else(|| StoreError::JournalCorrupt("store.created without base_currency".into()))?; + Currency::new(code).map_err(|_| StoreError::JournalCorrupt(format!("bad currency {code}"))) +} diff --git a/crates/ledger-store/tests/store.rs b/crates/ledger-store/tests/store.rs index 5511c65..f0239f3 100644 --- a/crates/ledger-store/tests/store.rs +++ b/crates/ledger-store/tests/store.rs @@ -1932,3 +1932,117 @@ fn a_retained_earnings_close_with_a_sweep_replays_it() { store.verify().unwrap(); assert_eq!(state(&store), (Decimal::ZERO, Decimal::from(300))); } + +/// The viewer tells a database that hasn't applied the newest lines +/// (Behind) from one that has applied lines the viewer hasn't read yet +/// (Ahead), and neither is a false "differs". +#[test] +fn the_viewer_tells_behind_from_ahead() { + use ledger_store::{DiskCheck, Snapshot}; + let dir = test_dir("behind-ahead"); + let store = build_books(&dir); + drop(store); + // Ahead: the viewer loads, then the writer appends and applies. + let snap = Snapshot::load(&dir).unwrap(); + let mut store = Store::open(&dir).unwrap(); + store.add_customer("Acme").unwrap(); + assert!( + matches!(snap.check_disk(), DiskCheck::Ahead { .. }), + "{:?}", + snap.check_disk() + ); + // Behind: a database restored from before that write. + drop(store); + let mut store = Store::open(&dir).unwrap(); + let before_write = std::fs::read(dir.join(ledger_store::DB_FILE)).unwrap(); + store.add_vendor("Supplier").unwrap(); + drop(store); + std::fs::write(dir.join(ledger_store::DB_FILE), &before_write).unwrap(); + let _ = std::fs::remove_file(dir.join("wanderbooks.db-wal")); + let _ = std::fs::remove_file(dir.join("wanderbooks.db-shm")); + let snap = Snapshot::load(&dir).unwrap(); + match snap.check_disk() { + DiskCheck::Behind { applied, journal } => assert!(applied < journal), + other => panic!("expected Behind, got {other:?}"), + } +} + +/// The table check says how many rows are missing and how many are +/// unexpected, so a person can tell a lost row from a stray one. +#[test] +fn the_table_check_counts_missing_and_unexpected_rows() { + let dir = test_dir("table-counts"); + let store = build_books(&dir); + drop(store); + let conn = rusqlite::Connection::open(dir.join(ledger_store::DB_FILE)).unwrap(); + // The events table mirrors the journal and isn't read back into the + // ledger, so only the table check sees this. + conn.execute( + "DELETE FROM events WHERE seq = (SELECT MAX(seq) FROM events)", + [], + ) + .unwrap(); + drop(conn); + let store = Store::open(&dir).unwrap(); + let err = store.verify().unwrap_err().to_string(); + assert!( + err.contains("table events: 1 rows missing, 0 unexpected"), + "{err}" + ); +} + +/// A journal writer for a path whose directory doesn't exist yet creates +/// it; the blob store reports a stored file as existing, and an unknown +/// one as not. +#[test] +fn a_new_journal_makes_its_directory_and_blobs_exist_once_stored() { + let dir = test_dir("new-dirs"); + let path = dir + .join("nested") + .join("deeper") + .join(ledger_store::JOURNAL_FILE); + let mut w = JournalWriter::open(&path).unwrap(); + w.append("test.line", serde_json::json!({"n": 1})).unwrap(); + assert!(path.exists()); + let blobs = ledger_store::BlobStore::open(&dir.join("blobs")).unwrap(); + let (sha, _) = blobs.put(b"receipt").unwrap(); + assert!(blobs.exists(&sha)); + assert!(!blobs.exists(&"0".repeat(64))); +} + +/// Store::open SCENARIO 6: a database that records a base currency other +/// than the journal's is the projection's fault, not the journal's: it is +/// kept aside and rebuilt (before 2026-10-13 the books refused to open, +/// blaming the journal). +#[test] +fn a_database_with_the_wrong_currency_is_rebuilt() { + let dir = test_dir("db-currency"); + let store = build_books(&dir); + let before = ledger_core::reports::snapshot(store.ledger()).unwrap(); + drop(store); + let conn = rusqlite::Connection::open(dir.join(ledger_store::DB_FILE)).unwrap(); + conn.execute( + "UPDATE meta SET value = 'EUR' WHERE key = 'base_currency'", + [], + ) + .unwrap(); + drop(conn); + let store = Store::open(&dir).unwrap(); + assert_eq!(store.base_currency(), Currency::USD); + assert_eq!( + ledger_core::reports::snapshot(store.ledger()).unwrap(), + before + ); + store.verify().unwrap(); + let kept = std::fs::read_dir(&dir) + .unwrap() + .filter(|e| { + e.as_ref() + .unwrap() + .file_name() + .to_string_lossy() + .contains(".corrupt.") + }) + .count(); + assert_eq!(kept, 1, "the wrong database is kept for forensics"); +}