nordisk/monetize

crates/monetize-server/src/estate_order/tests/drive_job.rs

//! **B6: the tick drives each estate order as a queued jera job** (owner rule: server background
//! work is a jera ChainQueue job, never a raw thread): one keyed chain PER ESTATE — an estate's
//! drives run one at a time, in order, and two estates never wait for each other; a second tick
//! while a drive is queued or running is that same job.

use super::*;

#[test]
fn the_tick_drives_each_estate_on_its_own_jera_chain() {
    let r = rig("drive-job");
    start(
        &r.state,
        "acme",
        "buyer@example.com",
        "",
        &EstateSizes::default(),
    )
    .expect("ordered");
    start(
        &r.state,
        "bolt",
        "buyer@example.com",
        "",
        &EstateSizes::default(),
    )
    .expect("ordered");
    r.pay.authorize(&handle_of(&r, "acme"));
    r.pay.authorize(&handle_of(&r, "bolt"));
    let rt = tokio::runtime::Builder::new_multi_thread()
        .worker_threads(2)
        .enable_all()
        .build()
        .unwrap();
    let state = r.state.clone();
    rt.block_on(async move {
        // Every lane the ticks made, kept here: a delivered order's lane is let go by the tick.
        let mut seen: BTreeMap<String, Arc<jera::jobs::chain::ChainQueue>> = BTreeMap::new();
        for _ in 0..400 {
            let s = state.clone();
            tokio::task::spawn_blocking(move || {
                // The estates call home, as an installed one does every minute.
                calls_home(&s, "acme");
                calls_home(&s, "bolt");
                super::tick(&s);
            })
            .await
            .unwrap();
            for (c, q) in state.estates.drive_lanes.lock().unwrap().iter() {
                seen.entry(c.clone()).or_insert_with(|| q.clone());
            }
            if ["acme", "bolt"].iter().all(|c| {
                state
                    .estates
                    .get(c)
                    .is_some_and(|r| r.phase == phase::CAPTURED)
            }) {
                break;
            }
            tokio::time::sleep(std::time::Duration::from_millis(25)).await;
        }
        assert_eq!(
            seen.keys().map(String::as_str).collect::<Vec<_>>(),
            ["acme", "bolt"],
            "one jera chain per estate"
        );
        // The drive that delivered an order is still finishing when its phase is read.
        for _ in 0..400 {
            if seen.values().all(|q| q.stats().done > 0) {
                break;
            }
            tokio::time::sleep(std::time::Duration::from_millis(25)).await;
        }
        assert!(
            seen.values().all(|q| q.stats().done > 0),
            "the drives ran as jera jobs"
        );
    });
    for c in ["acme", "bolt"] {
        let run = r.state.estates.get(c).unwrap();
        assert_eq!(run.phase, phase::CAPTURED, "{c}: {:#?}", run.history);
    }
}