estate/holger-live/tests/the_trial_job.rs
//! ★ **The trial confirm is a jera job** (L21, the owner's rule: server background work runs as
//! jera ChainQueue jobs with priority, never a bare task): a box on trial queues `live.trial`
//! looks at High priority; each look is one ledger row; the look that has seen the app serve for
//! [`CONFIRM_AFTER`] confirms the slot, and the look past [`CONFIRM_DEADLINE`] without serving
//! rolls it back (the re-exec hook). A box not on trial queues nothing.
use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
use std::sync::Arc;
use std::time::Duration;
use holger_live::bench::BenchBox;
use holger_live::trial::{self, TrialWatch, KIND_TRIAL};
use holger_live::{CONFIRM_AFTER, CONFIRM_DEADLINE};
use jera::jobs::{ChainQueue, JobSelector, JobStore, Priority};
fn dir(tag: &str) -> std::path::PathBuf {
let d = std::env::temp_dir().join(format!("holger-live-trial-{tag}-{}", std::process::id()));
let _ = std::fs::remove_dir_all(&d);
d
}
fn app(n: u8) -> Vec<u8> {
(0..64 * 1024u32).map(|i| (i % 251) as u8 ^ n).collect()
}
/// A box booted on trial in slot B.
fn on_trial(d: &std::path::Path) -> BenchBox {
let b = BenchBox::lay(&d.join("esp"), "kernel-1", 42, &app(1)).unwrap();
b.slot.stage(&b.stage_of(&app(2), 42)).unwrap();
b.slot.boot_as_korp_start().unwrap();
assert!(b.slot.on_trial());
b
}
/// A watch over `b` whose clock is `secs` and whose serving answer is `serving`.
fn watch(
b: BenchBox,
secs: Arc<AtomicU64>,
serving: Arc<AtomicBool>,
rolled: Arc<AtomicBool>,
) -> Arc<TrialWatch> {
Arc::new(TrialWatch::new(
Arc::new(b.slot),
Box::new(move || serving.load(Ordering::SeqCst)),
Box::new(move || Duration::from_secs(secs.load(Ordering::SeqCst))),
Box::new(move || rolled.store(true, Ordering::SeqCst)),
))
}
fn rows(store: &JobStore) -> Vec<(String, String, String)> {
let mut r: Vec<_> = store
.list(&JobSelector::All)
.unwrap()
.into_iter()
.filter(|r| r.kind == KIND_TRIAL)
.map(|r| (r.status, r.detail_json, r.target))
.collect();
r.sort();
r
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn a_trial_is_looked_at_by_queued_jobs_and_confirms_after_serving() {
let d = dir("confirm");
let store = JobStore::open(&d.join("jobs")).unwrap();
let q = ChainQueue::new(store.sink());
let (secs, serving, rolled) = (
Arc::new(AtomicU64::new(10)),
Arc::new(AtomicBool::new(false)),
Arc::new(AtomicBool::new(false)),
);
let w = watch(
on_trial(&d),
Arc::clone(&secs),
Arc::clone(&serving),
Arc::clone(&rolled),
);
// Not serving yet: the look is queued at High priority and waits.
let s = trial::spec(1);
assert_eq!((s.kind.as_str(), s.priority), (KIND_TRIAL, Priority::High));
assert!(trial::submit(&q, 1).is_new());
assert!(!trial::submit(&q, 1).is_new(), "one look per generation");
assert_eq!(q.pending_in_group(KIND_TRIAL), 1);
let w1 = Arc::clone(&w);
assert_eq!(q.drain_n(4, move |spec| trial::run(&w1, &spec)).await, 1);
assert_eq!(
w.slot().running().unwrap().slot,
"b booted",
"a look that waits writes nothing"
);
// It serves from 20 s on; a look at 90 s confirms the slot.
secs.store(20, Ordering::SeqCst);
serving.store(true, Ordering::SeqCst);
trial::submit(&q, 2);
let w2 = Arc::clone(&w);
q.drain_n(1, move |spec| trial::run(&w2, &spec)).await;
secs.store(20 + CONFIRM_AFTER.as_secs() + 10, Ordering::SeqCst);
trial::submit(&q, 3);
let w3 = Arc::clone(&w);
q.drain_n(1, move |spec| trial::run(&w3, &spec)).await;
assert_eq!(
w.slot().running().unwrap().slot,
"b",
"confirmed by the job"
);
assert!(w.finished() && !rolled.load(Ordering::SeqCst));
let r = rows(&store);
assert_eq!(r.len(), 3, "each look is one ledger row: {r:?}");
assert!(r.iter().all(|(st, _, _)| st == "done"), "{r:?}");
assert!(
r.iter()
.any(|(_, detail, _)| detail.contains("\"confirm\"")),
"{r:?}"
);
assert!(
r.iter()
.filter(|(_, detail, _)| detail.contains("\"wait\""))
.count()
== 2,
"{r:?}"
);
// Finished: no more looks are queued.
assert!(trial::submit_while(&q, &w, 4).is_none());
let _ = std::fs::remove_dir_all(&d);
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn a_trial_that_never_serves_is_rolled_back_by_its_job() {
let d = dir("back");
let store = JobStore::open(&d.join("jobs")).unwrap();
let q = ChainQueue::new(store.sink());
let (secs, serving, rolled) = (
Arc::new(AtomicU64::new(CONFIRM_DEADLINE.as_secs() + 1)),
Arc::new(AtomicBool::new(false)),
Arc::new(AtomicBool::new(false)),
);
let w = watch(on_trial(&d), secs, serving, Arc::clone(&rolled));
assert!(trial::submit_while(&q, &w, 1).is_some());
let w1 = Arc::clone(&w);
q.drain_n(1, move |spec| trial::run(&w1, &spec)).await;
assert!(
rolled.load(Ordering::SeqCst),
"the job asked for the re-exec (korp-start rolls back)"
);
assert!(w.finished());
let r = rows(&store);
assert_eq!(r.len(), 1);
assert!(r[0].1.contains("\"roll-back\""), "{r:?}");
assert_eq!(
w.slot().running().unwrap().slot,
"b booted",
"never confirmed: korp-start finds it booted"
);
let _ = std::fs::remove_dir_all(&d);
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn a_box_not_on_trial_queues_no_look() {
let d = dir("none");
let store = JobStore::open(&d.join("jobs")).unwrap();
let q = ChainQueue::new(store.sink());
let b = BenchBox::lay(&d.join("esp"), "kernel-1", 42, &app(1)).unwrap();
let w = watch(
b,
Arc::new(AtomicU64::new(0)),
Arc::new(AtomicBool::new(true)),
Arc::new(AtomicBool::new(false)),
);
assert!(w.finished(), "nothing on trial");
assert!(trial::submit_while(&q, &w, 1).is_none());
assert!(
trial::start(
Arc::clone(w.slot()),
Box::new(|| true),
&d.join("live-jobs"),
Box::new(|| {})
)
.is_none(),
"start queues nothing off trial"
);
assert_eq!(q.pending_in_group(KIND_TRIAL), 0);
let _ = std::fs::remove_dir_all(&d);
}
/// ★ **The deadline counts from the arming at boot** (lane B3, A9 row 6): a watch armed at the
/// app's first line, looked at first only after the deadline (the start waited for its console
/// token), rolls back at that first look — the late watch no longer restarts the clock.
#[test]
fn a_watch_armed_at_boot_rolls_back_at_its_first_late_look() {
let d = dir("armed");
let rolled = Arc::new(AtomicBool::new(false));
let r = Arc::clone(&rolled);
let armed = std::time::Instant::now() - CONFIRM_DEADLINE - Duration::from_secs(1);
let w = TrialWatch::armed(
Arc::new(on_trial(&d).slot),
Box::new(|| false),
armed,
Box::new(move || r.store(true, Ordering::SeqCst)),
);
assert_eq!(w.look().unwrap(), holger_live::ConfirmStep::RollBack);
assert!(rolled.load(Ordering::SeqCst));
let _ = std::fs::remove_dir_all(&d);
}