store: set aside a torn last journal line on open; server waits for lock

A crash or power loss mid-append can leave a partial last line. The
writer acknowledges an append only after its whole line is synced, so
that line was never acknowledged: opening now saves those bytes to
journal.jsonl.torn.<time>, truncates to the last newline, and journals
journal.tail_repaired (replay and projection ignore it; the GUI shows
it; books.digest reports it in journal_repairs). Damage before the last
newline is still corruption and still refuses to open. Tested at every
byte of the last append.

The MCP server now waits up to ten seconds for the writer lock: a
restarted server (a client reconnecting) raced the exiting one and died
with Locked, which also made mcp_schedules_and_recurring_over_stdio
flaky. The test harness now waits for each server to exit.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01GWVo7x3UHw5kwCUaWxiejX

crates/ledger-gui/src/pages/journal.rs

@@ -220,6 +220,11 @@ pub fn describe(b: &Books, line: &JournalLine) -> Option<Event> {
             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(),

crates/ledger-mcp/src/main.rs

@@ -116,7 +116,7 @@ fn main() -> Result<(), Box<dyn std::error::Error>> {
     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<Store, ledger_store::StoreError> {
+    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,
+        }
+    }
+}

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,
         }))
     }
 

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
     /// <https://github.com/swaits/wanderbooks>.
     pub stale_rules: Vec<StaleRule>,
+    /// 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<JournalRepair>,
+}
+
+/// 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`).

crates/ledger-mcp/tests/mcp_stdio.rs

@@ -9,10 +9,18 @@ struct Mcp {
     stdin: ChildStdin,
     stdout: BufReader<std::process::ChildStdout>,
     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());
+}

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
+/// every append. The sequence only advances after a successful fsync, and a
-/// torn tail can never exist in a file written by this writer.
+/// 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<Vec<JournalLine>, 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<Option<TornTail>, 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<u64, StoreError> {
     let raw = std::fs::read(path)?;
     let mut good_len = 0usize;

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<Vec<(String, u64, String)>, 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 (?,?,?,?,?,?,?)",

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 journal = dir.join(ledger_store::JOURNAL_FILE);
-        let w = JournalWriter::open(&dir.join(ledger_store::JOURNAL_FILE)).unwrap();
+    let before = ledger_core::reports::snapshot(store.ledger()).unwrap();
-        w.next_seq() - 1
+    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]

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.<time>`, truncates to the last
+  newline, and journals a `journal.tail_repaired` line; `books.digest`
+  reports it. Damage before the last newline is corruption and still
+  refuses to open. Better: the journal is
   git-friendly, so an optional `wanderbooks backup` that commits it to a
   private git repo gives versioned, distributed, deduplicated backup with
   full history.