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). The loop itself is jera's
//! (`jera::jobs::duty`, 0.2.3; owner 2026-11-03: ONE wrapper, in jera, for every consumer — this
//! module was the third copy beside gunnar-common `jobs` and holger-live `duty`). What stays
//! here is the server's spelling: the ledger under `<data>/jobs`, the lanes' sampling sink, the
//! subject rule of a one-shot, the clean stop, and a runtime for a caller on a plain 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): `jera::jobs::duty::once` on the subject's own lane, 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 (jera 0.2.3 leaves no
//! worker behind).
//! * [`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.
//!
//! ```text
//! once(kind, key, work) ─▶ subjects[kind/key] reserved ─▶ duty::once (lane, 1 look) ─▶ row done|FAILED ─▶ subject freed
//! every(kind, period, look) ─▶ duty::start(every(1)=0, every(n>1)=period) ─▶ look n ─▶ Again|After|Done
//! ```
//!
//! Either way the work is visible ([`running`]), a failure or a PANIC is an `ALERT jobs.failed <kind>` line
//! and a FAILED jera row (a duty's 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, BTreeSet};
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::{Arc, Mutex, OnceLock};
use std::time::{Duration, Instant};
use jera::jobs::JobSink;
pub use jera::jobs::duty::Round;
/// **One job, started**: jera's duty (its kind, its subject, its queue and what it has done).
pub type Job = jera::jobs::duty::Duty;
/// The ledger rows' domain.
const DOMAIN: &str = "monetize";
/// **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)()
}
}
/// The one-shot jobs whose work still runs, by `kind/key` (reserved before the lane starts, freed
/// by the work itself, so a second ask meanwhile is answered `Ok(false)`).
fn subjects() -> &'static Mutex<BTreeSet<String>> {
static S: OnceLock<Mutex<BTreeSet<String>>> = 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);
/// Where a job says its failures: the server's ALERT line (jera's words carry the kind, the look
/// and the cause).
fn say() -> jera::jobs::duty::Say {
Arc::new(|line: &str| {
// jera says `ALERT <kind> look <n> ...`; the server says it under its own code.
let line = line
.strip_prefix("ALERT")
.and_then(|r| r.strip_prefix(char::is_whitespace))
.unwrap_or(line);
eprintln!("monetize-server: ALERT jobs.failed {line}");
})
}
/// **Every job of this process that still runs.**
pub fn running() -> Vec<Arc<Job>> {
jera::jobs::duty::running()
}
/// The running job of `kind` (and subject `key`, `""` for a duty), if one runs.
pub fn find(kind: &str, key: &str) -> Option<Arc<Job>> {
jera::jobs::duty::find_keyed(kind, 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 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 server stopped before this job ended");
n
}
/// 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)
}
/// **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 mut s = subjects().lock().unwrap_or_else(|p| p.into_inner());
if !s.insert(id.clone()) {
return Ok(false);
}
}
let freed = id.clone();
let _in_rt = rt.enter();
let started = jera::jobs::duty::once(
kind,
key,
DOMAIN,
lane_sink("a one-shot"),
say(),
move || {
// The subject is free again at once (the next ask runs anew); the JOB is gone only
// once its terminal row is written (jera's rule since 0.2.3).
let _free = Finally(move || {
subjects()
.lock()
.unwrap_or_else(|p| p.into_inner())
.remove(&freed);
});
work()
},
);
match started {
Ok(_) => Ok(true),
Err(e) => {
subjects()
.lock()
.unwrap_or_else(|p| p.into_inner())
.remove(&id);
Err(e)
}
}
}
/// **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,
mut 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 sink = if every >= RECORD_MIN {
ledger_sink()
} else {
JobSink::noop()
};
let _in_rt = rt.enter();
jera::jobs::duty::start(
kind,
DOMAIN,
sink,
move |n| if n == 1 { Duration::ZERO } else { every },
say(),
move |_| look(),
)
}
#[cfg(test)]
mod tests {
use super::*;
use jera::jobs::chain::{ChainQueue, ChainSpec};
use std::sync::atomic::AtomicU64;
/// 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.5**: 0.2.1 is 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), 0.2.3 the duty that ends after Done (a one-shot per subject leaked its worker),
/// 0.2.5 a done duty leaves the registry only after its row is written (0.2.4: the ledger).
#[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, 5],
"the lock holds jera {locked}, before the duty-end fix"
);
assert!(
include_str!("../Cargo.toml").contains("jera = { version = \"0.2.5\""),
"monetize-server requires jera 0.2.5"
);
}
#[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"
);
}
}