store: a database with the wrong currency is rebuilt, not blamed on the journal
check / check (push) Canceled after 0s
check / check (push) Canceled after 0s
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 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01GWVo7x3UHw5kwCUaWxiejX
This commit is contained in:
1 parent
0cf8fb842f
commit
d67e897f1b
2 files changed
+173
-49
No files matched your search
@@ -284,8 +284,10 @@ impl Store {
|
|||||||
/// 4. IF a complete line fails its hash, parse, or `seq`: SHOULD refuse
|
/// 4. IF a complete line fails its hash, parse, or `seq`: SHOULD refuse
|
||||||
/// - NEVER repair or skip it (that is corruption, for a human).
|
/// - NEVER repair or skip it (that is corruption, for a human).
|
||||||
/// 5. IF the database is missing: SHOULD rebuild it from the journal.
|
/// 5. IF the database is missing: SHOULD rebuild it from the journal.
|
||||||
/// 6. IF the database fails SQLite's integrity check: SHOULD keep it
|
/// 6. IF the database fails SQLite's integrity check, or records a
|
||||||
/// aside for forensics and rebuild from the journal.
|
/// 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.
|
/// 7. IF the database is behind the journal: SHOULD apply the rest.
|
||||||
/// ```
|
/// ```
|
||||||
pub fn open(data_dir: &Path) -> Result<Self, StoreError> {
|
pub fn open(data_dir: &Path) -> Result<Self, StoreError> {
|
||||||
@@ -300,11 +302,28 @@ impl Store {
|
|||||||
// Single verified scan; the writer reuses it.
|
// Single verified scan; the writer reuses it.
|
||||||
let lines = read_and_verify_journal(&journal_path)?;
|
let lines = read_and_verify_journal(&journal_path)?;
|
||||||
let journal = JournalWriter::from_lines(&journal_path, &lines)?;
|
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 db_path = data_dir.join(DB_FILE);
|
||||||
let mut fresh_db = !db_path.exists();
|
let mut fresh_db = !db_path.exists();
|
||||||
if !fresh_db && !Self::db_file_healthy(&db_path) {
|
let damage = if fresh_db {
|
||||||
// The projection is corrupt. The journal is the source of truth,
|
None
|
||||||
// so quarantine the damaged file (for forensics) and rebuild.
|
} 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 stamp = chrono::Utc::now().format("%Y%m%dT%H%M%SZ").to_string();
|
||||||
let quarantine = db_path.with_extension(format!("corrupt.{stamp}"));
|
let quarantine = db_path.with_extension(format!("corrupt.{stamp}"));
|
||||||
std::fs::rename(&db_path, &quarantine).map_err(StoreError::Io)?;
|
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-wal"));
|
||||||
let _ = std::fs::remove_file(db_path.with_extension("db-shm"));
|
let _ = std::fs::remove_file(db_path.with_extension("db-shm"));
|
||||||
eprintln!(
|
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()
|
quarantine.display()
|
||||||
);
|
);
|
||||||
fresh_db = true;
|
fresh_db = true;
|
||||||
}
|
}
|
||||||
let conn = Self::open_conn(&db_path)?;
|
let conn = Self::open_conn(&db_path)?;
|
||||||
Self::migrate(&conn)?;
|
Self::migrate(&conn)?;
|
||||||
let base_currency = if fresh_db {
|
if fresh_db {
|
||||||
// Heal from the journal: currency comes from store.created.
|
Self::write_meta(&conn, "base_currency", base_currency.code())?;
|
||||||
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())?;
|
|
||||||
Self::write_meta(&conn, "last_applied_seq", "0")?;
|
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 {
|
let mut store = Self {
|
||||||
_lock: lock,
|
_lock: lock,
|
||||||
data_dir: data_dir.to_path_buf(),
|
data_dir: data_dir.to_path_buf(),
|
||||||
@@ -385,6 +374,20 @@ impl Store {
|
|||||||
Ok(store)
|
Ok(store)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/// The base currency an existing database records, if it can say.
|
||||||
|
fn db_currency(db_path: &Path) -> Option<Currency> {
|
||||||
|
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<Connection, StoreError> {
|
fn open_conn(db_path: &Path) -> Result<Connection, StoreError> {
|
||||||
let conn = Connection::open(db_path)?;
|
let conn = Connection::open(db_path)?;
|
||||||
conn.execute_batch(
|
conn.execute_batch(
|
||||||
@@ -445,16 +448,6 @@ impl Store {
|
|||||||
Ok(())
|
Ok(())
|
||||||
}
|
}
|
||||||
|
|
||||||
fn read_currency(conn: &Connection) -> Result<Currency, StoreError> {
|
|
||||||
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
|
/// Replay the journal: hydrate the in-memory ledger fully, and apply to
|
||||||
/// SQLite every line past `last_applied_seq`.
|
/// SQLite every line past `last_applied_seq`.
|
||||||
fn catch_up(&mut self, lines: &[JournalLine]) -> Result<(), StoreError> {
|
fn catch_up(&mut self, lines: &[JournalLine]) -> Result<(), StoreError> {
|
||||||
@@ -4270,3 +4263,20 @@ fn row_to_attachment(r: &rusqlite::Row) -> rusqlite::Result<Attachment> {
|
|||||||
withdrawn_reason: r.get(10)?,
|
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<Currency, StoreError> {
|
||||||
|
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}")))
|
||||||
|
}
|
||||||
@@ -1932,3 +1932,117 @@ fn a_retained_earnings_close_with_a_sweep_replays_it() {
|
|||||||
store.verify().unwrap();
|
store.verify().unwrap();
|
||||||
assert_eq!(state(&store), (Decimal::ZERO, Decimal::from(300)));
|
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");
|
||||||
|
}
|
||||||
Reference in new issue
Block a user