crates/monetize-server/src/jobs.rs
//! **The server's background work as named jera jobs** (A32-14, the owner's rule: background work
//! in a server is a jera ChainQueue job, never a raw thread).
//!
//! Two shapes, both on jera ChainQueue lanes on the caller's runtime (or, for a caller on a plain
//! thread, this module's own small one):
//!
//! * [`once`]: **one piece of work for one subject** (a wallet reading, a grow drive, a webhook's
//! settle). Its lane is the subject's own, so two subjects never wait on each other, and a
//! second ask for a subject whose work still runs is that work (`Ok(false)`), not a second
//! copy. Its lane goes when its work ends.
//! * [`every`]: **a duty that is due again and again** (a keeper, a watcher): a look at once, then
//! each next look after its period, or the look's own [`Round::After`], until it answers
//! [`Round::Done`] or is stopped.
//!
//! Either way the work is visible ([`running`]), a failure or a PANIC is an `ALERT jobs.failed <kind>` line
//! and a FAILED jera row (the next look runs as due), and [`stop_all`] (the server's clean stop)
//! ends every duty and lets no queued work start. A caller whose flag must be put back on every
//! path (a `running` bit) holds a [`Finally`] in its work: it runs after a panic too.
use std::collections::BTreeMap;
use std::panic::{catch_unwind, AssertUnwindSafe};
use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
use std::sync::{Arc, Mutex, OnceLock};
use std::time::{Duration, Instant};
use jera::jobs::chain::{ChainQueue, ChainSpec, Priority};
use jera::jobs::JobSink;
/// The ledger rows' domain.
const DOMAIN: &str = "monetize";
/// What a look of a duty answers.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum Round {
/// Due again after the duty's period.
Again,
/// Due again after this wait (a backoff).
After(Duration),
/// The duty is done.
Done,
}
/// **Runs its closure when dropped** — on every path out of a job's work, a panic included.
pub struct Finally<F: FnMut()>(pub F);
impl<F: FnMut()> Drop for Finally<F> {
fn drop(&mut self) {
(self.0)()
}
}
/// **One job, started**: its kind, its subject, its queue and what it has done.
pub struct Job {
kind: String,
key: String,
queue: Arc<ChainQueue>,
stopped: AtomicBool,
in_look: AtomicBool,
looks: AtomicU64,
failures: AtomicU64,
last_failure: Mutex<Option<String>>,
wake: tokio::sync::Notify,
tasks: Mutex<Vec<tokio::task::JoinHandle<()>>>,
}
impl Job {
/// The job's kind (its jera rows' kind).
pub fn kind(&self) -> &str {
&self.kind
}
/// The job's subject (`""` for a duty).
pub fn key(&self) -> &str {
&self.key
}
/// The looks (or the one run) so far, the failed ones too.
pub fn looks(&self) -> u64 {
self.looks.load(Ordering::SeqCst)
}
/// The looks that failed or panicked.
pub fn failures(&self) -> u64 {
self.failures.load(Ordering::SeqCst)
}
/// The last failure, in words.
pub fn last_failure(&self) -> Option<String> {
self.last_failure
.lock()
.unwrap_or_else(|p| p.into_inner())
.clone()
}
/// Stopped: asked to, done, or (a [`once`]) its work ended.
pub fn is_stopped(&self) -> bool {
self.stopped.load(Ordering::SeqCst)
}
/// The job's jera queue (its stats: done, failed).
pub fn queue(&self) -> &Arc<ChainQueue> {
&self.queue
}
/// **Stop**, waiting up to `grace` for the look in flight. True when no look runs any more.
pub fn stop(&self, grace: Duration) -> bool {
self.tell_stop();
let until = Instant::now() + grace;
while self.in_look.load(Ordering::SeqCst) {
if Instant::now() >= until {
return false;
}
std::thread::sleep(Duration::from_millis(5));
}
true
}
fn tell_stop(&self) {
self.stopped.store(true, Ordering::SeqCst);
self.wake.notify_waiters();
for t in self
.tasks
.lock()
.unwrap_or_else(|p| p.into_inner())
.drain(..)
{
t.abort();
}
}
fn new(kind: &str, key: &str, sink: JobSink) -> Arc<Job> {
Arc::new(Job {
kind: kind.to_string(),
key: key.to_string(),
// RA3-8: jera's own re-ask; since jera 0.2.1 a look at the exclusive gate never takes
// it, so lanes starting at the same instant all start at once.
queue: ChainQueue::new(sink),
stopped: AtomicBool::new(false),
in_look: AtomicBool::new(false),
looks: AtomicU64::new(0),
failures: AtomicU64::new(0),
last_failure: Mutex::new(None),
wake: tokio::sync::Notify::new(),
tasks: Mutex::new(Vec::new()),
})
}
/// One look, a panic caught: counted and, failed, said. `Err` is the FAILED row's words.
fn look<T>(&self, n: u64, f: impl FnOnce() -> Result<T, String>) -> Result<T, String> {
self.in_look.store(true, Ordering::SeqCst);
self.looks.fetch_add(1, Ordering::SeqCst);
let out = match catch_unwind(AssertUnwindSafe(f)) {
Ok(r) => r,
Err(p) => Err(format!("panicked: {}", panic_words(p.as_ref()))),
};
self.in_look.store(false, Ordering::SeqCst);
if let Err(why) = &out {
self.failures.fetch_add(1, Ordering::SeqCst);
*self.last_failure.lock().unwrap_or_else(|p| p.into_inner()) = Some(why.clone());
let subject = if self.key.is_empty() {
String::new()
} else {
format!(" ({})", self.key)
};
eprintln!("monetize-server: ALERT jobs.failed {}{subject}: look {n} failed: {why}; the next runs as due", self.kind);
}
out
}
}
/// A panic's words.
fn panic_words(p: &(dyn std::any::Any + Send)) -> String {
p.downcast_ref::<&str>()
.map(|s| s.to_string())
.or_else(|| p.downcast_ref::<String>().cloned())
.unwrap_or_else(|| "(no words)".to_string())
}
fn registry() -> &'static Mutex<Vec<Arc<Job>>> {
static R: OnceLock<Mutex<Vec<Arc<Job>>>> = OnceLock::new();
R.get_or_init(Mutex::default)
}
/// The one-shot jobs whose work still runs, by `kind/key`.
fn subjects() -> &'static Mutex<BTreeMap<String, Arc<Job>>> {
static S: OnceLock<Mutex<BTreeMap<String, Arc<Job>>>> = OnceLock::new();
S.get_or_init(Mutex::default)
}
/// Set by [`stop_all`]: no new work starts after the server's clean stop.
static STOPPING: AtomicBool = AtomicBool::new(false);
/// **Every job of this process that still runs.**
pub fn running() -> Vec<Arc<Job>> {
registry()
.lock()
.unwrap_or_else(|p| p.into_inner())
.iter()
.filter(|j| !j.is_stopped())
.cloned()
.collect()
}
/// The running job of `kind` (and subject `key`, `""` for a duty), if one runs.
pub fn find(kind: &str, key: &str) -> Option<Arc<Job>> {
running()
.into_iter()
.find(|j| j.kind == kind && j.key == key)
}
/// **Stop every job** (the server's clean stop), all within the one `grace`; nothing new starts
/// after it. Answers how many ran.
pub fn stop_all(grace: Duration) -> usize {
STOPPING.store(true, Ordering::SeqCst);
let all = running();
for j in &all {
j.tell_stop();
}
let until = Instant::now() + grace;
for j in &all {
j.stop(until.saturating_duration_since(Instant::now()));
}
// The lanes were cut mid-look: their rows say stopped, not running for ever (RA4-5).
close_orphans("the server stopped before this job ended");
all.len()
}
/// Let work start again after [`stop_all`] (an in-process test that serves twice).
pub fn resume() {
STOPPING.store(false, Ordering::SeqCst);
}
/// A duty looked at more often than this writes no ledger rows (a row every few seconds for the
/// server's life would be redb commits for nothing); its looks are still counted and failures said.
pub const RECORD_MIN: Duration = Duration::from_secs(30);
/// **Keep this process's jera ledger under `dir`** (the server's `<data>/jobs`, RA3-5): from then
/// on every one-shot job, and the looks of every duty looked at no more often than
/// [`RECORD_MIN`], are rows there (queued, running, then done or FAILED).
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 server restarted before this job ended");
if closed > 0 {
eprintln!("monetize-server: the jobs' ledger: {closed} row(s) a stopped run left running are marked stopped");
}
Ok(())
}
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)
}
/// **A lane's sink that never floods the ledger** (RA4-4, RA4-6): a FAILED row is always written;
/// any other row of a (kind, subject) at most once per [`RECORD_MIN`] (a job picked keeps every
/// one of its rows, queued to done). A per-tick lane (the estate drives, the custody step, 2 s
/// apart) would otherwise fill jera's capped table and push the failures it should keep out.
/// `what` names the lane on its failure line.
pub fn lane_sink(what: &'static str) -> JobSink {
lane_sink_to(what, ledger_sink)
}
/// [`lane_sink`] writing to `to()` (the ledger's sink; a test's own).
fn lane_sink_to(what: &'static str, to: impl Fn() -> JobSink + Send + Sync + 'static) -> JobSink {
JobSink::remote(move |rec: &jera::jobs::JobRecord| {
// RA4-25: the decision and its write are ONE step. jera's submitter and the job's worker
// write the same job's rows from two threads; decided under the locks but written after
// them, a late QUEUED row was written over the done row it had been checked against.
// redb takes one write at a time anyway, so this costs no parallel write.
static WRITE: Mutex<()> = Mutex::new(());
let _write = WRITE.lock().unwrap_or_else(|p| p.into_inner());
// Process-wide: every lane (and every one-shot job's own lane) samples one subject once.
// (when each subject last wrote a row, the jobs picked to keep every row)
type Picked = (
BTreeMap<String, Instant>,
std::collections::BTreeSet<String>,
);
static PICKED: OnceLock<Mutex<Picked>> = OnceLock::new();
let picked = PICKED.get_or_init(Mutex::default);
// jera's submit writes a job's QUEUED row after the job is visible to its worker, so a
// quick job's terminal row can come first: a late non-terminal row never overwrites it.
static ENDED: OnceLock<Mutex<std::collections::VecDeque<String>>> = OnceLock::new();
let ended = ENDED.get_or_init(Mutex::default);
{
let mut e = ended.lock().unwrap_or_else(|p| p.into_inner());
if rec.is_terminal() {
e.push_back(rec.job_id.clone());
while e.len() > 4096 {
e.pop_front();
}
} else if e.contains(&rec.job_id) {
return;
}
}
if rec.status == "failed" {
eprintln!(
"monetize-server: {what} job {} failed: {}",
rec.job_id, rec.detail_json
);
}
let take = {
let mut g = picked.lock().unwrap_or_else(|p| p.into_inner());
let subject = format!("{}|{}", rec.kind, rec.target);
let take = rec.status == "failed"
|| g.1.contains(&rec.job_id)
|| g.0.get(&subject).is_none_or(|t| t.elapsed() >= RECORD_MIN);
if take && !g.1.contains(&rec.job_id) {
g.0.insert(subject, Instant::now());
g.1.insert(rec.job_id.clone());
}
if rec.is_terminal() {
g.1.remove(&rec.job_id);
}
take
};
if take {
to().record(rec);
}
})
}
/// **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)
}
}
/// The ledger's rows of `kind` (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 this module's own small one.
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("monetize-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)
}
fn listed(job: &Arc<Job>) {
let mut all = registry().lock().unwrap_or_else(|p| p.into_inner());
all.retain(|j| !j.is_stopped());
all.push(Arc::clone(job));
}
/// **The subject `key` of `state`**: a job's key within one server's state. One server runs per
/// process, but tests run many side by side, each with the same product names: without the
/// state in the key one test's drive would answer for another's.
pub fn of(state: &crate::state::AppState, key: &str) -> String {
format!(
"{key}@{:x}",
state as *const crate::state::AppState as usize
)
}
/// **Run `work` for `key` as a jera job of `kind`**, on the subject's own lane. `Ok(true)`: it
/// is queued; `Ok(false)`: work for `key` still runs, and that is the answer; `Err`: it cannot
/// start (no runtime, or the server is stopping), and nothing ran.
pub fn once(
kind: &str,
key: &str,
work: impl FnOnce() -> Result<(), String> + Send + 'static,
) -> Result<bool, String> {
if STOPPING.load(Ordering::SeqCst) {
return Err(format!(
"the job `{kind}` ({key}) does not start: the server is stopping"
));
}
let rt = handle().map_err(|e| format!("the job `{kind}` ({key}) cannot start: {e}"))?;
let id = format!("{kind}/{key}");
let job = {
let mut subjects = subjects().lock().unwrap_or_else(|p| p.into_inner());
if subjects.contains_key(&id) {
return Ok(false);
}
let job = Job::new(kind, key, lane_sink("a one-shot"));
subjects.insert(id.clone(), Arc::clone(&job));
job
};
let work = Mutex::new(Some(work));
let j = Arc::clone(&job);
let worker = rt.spawn(Arc::clone(&job.queue).run(move |_spec| {
let out = match work.lock().unwrap_or_else(|p| p.into_inner()).take() {
Some(work) => j.look(1, work),
None => Ok(()),
};
// The subject is free again at once (the next ask runs anew); the JOB is gone only once
// its terminal row is written (the reaper below).
subjects()
.lock()
.unwrap_or_else(|p| p.into_inner())
.remove(&format!("{}/{}", j.kind, j.key));
j.wake.notify_waiters();
out.map(|()| serde_json::json!({ "key": j.key }))
.map_err(|why| anyhow::anyhow!(why))
}));
// The lane ends once its one job's TERMINAL ROW is written (the queue counts it done or failed
// only after that write): then the subject is free, the job is gone, and its worker ends.
let (j, q) = (Arc::clone(&job), Arc::clone(&job.queue));
let reaper = rt.spawn(async move {
loop {
let st = q.stats();
if st.done + st.failed >= 1 || j.is_stopped() {
break;
}
let _ = tokio::time::timeout(Duration::from_millis(20), j.wake.notified()).await;
}
j.stopped.store(true, Ordering::SeqCst);
for t in j.tasks.lock().unwrap_or_else(|p| p.into_inner()).drain(..) {
t.abort();
}
});
job.tasks
.lock()
.unwrap_or_else(|p| p.into_inner())
.push(worker);
// A32-14, stays a task: the lane's reaper, bounded by its one job (it ends the lane's worker).
drop(reaper);
listed(&job);
let spec = ChainSpec::new(
kind,
id.clone(),
id,
0,
DOMAIN,
serde_json::json!({ "key": key }),
)
.with_target(format!("{kind} {key}"))
.with_priority(Priority::Normal);
let _ = job.queue.submit(spec);
Ok(true)
}
/// **Start a duty 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 failed or
/// panicked look is said and the next runs after `every`.
pub fn every(
kind: &str,
every: Duration,
look: impl FnMut() -> Result<Round, String> + Send + 'static,
) -> Result<Arc<Job>, String> {
if STOPPING.load(Ordering::SeqCst) {
return Err(format!(
"the duty `{kind}` does not start: the server is stopping"
));
}
let rt = handle().map_err(|e| format!("the duty `{kind}` cannot start: {e}"))?;
let job = Job::new(
kind,
"",
if every >= RECORD_MIN {
ledger_sink()
} else {
JobSink::noop()
},
);
// The look's answer goes back to the ticker, which waits for it before it times the next.
let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel::<Round>();
let look = Mutex::new(look);
let j = Arc::clone(&job);
let worker = rt.spawn(Arc::clone(&job.queue).run(move |spec| {
let n = u64::try_from(spec.generation).unwrap_or(0);
if j.is_stopped() {
let _ = tx.send(Round::Done);
return Ok(serde_json::json!({ "look": n, "ran": false }));
}
match j.look(n, || (look.lock().unwrap_or_else(|p| p.into_inner()))()) {
Ok(round) => {
let _ = tx.send(round);
Ok(serde_json::json!({ "look": n, "done": round == Round::Done }))
}
Err(why) => {
let _ = tx.send(Round::Again);
Err(anyhow::anyhow!(why))
}
}
}));
let j = Arc::clone(&job);
let ticker = rt.spawn(async move {
let mut n = 0u64;
loop {
n += 1;
let generation = i64::try_from(n).unwrap_or(i64::MAX);
let spec = ChainSpec::new(
j.kind.as_str(),
format!("{}-{n}", j.kind),
j.kind.as_str(),
generation,
DOMAIN,
serde_json::json!({ "look": n }),
)
.with_target(format!("look {n} of {}", j.kind))
.with_priority(Priority::Normal);
let _ = j.queue.submit(spec);
let next = match rx.recv().await {
Some(Round::Done) | None => {
j.stopped.store(true, Ordering::SeqCst);
return;
}
Some(Round::Again) => every,
Some(Round::After(d)) => d,
};
tokio::select! {
_ = tokio::time::sleep(next) => {}
_ = j.wake.notified() => {}
}
if j.is_stopped() {
return;
}
}
});
job.tasks
.lock()
.unwrap_or_else(|p| p.into_inner())
.extend([worker, ticker]);
listed(&job);
Ok(job)
}
#[cfg(test)]
mod tests {
use super::*;
/// The jobs share one registry and one stop: one test at a time.
static ONE: Mutex<()> = Mutex::new(());
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(5));
}
false
}
const FAST: Duration = Duration::from_millis(20);
/// jera's own re-ask of its exclusive gate (2 s by default, RA3-8).
const JERA_REASK: Duration = Duration::from_secs(2);
/// ★ **A one-shot job runs once, on its subject's lane, and is gone after.**
#[test]
fn a_one_shot_job_runs_once_and_is_gone() {
let _one = ONE.lock().unwrap_or_else(|p| p.into_inner());
let ran = Arc::new(AtomicU64::new(0));
let r = Arc::clone(&ran);
assert_eq!(
once("test.once", "a", move || {
r.fetch_add(1, Ordering::SeqCst);
Ok(())
}),
Ok(true)
);
assert!(
until(|| ran.load(Ordering::SeqCst) == 1 && find("test.once", "a").is_none()),
"ran once and is gone"
);
}
/// ★ **Jobs started at the same instant all start at once** (jera's exclusive gate is a
/// one-permit probe: two lanes asking together, one lost and waited the queue's re-ask,
/// 2 s by default — as long as a shop window waits for its reading).
#[test]
fn jobs_started_at_the_same_instant_all_start_at_once() {
let _one = ONE.lock().unwrap_or_else(|p| p.into_inner());
// RA3-8: many lanes, released together by a barrier, five times: a look at the gate
// that takes it makes some lane lose and sleep its whole re-ask (2 s by default).
const LANES: u64 = 64;
for round in 0..5 {
let ran = Arc::new(AtomicU64::new(0));
let gate = Arc::new(std::sync::Barrier::new(LANES as usize));
let began = Instant::now();
let starters: Vec<_> = (0..LANES)
.map(|i| {
let (r, gate) = (Arc::clone(&ran), Arc::clone(&gate));
std::thread::spawn(move || {
gate.wait();
once("test.burst", &format!("r{round}k{i}"), move || {
r.fetch_add(1, Ordering::SeqCst);
Ok(())
})
})
})
.collect();
for s in starters {
assert_eq!(s.join().unwrap(), Ok(true));
}
assert!(until(|| ran.load(Ordering::SeqCst) == LANES));
// MHUNT-5 (OX-MONHUNT 2026-10-28): the bound is what a lane that lost the gate cannot
// beat, jera's re-ask (2 s): 1 s went red on a loaded host (64 threads at load ~100
// took 1.35 s, every one of them started), while a lost lane still sleeps 2 s.
assert!(began.elapsed() < JERA_REASK - Duration::from_millis(100), "round {round}: {LANES} jobs took {:?} to run: a lane lost the gate and slept jera's re-ask", began.elapsed());
}
}
/// ★ **A job that is gone has its row written** (MERGE-TRAIN's batch gate under load: "two
/// finished rows: queued, done"): the subject was let go before the lane wrote its terminal
/// row, so a reader that waited on the job's end could still read it queued.
#[test]
fn a_job_that_is_gone_has_its_terminal_row_written() {
let _one = ONE.lock().unwrap_or_else(|p| p.into_inner());
let dir = std::env::temp_dir().join(format!("monetize-jobs-gone-{}", std::process::id()));
let _ = std::fs::remove_dir_all(&dir);
set_ledger(&dir).expect("the ledger opens");
for i in 0..50 {
assert_eq!(once("test.gone", &format!("k{i}"), || Ok(())), Ok(true));
}
for i in 0..50 {
assert!(
until(|| find("test.gone", &format!("k{i}")).is_none()),
"job {i} ends"
);
}
let rows = ledger_rows("test.gone");
let open: Vec<_> = rows
.iter()
.filter(|r| r.status != "done")
.map(|r| (r.job_id.clone(), r.status.clone()))
.collect();
assert!(
open.is_empty() && rows.len() == 50,
"every gone job's row is done: {} rows, open {open:?}",
rows.len()
);
}
/// ★ **A late QUEUED row never overwrites its job's terminal row** (RA4-25): jera's submit
/// writes the QUEUED row after the job is visible to its worker, so the submitter's write and
/// the worker's running + done can interleave. The submitter is held INSIDE its write here
/// until the worker's done row has been written (or half a second, when the worker cannot
/// write meanwhile): the row the ledger keeps must be the done one.
#[test]
fn a_late_queued_row_never_overwrites_the_terminal_row() {
use std::sync::mpsc;
let rows: Arc<Mutex<BTreeMap<String, String>>> = Arc::default();
let (in_write, a_in_write) = mpsc::channel::<()>();
let (worker_done, b_done) = mpsc::channel::<()>();
let (in_write, b_done) = (Mutex::new(in_write), Mutex::new(b_done));
let r = Arc::clone(&rows);
let held = JobSink::remote(move |rec: &jera::jobs::JobRecord| {
if rec.status == "queued" {
let _ = in_write.lock().unwrap().send(());
let _ = b_done
.lock()
.unwrap()
.recv_timeout(Duration::from_millis(500));
}
r.lock()
.unwrap()
.insert(rec.job_id.clone(), rec.status.clone());
});
let sink = lane_sink_to("a test lane", move || held.clone());
let id = format!("ra4-25-{}", std::process::id());
let rec = |status: &str| jera::jobs::JobRecord {
job_id: id.clone(),
kind: "test.late".into(),
target: format!("test.late {id}"),
workspace: DOMAIN.into(),
status: status.into(),
ts_start_micros: 0,
ts_end_micros: None,
elapsed_ms: None,
detail_json: "{}".into(),
result_ref: String::new(),
parent_id: None,
node: None,
phase: None,
actor: None,
actor_method: None,
};
let (queued, running, done) = (rec("queued"), rec("running"), rec("done"));
let (s1, s2) = (sink.clone(), sink);
let submitter = std::thread::spawn(move || s1.record(&queued));
a_in_write
.recv_timeout(Duration::from_secs(10))
.expect("the submitter writes its queued row");
let worker = std::thread::spawn(move || {
s2.record(&running);
s2.record(&done);
let _ = worker_done.send(());
});
submitter.join().unwrap();
worker.join().unwrap();
assert_eq!(
rows.lock().unwrap().get(&id).map(String::as_str),
Some("done"),
"the ledger keeps the terminal row"
);
}
/// ★ **A per-tick subject leaves at most one row per RECORD_MIN, a FAILED one always** (RA4-4:
/// the custody step on the 2 s tick wrote a row each time and pushed every older row out).
#[test]
fn a_per_tick_subject_leaves_one_row_and_its_failures() {
let _one = ONE.lock().unwrap_or_else(|p| p.into_inner());
let dir = std::env::temp_dir().join(format!("monetize-jobs-tick-{}", std::process::id()));
let _ = std::fs::remove_dir_all(&dir);
set_ledger(&dir).expect("the ledger opens");
for _ in 0..10 {
assert_eq!(once("test.tick", "acme", || Ok(())), Ok(true));
assert!(until(|| find("test.tick", "acme").is_none()));
}
assert_eq!(
once("test.tick", "acme", || Err("the hand-off failed".into())),
Ok(true)
);
assert!(
until(|| ledger_rows("test.tick")
.iter()
.any(|r| r.status == "failed")),
"the failure is a row"
);
let done = ledger_rows("test.tick")
.iter()
.filter(|r| r.status == "done")
.count();
assert_eq!(done, 1, "ten ticks on one subject leave one done row");
}
/// ★ **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!("monetize-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<_>>()
);
}
/// ★ **With a ledger, a one-shot job is a jera row, a failed one a FAILED row; a frequent
/// duty writes none** (RA3-5: the lanes wrote to `JobSink::noop()`).
#[test]
fn with_a_ledger_jobs_are_rows_and_a_failed_one_a_failed_row() {
let _one = ONE.lock().unwrap_or_else(|p| p.into_inner());
let dir = std::env::temp_dir().join(format!("monetize-jobs-ledger-{}", std::process::id()));
let _ = std::fs::remove_dir_all(&dir);
set_ledger(&dir).expect("the ledger opens");
assert_eq!(once("test.rows", "ok", || Ok(())), Ok(true));
assert_eq!(
once("test.rows", "bad", || Err("it fails".into())),
Ok(true)
);
// Wait on the jobs' end, never on timing: a job that is gone has its row written.
assert!(
until(|| find("test.rows", "ok").is_none() && find("test.rows", "bad").is_none()),
"both jobs end"
);
let finished = ledger_rows("test.rows")
.into_iter()
.filter(|r| r.status == "done" || r.status == "failed")
.count();
assert_eq!(
finished,
2,
"two finished rows: {:?}",
ledger_rows("test.rows")
.iter()
.map(|r| r.status.clone())
.collect::<Vec<_>>()
);
let rows = ledger_rows("test.rows");
assert!(
rows.iter().any(|r| r.status == "done") && rows.iter().any(|r| r.status == "failed")
);
let quick = every("test.quick", FAST, || Ok(Round::Again)).expect("a duty starts");
assert!(until(|| quick.looks() >= 3));
assert!(
ledger_rows("test.quick").is_empty(),
"a frequent duty writes no rows"
);
quick.stop(Duration::from_secs(1));
}
/// ★ **A second ask while the work runs is that work**, never a second copy; another subject
/// does not wait for it.
#[test]
fn a_second_ask_while_it_runs_is_that_work_and_another_subject_does_not_wait() {
let _one = ONE.lock().unwrap_or_else(|p| p.into_inner());
let (tx, rx) = std::sync::mpsc::channel::<()>();
let rx = Mutex::new(rx);
assert_eq!(
once("test.dup", "slow", move || {
let _ = rx.lock().unwrap().recv_timeout(Duration::from_secs(10));
Ok(())
}),
Ok(true)
);
assert!(
until(|| find("test.dup", "slow").is_some_and(|j| j.looks() == 1)),
"the slow work runs"
);
assert_eq!(
once("test.dup", "slow", || Ok(())),
Ok(false),
"the same subject's work still runs: that is the answer"
);
let other = Arc::new(AtomicBool::new(false));
let o = Arc::clone(&other);
assert_eq!(
once("test.dup", "quick", move || {
o.store(true, Ordering::SeqCst);
Ok(())
}),
Ok(true)
);
assert!(
until(|| other.load(Ordering::SeqCst)),
"another subject runs while the slow one still does"
);
tx.send(()).unwrap();
assert!(until(|| find("test.dup", "slow").is_none()));
}
/// ★ **Work that panics is said, a FAILED jera job, and its Finally still runs** (the raw
/// thread it replaces left its `running` flag set for good: the subject never ran again).
#[test]
fn work_that_panics_is_a_failed_job_and_its_finally_runs() {
let _one = ONE.lock().unwrap_or_else(|p| p.into_inner());
let flag = Arc::new(AtomicBool::new(true));
let f = Arc::clone(&flag);
assert_eq!(
once("test.panics", "p", move || {
let _reset = Finally(move || f.store(false, Ordering::SeqCst));
panic!("the work breaks")
}),
Ok(true)
);
assert!(
until(|| !flag.load(Ordering::SeqCst)),
"the Finally ran after the panic"
);
assert!(until(|| find("test.panics", "p").is_none()));
let ran = Arc::new(AtomicBool::new(false));
let r = Arc::clone(&ran);
assert_eq!(
once("test.panics", "p", move || {
r.store(true, Ordering::SeqCst);
Ok(())
}),
Ok(true),
"the subject can run again"
);
assert!(until(|| ran.load(Ordering::SeqCst)));
}
/// ★ **A duty that panics is said, a FAILED row, and goes on**; Done ends it; a stopped duty
/// looks no more.
#[test]
fn a_duty_goes_on_after_a_panic_done_ends_it_and_a_stop_stops_it() {
let _one = ONE.lock().unwrap_or_else(|p| p.into_inner());
let mut n = 0;
let d = every("test.every", FAST, move || {
n += 1;
if n == 1 {
panic!("the first look breaks")
}
Ok(Round::Again)
})
.expect("a duty starts");
assert!(until(|| d.looks() >= 3), "the looks after the panic run");
assert_eq!(d.failures(), 1);
assert!(
until(|| d.queue().stats().failed == 1 && d.queue().stats().done >= 2),
"{:?}",
d.queue().stats()
);
let mut m = 0;
let done = every("test.done", Duration::from_secs(3600), move || {
m += 1;
Ok(if m >= 2 {
Round::Done
} else {
Round::After(Duration::from_millis(5))
})
})
.expect("a duty starts");
assert!(until(|| done.is_stopped()), "Done ends it");
assert_eq!(done.looks(), 2);
// The stop (`stop_all` is the server's, process-wide: tests in this binary share it, so
// one duty is stopped by hand here).
assert!(d.stop(Duration::from_secs(1)));
assert!(
d.is_stopped() && find("test.every", "").is_none(),
"a stopped duty is not listed"
);
let at = d.looks();
std::thread::sleep(Duration::from_millis(100));
assert_eq!(d.looks(), at, "no look after the stop");
}
}
/// ★ **The server's background work runs as named jera jobs, not raw threads** (A32-14). Each
/// site, read off its source: the job it starts, and the raw thread it no longer starts.
#[cfg(test)]
mod sites {
/// The production part of a file (before its first test module).
fn code(src: &str) -> &str {
&src[..src.find("#[cfg(test)]\nmod").unwrap_or(src.len())]
}
/// Whitespace-blind: rustfmt may break a call over lines (stink-mon 2026-11-03).
fn squash(s: &str) -> String {
s.chars().filter(|c| !c.is_whitespace()).collect()
}
fn holds(src: &str, job: &str, thread: &str) {
let c = squash(code(src));
let (job, thread) = (&squash(job), &squash(thread));
assert!(
c.contains(job.as_str()),
"the work starts as a jera job: `{job}` is not in the source"
);
assert!(
!c.contains(thread.as_str()),
"the raw thread `{thread}` is still started"
);
}
#[test]
fn the_wallet_reading_and_its_keeper_are_jera_jobs() {
holds(
include_str!("grow.rs"),
"crate::jobs::once(WALLET_JOB",
".name(format!(\"wallet-{product}\"))",
);
holds(
include_str!("grow.rs"),
"crate::jobs::every(WALLET_KEEPER",
".name(\"wallet-keeper\".into())",
);
}
#[test]
fn the_plan_basis_and_its_keeper_are_jera_jobs() {
holds(
include_str!("plans.rs"),
"crate::jobs::once(BASIS_JOB",
".name(\"plan-basis\".into())",
);
holds(
include_str!("plans.rs"),
"crate::jobs::every(BASIS_KEEPER",
".name(\"plan-keeper\".into())",
);
}
#[test]
fn the_callbacks_settle_and_grant_as_jera_jobs() {
holds(
include_str!("callback.rs"),
"crate::jobs::once(SIGNUP_SETTLE_JOB",
".name(\"callback-signup\".into())",
);
holds(
include_str!("callback.rs"),
"crate::jobs::once(GRANT_JOB",
".name(\"callback-grant\".into())",
);
}
#[test]
fn the_guest_watch_is_a_jera_job() {
holds(
include_str!("guests.rs"),
"crate::jobs::every(GUEST_WATCH",
".name(\"monetize-guest-watch\".into())",
);
}
#[test]
fn a_custody_step_runs_on_its_estates_own_lane() {
// A step waits up to 20 min between its polls: on the shared front-keep lane that wait
// held every other estate's front keep boot.
holds(
include_str!("estate_order/update.rs"),
"crate::jobs::once(CUSTODY_STEP_JOB",
"if spec.kind == CUSTODY_STEP_JOB {",
);
}
/// ★ **Every jera lane in the server writes the ledger** (RA4-6: six hand-built lanes printed
/// their failures and kept no row, so the estate drives were nowhere in `nornir jobs`).
#[test]
fn every_lane_in_the_server_writes_the_ledger() {
for (file, src) in [
("estate_order.rs", include_str!("estate_order.rs")),
(
"estate_order/update.rs",
include_str!("estate_order/update.rs"),
),
(
"estate_order/narrate.rs",
include_str!("estate_order/narrate.rs"),
),
("fleet.rs", include_str!("fleet.rs")),
("plugins.rs", include_str!("plugins.rs")),
] {
let c = code(src);
let printing_only = c.matches("JobSink::remote(").count();
let teed = c.matches("crate::jobs::ledger_sink().record(").count();
assert!(printing_only <= teed, "{file}: a lane's sink only prints; build it on crate::jobs::lane_sink (or tee crate::jobs::ledger_sink())");
}
}
/// ★ **No lane in monetize re-asks jera's gate on a poll of its own** (RA3-8): jera 0.2.1's
/// look at the exclusive gate never takes it, so the 20 ms (and the console's 100 ms) copies
/// of that fix are gone; every lane keeps jera's own re-ask.
#[test]
fn no_lane_polls_jeras_gate_on_its_own() {
for (file, src) in [
("monetize-server jobs.rs", include_str!("jobs.rs")),
(
"monetize-site-buy ladders.rs",
include_str!("../../monetize-site-buy/src/ladders.rs"),
),
(
"monetize-ui-server estates.rs",
include_str!("../../monetize-ui-server/src/estates.rs"),
),
] {
let c = code(src);
assert!(
!c.contains("new_with_poll(")
&& !c.contains("DEFER_POLL")
&& !c.contains("GATE_POLL"),
"{file}: a lane polls jera's gate on its own"
);
}
}
/// ★ **monetize requires jera 0.2.1, the gate fix** (RA3-8): with jera's own re-ask (2 s) a
/// lane is only as quick as jera's look at its exclusive gate; 0.2.0's look took the gate.
#[test]
fn monetize_requires_the_jera_with_the_gate_fix() {
let lock = include_str!("../../../Cargo.lock");
let at = lock
.find("name = \"jera\"\nversion = \"")
.expect("jera in the lock")
+ "name = \"jera\"\nversion = \"".len();
let locked = &lock[at..at + lock[at..].find('"').unwrap()];
let v: Vec<u32> = locked.split('.').map(|n| n.parse().unwrap()).collect();
assert!(
v >= vec![0, 2, 1],
"the lock holds jera {locked}, before the gate fix"
);
assert!(
include_str!("../Cargo.toml").contains("jera = { version = \"0.2.1\""),
"monetize-server requires jera 0.2.1"
);
}
#[test]
fn the_server_stops_its_jobs_at_the_clean_stop() {
let cli = include_str!("cli.rs");
let serve = &cli[cli.find("crate::serve_on_with(listener").expect("serve")..];
assert!(
serve[..serve.find("Ok::<(), anyhow::Error>(())").unwrap()]
.contains("crate::jobs::stop_all("),
"the jobs stop when the server has served"
);
}
}