crates/gunnar-common/src/jobs.rs
//! **A box's own background duty as named jera jobs** (feature `jobs`; A32-14, the owner's rule:
//! background work in a server runs as jera ChainQueue jobs, never a raw thread).
//!
//! [`nordisk_pairing::duty`] is the same contract on a named thread of its own; it stays for a loop whose
//! state is not `Send` (the pairing loop's tunnel driver, the twin follower). Everything else
//! runs here, on `jera::jobs::duty` (owner 2026-11-03: ONE wrapper, in jera, for every consumer;
//! this module was one of three copies): a ticker queues one look at a time on the duty's
//! ChainQueue, and the queue's worker runs it on the runtime's blocking pool.
//!
//! * **Visible.** Every job this process started is in [`running`], with its looks and its
//! failures; its queue's stats count the done and failed looks.
//! * **A failure is said.** A look that PANICS is an `ALERT <kind>` line (through the job's
//! [`Say`]), counted, and a FAILED row; the next look runs as due.
//! * **It stops.** At [`Job::stop`] and [`stop_all`] (the box's clean stop), and when a look
//! answers [`Round::Done`]. The wait between looks ends at the stop.
//! * **The rhythm is the duty's.** The first look runs at once; after a look the next waits the
//! job's period ([`Round::Again`]) or the look's own wait ([`Round::After`], a backoff).
use std::sync::{Arc, Mutex, OnceLock};
use std::time::Duration;
use jera::jobs::JobSink;
pub use nordisk_pairing::duty::{Round, Say};
/// The ledger rows' domain.
const DOMAIN: &str = "gunnar";
/// **One job, started**: jera's duty (its kind, its queue and what it has done).
pub type Job = jera::jobs::duty::Duty;
/// **Every job of this process that still runs.**
pub fn running() -> Vec<Arc<Job>> {
jera::jobs::duty::running()
}
/// The running job of `kind`, if one runs.
pub fn find(kind: &str) -> Option<Arc<Job>> {
jera::jobs::duty::find(kind)
}
/// **Stop every job** (the box's clean stop), each within `grace`. Answers how many ran.
pub fn stop_all(grace: Duration) -> usize {
let n = jera::jobs::duty::stop_all(grace);
// The lanes were cut mid-look: their rows say stopped, not running for ever (RA4-5).
close_orphans("the box stopped before this job ended");
n
}
/// A duty looked at more often than this writes no ledger rows (a row every two seconds for the
/// box's life would be redb commits for nothing); its looks are still counted and its failures said.
pub const RECORD_MIN: Duration = Duration::from_secs(30);
/// **Keep this process's jera ledger under `dir`** (the box's `<data>/jobs`, RA3-5): from now on
/// the job queue's jobs, and the looks of every duty looked at no more often than [`RECORD_MIN`],
/// are rows there (queued, running, then done or FAILED), read by `nornir jobs` and the Jera tab.
pub fn set_ledger(dir: &std::path::Path) -> Result<(), String> {
let store = jera::jobs::JobStore::open(dir).map_err(|e| {
format!(
"the jobs' ledger under {} does not open: {e:#}",
dir.display()
)
})?;
*ledger().lock().unwrap_or_else(|p| p.into_inner()) = Some(Arc::new(store));
let closed = close_orphans("the box restarted before this job ended");
if closed > 0 {
tracing::warn!(
"the jobs' ledger: {closed} row(s) a stopped run left running are marked stopped"
);
}
Ok(())
}
/// **Close the rows a stopped or crashed run left running** (RA4-5): every non-terminal row in
/// the ledger is marked stopped with `why`; jera prunes only finished rows, so a row a killed
/// lane never finished would show running for ever. Answers how many it closed.
pub fn close_orphans(why: &str) -> usize {
let Some(store) = ledger().lock().unwrap_or_else(|p| p.into_inner()).clone() else {
return 0;
};
let open: std::collections::BTreeSet<String> = store
.list(&jera::jobs::JobSelector::All)
.unwrap_or_default()
.into_iter()
.filter(|r| !r.is_terminal())
.map(|r| r.kind)
.collect();
let kinds: Vec<&str> = open.iter().map(String::as_str).collect();
if kinds.is_empty() {
0
} else {
store.stop_orphans(&kinds, why)
}
}
fn ledger() -> &'static Mutex<Option<Arc<jera::jobs::JobStore>>> {
static L: OnceLock<Mutex<Option<Arc<jera::jobs::JobStore>>>> = OnceLock::new();
L.get_or_init(Mutex::default)
}
/// The sink a new lane writes to: the ledger's, or none.
pub fn ledger_sink() -> JobSink {
ledger()
.lock()
.unwrap_or_else(|p| p.into_inner())
.as_ref()
.map(|s| s.sink())
.unwrap_or_else(JobSink::noop)
}
/// The ledger's rows of `kind`, newest last (empty with no ledger).
pub fn ledger_rows(kind: &str) -> Vec<jera::jobs::JobRecord> {
let store = ledger().lock().unwrap_or_else(|p| p.into_inner()).clone();
store
.and_then(|s| {
s.list(&jera::jobs::JobSelector::Kind(kind.to_string()))
.ok()
})
.unwrap_or_default()
}
/// **The runtime a job runs on**: the caller's, or (a caller on a plain thread) this module's own
/// small one, started once. `Err` only when no runtime can be built at all.
pub fn handle() -> Result<tokio::runtime::Handle, String> {
if let Ok(h) = tokio::runtime::Handle::try_current() {
return Ok(h);
}
static OWN: OnceLock<Result<tokio::runtime::Runtime, String>> = OnceLock::new();
OWN.get_or_init(|| {
tokio::runtime::Builder::new_multi_thread()
.worker_threads(2)
.thread_name("gunnar-jobs")
.enable_all()
.build()
.map_err(|e| format!("the jobs' runtime cannot start: {e}"))
})
.as_ref()
.map(|rt| rt.handle().clone())
.map_err(Clone::clone)
}
/// **Start a job of `kind`**: a look at once, then each next look after `every` (or the look's
/// own [`Round::After`]), until it is stopped or a look answers [`Round::Done`]. A job that
/// cannot start is an answer, never a panic.
pub fn start(
kind: &str,
every: Duration,
say: Say,
mut look: impl FnMut() -> Round + Send + 'static,
) -> Result<Arc<Job>, String> {
let rt = handle().map_err(|e| format!("the job `{kind}` cannot start: {e}"))?;
let _on = rt.enter();
// jera's own re-ask (RA3-8: jera 0.2.1 looks at the exclusive gate and never takes it).
let sink = if every >= RECORD_MIN {
ledger_sink()
} else {
JobSink::noop()
};
let rhythm = move |n: u64| if n <= 1 { Duration::ZERO } else { every };
jera::jobs::duty::start(kind, DOMAIN, sink, rhythm, say, move |_| {
Ok(match look() {
Round::Again => jera::jobs::duty::Round::Again,
Round::After(d) => jera::jobs::duty::Round::After(d),
Round::Done => jera::jobs::duty::Round::Done,
})
})
}
#[cfg(test)]
mod tests {
use super::*;
use jera::jobs::chain::{ChainQueue, ChainSpec};
use std::sync::atomic::{AtomicU64, Ordering};
use std::time::Instant;
/// The jobs 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
}
const FAST: Duration = Duration::from_millis(20);
/// ★ **A look that panics is said, counted, a FAILED jera job, and the job goes on** (A32-14).
/// ★ **A Gunnar job is jera's duty** (lane hol-312fix; owner 2026-11-03: ONE wrapper, in
/// jera, for every consumer): the job this module starts is the one `jera::jobs::duty` lists
/// and stops, not a second registry with a second copy of the loop.
#[test]
fn a_gunnar_job_is_jeras_duty() {
let _one = ONE.lock().unwrap_or_else(|p| p.into_inner());
let (say, _) = heard();
let j = start("test-jeras", Duration::from_millis(20), say, || {
Round::Again
})
.expect("a job starts");
assert!(until(|| j.looks() >= 1));
assert!(
jera::jobs::duty::find("test-jeras").is_some(),
"jera lists the job Gunnar started"
);
assert!(stop_all(Duration::from_secs(1)) >= 1);
assert!(jera::jobs::duty::find("test-jeras").is_none());
}
#[test]
fn a_look_that_panics_is_said_counted_failed_and_the_job_goes_on() {
let _one = ONE.lock().unwrap_or_else(|p| p.into_inner());
let (say, lines) = heard();
let mut n = 0;
let j = start("test.panics", FAST, say, move || {
n += 1;
if n == 1 {
panic!("the first look breaks")
}
Round::Again
})
.expect("a job starts");
assert!(
until(|| j.looks() >= 3),
"the looks after the panic run (looks {})",
j.looks()
);
assert_eq!(j.failures(), 1, "the panic is counted");
assert!(
j.last_failure()
.unwrap_or_default()
.contains("the first look breaks"),
"{:?}",
j.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(|| j.queue().stats().failed >= 1 && j.queue().stats().done >= 2),
"the panicked look is a FAILED job, the others DONE: {:?}",
j.queue().stats()
);
j.stop(Duration::from_secs(1));
}
/// ★ **A stopped job looks no more, and is not listed** (the box's clean stop).
#[test]
fn a_stopped_job_looks_no_more() {
let _one = ONE.lock().unwrap_or_else(|p| p.into_inner());
let (say, _) = heard();
let j = start("test.stops", FAST, say, || Round::Again).expect("a job starts");
assert!(until(|| j.looks() >= 1));
assert!(find("test.stops").is_some(), "a running job is listed");
assert!(stop_all(Duration::from_secs(1)) >= 1);
assert!(
j.is_stopped() && find("test.stops").is_none(),
"a stopped job is not listed"
);
std::thread::sleep(Duration::from_millis(100));
let at = j.looks();
std::thread::sleep(Duration::from_millis(200));
assert_eq!(j.looks(), at, "no look runs after the stop");
}
/// ★ **The stop ends a long wait at once** (a period of an hour does not hold the stop).
#[test]
fn the_stop_ends_a_long_wait_at_once() {
let _one = ONE.lock().unwrap_or_else(|p| p.into_inner());
let (say, _) = heard();
let j = start("test.long-wait", Duration::from_secs(3600), say, || {
Round::Again
})
.expect("a job starts");
assert!(until(|| j.looks() == 1), "the first look runs at once");
let t = Instant::now();
assert!(j.stop(Duration::from_secs(2)), "stopped");
assert!(
t.elapsed() < Duration::from_secs(1),
"the stop did not wait out the period"
);
}
/// ★ **Done ends the job; After is the look's own wait** (a retry with a backoff).
#[test]
fn done_ends_the_job_and_after_is_the_looks_own_wait() {
let _one = ONE.lock().unwrap_or_else(|p| p.into_inner());
let (say, _) = heard();
let mut n = 0;
// A period of an hour: only the looks' own short waits can bring the next looks.
let j = start("test.after", Duration::from_secs(3600), say, move || {
n += 1;
if n >= 3 {
Round::Done
} else {
Round::After(Duration::from_millis(10))
}
})
.expect("a job starts");
assert!(
until(|| j.is_stopped()),
"Done stops the job (looks {})",
j.looks()
);
std::thread::sleep(Duration::from_millis(150));
assert_eq!(j.looks(), 3, "no look after Done");
assert!(find("test.after").is_none());
}
/// ★ **Jobs started at the same instant all look at once**, on jera's own 2 s re-ask (jera
/// 0.2.0's gate was a one-permit probe: two lanes asking together, one lost and waited the
/// re-ask; 0.2.1 only looks at the gate, RA3-8).
#[test]
fn jobs_started_at_the_same_instant_all_look_at_once() {
let _one = ONE.lock().unwrap_or_else(|p| p.into_inner());
let looked = Arc::new(AtomicU64::new(0));
let began = Instant::now();
// A server's runtime: many workers, so the lanes' first probes of the gate meet.
let rt = tokio::runtime::Builder::new_multi_thread()
.worker_threads(8)
.enable_all()
.build()
.unwrap();
let together = Arc::new(std::sync::Barrier::new(8));
let starters: Vec<_> = (0..8)
.map(|i| {
let l = Arc::clone(&looked);
let together = Arc::clone(&together);
let h = rt.handle().clone();
std::thread::spawn(move || {
let _on = h.enter();
let (say, _) = heard();
together.wait();
start(
&format!("test.burst-{i}"),
Duration::from_secs(3600),
say,
move || {
l.fetch_add(1, Ordering::SeqCst);
Round::Done
},
)
.expect("a job starts")
})
})
.collect();
let jobs: Vec<_> = starters.into_iter().map(|t| t.join().unwrap()).collect();
assert!(until(|| looked.load(Ordering::SeqCst) == 8));
assert!(
began.elapsed() < Duration::from_millis(1000),
"eight first looks took {:?}",
began.elapsed()
);
drop(jobs);
}
/// ★ **With a ledger, a recorded duty's looks are jera rows, a panicked one a FAILED row**
/// (RA3-5: the lanes wrote to `JobSink::noop()`, so nothing showed in `nornir jobs`).
#[test]
fn with_a_ledger_a_recorded_dutys_looks_are_rows_and_a_panic_a_failed_row() {
let _one = ONE.lock().unwrap_or_else(|p| p.into_inner());
let dir = std::env::temp_dir().join(format!("gunnar-jobs-ledger-{}", std::process::id()));
let _ = std::fs::remove_dir_all(&dir);
set_ledger(&dir).expect("the ledger opens");
let (say, _) = heard();
let mut n = 0;
let j = start("test.recorded", RECORD_MIN, say, move || {
n += 1;
match n {
1 => Round::After(Duration::from_millis(10)),
_ => panic!("the second look breaks"),
}
})
.expect("a job starts");
assert!(
until(|| ledger_rows("test.recorded")
.iter()
.filter(|r| r.status == "done" || r.status == "failed")
.count()
>= 2),
"two finished rows: {:?}",
ledger_rows("test.recorded")
.iter()
.map(|r| r.status.clone())
.collect::<Vec<_>>()
);
let rows = ledger_rows("test.recorded");
assert!(
rows.iter().any(|r| r.status == "done") && rows.iter().any(|r| r.status == "failed"),
"a done row and a FAILED row"
);
j.stop(Duration::from_secs(1));
// A duty looked at every 20 ms writes nothing.
let (say, _) = heard();
let quick = start("test.unrecorded", FAST, say, || Round::Again).expect("a job starts");
assert!(until(|| quick.looks() >= 3));
assert!(
ledger_rows("test.unrecorded").is_empty(),
"a frequent duty writes no rows"
);
quick.stop(Duration::from_secs(1));
}
/// ★ **A row a stopped run left running reads stopped once the ledger opens again** (RA4-5).
#[test]
fn a_row_left_running_reads_stopped_after_the_ledger_opens() {
let _one = ONE.lock().unwrap_or_else(|p| p.into_inner());
let dir = std::env::temp_dir().join(format!("gunnar-jobs-orphan-{}", std::process::id()));
let _ = std::fs::remove_dir_all(&dir);
{
let store = jera::jobs::JobStore::open(&dir).expect("a ledger");
let q = ChainQueue::new(store.sink());
let _ = q.submit(ChainSpec::new(
"test.orphan",
"k",
"test.orphan",
0,
DOMAIN,
serde_json::json!({}),
));
}
set_ledger(&dir).expect("the ledger opens");
let rows = ledger_rows("test.orphan");
assert!(
!rows.is_empty() && rows.iter().all(|r| r.status == "stopped"),
"{:?}",
rows.iter().map(|r| r.status.clone()).collect::<Vec<_>>()
);
}
/// **A job started from a plain thread runs** on the module's own runtime.
#[test]
fn a_job_started_with_no_runtime_runs() {
let _one = ONE.lock().unwrap_or_else(|p| p.into_inner());
let (say, _) = heard();
let j = std::thread::spawn(move || start("test.no-runtime", FAST, say, || Round::Done))
.join()
.unwrap()
.expect("a job starts");
assert!(until(|| j.is_stopped() && j.looks() == 1));
}
/// **A job started on the caller's runtime runs there.**
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
#[allow(clippy::await_holding_lock)] // the registry lock is the point: one test at a time
async fn a_job_started_on_a_runtime_runs_there() {
let _one = ONE.lock().unwrap_or_else(|p| p.into_inner());
let (say, _) = heard();
let j = start("test.on-runtime", FAST, say, || Round::Done).expect("a job starts");
let t = Instant::now();
while !(j.is_stopped() && j.looks() == 1) && t.elapsed() < Duration::from_secs(10) {
tokio::time::sleep(Duration::from_millis(10)).await;
}
assert!(j.is_stopped() && j.looks() == 1);
}
}