feat: end-to-end integrity checks with self-healing; website; installer and Windows fixes
Store::verify compares the whole SQLite projection, every table row for row, against a fresh projection of the journal; verify_and_heal rebuilds a drifted database (keeping the old file) at every server start and via the new books.verify tool, which also re-hashes every blob. The GUI checks the server's database read-only and every stored file, and shows the result. rebuild() also drops stale WAL/SHM files. site/: the wanderbooks website (just site, just publish). install.sh: OS detection (Windows points to WSL2), C compiler and git checks, WANDERBOOKS_REPO. Windows: no directory fsync off Unix; .exe-aware lookup. Repo justfile: check, build, install, demo, site. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01MtYEaR6M6Ybdhm6X98BWNu
This commit is contained in:
25 files changed
+808
-41
No files matched your search
@@ -169,6 +169,8 @@ fn print_usage() {
|
||||
/// A binary next to this one (cargo installs them side by side), else on
|
||||
/// the PATH.
|
||||
fn find_binary(name: &str) -> Option<PathBuf> {
|
||||
let name = format!("{name}{}", env::consts::EXE_SUFFIX);
|
||||
let name = name.as_str();
|
||||
if let Some(dir) = env::current_exe()
|
||||
.ok()
|
||||
.and_then(|e| e.parent().map(Path::to_path_buf))
|
||||
|
||||
@@ -361,16 +361,33 @@ fn live_status(ui: &mut Ui, b: &Books, since_update: f64, live: bool) {
|
||||
.color(p.muted),
|
||||
);
|
||||
});
|
||||
ui.label(
|
||||
RichText::new(format!("{} Journal chain verified", icon::SEAL_CHECK))
|
||||
.font(theme::regular(11.5))
|
||||
.color(p.muted),
|
||||
)
|
||||
.on_hover_text(format!(
|
||||
"Every journal event's hash chains to the one before it.\nHead: {}\nData: {}",
|
||||
b.head,
|
||||
b.data_dir.display()
|
||||
));
|
||||
let check = |ui: &mut Ui, ok: bool, text: String, tip: String| {
|
||||
let (glyph, color) = if ok { (icon::SEAL_CHECK, p.muted) } else { (icon::WARNING_CIRCLE, p.negative) };
|
||||
ui.label(RichText::new(format!("{glyph} {text}")).font(theme::regular(11.5)).color(color))
|
||||
.on_hover_text(tip);
|
||||
};
|
||||
check(
|
||||
ui,
|
||||
true,
|
||||
"Journal chain verified".into(),
|
||||
format!("Every journal event's hash chains to the one before it.\nHead: {}\nData: {}", b.head, b.data_dir.display()),
|
||||
);
|
||||
use ledger_store::DiskCheck;
|
||||
let (ok, text, tip) = match &b.integrity.disk {
|
||||
DiskCheck::Matches => (true, "Database matches journal".to_string(), "The server's SQLite database matches a fresh projection of the journal, every table, row for row.".to_string()),
|
||||
DiskCheck::Behind { applied, journal } => (true, "Database catching up".to_string(), format!("The database has applied {applied} of {journal} journal events; the server applies the rest right after writing them.")),
|
||||
DiskCheck::Missing => (true, "Database will rebuild".to_string(), "No database file yet; the server builds it from the journal when it starts.".to_string()),
|
||||
DiskCheck::Differs(why) => (false, "Database differs".to_string(), format!("{why}. The server rebuilds it from the journal (books.verify, or on its next start).")),
|
||||
DiskCheck::Unreadable(why) => (false, "Database unreadable".to_string(), why.clone()),
|
||||
};
|
||||
check(ui, ok, text, tip);
|
||||
let bad = b.integrity.file_failures.len();
|
||||
check(
|
||||
ui,
|
||||
bad == 0,
|
||||
if bad == 0 { format!("{} files verified", b.integrity.files_ok) } else { format!("{bad} files damaged") },
|
||||
"Every stored receipt and document re-hashed against its recorded SHA-256.".into(),
|
||||
);
|
||||
});
|
||||
}
|
||||
|
||||
|
||||
@@ -260,6 +260,43 @@ pub fn find(b: &Books) -> Vec<Attention> {
|
||||
}
|
||||
}
|
||||
|
||||
use ledger_store::DiskCheck;
|
||||
let verify_prompt = "In wanderbooks, run books.verify and tell me what it found.".to_string();
|
||||
match &b.integrity.disk {
|
||||
DiskCheck::Differs(why) => out.push(Attention {
|
||||
severity: Severity::Problem,
|
||||
title: "The database doesn't match the journal".into(),
|
||||
detail: format!("{why}. The journal is the record; the server rebuilds the database from it (books.verify, or on its next start)."),
|
||||
target: Some(Route::Journal),
|
||||
prompt: verify_prompt.clone(),
|
||||
}),
|
||||
DiskCheck::Unreadable(why) => out.push(Attention {
|
||||
severity: Severity::Problem,
|
||||
title: "Couldn't check the database".into(),
|
||||
detail: format!("{why}."),
|
||||
target: Some(Route::Journal),
|
||||
prompt: verify_prompt.clone(),
|
||||
}),
|
||||
_ => {}
|
||||
}
|
||||
if !b.integrity.file_failures.is_empty() {
|
||||
let ids: Vec<String> = b
|
||||
.integrity
|
||||
.file_failures
|
||||
.iter()
|
||||
.map(|(id, _)| id.to_string())
|
||||
.collect();
|
||||
out.push(Attention {
|
||||
severity: Severity::Problem,
|
||||
title: format!("{} damaged or missing", fmt::count(ids.len(), "stored file is", "stored files are")),
|
||||
detail: format!(
|
||||
"Attachment {}: the bytes on disk don't match their recorded hash. Restore them from a backup (the books dir's git history).",
|
||||
ids.join(", ")
|
||||
),
|
||||
target: Some(Route::Documents),
|
||||
prompt: verify_prompt,
|
||||
});
|
||||
}
|
||||
out.sort_by_key(|a| a.severity);
|
||||
out
|
||||
}
|
||||
@@ -85,6 +85,46 @@ pub struct Books {
|
||||
pub receipt_threshold: Decimal,
|
||||
pub attention: Vec<Attention>,
|
||||
pub kpis: Kpis,
|
||||
pub integrity: Integrity,
|
||||
}
|
||||
|
||||
/// Whether what's on disk is what the journal says.
|
||||
#[derive(Clone, Debug)]
|
||||
pub struct Integrity {
|
||||
/// The writer's SQLite database against the journal.
|
||||
pub disk: ledger_store::DiskCheck,
|
||||
/// Stored files whose bytes match their hash.
|
||||
pub files_ok: u64,
|
||||
/// Files that are missing or damaged: (attachment id, why).
|
||||
pub file_failures: Vec<(u64, String)>,
|
||||
}
|
||||
|
||||
impl Integrity {
|
||||
/// Check the database and every file not already verified (by hash, so
|
||||
/// a file is read once however often the books change).
|
||||
pub fn check(snap: &Snapshot, verified: &mut HashSet<String>) -> Integrity {
|
||||
let store = ledger_store::BlobStore::open_read_only(&snap.data_dir().join("blobs"));
|
||||
let mut ok = 0;
|
||||
let mut failures = Vec::new();
|
||||
for a in snap.attachments(None).unwrap_or_default() {
|
||||
if verified.contains(&a.blob_sha) {
|
||||
ok += 1;
|
||||
continue;
|
||||
}
|
||||
match store.get(&a.blob_sha) {
|
||||
Ok(_) => {
|
||||
verified.insert(a.blob_sha.clone());
|
||||
ok += 1;
|
||||
}
|
||||
Err(e) => failures.push((a.id, e.to_string())),
|
||||
}
|
||||
}
|
||||
Integrity {
|
||||
disk: snap.check_disk(),
|
||||
files_ok: ok,
|
||||
file_failures: failures,
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// The headline numbers.
|
||||
@@ -113,6 +153,7 @@ impl Books {
|
||||
settings: BooksSettings,
|
||||
today: NaiveDate,
|
||||
prev_seq: Option<u64>,
|
||||
integrity: Integrity,
|
||||
) -> Result<Books, String> {
|
||||
let ledger = snap.ledger().clone();
|
||||
let attachments = snap.attachments(None).map_err(|e| e.to_string())?;
|
||||
@@ -268,6 +309,7 @@ impl Books {
|
||||
receipt_threshold,
|
||||
attention: Vec::new(),
|
||||
kpis: Kpis::default(),
|
||||
integrity,
|
||||
};
|
||||
books.kpis = books.compute_kpis();
|
||||
books.attention = crate::attention::find(&books);
|
||||
|
||||
@@ -5,11 +5,12 @@
|
||||
//! two. Filesystem events are the trigger; a slow poll backs them up in case
|
||||
//! an event is missed (network drives, odd editors).
|
||||
|
||||
use crate::books::Books;
|
||||
use crate::books::{Books, Integrity};
|
||||
use chrono::NaiveDate;
|
||||
use ledger_config::BooksSettings;
|
||||
use ledger_store::Snapshot;
|
||||
use notify::{RecursiveMode, Watcher};
|
||||
use std::collections::HashSet;
|
||||
use std::path::PathBuf;
|
||||
use std::sync::Arc;
|
||||
use std::sync::mpsc::{Receiver, RecvTimeoutError, Sender, channel};
|
||||
@@ -63,7 +64,8 @@ impl Loader {
|
||||
/// Load once, synchronously (tests and one-shot tools).
|
||||
pub fn load_once(source: &Source) -> Result<Books, String> {
|
||||
let snap = Snapshot::load(&source.data_dir).map_err(|e| e.to_string())?;
|
||||
Books::build(&snap, settings(source)?, today(source), None)
|
||||
let integrity = Integrity::check(&snap, &mut HashSet::new());
|
||||
Books::build(&snap, settings(source)?, today(source), None, integrity)
|
||||
}
|
||||
|
||||
fn today(source: &Source) -> NaiveDate {
|
||||
@@ -102,13 +104,22 @@ fn run(source: Source, ctx: egui::Context, tx: Sender<Update>) {
|
||||
}
|
||||
}
|
||||
};
|
||||
let build = |snap: &Snapshot, prev: Option<u64>| -> Update {
|
||||
match settings(&source).and_then(|s| Books::build(snap, s, today(&source), prev)) {
|
||||
let mut verified: HashSet<String> = HashSet::new();
|
||||
// Build the picture; also return the database check it was built with.
|
||||
let mut build = |snap: &Snapshot, prev: Option<u64>| -> (Update, ledger_store::DiskCheck) {
|
||||
let integrity = Integrity::check(snap, &mut verified);
|
||||
let disk = integrity.disk.clone();
|
||||
let update = match settings(&source)
|
||||
.and_then(|s| Books::build(snap, s, today(&source), prev, integrity))
|
||||
{
|
||||
Ok(b) => Update::Books(Arc::new(b)),
|
||||
Err(e) => Update::Error(e),
|
||||
}
|
||||
};
|
||||
(update, disk)
|
||||
};
|
||||
send(build(&snap, None));
|
||||
let (first, disk) = build(&snap, None);
|
||||
let mut last_disk = Some(disk);
|
||||
send(first);
|
||||
|
||||
let (ev_tx, ev_rx) = channel();
|
||||
let mut watcher = notify::recommended_watcher(move |res: notify::Result<notify::Event>| {
|
||||
@@ -149,10 +160,19 @@ fn run(source: Source, ctx: egui::Context, tx: Sender<Update>) {
|
||||
};
|
||||
let toml_now = mtime(&source.books_toml);
|
||||
let new_day = today(&source);
|
||||
if grew || toml_now != toml_mtime || new_day != day {
|
||||
// The writer journals first and updates its database right after:
|
||||
// a "behind" database usually catches up a moment later, so look
|
||||
// again rather than leave a stale warning up.
|
||||
let disk_changed = !grew
|
||||
&& last_disk
|
||||
.as_ref()
|
||||
.is_some_and(|d| *d != ledger_store::DiskCheck::Matches && snap.check_disk() != *d);
|
||||
if grew || disk_changed || toml_now != toml_mtime || new_day != day {
|
||||
toml_mtime = toml_now;
|
||||
day = new_day;
|
||||
send(build(&snap, grew.then_some(before)));
|
||||
let (update, disk) = build(&snap, grew.then_some(before));
|
||||
last_disk = Some(disk);
|
||||
send(update);
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -115,7 +115,7 @@ fn main() -> Result<(), Box<dyn std::error::Error>> {
|
||||
};
|
||||
let company = settings.company;
|
||||
|
||||
let store = if data_dir.join(ledger_store::JOURNAL_FILE).exists() {
|
||||
let mut store = if data_dir.join(ledger_store::JOURNAL_FILE).exists() {
|
||||
Store::open(&data_dir)?
|
||||
} else {
|
||||
let code = config.base_currency.as_deref().unwrap_or("USD");
|
||||
@@ -123,11 +123,15 @@ fn main() -> Result<(), Box<dyn std::error::Error>> {
|
||||
eprintln!("ledger-mcp: creating new store (base currency {code})");
|
||||
Store::create(&data_dir, currency)?
|
||||
};
|
||||
// The journal is the source of truth: verify the chain and the
|
||||
// projection before serving a single tool call.
|
||||
store
|
||||
.verify()
|
||||
// The journal is the source of truth: verify the chain, the replay, and
|
||||
// the whole projection before serving a single tool call, rebuilding
|
||||
// the projection if it drifted.
|
||||
let report = store
|
||||
.verify_and_heal()
|
||||
.map_err(|e| format!("store verify failed: {e}"))?;
|
||||
if let Some(why) = &report.healed {
|
||||
eprintln!("ledger-mcp: rebuilt the database from the journal ({why})");
|
||||
}
|
||||
|
||||
let rt = tokio::runtime::Builder::new_multi_thread()
|
||||
.enable_all()
|
||||
|
||||
@@ -1375,6 +1375,43 @@ impl BooksServer {
|
||||
Ok(Json(SearchEntriesOut { entries }))
|
||||
}
|
||||
|
||||
#[tool(
|
||||
name = "books.verify",
|
||||
description = "Check the books end to end: the journal's hash chain; a fresh replay against the live ledger; every table of the SQLite database, row for row, against a fresh projection of the journal (rebuilding the database from the journal if it drifted, keeping the old file); and, by default, every stored receipt and document against its SHA-256. Run it whenever a number looks wrong or a human asks whether the books are intact."
|
||||
)]
|
||||
async fn verify(&self, params: Parameters<VerifyParams>) -> Result<Json<VerifyOut>, String> {
|
||||
let mut store = self.lock()?;
|
||||
let report = store.verify_and_heal().map_err(|e| e.to_string())?;
|
||||
let blobs = if params.0.blobs {
|
||||
Some(store.verify_blobs().map_err(|e| e.to_string())?)
|
||||
} else {
|
||||
None
|
||||
};
|
||||
let mut checked = vec![
|
||||
format!("journal hash chain ({} events)", report.events),
|
||||
"fresh replay matches the live ledger (every statement, assets, classes, payroll)"
|
||||
.to_string(),
|
||||
"SQLite database matches a fresh projection of the journal, every table, row for row"
|
||||
.to_string(),
|
||||
];
|
||||
if let Some(n) = blobs {
|
||||
checked.push(format!("{n} stored files match their SHA-256"));
|
||||
}
|
||||
Ok(Json(VerifyOut {
|
||||
status: if report.healed.is_some() {
|
||||
"healed"
|
||||
} else {
|
||||
"ok"
|
||||
}
|
||||
.to_string(),
|
||||
events: report.events,
|
||||
head: report.head,
|
||||
healed: report.healed,
|
||||
blobs_verified: blobs,
|
||||
checked,
|
||||
}))
|
||||
}
|
||||
|
||||
#[tool(
|
||||
name = "books.digest",
|
||||
description = "What changed since a date: entries posted, per-account movement, and receipt coverage. Built for human review.",
|
||||
|
||||
@@ -375,6 +375,35 @@ pub struct Vendor1099Out {
|
||||
pub notes: Vec<String>,
|
||||
}
|
||||
|
||||
/// Parameters for `books.verify`.
|
||||
#[derive(Debug, Deserialize, JsonSchema)]
|
||||
pub struct VerifyParams {
|
||||
/// Also re-hash every stored receipt and document (default true).
|
||||
#[serde(default = "default_true")]
|
||||
pub blobs: bool,
|
||||
}
|
||||
|
||||
fn default_true() -> bool {
|
||||
true
|
||||
}
|
||||
|
||||
/// Result of `books.verify`.
|
||||
#[derive(Debug, Serialize, JsonSchema)]
|
||||
pub struct VerifyOut {
|
||||
/// "ok", or "healed" when the database was rebuilt from the journal.
|
||||
pub status: String,
|
||||
/// Journal events whose hash chain verified.
|
||||
pub events: u64,
|
||||
/// SHA-256 of the last journal line: the verified head.
|
||||
pub head: String,
|
||||
/// Why the database was rebuilt, if it was.
|
||||
pub healed: Option<String>,
|
||||
/// Stored files re-hashed, when checked.
|
||||
pub blobs_verified: Option<u64>,
|
||||
/// What was checked, in words.
|
||||
pub checked: Vec<String>,
|
||||
}
|
||||
|
||||
/// Parameters for `books.chart_template`.
|
||||
#[derive(Debug, Deserialize, JsonSchema)]
|
||||
pub struct ChartTemplateParams {
|
||||
|
||||
@@ -139,7 +139,7 @@ fn mcp_end_to_end_over_stdio() {
|
||||
// Tool listing: 23 narrow tools, chart is read-only.
|
||||
let list = mcp.request("tools/list", json!({}));
|
||||
let tools = list["result"]["tools"].as_array().unwrap();
|
||||
assert_eq!(tools.len(), 64, "expected 64 tools, got {}", tools.len());
|
||||
assert_eq!(tools.len(), 65, "expected 65 tools, got {}", tools.len());
|
||||
let chart = tools.iter().find(|t| t["name"] == "books.chart").unwrap();
|
||||
assert_eq!(chart["annotations"]["readOnlyHint"], true);
|
||||
assert!(
|
||||
@@ -1548,4 +1548,9 @@ fn mcp_vendor_1099_over_stdio() {
|
||||
assert_eq!(r["vendors"][0]["paid"], "0.00");
|
||||
let vendors = mcp.call("books.list_vendors", json!({}));
|
||||
assert_eq!(vendors["vendors"][0]["tin_on_file"], true);
|
||||
// The books check out end to end, receipt included.
|
||||
let v = mcp.call("books.verify", json!({}));
|
||||
assert_eq!(v["status"], "ok");
|
||||
assert_eq!(v["blobs_verified"], 1);
|
||||
assert!(v["events"].as_u64().unwrap() > 10);
|
||||
}
|
||||
@@ -60,9 +60,14 @@ impl BlobStore {
|
||||
f.sync_all()?;
|
||||
}
|
||||
fs::rename(&tmp, &path)?;
|
||||
// Fsync the directory so the rename itself is durable.
|
||||
let dir = fs::File::open(path.parent().unwrap())?;
|
||||
dir.sync_all()?;
|
||||
// Fsync the directory so the rename itself is durable. Unix only:
|
||||
// Windows can't open a directory as a file, and NTFS journals the
|
||||
// rename's metadata itself.
|
||||
#[cfg(unix)]
|
||||
{
|
||||
let dir = fs::File::open(path.parent().unwrap())?;
|
||||
dir.sync_all()?;
|
||||
}
|
||||
Ok((sha, false))
|
||||
}
|
||||
|
||||
|
||||
@@ -28,8 +28,8 @@ pub mod store;
|
||||
pub use blob::BlobStore;
|
||||
pub use error::StoreError;
|
||||
pub use journal::{GENESIS_PREV, JournalLine, JournalWriter};
|
||||
pub use snapshot::Snapshot;
|
||||
pub use snapshot::{DiskCheck, Snapshot};
|
||||
pub use store::{
|
||||
Attachment, DB_FILE, DepreciationRun, DisposalPreview, EntryRow, EventRow, JOURNAL_FILE,
|
||||
PayrollLineInput, Store,
|
||||
PayrollLineInput, Store, VerifyReport,
|
||||
};
|
||||
@@ -18,14 +18,36 @@ use crate::blob::BlobStore;
|
||||
use crate::error::StoreError;
|
||||
use crate::journal::{GENESIS_PREV, JournalLine, sha256_hex};
|
||||
use crate::store::{
|
||||
Attachment, BLOBS_DIR, JOURNAL_FILE, Store, apply_line_to_sqlite, query_attachments,
|
||||
read_attachment_bytes, replay_line_into_ledger,
|
||||
Attachment, BLOBS_DIR, DB_FILE, JOURNAL_FILE, Store, apply_line_to_sqlite, compare_projections,
|
||||
projection_rows, query_attachments, read_attachment_bytes, replay_line_into_ledger,
|
||||
};
|
||||
use ledger_core::{Currency, EntryId, Ledger};
|
||||
use rusqlite::Connection;
|
||||
use std::io::{Read, Seek, SeekFrom};
|
||||
use std::path::{Path, PathBuf};
|
||||
|
||||
/// How the writer's on-disk SQLite database compares with the journal.
|
||||
#[derive(Clone, Debug, PartialEq, Eq)]
|
||||
pub enum DiskCheck {
|
||||
/// Every table matches a fresh projection of the journal.
|
||||
Matches,
|
||||
/// The database hasn't applied the newest journal lines yet (normal
|
||||
/// for a moment after a write; the writer applies right after
|
||||
/// journaling, and heals on its next start).
|
||||
Behind {
|
||||
/// Journal lines the database has applied.
|
||||
applied: u64,
|
||||
/// Journal lines there are.
|
||||
journal: u64,
|
||||
},
|
||||
/// No database file (the writer rebuilds it on start).
|
||||
Missing,
|
||||
/// The database differs from the journal: what differs.
|
||||
Differs(String),
|
||||
/// The database couldn't be read.
|
||||
Unreadable(String),
|
||||
}
|
||||
|
||||
/// A read-only, live-refreshable view of a data directory.
|
||||
pub struct Snapshot {
|
||||
data_dir: PathBuf,
|
||||
@@ -168,6 +190,58 @@ impl Snapshot {
|
||||
query_attachments(&self.conn, entry_id)
|
||||
}
|
||||
|
||||
/// Compare the writer's SQLite database (opened read-only) with this
|
||||
/// snapshot's own projection of the journal, every table, row for row.
|
||||
pub fn check_disk(&self) -> DiskCheck {
|
||||
use rusqlite::OpenFlags;
|
||||
let path = self.data_dir.join(DB_FILE);
|
||||
if !path.exists() {
|
||||
return DiskCheck::Missing;
|
||||
}
|
||||
let conn = match Connection::open_with_flags(
|
||||
&path,
|
||||
OpenFlags::SQLITE_OPEN_READ_ONLY | OpenFlags::SQLITE_OPEN_NO_MUTEX,
|
||||
) {
|
||||
Ok(c) => c,
|
||||
Err(e) => return DiskCheck::Unreadable(e.to_string()),
|
||||
};
|
||||
let applied: u64 = match conn.query_row(
|
||||
"SELECT value FROM meta WHERE key='last_applied_seq'",
|
||||
[],
|
||||
|r| r.get::<_, String>(0),
|
||||
) {
|
||||
Ok(v) => v.parse().unwrap_or(0),
|
||||
Err(e) => return DiskCheck::Unreadable(e.to_string()),
|
||||
};
|
||||
let journal = self.journal_seq();
|
||||
if applied != journal {
|
||||
return DiskCheck::Behind { applied, journal };
|
||||
}
|
||||
let (disk, ours) = match (projection_rows(&conn), projection_rows(&self.conn)) {
|
||||
(Ok(d), Ok(o)) => (d, o),
|
||||
(Err(e), _) | (_, Err(e)) => return DiskCheck::Unreadable(e.to_string()),
|
||||
};
|
||||
match compare_projections(&disk, &ours) {
|
||||
Ok(()) => DiskCheck::Matches,
|
||||
Err(StoreError::VerifyMismatch(why)) => DiskCheck::Differs(why),
|
||||
Err(e) => DiskCheck::Unreadable(e.to_string()),
|
||||
}
|
||||
}
|
||||
|
||||
/// Re-hash every stored file. Returns how many verified, and each
|
||||
/// failure as (attachment id, why).
|
||||
pub fn verify_blobs(&self) -> (u64, Vec<(u64, String)>) {
|
||||
let mut ok = 0;
|
||||
let mut bad = Vec::new();
|
||||
for a in self.attachments(None).unwrap_or_default() {
|
||||
match self.blobs.get(&a.blob_sha) {
|
||||
Ok(_) => ok += 1,
|
||||
Err(e) => bad.push((a.id, e.to_string())),
|
||||
}
|
||||
}
|
||||
(ok, bad)
|
||||
}
|
||||
|
||||
/// An attachment's bytes, verified against its hash.
|
||||
pub fn attachment_bytes(&self, attachment_id: u64) -> Result<Vec<u8>, StoreError> {
|
||||
read_attachment_bytes(&self.conn, &self.blobs, attachment_id)
|
||||
|
||||
@@ -8,7 +8,9 @@
|
||||
|
||||
use crate::blob::BlobStore;
|
||||
use crate::error::StoreError;
|
||||
use crate::journal::{JournalLine, JournalWriter, read_and_verify_journal};
|
||||
use crate::journal::{
|
||||
GENESIS_PREV, JournalLine, JournalWriter, read_and_verify_journal, sha256_hex,
|
||||
};
|
||||
use crate::schema;
|
||||
use chrono::{Months, NaiveDate};
|
||||
use ledger_core::{
|
||||
@@ -2257,7 +2259,44 @@ impl Store {
|
||||
"payroll runs diverged between journal, live ledger, and sqlite projection".into(),
|
||||
));
|
||||
}
|
||||
Ok(())
|
||||
// Finally, the whole projection, not just what reports read: every
|
||||
// table, row for row, against a fresh projection of the journal.
|
||||
compare_projections(
|
||||
&projection_rows(&self.conn)?,
|
||||
&projection_rows(&fresh_projection(&lines)?)?,
|
||||
)
|
||||
}
|
||||
|
||||
/// [`Self::verify`], and if the SQLite projection is what diverged,
|
||||
/// heal it: keep the divergent file for forensics, rebuild from the
|
||||
/// journal, and verify again. A journal that fails its own chain is
|
||||
/// never "healed": that's an error for a human.
|
||||
pub fn verify_and_heal(&mut self) -> Result<VerifyReport, StoreError> {
|
||||
let healed = match self.verify() {
|
||||
Ok(()) => None,
|
||||
Err(StoreError::VerifyMismatch(why)) => {
|
||||
let stamp = chrono::Utc::now().format("%Y%m%dT%H%M%SZ").to_string();
|
||||
let keep = self.db_path.with_extension(format!("mismatch.{stamp}"));
|
||||
std::fs::copy(&self.db_path, &keep)?;
|
||||
eprintln!(
|
||||
"wanderbooks: {why}; kept the old database as {} and rebuilt from the journal",
|
||||
keep.display()
|
||||
);
|
||||
self.rebuild()?;
|
||||
self.verify()?;
|
||||
Some(why)
|
||||
}
|
||||
Err(e) => return Err(e),
|
||||
};
|
||||
let lines = read_and_verify_journal(self.journal.path())?;
|
||||
Ok(VerifyReport {
|
||||
events: lines.len() as u64,
|
||||
head: lines
|
||||
.last()
|
||||
.map(|l| sha256_hex(l.to_canonical().as_bytes()))
|
||||
.unwrap_or_else(|| GENESIS_PREV.to_string()),
|
||||
healed,
|
||||
})
|
||||
}
|
||||
|
||||
/// Verify every blob's bytes against its SHA-256. Returns the count
|
||||
@@ -2282,6 +2321,10 @@ impl Store {
|
||||
let old = std::mem::replace(&mut self.conn, Connection::open_in_memory()?);
|
||||
old.close().map_err(|(_, e)| StoreError::Rusqlite(e))?;
|
||||
std::fs::remove_file(&self.db_path)?;
|
||||
// WAL and SHM are derived state too; a stale WAL must not leak into
|
||||
// the rebuilt database.
|
||||
let _ = std::fs::remove_file(self.db_path.with_extension("db-wal"));
|
||||
let _ = std::fs::remove_file(self.db_path.with_extension("db-shm"));
|
||||
self.conn = Self::open_conn(&self.db_path)?;
|
||||
Self::migrate(&self.conn)?;
|
||||
Self::write_meta(&self.conn, "base_currency", self.base_currency.code())?;
|
||||
@@ -3774,6 +3817,93 @@ pub(crate) fn apply_line_to_sqlite(
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// What [`Store::verify_and_heal`] found.
|
||||
#[derive(Clone, Debug, PartialEq, Eq)]
|
||||
pub struct VerifyReport {
|
||||
/// Journal events verified.
|
||||
pub events: u64,
|
||||
/// Hash of the last journal line: the verified head of the chain.
|
||||
pub head: String,
|
||||
/// Why the projection was rebuilt, if it was.
|
||||
pub healed: Option<String>,
|
||||
}
|
||||
|
||||
/// A projection of `lines` into a private in-memory database, built with
|
||||
/// exactly the code the writer uses.
|
||||
pub(crate) fn fresh_projection(lines: &[JournalLine]) -> Result<Connection, StoreError> {
|
||||
let conn = Connection::open_in_memory()?;
|
||||
Store::migrate(&conn)?;
|
||||
for line in lines {
|
||||
apply_line_to_sqlite(&conn, line)?;
|
||||
}
|
||||
Ok(conn)
|
||||
}
|
||||
|
||||
/// Every row of every projected table, as canonical strings, sorted: two
|
||||
/// projections hold the same data exactly when these are equal. The `meta`
|
||||
/// table (bookkeeping about the projection itself) is left out.
|
||||
pub(crate) fn projection_rows(
|
||||
conn: &Connection,
|
||||
) -> Result<std::collections::BTreeMap<String, Vec<String>>, StoreError> {
|
||||
use rusqlite::types::ValueRef;
|
||||
let tables: Vec<String> = conn
|
||||
.prepare(
|
||||
"SELECT name FROM sqlite_master WHERE type='table' AND name NOT LIKE 'sqlite_%' AND name <> 'meta' ORDER BY name",
|
||||
)?
|
||||
.query_map([], |r| r.get(0))?
|
||||
.collect::<Result<_, _>>()?;
|
||||
let mut out = std::collections::BTreeMap::new();
|
||||
for t in tables {
|
||||
let mut stmt = conn.prepare(&format!("SELECT * FROM \"{t}\""))?;
|
||||
let n = stmt.column_count();
|
||||
let mut rows: Vec<String> = stmt
|
||||
.query_map([], |r| {
|
||||
let mut s = String::new();
|
||||
for i in 0..n {
|
||||
let v = match r.get_ref(i)? {
|
||||
ValueRef::Null => "∅".to_string(),
|
||||
ValueRef::Integer(x) => format!("i{x}"),
|
||||
ValueRef::Real(x) => format!("r{x}"),
|
||||
ValueRef::Text(b) => format!("t{}", String::from_utf8_lossy(b)),
|
||||
ValueRef::Blob(b) => format!("b{}", sha256_hex(b)),
|
||||
};
|
||||
s.push_str(&v);
|
||||
s.push('\u{1f}');
|
||||
}
|
||||
Ok(s)
|
||||
})?
|
||||
.collect::<Result<_, _>>()?;
|
||||
rows.sort();
|
||||
out.insert(t, rows);
|
||||
}
|
||||
Ok(out)
|
||||
}
|
||||
|
||||
/// Compare two projections; name the first table that differs.
|
||||
pub(crate) fn compare_projections(
|
||||
have: &std::collections::BTreeMap<String, Vec<String>>,
|
||||
want: &std::collections::BTreeMap<String, Vec<String>>,
|
||||
) -> Result<(), StoreError> {
|
||||
for (table, rows) in want {
|
||||
match have.get(table) {
|
||||
None => {
|
||||
return Err(StoreError::VerifyMismatch(format!(
|
||||
"sqlite projection is missing table {table}"
|
||||
)));
|
||||
}
|
||||
Some(h) if h != rows => {
|
||||
let missing = rows.iter().filter(|r| !h.contains(r)).count();
|
||||
let extra = h.iter().filter(|r| !rows.contains(r)).count();
|
||||
return Err(StoreError::VerifyMismatch(format!(
|
||||
"sqlite projection differs from the journal in table {table}: {missing} rows missing, {extra} unexpected"
|
||||
)));
|
||||
}
|
||||
_ => {}
|
||||
}
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// Attachments in the projection, optionally only one entry's, by id.
|
||||
pub(crate) fn query_attachments(
|
||||
conn: &Connection,
|
||||
|
||||
@@ -143,3 +143,45 @@ fn walk(dir: &std::path::Path) -> Vec<(String, u64)> {
|
||||
}
|
||||
out
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn a_snapshot_checks_the_writers_database_and_files() {
|
||||
use ledger_store::DiskCheck;
|
||||
let dir = test_dir("disk");
|
||||
let mut store = Store::create(&dir, Currency::USD).unwrap();
|
||||
let cash = store
|
||||
.add_account("1000", "Cash", AccountType::Asset, None, false)
|
||||
.unwrap();
|
||||
let equity = store
|
||||
.add_account("3000", "Equity", AccountType::Equity, None, false)
|
||||
.unwrap();
|
||||
let e = post(&mut store, 1, cash, equity, 100);
|
||||
store
|
||||
.attach_receipt(Some(e), "r.png", "image/png", b"receipt-bytes", None)
|
||||
.unwrap();
|
||||
drop(store);
|
||||
let snap = Snapshot::load(&dir).unwrap();
|
||||
assert_eq!(snap.check_disk(), DiskCheck::Matches);
|
||||
assert_eq!(snap.verify_blobs(), (1, vec![]));
|
||||
// Someone edits the database behind the journal's back.
|
||||
let conn = rusqlite::Connection::open(dir.join(ledger_store::DB_FILE)).unwrap();
|
||||
conn.execute(
|
||||
"UPDATE accounts SET name = 'Petty cash' WHERE code = '1000'",
|
||||
[],
|
||||
)
|
||||
.unwrap();
|
||||
drop(conn);
|
||||
match snap.check_disk() {
|
||||
DiskCheck::Differs(why) => assert!(why.contains("accounts"), "{why}"),
|
||||
other => panic!("drift not seen: {other:?}"),
|
||||
}
|
||||
// A damaged file is caught too.
|
||||
let blob = walk(&dir.join("blobs"))
|
||||
.into_iter()
|
||||
.map(|(p, _)| p)
|
||||
.find(|p| !p.ends_with(".tmp"))
|
||||
.unwrap();
|
||||
std::fs::write(&blob, b"tampered").unwrap();
|
||||
let (ok, bad) = snap.verify_blobs();
|
||||
assert_eq!((ok, bad.len()), (0, 1));
|
||||
}
|
||||
@@ -683,3 +683,47 @@ fn vendor_1099_tags_replay_and_project() {
|
||||
assert_eq!(form, "nec");
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn the_whole_projection_is_checked_and_healed() {
|
||||
let dir = test_dir("heal");
|
||||
let mut store = build_books(&dir);
|
||||
let e = store
|
||||
.ledger()
|
||||
.posted_entries()
|
||||
.next()
|
||||
.map(|e| e.id())
|
||||
.unwrap();
|
||||
store
|
||||
.attach_receipt(Some(e), "r.png", "image/png", b"receipt", None)
|
||||
.unwrap();
|
||||
store.verify().unwrap();
|
||||
drop(store);
|
||||
// Drift no accounting report would notice: a receipt quietly unlinked.
|
||||
let db = dir.join(ledger_store::DB_FILE);
|
||||
let conn = rusqlite::Connection::open(&db).unwrap();
|
||||
conn.execute("UPDATE attachments SET entry_id = NULL", [])
|
||||
.unwrap();
|
||||
drop(conn);
|
||||
let mut store = Store::open(&dir).unwrap();
|
||||
match store.verify() {
|
||||
Err(StoreError::VerifyMismatch(why)) => assert!(why.contains("attachments"), "{why}"),
|
||||
other => panic!("drift not caught: {other:?}"),
|
||||
}
|
||||
let report = store.verify_and_heal().unwrap();
|
||||
assert!(report.healed.is_some(), "healed");
|
||||
assert!(report.events > 0);
|
||||
store.verify().unwrap();
|
||||
assert_eq!(
|
||||
store.list_attachments(Some(e)).unwrap().len(),
|
||||
1,
|
||||
"link restored"
|
||||
);
|
||||
let kept = std::fs::read_dir(&dir)
|
||||
.unwrap()
|
||||
.filter_map(|d| d.ok())
|
||||
.any(|d| d.file_name().to_string_lossy().contains(".mismatch."));
|
||||
assert!(kept, "the divergent database is kept for forensics");
|
||||
// A healthy store reports nothing to heal.
|
||||
assert_eq!(store.verify_and_heal().unwrap().healed, None);
|
||||
}
|
||||
Reference in new issue
Block a user