wanderbooks/crates/ledger-cli/tests/http_concurrency.rs

//! Concurrency: several agents at once against one book's store.
//!
//! The books server gives every session its own service over the book's
//! one shared, locked store (ledger-mcp local.rs INVARIANT 4). Oracle (server.rs INVARIANTS 2 and
//! 5, under contention): clients posting at the same time leave books
//! that verify, a trial balance that balances, every posted entry with
//! its receipt (threshold 0.00), and each draft posted exactly once even
//! when two clients race to post it; a restart replays the same books.

use common::{Client, call};
use serde_json::json;

mod common;

const CLIENTS: usize = 6;
const ENTRIES: usize = 12;

/// Draft a `cents` entry tagged `tag` on `c` (a session with `token`, of
/// the server whose `run/` folder is `run`), with a receipt uploaded for
/// it as an agent uploads one, if `with_receipt`.
async fn draft(
    (c, run, token): (&Client, &std::path::Path, &str),
    tag: &str,
    cents: u64,
    with_receipt: bool,
) -> u64 {
    let e = call(
        c,
        "books.draft_entry",
        json!({"date": "2026-03-01", "description": tag}),
    )
    .await
    .unwrap();
    let id = e["entry_id"].as_u64().unwrap();
    let amount = format!("{}.{:02}", cents / 100, cents % 100);
    for (account, side) in [("6000", "debit"), ("1000", "credit")] {
        call(
            c,
            "books.add_line",
            json!({"entry_id": id, "account": account, "side": side, "amount": amount}),
        )
        .await
        .unwrap();
    }
    if with_receipt {
        let upload = common::upload_on(
            c,
            run,
            common::BOOK,
            token,
            tag.as_bytes(),
            (&format!("{tag}.pdf"), "application/pdf", "receipt"),
        )
        .await;
        call(
            c,
            "books.attach_receipt",
            json!({"entry_id": id, "upload_id": upload}),
        )
        .await
        .unwrap();
    }
    id
}

#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn agents_posting_at_once_leave_whole_books() {
    // Made on a thread of its own: the settings go in through a session
    // with a runtime of its own (common.rs), never inside the test's.
    let books = std::thread::spawn(|| {
        common::Books::with(
            "http-concurrency",
            "USD",
            Some("[books]\nreceipt_threshold = \"0.00\"\n"),
        )
    })
    .join()
    .unwrap();
    let (run, token) = (books.home.run(), books.home.token(common::BOOK));
    let connect = |run: std::path::PathBuf, token: String| async move {
        common::connect_at(&run, common::BOOK, Some(token))
            .await
            .unwrap()
    };
    let setup = connect(run.clone(), token.clone()).await;
    for (code, name, ty) in [
        ("1000", "Cash", "asset"),
        ("3000", "Capital", "equity"),
        ("6000", "Supplies", "expense"),
    ] {
        call(
            &setup,
            "books.add_account",
            json!({"code": code, "name": name, "account_type": ty}),
        )
        .await
        .unwrap();
    }
    // Drafts every client will race to post.
    let mut contested = Vec::new();
    for k in 0..ENTRIES {
        contested.push(
            draft(
                (&setup, &run, &token),
                &format!("contested {k}"),
                100 + k as u64,
                true,
            )
            .await,
        );
    }

    let mut tasks = Vec::new();
    for n in 0..CLIENTS {
        let (run, token) = (run.clone(), token.clone());
        let contested = contested.clone();
        tasks.push(tokio::spawn(async move {
            let c = connect(run.clone(), token.clone()).await;
            let (mut posted, mut refused, mut won) = (0usize, 0usize, 0usize);
            for (k, &race) in contested.iter().enumerate() {
                // A receipted entry posts; a bare one is refused.
                let receipted = k % 3 != 0;
                let id = draft(
                    (&c, &run, &token),
                    &format!("c{n} e{k}"),
                    (n * 1000 + k + 1) as u64,
                    receipted,
                )
                .await;
                match call(
                    &c,
                    "books.post_entry",
                    json!({"entry_id": id, "confirmed": true}),
                )
                .await
                {
                    Ok(_) => posted += 1,
                    Err(_) => refused += 1,
                }
                assert_eq!(posted + refused, k + 1);
                // Race everyone for the contested draft.
                if call(
                    &c,
                    "books.post_entry",
                    json!({"entry_id": race, "confirmed": true}),
                )
                .await
                .is_ok()
                {
                    won += 1;
                }
            }
            let _ = c.cancel().await;
            (posted, refused, won)
        }));
    }
    let (mut posted, mut refused, mut won) = (0, 0, 0);
    for t in tasks {
        let (p, r, w) = t.await.unwrap();
        posted += p;
        refused += r;
        won += w;
    }
    let per_client_bare = (0..ENTRIES).filter(|k| k % 3 == 0).count();
    assert_eq!(
        refused,
        CLIENTS * per_client_bare,
        "every bare entry refused"
    );
    assert_eq!(posted, CLIENTS * (ENTRIES - per_client_bare));
    assert_eq!(won, ENTRIES, "each contested draft posted exactly once");

    let v = call(&setup, "books.verify", json!({})).await.unwrap();
    assert_eq!(v["status"], "ok", "{v}");
    let tb = call(&setup, "books.trial_balance", json!({}))
        .await
        .unwrap();
    assert_eq!(tb["balanced"], true, "{tb}");
    // Every posted entry has its receipt (threshold 0.00).
    for id in 1..=(ENTRIES + CLIENTS * ENTRIES) as u64 {
        let e = call(&setup, "books.get_entry", json!({"entry_id": id}))
            .await
            .unwrap();
        if e["status"] == "posted" {
            assert!(
                e["receipts"].as_array().is_some_and(|r| !r.is_empty()),
                "entry {id} posted without a receipt: {e}"
            );
        }
    }
    let head = v["head"].clone();
    let _ = setup.cancel().await;

    // A restart replays the same books.
    books.restart();
    let c = connect(run.clone(), token.clone()).await;
    let v2 = call(&c, "books.verify", json!({})).await.unwrap();
    assert_eq!(v2["status"], "ok");
    assert_eq!(
        v2["head"], head,
        "the restarted server sees the same journal"
    );
    let _ = c.cancel().await;
}