viewer: no false "database differs" while the books are being written

New concurrency test: one thread writes entries, drafts and receipts as
fast as it can while a Snapshot (the viewer) refreshes in a loop. At
every refresh the snapshot's books must balance, never go backward, and
never see the writer's database as differing from the journal; at the
end it must equal the writer in full. It failed 30 runs in 50:
check_disk read the applied position and the tables in separate
statements, so a write landing between them made a consistent database
look like it "differs", which the viewer shows as a problem in Needs
attention.

- check_disk reads in one read transaction (one consistent view).
- A database that has applied lines the viewer hasn't read yet is now
  DiskCheck::Ahead ("viewer catching up"), not Behind.
- Without the transaction the test fails 14 runs in 20; with it, 50/50.

Also: members' equity spec 3b states the exact bound (a member's period
income is within two cents of their share, one cent of rounding at each
end; the running balance chains exactly), and the spec-blind property
checks that bound. docs/testing.md gains the concurrency layer and item
12: each close into members' capital still rounds on its own (pending).

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

crates/ledger-store/tests/concurrency.rs

@@ -0,0 +1,127 @@
+//! Concurrency (rank 2): the viewer reads while the agent writes.
+//!
+//! The viewer ([`Snapshot`]) follows the live journal of a [`Store`] that
+//! another thread (in production, another process) is writing as fast as
+//! it can. Its own spec (snapshot.rs): a line still being appended is left
+//! for the next refresh, never treated as corruption; it never writes.
+//!
+//! Oracle, at every refresh: the snapshot's books are whole (the trial
+//! balance balances, every posted entry is valid), they never go backward,
+//! and the writer's database is never seen to differ from the journal
+//! (behind is fine: the writer applies right after journaling). At the end
+//! the snapshot equals the writer, every part of the ledger, attachments
+//! included.
+
+use chrono::NaiveDate;
+use ledger_core::{AccountType, Currency, Money, Side};
+use ledger_store::{DiskCheck, Snapshot, Store};
+use rust_decimal::Decimal;
+use std::sync::atomic::{AtomicBool, Ordering};
+use std::sync::{Arc, Barrier};
+
+fn test_dir(name: &str) -> std::path::PathBuf {
+    let dir = std::env::temp_dir().join(format!("wanderbooks-conc-{name}-{}", std::process::id()));
+    let _ = std::fs::remove_dir_all(&dir);
+    std::fs::create_dir_all(&dir).unwrap();
+    dir
+}
+
+const ENTRIES: u32 = 400;
+
+#[test]
+fn a_viewer_reading_while_the_books_are_written_sees_only_whole_books() {
+    let dir = test_dir("viewer");
+    let mut store = Store::create(&dir, Currency::USD).unwrap();
+    let cash = store
+        .add_account("1000", "Cash", AccountType::Asset, None, false)
+        .unwrap();
+    let sales = store
+        .add_account("4000", "Sales", AccountType::Revenue, None, false)
+        .unwrap();
+    let mut snapshot = Snapshot::load(&dir).unwrap();
+
+    let done = Arc::new(AtomicBool::new(false));
+    let start = Arc::new(Barrier::new(2));
+    let writer = {
+        let (done, start) = (done.clone(), start.clone());
+        std::thread::spawn(move || {
+            start.wait();
+            for n in 0..ENTRIES {
+                let day = NaiveDate::from_ymd_opt(2026, 1, 1).unwrap()
+                    + chrono::Days::new(u64::from(n % 365));
+                let e = store.create_entry(day, &format!("sale {n}")).unwrap();
+                let m = Money::new(Decimal::new(i64::from(n) * 100 + 1, 2), Currency::USD);
+                store.add_line(e, cash, Side::Debit, m).unwrap();
+                store.add_line(e, sales, Side::Credit, m).unwrap();
+                if n % 3 == 0 {
+                    // Leave every third a draft for a while: drafts are
+                    // journaled too, and must read back whole.
+                    continue;
+                }
+                if n % 10 == 1 {
+                    store
+                        .attach_receipt(
+                            Some(e),
+                            &format!("r{n}.pdf"),
+                            "application/pdf",
+                            format!("receipt {n}").as_bytes(),
+                            None,
+                        )
+                        .unwrap();
+                }
+                store.post_entry(e).unwrap();
+            }
+            done.store(true, Ordering::SeqCst);
+            store
+        })
+    };
+
+    start.wait();
+    let (mut last_seq, mut last_posted, mut refreshes) = (0u64, 0usize, 0u64);
+    loop {
+        let finished = done.load(Ordering::SeqCst);
+        snapshot.refresh().unwrap();
+        refreshes += 1;
+        let l = snapshot.ledger();
+        let tb = ledger_core::reports::trial_balance(l);
+        assert_eq!(tb.total_debits, tb.total_credits, "refresh {refreshes}");
+        for e in l.posted_entries() {
+            assert!(
+                e.validate(l.chart()).is_ok(),
+                "entry {} at refresh {refreshes}",
+                e.id()
+            );
+        }
+        let posted = l.posted_entries().count();
+        assert!(
+            snapshot.journal_seq() >= last_seq,
+            "the journal went backward"
+        );
+        assert!(posted >= last_posted, "posted entries went backward");
+        (last_seq, last_posted) = (snapshot.journal_seq(), posted);
+        match snapshot.check_disk() {
+            DiskCheck::Matches | DiskCheck::Behind { .. } | DiskCheck::Ahead { .. } => {}
+            other => panic!("refresh {refreshes}: the database {other:?}"),
+        }
+        if finished {
+            break;
+        }
+    }
+    let store = writer.join().unwrap();
+    snapshot.refresh().unwrap();
+    let parts = store.ledger().differs_from(snapshot.ledger());
+    assert!(
+        parts.is_empty(),
+        "the viewer differs from the writer in {parts:?}"
+    );
+    assert_eq!(
+        snapshot.attachments(None).unwrap().len(),
+        store.list_attachments(None).unwrap().len()
+    );
+    assert_eq!(snapshot.check_disk(), DiskCheck::Matches);
+    assert!(refreshes > 1, "the reader never overlapped the writer");
+    if std::env::var("CONC_STATS").is_ok() {
+        eprintln!("REFRESHES {refreshes}");
+    }
+    let _ = std::fs::remove_dir_all(&dir);
+}