diff --git a/crates/ledger-gui/src/pages/journal.rs b/crates/ledger-gui/src/pages/journal.rs index d1e1340..ebbf27a 100644 --- a/crates/ledger-gui/src/pages/journal.rs +++ b/crates/ledger-gui/src/pages/journal.rs @@ -220,6 +220,11 @@ pub fn describe(b: &Books, line: &JournalLine) -> Option { format!("Recurring: {}", s(p, "description")), Some(Route::Company), ), + "journal.tail_repaired" => ev( + icon::WARNING, + format!("Cut-off journal line set aside ({} bytes)", p["bytes"]), + None, + ), "recurring.generated" => ev( icon::REPEAT, "Recurring draft created".into(), diff --git a/crates/ledger-mcp/src/main.rs b/crates/ledger-mcp/src/main.rs index 0144cd6..b4e0c3f 100644 --- a/crates/ledger-mcp/src/main.rs +++ b/crates/ledger-mcp/src/main.rs @@ -116,7 +116,7 @@ fn main() -> Result<(), Box> { let company = settings.company; let mut store = if data_dir.join(ledger_store::JOURNAL_FILE).exists() { - Store::open(&data_dir)? + open_waiting_for_lock(&data_dir)? } else { let code = config.base_currency.as_deref().unwrap_or("USD"); let currency = Currency::new(code).map_err(|_| format!("bad base currency {code:?}"))?; @@ -180,3 +180,20 @@ async fn serve_http( axum::serve(listener, router).await?; Ok(()) } + +/// Open the books, waiting up to ten seconds for another writer to let go. +/// A restarted server (a client reconnecting) often starts while the old +/// process is still exiting with the lock; waiting turns that race into a +/// short delay. A writer that keeps the books longer is a real conflict, +/// and the refusal says so. +fn open_waiting_for_lock(data_dir: &std::path::Path) -> Result { + let deadline = std::time::Instant::now() + std::time::Duration::from_secs(10); + loop { + match Store::open(data_dir) { + Err(ledger_store::StoreError::Locked(_)) if std::time::Instant::now() < deadline => { + std::thread::sleep(std::time::Duration::from_millis(100)); + } + other => return other, + } + } +} diff --git a/crates/ledger-mcp/src/server.rs b/crates/ledger-mcp/src/server.rs index 2b20cd6..5903052 100644 --- a/crates/ledger-mcp/src/server.rs +++ b/crates/ledger-mcp/src/server.rs @@ -1623,11 +1623,23 @@ impl BooksServer { why: r.review_why.to_string(), }) .collect(); + let journal_repairs = store + .tail_repairs() + .map_err(|e| e.to_string())? + .into_iter() + .filter(|(at, _, _)| at.get(..10).is_some_and(|d| d >= p.since.as_str())) + .map(|(at, bytes, saved_as)| JournalRepair { + at, + bytes, + saved_as, + }) + .collect(); Ok(Json(DigestOut { since: since.to_string(), entries, totals_by_account: totals, stale_rules, + journal_repairs, })) } diff --git a/crates/ledger-mcp/src/types.rs b/crates/ledger-mcp/src/types.rs index b02d525..929c895 100644 --- a/crates/ledger-mcp/src/types.rs +++ b/crates/ledger-mcp/src/types.rs @@ -1266,6 +1266,22 @@ pub struct DigestOut { /// for everyone with a pull request (or at least an issue) at /// . pub stale_rules: Vec, + /// Appends cut off mid-line (a crash or power loss) that were set aside + /// when the books were opened, since `since`. Nothing acknowledged was + /// lost; tell the human, and if they had just asked for a change, + /// check it is there and redo it if not. + pub journal_repairs: Vec, +} + +/// A torn journal tail set aside on open. +#[derive(Debug, Serialize, JsonSchema)] +pub struct JournalRepair { + /// When it was set aside (UTC timestamp). + pub at: String, + /// How many bytes were cut off. + pub bytes: u64, + /// The file beside the journal they were saved to. + pub saved_as: String, } /// A dated rule due for review (see `ledger_core::rules`). diff --git a/crates/ledger-mcp/tests/mcp_stdio.rs b/crates/ledger-mcp/tests/mcp_stdio.rs index 9aa966a..d70b567 100644 --- a/crates/ledger-mcp/tests/mcp_stdio.rs +++ b/crates/ledger-mcp/tests/mcp_stdio.rs @@ -9,10 +9,18 @@ struct Mcp { stdin: ChildStdin, stdout: BufReader, next_id: u64, - #[allow(dead_code)] child: Child, } +/// Closing stdin ends the server; wait for it to exit, so the next server +/// on the same books never races it for the writer lock. +impl Drop for Mcp { + fn drop(&mut self) { + let _ = self.child.kill(); + let _ = self.child.wait(); + } +} + impl Mcp { fn spawn(data_dir: &std::path::Path) -> Self { Self::spawn_with(data_dir, &[]) @@ -1673,3 +1681,29 @@ fn mcp_receipts_always_required_when_threshold_is_zero() { "a $5 entry needs a receipt too: {err}" ); } + +/// A client reconnecting starts a new server while the old one still has +/// the books: the new one waits for the lock instead of dying. +#[test] +fn a_restarted_server_waits_for_the_old_one_to_let_go() { + let dir = test_dir("restart-race"); + let init = |mcp: &mut Mcp| { + mcp.request( + "initialize", + json!({ + "protocolVersion": "2025-06-18", + "capabilities": {}, + "clientInfo": {"name": "smoke-test", "version": "0.1.0"}, + }), + ); + mcp.notify("notifications/initialized"); + }; + let mut old = Mcp::spawn(&dir); + init(&mut old); + let mut new = Mcp::spawn(&dir); + std::thread::sleep(std::time::Duration::from_millis(500)); + drop(old); + init(&mut new); + let chart = new.call("books.chart", json!({})); + assert!(chart.is_object()); +} diff --git a/crates/ledger-store/src/journal.rs b/crates/ledger-store/src/journal.rs index 8e11f32..e92cf8c 100644 --- a/crates/ledger-store/src/journal.rs +++ b/crates/ledger-store/src/journal.rs @@ -112,8 +112,11 @@ pub fn sha256_hex(bytes: &[u8]) -> String { } /// Append-only writer. Assigns `seq`, stamps `ts`, chains `prev`, fsyncs -/// every append. The sequence only advances after a successful fsync, so a -/// torn tail can never exist in a file written by this writer. +/// every append. The sequence only advances after a successful fsync, and a +/// failed write is truncated away, so an append is acknowledged only once +/// its whole line is on disk. A crash or power loss mid-append can still +/// leave a partial last line; it was never acknowledged, and +/// [`repair_torn_tail`] sets it aside when the books are next opened. pub struct JournalWriter { path: PathBuf, file: File, @@ -224,10 +227,74 @@ pub fn read_and_verify_journal(path: &Path) -> Result, StoreErr Ok(lines) } +/// A torn final journal line, set aside when the books were opened. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct TornTail { + /// How many bytes followed the last complete line. + pub bytes: u64, + /// SHA-256 of those bytes. + pub sha256: String, + /// Where they were saved, beside the journal. + pub saved_as: PathBuf, +} + +/// Set aside a torn final line: bytes after the journal's last newline. +/// +/// ```text +/// GOAL: open books whose last append was cut off (power loss, crash). +/// SOURCE: docs/receipt-store.md (a command is acknowledged only after its +/// whole line, newline included, is synced); the same rule write-ahead +/// logs use for a torn tail record (SQLite, PostgreSQL: recovery stops +/// at the first incomplete record). +/// SCENARIOS: +/// 1. IF the journal ends with a newline: SHOULD change nothing. +/// 2. IF bytes follow the last newline: SHOULD save them to a file +/// beside the journal, truncate the journal to the last newline, and +/// report them +/// BECAUSE [`JournalWriter::append`] syncs the newline before it +/// returns, so a line without one was never acknowledged: no one was +/// told it happened, and dropping it loses nothing. +/// 3. NEVER touch a complete line: damage before the last newline is +/// corruption, and opening refuses it. +/// ``` +pub fn repair_torn_tail(path: &Path) -> Result, StoreError> { + let raw = std::fs::read(path)?; + let keep = raw.iter().rposition(|&b| b == b'\n').map_or(0, |i| i + 1); + if keep == raw.len() { + return Ok(None); + } + let tail = &raw[keep..]; + let stamp = chrono::Utc::now().format("%Y%m%dT%H%M%S%.fZ").to_string(); + let mut saved_as = path.as_os_str().to_owned(); + saved_as.push(format!(".torn.{stamp}")); + let saved_as = PathBuf::from(saved_as); + { + let mut f = OpenOptions::new() + .write(true) + .create_new(true) + .open(&saved_as)?; + f.write_all(tail)?; + f.sync_all()?; + } + if let Some(dir) = path.parent() + && let Ok(d) = File::open(dir) + { + let _ = d.sync_all(); + } + let file = OpenOptions::new().write(true).open(path)?; + file.set_len(keep as u64)?; + file.sync_all()?; + Ok(Some(TornTail { + bytes: tail.len() as u64, + sha256: sha256_hex(tail), + saved_as, + })) +} + /// Truncate a journal file to its last good line. For recovering *copies* /// with a torn tail (e.g. a backup taken mid-write): the projection never /// saw the torn line, so dropping it is safe. Returns the last good seq. -/// The live journal written by [`JournalWriter`] cannot have a torn tail. +/// The live journal's torn tail is handled on open by [`repair_torn_tail`]. pub fn truncate_to_last_good_seq(path: &Path) -> Result { let raw = std::fs::read(path)?; let mut good_len = 0usize; diff --git a/crates/ledger-store/src/store.rs b/crates/ledger-store/src/store.rs index 0fd4b92..0dbc048 100644 --- a/crates/ledger-store/src/store.rs +++ b/crates/ledger-store/src/store.rs @@ -237,6 +237,8 @@ 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)?; @@ -305,6 +307,23 @@ impl Store { base_currency, }; store.catch_up(&lines)?; + if let Some(t) = torn { + // On the record, so the repair is part of the audit trail. + let line = store.append_journal( + "journal.tail_repaired", + json!({ + "bytes": t.bytes, + "sha256": t.sha256, + "saved_as": t.saved_as.file_name().map(|n| n.to_string_lossy().into_owned()), + }), + )?; + apply_line_to_sqlite(&store.conn, &line)?; + eprintln!( + "wanderbooks: the journal's last line was cut off (an append that never finished); set aside {} bytes in {}", + t.bytes, + t.saved_as.display() + ); + } Ok(store) } @@ -414,6 +433,23 @@ impl Store { self.base_currency } + /// Every torn journal tail set aside on open, oldest first, as + /// (when, bytes, file it was saved to). + pub fn tail_repairs(&self) -> Result, StoreError> { + let lines = crate::journal::read_and_verify_journal(&self.data_dir.join(JOURNAL_FILE))?; + Ok(lines + .iter() + .filter(|l| l.line_type == "journal.tail_repaired") + .map(|l| { + ( + l.ts.clone(), + l.payload["bytes"].as_u64().unwrap_or(0), + l.payload["saved_as"].as_str().unwrap_or("").to_string(), + ) + }) + .collect()) + } + /// Data directory. pub fn data_dir(&self) -> &Path { &self.data_dir @@ -3340,6 +3376,7 @@ pub(crate) fn replay_line_into_ledger( p.get("class_id").and_then(|v| v.as_u64()).map(|v| v as u32), )?; } + "journal.tail_repaired" => {} other => { return Err(StoreError::JournalCorrupt(format!( "unknown line type: {other}" @@ -3545,7 +3582,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" => {} + "store.created" | "journal.tail_repaired" => {} "account.added" => { conn.execute( "INSERT INTO accounts(id, code, name, type, parent, placeholder, currency) VALUES (?,?,?,?,?,?,?)", diff --git a/crates/ledger-store/tests/store.rs b/crates/ledger-store/tests/store.rs index 04acb2d..bab36c2 100644 --- a/crates/ledger-store/tests/store.rs +++ b/crates/ledger-store/tests/store.rs @@ -154,35 +154,112 @@ fn journal_tamper_is_detected() { ); } +/// Oracle (rank 1, journal::repair_torn_tail's spec): an append cut off +/// mid-line was never acknowledged, so opening sets it aside, keeps every +/// acknowledged line, puts the repair on the record, and reports it. #[test] -fn torn_tail_is_truncatable() { +fn a_torn_tail_is_set_aside_on_open() { let dir = test_dir("torntail"); let store = build_books(&dir); - let seq_before = { - let w = JournalWriter::open(&dir.join(ledger_store::JOURNAL_FILE)).unwrap(); - w.next_seq() - 1 - }; + let journal = dir.join(ledger_store::JOURNAL_FILE); + let before = ledger_core::reports::snapshot(store.ledger()).unwrap(); + let seq_before = JournalWriter::open(&journal).unwrap().next_seq() - 1; drop(store); - // Simulate a torn backup: append half a line. use std::io::Write; + let torn = b"{\"seq\": 999, \"trunca"; let mut f = std::fs::OpenOptions::new() .append(true) - .open(dir.join(ledger_store::JOURNAL_FILE)) + .open(&journal) .unwrap(); - f.write_all(b"{\"seq\": 999, \"trunca").unwrap(); + f.write_all(torn).unwrap(); f.sync_all().unwrap(); drop(f); - // Strict open now fails... - assert!(Store::open(&dir).is_err()); - // ...but truncation recovers to the last good seq. - let last_good = - ledger_store::journal::truncate_to_last_good_seq(&dir.join(ledger_store::JOURNAL_FILE)) - .unwrap(); - assert_eq!(last_good, seq_before); let store = Store::open(&dir).unwrap(); store.verify().unwrap(); + assert_eq!( + ledger_core::reports::snapshot(store.ledger()).unwrap(), + before + ); + let repairs = store.tail_repairs().unwrap(); + assert_eq!(repairs.len(), 1); + assert_eq!(repairs[0].1, torn.len() as u64); + // The cut-off bytes are kept, exactly, beside the journal. + assert_eq!(std::fs::read(dir.join(&repairs[0].2)).unwrap(), torn); + // The repair is the one new line, right after the last good one. + let lines = ledger_store::journal::read_and_verify_journal(&journal).unwrap(); + assert_eq!(lines.len() as u64, seq_before + 1); + assert_eq!(lines.last().unwrap().line_type, "journal.tail_repaired"); + // Opening again repairs nothing more. + drop(store); + let store = Store::open(&dir).unwrap(); + assert_eq!(store.tail_repairs().unwrap().len(), 1); + store.verify().unwrap(); +} + +/// A damaged *complete* line is corruption, not a torn tail: opening still +/// refuses it, and the journal is left as it was. +#[test] +fn damage_before_the_last_newline_still_refuses_to_open() { + let dir = test_dir("tornmid"); + drop(build_books(&dir)); + let journal = dir.join(ledger_store::JOURNAL_FILE); + let mut raw = std::fs::read(&journal).unwrap(); + let cut = raw.len() - 2; // inside the last complete line + raw.remove(cut - 5); + std::fs::write(&journal, &raw).unwrap(); + assert!(Store::open(&dir).is_err()); + assert_eq!(std::fs::read(&journal).unwrap(), raw); +} + +/// Power loss at any byte of the last append: whatever prefix reached the +/// disk, the books open with exactly what was acknowledged before it. +#[test] +fn power_loss_at_any_byte_of_the_last_append_is_recovered() { + let dir = test_dir("tornevery"); + let store = build_books(&dir); + let journal = dir.join(ledger_store::JOURNAL_FILE); + let before_bytes = std::fs::read(&journal).unwrap(); + let before = ledger_core::reports::snapshot(store.ledger()).unwrap(); + drop(store); + // The books as they were on disk before the last append (the database + // is never ahead of the journal: the line is written first). + let pre = test_dir("tornevery-pre"); + for e in std::fs::read_dir(&dir).unwrap() { + let e = e.unwrap(); + if e.file_type().unwrap().is_file() { + std::fs::copy(e.path(), pre.join(e.file_name())).unwrap(); + } + } + let mut store = Store::open(&dir).unwrap(); + store.add_customer("Acme").unwrap(); + drop(store); + let full = std::fs::read(&journal).unwrap(); + let last = &full[before_bytes.len()..]; + assert_eq!(last.last(), Some(&b'\n')); + for cut in 1..last.len() { + let case = test_dir(&format!("tornevery-{cut}")); + for e in std::fs::read_dir(&pre).unwrap() { + let e = e.unwrap(); + if e.file_type().unwrap().is_file() { + std::fs::copy(e.path(), case.join(e.file_name())).unwrap(); + } + } + let mut torn = before_bytes.clone(); + torn.extend_from_slice(&last[..cut]); + std::fs::write(case.join(ledger_store::JOURNAL_FILE), &torn).unwrap(); + let store = Store::open(&case).unwrap_or_else(|e| panic!("cut at {cut}: {e}")); + store.verify().unwrap(); + assert_eq!( + ledger_core::reports::snapshot(store.ledger()).unwrap(), + before, + "cut at {cut}" + ); + assert!(store.ledger().list_customers().is_empty(), "cut at {cut}"); + drop(store); + let _ = std::fs::remove_dir_all(&case); + } } #[test] diff --git a/docs/receipt-store.md b/docs/receipt-store.md index ef4355c..77f0a55 100644 --- a/docs/receipt-store.md +++ b/docs/receipt-store.md @@ -23,7 +23,13 @@ easily backupable, journable": journal file plus the blob directory. Plain files, no WAL-file timing windows, no special backup API. A torn tail from a mid-copy is detectable (the last line fails hash/parse) and recoverable (truncate to the last - good seq — the projection never saw it). Better: the journal is + good seq — the projection never saw it). The live journal recovers the + same way on its own: an append cut off by a crash or power loss (bytes + after the last newline) was never acknowledged, so opening the books + saves those bytes to `journal.jsonl.torn.