estate/holger-live/src/duty.rs
//! **A box's own background duty as jera jobs** (feature `jobs`; A28 M18, the owner's rule:
//! server background work runs as jera ChainQueue jobs, never a raw thread or a bare task).
//!
//! A duty is a look that is due again and again: a watch (the memory line, the pair status),
//! a retry (the pairing), the front's role. The loop itself is `jera::jobs::duty` (owner
//! 2026-11-03: ONE wrapper, in jera, for every consumer; this module was one of three copies).
//! What stays here is Holger's spelling of it: a look answers [`Look`] with an `anyhow` error,
//! a duty's ledger is a jera store at a path, and [`stop_all`] is the box's clean stop.
//!
//! * **Visible.** Every duty this process started is in [`running`], with its looks and its
//! failures; with a ledger, each look is one jera row of the duty's kind.
//! * **A failure is said.** A look that fails or PANICS is an `ALERT <kind>` line (through the
//! duty's [`Say`]), counted, and a FAILED row; the next look runs as due.
//! * **It stops.** At [`Duty::stop`] and [`stop_all`] (the box's clean stop), and when a look
//! answers [`Look::Done`].
use std::path::Path;
use std::sync::Arc;
use std::time::Duration;
use jera::jobs::duty::Round;
use jera::jobs::JobSink;
pub use jera::jobs::duty::{find, running, Duty, Say};
/// How long the box's clean stop gives each duty's look in flight.
pub const STOP_GRACE: Duration = Duration::from_secs(5);
/// The jera domain of a Holger box's duties (their ledger rows).
const DOMAIN: &str = "holger";
/// What a look answers: due again, or the duty is done (a retry that succeeded).
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum Look {
Again,
Done,
}
/// **Stop every duty** (the box's clean stop), each within [`STOP_GRACE`]. Answers how many ran.
pub fn stop_all() -> usize {
jera::jobs::duty::stop_all(STOP_GRACE)
}
/// **Start a duty** on the current runtime. Before look `n` (from 1) it waits `every(n)`; with
/// a `ledger` each look is a jera row there (a ledger that cannot open is said, and the duty
/// runs unrecorded). No runtime: an answer, never a panic.
pub fn start(
kind: &str,
ledger: Option<&Path>,
every: impl Fn(u64) -> Duration + Send + Sync + 'static,
say: Say,
look: impl Fn(u64) -> anyhow::Result<Look> + Send + Sync + 'static,
) -> Result<Arc<Duty>, String> {
if tokio::runtime::Handle::try_current().is_err() {
return Err(format!("the duty `{kind}` has no runtime to run on"));
}
let sink = match ledger.map(jera::jobs::JobStore::open) {
None => JobSink::noop(),
Some(Ok(store)) => store.sink(),
Some(Err(e)) => {
say(&format!(
"ALERT {kind} its ledger cannot open ({e:#}); the duty runs, unrecorded"
));
JobSink::noop()
}
};
jera::jobs::duty::start(kind, DOMAIN, sink, every, say, move |n| match look(n) {
Ok(Look::Again) => Ok(Round::Again),
Ok(Look::Done) => Ok(Round::Done),
Err(e) => Err(format!("{e:#}")),
})
}
#[cfg(test)]
mod tests {
use super::*;
use std::sync::Mutex;
use std::time::Instant;
/// The duties share one registry, and [`stop_all`] stops every one: one test at a time.
static ONE: Mutex<()> = Mutex::new(());
fn heard() -> (Say, Arc<Mutex<Vec<String>>>) {
let lines = Arc::new(Mutex::new(Vec::new()));
let l = Arc::clone(&lines);
(
Arc::new(move |s: &str| l.lock().unwrap().push(s.to_string())),
lines,
)
}
fn until(what: impl Fn() -> bool) -> bool {
let t = Instant::now();
while t.elapsed() < Duration::from_secs(10) {
if what() {
return true;
}
std::thread::sleep(Duration::from_millis(10));
}
false
}
fn fast(_: u64) -> Duration {
Duration::from_millis(20)
}
/// ★ **A look that panics is said, counted, and the duty goes on** (M18): the raw thread
/// it replaces ended at the panic, silently, and the watch with it.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn a_look_that_panics_is_said_counted_and_the_duty_goes_on() {
let _one = ONE.lock().unwrap_or_else(|p| p.into_inner());
let (say, lines) = heard();
let d = start("test.panics", None, fast, say, |n| {
if n == 1 {
panic!("the first look breaks")
} else {
Ok(Look::Again)
}
})
.expect("a duty starts");
assert!(
until(|| d.looks() >= 3),
"the looks after the panic run (looks {})",
d.looks()
);
assert_eq!(d.failures(), 1, "the panic is counted");
assert!(
d.last_failure()
.unwrap_or_default()
.contains("the first look breaks"),
"the panic's words are kept: {:?}",
d.last_failure()
);
let said = lines.lock().unwrap().join("\n");
assert!(
said.contains("ALERT test.panics") && said.contains("panicked"),
"the panic is an ALERT line: {said:?}"
);
assert!(
until(|| d.queue().stats().failed >= 1),
"the panicked look is a FAILED job"
);
d.stop(STOP_GRACE);
}
/// ★ **A failed look is said and is a FAILED jera job**; the duty goes on.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn a_failed_look_is_said_and_is_a_failed_job() {
let _one = ONE.lock().unwrap_or_else(|p| p.into_inner());
let (say, lines) = heard();
let d = start("test.fails", None, fast, say, |n| {
if n == 1 {
Err(anyhow::anyhow!("no answer"))
} else {
Ok(Look::Again)
}
})
.expect("a duty starts");
assert!(
until(|| d.looks() >= 2 && d.queue().stats().done >= 1),
"the duty goes on after a failed look"
);
assert_eq!((d.failures(), d.queue().stats().failed), (1, 1));
assert!(
lines
.lock()
.unwrap()
.iter()
.any(|l| l.contains("ALERT test.fails") && l.contains("no answer")),
"the failure is said"
);
d.stop(STOP_GRACE);
}
/// ★ **A stopped duty looks no more** (the box's clean stop): the raw thread looked on for
/// the life of the process, whatever was asked.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn a_stopped_duty_looks_no_more() {
let _one = ONE.lock().unwrap_or_else(|p| p.into_inner());
let (say, _) = heard();
let d = start("test.stops", None, fast, say, |_| Ok(Look::Again)).expect("a duty starts");
assert!(until(|| d.looks() >= 1));
assert!(find("test.stops").is_some(), "a running duty is listed");
assert!(stop_all() >= 1);
assert!(
d.is_stopped() && find("test.stops").is_none(),
"a stopped duty is not listed"
);
std::thread::sleep(Duration::from_millis(100));
let at = d.looks();
std::thread::sleep(Duration::from_millis(200));
assert_eq!(d.looks(), at, "no look runs after the stop");
}
/// ★ **A look that answers Done ends the duty** (a retry that succeeded).
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn a_done_look_ends_the_duty() {
let _one = ONE.lock().unwrap_or_else(|p| p.into_inner());
let (say, _) = heard();
let d = start("test.done", None, fast, say, |n| {
Ok(if n >= 2 { Look::Done } else { Look::Again })
})
.expect("a duty starts");
assert!(until(|| d.is_stopped()), "Done stops the duty");
std::thread::sleep(Duration::from_millis(150));
assert_eq!(d.looks(), 2, "no look after Done");
}
/// ★ **A ledger keeps each look as a jera row of the duty's kind.**
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn a_ledger_keeps_each_look_as_a_row() {
let _one = ONE.lock().unwrap_or_else(|p| p.into_inner());
let dir = std::env::temp_dir().join(format!(
"holger-live-duty-{}-{}",
std::process::id(),
line!()
));
let _ = std::fs::remove_dir_all(&dir);
let (say, _) = heard();
let d = start("test.ledger", Some(&dir), fast, say, |n| {
Ok(if n >= 2 { Look::Done } else { Look::Again })
})
.expect("a duty starts");
assert!(until(|| d.is_stopped() && d.queue().stats().done >= 2));
assert!(
jera::jobs::store_path(&dir).is_file(),
"the ledger is written at {}",
dir.display()
);
let _ = std::fs::remove_dir_all(&dir);
}
/// ★ **A Holger duty is jera's duty** (lane hol-312fix; owner 2026-11-03: ONE wrapper, in
/// jera, for every consumer). The duty this module starts is the one `jera::jobs::duty`
/// lists and stops: no second registry, no second copy of the loop.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn a_holger_duty_is_jeras_duty() {
let _one = ONE.lock().unwrap_or_else(|p| p.into_inner());
let (say, _) = heard();
let d = start("test.jeras", None, fast, say, |_| Ok(Look::Again)).expect("a duty starts");
assert!(until(|| d.looks() >= 1));
assert!(
jera::jobs::duty::find("test.jeras").is_some(),
"jera lists the duty Holger started"
);
assert!(stop_all() >= 1);
assert!(jera::jobs::duty::find("test.jeras").is_none());
}
/// **No runtime is an answer**, never a panic.
#[test]
fn no_runtime_is_an_answer() {
let _one = ONE.lock().unwrap_or_else(|p| p.into_inner());
let (say, _) = heard();
let r = start("test.no-runtime", None, fast, say, |_| Ok(Look::Done));
assert!(
r.err().unwrap_or_default().contains("runtime"),
"refused by name"
);
}
}