estate/holger-live/src/trial.rs
//! **The trial confirm as jera jobs** (feature `jobs`, L21 2026-10-18; the owner's rule: server
//! background work runs as jera ChainQueue jobs with priority, never a raw thread or a bare task).
//! A box (main, twin, the front) that re-exec'd into a live update runs its new app on TRIAL. A
//! ticker queues one [`KIND_TRIAL`] look per [`LOOK_EVERY`] at [`Priority::High`] (the box's own
//! health, before anything else it does) under one idempotency key per generation, as the front's
//! `front.recharge`; each look is one ledger row: `wait`, `confirm` (served for
//! [`crate::CONFIRM_AFTER`]: the slot is confirmed) or `roll-back` (not serving by
//! [`crate::CONFIRM_DEADLINE`]: the re-exec hook runs and korp-start rolls the slot back). A box
//! not on trial queues nothing; a finished trial queues no more looks.
use std::path::Path;
use std::sync::{Arc, Mutex};
use std::time::Duration;
use jera::jobs::{ChainQueue, ChainSpec, Priority, Submitted};
use crate::{ConfirmStep, LiveSlot};
/// The ledger kind of one look at the trial.
pub const KIND_TRIAL: &str = "live.trial";
/// How often the trial is looked at.
pub const LOOK_EVERY: Duration = Duration::from_secs(5);
/// Where a box keeps its live jobs, under its data root (its own ledger: the server's other
/// ledgers, `fleet/jobs` and the front's `jobs`, are opened by their own duties).
pub const JOBS_REL: &str = "live/jobs";
/// Whether the app serves now (a blocking probe: runs on the queue's blocking thread).
pub type Serving = Box<dyn Fn() -> bool + Send + Sync>;
/// How long this process has run.
pub type Clock = Box<dyn Fn() -> Duration + Send + Sync>;
/// The re-exec a rollback asks for (korp-start then boots the other slot).
pub type RollBack = Box<dyn Fn() + Send + Sync>;
/// **One trial, watched**: the slot, how serving is probed, the clock, the rollback hook, and
/// since when the app serves.
pub struct TrialWatch {
slot: Arc<LiveSlot>,
serving: Serving,
clock: Clock,
roll_back: RollBack,
since: Mutex<Option<Duration>>,
finished: Mutex<bool>,
}
impl TrialWatch {
pub fn new(
slot: Arc<LiveSlot>,
serving: Serving,
clock: Clock,
roll_back: RollBack,
) -> TrialWatch {
TrialWatch {
slot,
serving,
clock,
roll_back,
since: Mutex::new(None),
finished: Mutex::new(false),
}
}
/// **A watch whose clock started at `armed`** (lane B3, A9 row 6): the app's first line
/// (korp `live::arm_trial`), not the moment the watch starts, so the deadline holds even
/// when the start waited (a console token, a refused declaration retried).
pub fn armed(
slot: Arc<LiveSlot>,
serving: Serving,
armed: std::time::Instant,
roll_back: RollBack,
) -> TrialWatch {
TrialWatch::new(slot, serving, Box::new(move || armed.elapsed()), roll_back)
}
/// The slot watched.
pub fn slot(&self) -> &Arc<LiveSlot> {
&self.slot
}
/// Whether nothing is left to look at (confirmed, rolled back, or never on trial).
pub fn finished(&self) -> bool {
*self.finished.lock().unwrap_or_else(|e| e.into_inner()) || !self.slot.on_trial()
}
/// **One look**: has the app served, then korp's step (the confirm is written by
/// [`LiveSlot::confirm_step`]); a rollback runs the re-exec hook. A finished trial looks no more.
pub fn look(&self) -> Result<ConfirmStep, String> {
if self.finished() {
return Ok(ConfirmStep::Nothing);
}
let up = (self.clock)();
let since = {
let mut since = self.since.lock().map_err(|e| e.to_string())?;
if since.is_none() && (self.serving)() {
*since = Some(up);
}
*since
};
let step = self.slot.confirm_step(up, since);
let done = |d: bool| *self.finished.lock().unwrap_or_else(|e| e.into_inner()) = d;
match step {
Ok(ConfirmStep::Confirm) | Ok(ConfirmStep::Nothing) => done(true),
Ok(ConfirmStep::RollBack) => {
done(true);
(self.roll_back)();
}
Ok(ConfirmStep::Wait) => {}
// A confirm NOT written: never looked at again — the next start rolls the slot back.
Err(_) => done(true),
}
step
}
}
/// The ledger word of a step.
pub fn word(step: ConfirmStep) -> &'static str {
match step {
ConfirmStep::Nothing => "nothing",
ConfirmStep::Wait => "wait",
ConfirmStep::Confirm => "confirm",
ConfirmStep::RollBack => "roll-back",
}
}
/// The look of `generation`, at High priority (the box's own health drains first).
pub fn spec(generation: i64) -> ChainSpec {
ChainSpec::new(
KIND_TRIAL,
format!("{KIND_TRIAL}-{generation}"),
KIND_TRIAL,
generation,
"holger-live",
serde_json::json!({ "generation": generation }),
)
.with_target("the live slot on trial")
.with_priority(Priority::High)
}
/// Queue the look of `generation` (one per generation: a resubmit is a no-op).
pub fn submit(q: &Arc<ChainQueue>, generation: i64) -> Submitted {
q.submit(spec(generation))
}
/// Queue the look of `generation` while the trial is not finished; `None` once it is.
pub fn submit_while(q: &Arc<ChainQueue>, w: &TrialWatch, generation: i64) -> Option<Submitted> {
(!w.finished()).then(|| submit(q, generation))
}
/// **The job**: one look, its step as the ledger row's detail. A confirm that could not be
/// written is the job's failure (the row says why; the next start rolls the slot back).
pub fn run(w: &TrialWatch, spec: &ChainSpec) -> anyhow::Result<serde_json::Value> {
if spec.kind != KIND_TRIAL {
anyhow::bail!("no live job {:?}", spec.kind);
}
let step = w.look().map_err(|e| {
anyhow::anyhow!("the confirm was NOT written ({e}): the next start rolls this slot back")
})?;
Ok(
serde_json::json!({ "step": word(step), "slot": w.slot.running().map(|r| r.slot).unwrap_or_default() }),
)
}
/// **Start the trial's jobs** on the current runtime: nothing on trial, nothing starts (`None`).
/// On trial: the ledger at `jobs_dir`, its ChainQueue's worker, and the ticker that queues one
/// look per [`LOOK_EVERY`] until the trial is finished. The clock starts now ([`start_armed`]:
/// at the arming).
pub fn start(
slot: Arc<LiveSlot>,
serving: Serving,
jobs_dir: &Path,
roll_back: RollBack,
) -> Option<Arc<ChainQueue>> {
start_armed(
slot,
serving,
jobs_dir,
roll_back,
std::time::Instant::now(),
)
}
/// [`start`] with the clock started at `armed` (the app's first line, lane B3).
pub fn start_armed(
slot: Arc<LiveSlot>,
serving: Serving,
jobs_dir: &Path,
roll_back: RollBack,
armed: std::time::Instant,
) -> Option<Arc<ChainQueue>> {
if !slot.on_trial() {
return None;
}
let store = match jera::jobs::JobStore::open(jobs_dir) {
Ok(s) => s,
Err(e) => {
eprintln!("holger-live: the live job ledger {} cannot open ({e:#}): the trial is not confirmed — the next start rolls this slot back", jobs_dir.display());
return None;
}
};
let q = ChainQueue::new(store.sink());
let w = Arc::new(TrialWatch::armed(slot, serving, armed, roll_back));
let handle = tokio::runtime::Handle::try_current().ok()?;
let worker = Arc::clone(&w);
// A32-14, stays a task: the trial lane's jera worker.
handle.spawn(Arc::clone(&q).run(move |spec| run(&worker, &spec)));
let (ticker, tq) = (Arc::clone(&w), Arc::clone(&q));
// The ticker only queues; the looks run on the queue's worker (the sink holds the ledger).
// A32-14, stays a task: the trial lane's ticker (it submits the jera looks).
handle.spawn(async move {
let mut generation = 0i64;
loop {
tokio::time::sleep(LOOK_EVERY).await;
generation += 1;
if submit_while(&tq, &ticker, generation).is_none() {
return;
}
}
});
Some(q)
}