crates/jera/src/jobs/chain.rs
//! **The background CHAIN** — submit-now, run-when-the-machine-gets-to-it.
//!
//! [`super`] is a *ledger*: it records that an op ran. What it never had is a
//! **queue** — somewhere to put derived work that MUST happen after a fact
//! lands, but must not happen on the request that asks for the result, and must
//! not shove the machine's real work aside to do it.
//!
//! The motivating case (owner, 2026-08-22) is the Mímir lens chain: a
//! knowledge snapshot lands, eleven expensive folds become due, and the answer
//! is read from a table in milliseconds instead of computed in seconds on the
//! asking thread. But nothing here knows what a lens is. A chain job is a
//! `(kind, key, group, generation, payload)` and a runner closure; the domain
//! owns the meaning.
//!
//! Four properties, and they are the whole module:
//!
//! * **Idempotent.** Submitting the same `key` twice while the first is still
//! pending or running returns the SAME job id and queues nothing. Eleven
//! populate ticks in a row are eleven no-ops, not 121 folds.
//! * **Superseding.** A submission with `generation` G cancels every PENDING
//! job in the same `group` with a LOWER generation. Yesterday's snapshot is
//! not worth computing once today's has landed. A cancelled job records
//! [`status::STOPPED`] — terminal, but explicitly **not a failure**, the
//! distinction this ledger already draws for daemons.
//! * **Prioritised.** [`Priority::High`] drains before [`Priority::Normal`]
//! before [`Priority::Low`]; within a priority it is FIFO.
//! * **Deferential — but not forever.** The worker asks
//! [`ChainQueue::with_defer_while`] before every job and waits while the
//! machine is busy. It also yields to the ledger's own
//! [`JobClass::Exclusive`](super::JobClass::Exclusive) gate — by LOOKING at
//! it (`available_permits`), never taking it, so a background chain can *lose*
//! the race for the box but can never make a bench wait for it, nor another
//! lane that looks at the same instant (RA3-8: a `try_acquire` look did).
//!
//! The domain predicate has a CEILING ([`ChainQueue::with_max_defer`],
//! default 10 minutes) and this is not a nicety. Measured on oden's live
//! ledger 2026-08-22: a non-terminal `workspace_populate` row **5.5 days**
//! old, beside `workspace_republish` and `scip_ingest` rows of 21 h and 1 h —
//! every one of them a process that was KILLED and so never wrote its
//! terminal record. A domain predicate reading "is a heavy job non-terminal"
//! was therefore permanently `true`, and the chain queued, superseded and
//! queued again for an hour without ever folding once. "As the machine gets
//! to it" must still mean it gets to it. The EXCLUSIVE gate is never
//! overridden — a bench stays isolated; only the domain's own signal expires.
//!
//! Two more, added 2026-09-10 after the lens chain was MEASURED starving on oden
//! (republishes every 2–3 min, lens rows `done 29 / stopped 75`, one lens per ten
//! minutes getting through the deferral). Both serve a job that means **"make X
//! current"** — keyed WITHOUT a generation, folding whatever the warehouse holds
//! when it runs — which is what a job that reads live tables always was:
//!
//! * **Re-arm on duplicate-while-running.** A submit against a key that is
//! RUNNING (not merely pending) is still `Duplicate`, but the queue remembers
//! it and pushes ONE fresh pending copy when that run ends. A generation that
//! moved mid-fold is folded again, once, however many times it was asked for
//! meanwhile. A duplicate against a PENDING key stays a pure no-op — that copy
//! has not started and will read the warehouse as it is when it does.
//! * **The deferral ceiling is cumulative.** [`ChainQueue::with_max_defer`] is
//! measured from the start of the current busy stretch, across jobs — not
//! restarted for every job. One continuous busy signal costs the chain the
//! ceiling ONCE, then it drains; it yields again only after the predicate has
//! actually been false in between. The exclusive gate is checked first and
//! alone, and while it is held the domain predicate is not consulted at all.
//!
//! **One at a time.** The worker is a single loop by design: the folds it was
//! built for read big shared table previews, and running two at once doubles
//! the resident set to halve nothing. Run a second [`ChainQueue`] if a domain
//! genuinely wants two lanes.
//!
//! **The queue is in-memory; the ledger is durable.** A restart loses the
//! pending list — deliberately: the domain that submits knows how to work out
//! what is still due (for the lens chain, "which lenses have no `lens_runs` row
//! at the current generation"), and that recomputation is always more truthful
//! than a replayed queue whose reason to exist may have expired. What survives
//! a restart is every job's *record*.
use std::collections::HashMap;
use std::sync::{Arc, Mutex};
use serde_json::Value;
use super::{JobRecord, JobSink, status};
/// Drain order. Within one priority the queue is FIFO, so a chain submitted as
/// a unit runs in the order the domain listed it.
#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord)]
pub enum Priority {
/// Behind everything else — the default for derived/refresh work.
Low,
/// Ordinary background work.
Normal,
/// Drains first. For work a human is visibly waiting on.
High,
}
impl Default for Priority {
fn default() -> Self {
Priority::Low
}
}
/// One unit of chain work: what to run, and the three keys that make a chain
/// safe to submit repeatedly.
#[derive(Debug, Clone)]
pub struct ChainSpec {
/// The ledger [`kind`](super::kind) this job records as (e.g.
/// [`super::kind::LENS`]).
pub kind: String,
/// **Idempotency key.** Unique per unit of work — for the lens chain,
/// `(workspace, repo-scope, snapshot, lens)`. A resubmit while this key is
/// pending or running is a no-op.
pub key: String,
/// **Supersede group.** A newer generation in the same group cancels this
/// one while it is still pending — for the lens chain, the `(workspace,
/// repo-scope)` whose snapshot moved.
pub group: String,
/// Higher wins. Work derived from a snapshot should use something that
/// only moves when the snapshot does.
pub generation: i64,
/// Human-readable subject for the ledger row.
pub target: String,
pub workspace: String,
/// The runner's input, verbatim. Also stored as the ledger row's
/// `detail_json`, so a queued job is legible without the queue.
pub payload: Value,
pub priority: Priority,
}
impl ChainSpec {
/// A [`Priority::Low`] spec — the common case.
pub fn new(
kind: impl Into<String>,
key: impl Into<String>,
group: impl Into<String>,
generation: i64,
workspace: impl Into<String>,
payload: Value,
) -> Self {
let key = key.into();
ChainSpec {
kind: kind.into(),
target: key.clone(),
key,
group: group.into(),
generation,
workspace: workspace.into(),
payload,
priority: Priority::Low,
}
}
/// Set the ledger row's human-readable target (defaults to the key).
pub fn with_target(mut self, target: impl Into<String>) -> Self {
self.target = target.into();
self
}
/// Set the drain priority.
pub fn with_priority(mut self, p: Priority) -> Self {
self.priority = p;
self
}
}
/// What a [`ChainQueue::submit`] did. Never an error for "already there" —
/// a chain is *meant* to be submitted repeatedly.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum Submitted {
/// Newly queued under this `job_id`.
Accepted(String),
/// The key was already pending or running; this is that job's id and
/// nothing was queued.
Duplicate(String),
/// Refused: a job for this `group` at a HIGHER generation is already
/// pending or running, so this submission is already stale. Carries the
/// generation that beat it.
Stale(i64),
}
impl Submitted {
/// The job id, when one exists.
pub fn job_id(&self) -> Option<&str> {
match self {
Submitted::Accepted(id) | Submitted::Duplicate(id) => Some(id),
Submitted::Stale(_) => None,
}
}
/// Whether this submission put new work on the queue.
pub fn is_new(&self) -> bool {
matches!(self, Submitted::Accepted(_))
}
}
/// How long the domain predicate may hold the chain off before the worker
/// proceeds anyway (see the module docs).
pub const DEFAULT_MAX_DEFER: std::time::Duration = std::time::Duration::from_secs(600);
/// A queued (not yet started) unit.
#[derive(Debug, Clone)]
struct Pending {
job_id: String,
spec: ChainSpec,
seq: u64,
/// When the unit was queued — the ledger row's `ts_start_micros` for a
/// superseded job (it never ran), and the start of the wait otherwise.
queued_micros: i64,
/// When the worker claimed it, so `elapsed_ms` measures the WORK and not
/// the queue wait (the `start_scheduled` discipline, verbatim).
started_micros: Option<i64>,
}
#[derive(Default)]
struct Inner {
pending: Vec<Pending>,
/// key → job_id for everything pending OR running (the idempotency set).
live: HashMap<String, String>,
/// The spec currently being run, if any.
running: Option<Pending>,
/// **Re-arm on duplicate-while-running.** The LATEST spec submitted for the
/// running key while it was running. When the run ends, one fresh pending
/// copy is pushed from this — so a generation that moved mid-fold is folded
/// again, once, however many times it was asked for meanwhile. See the
/// module docs, "make current".
rearm: Option<ChainSpec>,
seq: u64,
/// Terminal counts, for [`ChainQueue::stats`].
done: u64,
failed: u64,
superseded: u64,
}
/// The queue. Cheap to clone via [`Arc`]; submit from anywhere, drain with one
/// [`ChainQueue::run`] task.
pub struct ChainQueue {
inner: Mutex<Inner>,
sink: JobSink,
wake: tokio::sync::Notify,
defer_while: Mutex<Option<Arc<dyn Fn() -> bool + Send + Sync>>>,
/// How long the worker waits before re-asking a `true` deferral.
defer_poll: std::time::Duration,
/// How long the DOMAIN predicate may hold the chain off before the worker
/// proceeds anyway. See the module docs — a zombie ledger row would
/// otherwise defer it for ever. `None` = no ceiling.
max_defer: Mutex<Option<std::time::Duration>>,
/// When the domain predicate LAST said busy after having said quiet — the
/// start of the current busy stretch. The ceiling is measured from here and
/// is therefore CUMULATIVE across jobs: a chain held off for `max_defer` by
/// one continuous busy signal then drains, and only starts yielding again
/// once the predicate has been false in between. Measured on oden
/// 2026-09-10: with a per-job clock, republishes every 2–3 min let one lens
/// through every ten minutes — thirteen lenses became two hours.
busy_since: Mutex<Option<std::time::Instant>>,
}
/// A point-in-time count of what the queue is doing — the number the pane's
/// "queued in jera (#n)" renders.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
pub struct ChainStats {
pub pending: usize,
pub running: usize,
pub done: u64,
pub failed: u64,
pub superseded: u64,
}
impl ChainQueue {
/// A queue whose lifecycle rows land in `sink` (the process's
/// [`JobStore::sink`](super::JobStore::sink)).
pub fn new(sink: JobSink) -> Arc<Self> {
Arc::new(ChainQueue {
inner: Mutex::new(Inner::default()),
sink,
wake: tokio::sync::Notify::new(),
defer_while: Mutex::new(None),
defer_poll: std::time::Duration::from_secs(2),
max_defer: Mutex::new(Some(DEFAULT_MAX_DEFER)),
busy_since: Mutex::new(None),
})
}
/// **Yield to the machine's real work.** The worker calls `pred` before
/// every job and, while it returns `true`, waits [`defer_poll`] and asks
/// again — it never queues behind the busy work, so it can lose the box
/// indefinitely without ever delaying what it deferred to.
///
/// The exclusive gate is checked separately and unconditionally (see the
/// module docs); this is the domain's *additional* "is the machine busy"
/// signal — for nornir, "is a populate or a build running".
pub fn with_defer_while(
self: &Arc<Self>,
pred: impl Fn() -> bool + Send + Sync + 'static,
) -> Arc<Self> {
*self.defer_while.lock().expect("chain lock") = Some(Arc::new(pred));
self.clone()
}
/// Cap how long the DOMAIN predicate may hold the chain off (default
/// [`DEFAULT_MAX_DEFER`]). `None` removes the ceiling — only do that where
/// the predicate is PROVEN to go false, which a ledger-derived one is not.
pub fn with_max_defer(self: &Arc<Self>, d: Option<std::time::Duration>) -> Arc<Self> {
*self.max_defer.lock().expect("chain lock") = d;
self.clone()
}
/// A queue with a non-default deferral re-ask interval.
pub fn new_with_poll(sink: JobSink, defer_poll: std::time::Duration) -> Arc<Self> {
Arc::new(ChainQueue {
inner: Mutex::new(Inner::default()),
sink,
wake: tokio::sync::Notify::new(),
defer_while: Mutex::new(None),
defer_poll,
max_defer: Mutex::new(Some(DEFAULT_MAX_DEFER)),
busy_since: Mutex::new(None),
})
}
/// **Submit one unit.** Idempotent on `spec.key`, superseding on
/// `(spec.group, spec.generation)`. Writes a [`status::QUEUED`] ledger row
/// for an accepted job, so the wait is visible in `nornir jobs` / the Jera
/// tab before anything runs.
pub fn submit(self: &Arc<Self>, spec: ChainSpec) -> Submitted {
let out = {
let mut inner = self.inner.lock().expect("chain lock");
// Idempotency first: an identical unit already in flight wins.
if let Some(id) = inner.live.get(&spec.key) {
let id = id.clone();
// …but a duplicate against the RUNNING key is remembered, so the
// run's end pushes one fresh copy (the latest spec asked for).
// A duplicate against a PENDING key is a pure no-op: that copy has
// not started and will read the warehouse as it is when it does.
if inner.running.as_ref().is_some_and(|r| r.spec.key == spec.key) {
inner.rearm = Some(spec.clone());
}
return Submitted::Duplicate(id);
}
// Is this submission already stale? A pending or running job in the
// same group at a HIGHER generation means the caller is asking for
// yesterday's answer.
let newest = inner
.pending
.iter()
.chain(inner.running.iter())
.filter(|p| p.spec.group == spec.group)
.map(|p| p.spec.generation)
.max();
if let Some(g) = newest {
if g > spec.generation {
return Submitted::Stale(g);
}
}
// Supersede: cancel every PENDING job in this group at a lower
// generation. A RUNNING job is left to finish — killing work
// already half-done buys nothing, and its result is still a
// truthful reading of the generation it names.
let mut cancelled: Vec<Pending> = Vec::new();
inner.pending.retain(|p| {
if p.spec.group == spec.group && p.spec.generation < spec.generation {
cancelled.push(p.clone());
false
} else {
true
}
});
for c in &cancelled {
inner.live.remove(&c.spec.key);
inner.superseded += 1;
}
inner.seq += 1;
let seq = inner.seq;
let job_id = uuid::Uuid::new_v4().to_string();
inner.live.insert(spec.key.clone(), job_id.clone());
inner.pending.push(Pending {
job_id: job_id.clone(),
spec: spec.clone(),
seq,
queued_micros: chrono::Utc::now().timestamp_micros(),
started_micros: None,
});
(job_id, cancelled)
};
let (job_id, cancelled) = out;
// Ledger writes happen OUTSIDE the lock (the sink may touch redb).
for c in cancelled {
self.write_terminal(
&c,
status::STOPPED,
serde_json::json!({
"superseded_by_generation": spec.generation,
"generation": c.spec.generation,
"group": c.spec.group,
"why": "a newer generation landed before this job ran; \
computing the older one would produce a stale answer",
}),
);
}
self.sink.upsert(&self.record(&job_id, &spec, status::QUEUED, None));
self.notify();
Submitted::Accepted(job_id)
}
/// Submit a whole chain in one call — the units drain in list order within
/// their priority. Returns one [`Submitted`] per spec, positionally.
pub fn submit_all(self: &Arc<Self>, specs: Vec<ChainSpec>) -> Vec<Submitted> {
specs.into_iter().map(|s| self.submit(s)).collect()
}
/// Cancel every PENDING job in `group` below `generation` without
/// submitting anything. Returns how many were stopped.
pub fn cancel_group_below(self: &Arc<Self>, group: &str, generation: i64) -> usize {
let cancelled: Vec<Pending> = {
let mut inner = self.inner.lock().expect("chain lock");
let mut out = Vec::new();
inner.pending.retain(|p| {
if p.spec.group == group && p.spec.generation < generation {
out.push(p.clone());
false
} else {
true
}
});
for c in &out {
inner.live.remove(&c.spec.key);
inner.superseded += 1;
}
out
};
let n = cancelled.len();
for c in cancelled {
self.write_terminal(
&c,
status::STOPPED,
serde_json::json!({
"superseded_by_generation": generation,
"generation": c.spec.generation,
"group": c.spec.group,
}),
);
}
n
}
/// What the queue is doing right now.
pub fn stats(&self) -> ChainStats {
let inner = self.inner.lock().expect("chain lock");
ChainStats {
pending: inner.pending.len(),
running: usize::from(inner.running.is_some()),
done: inner.done,
failed: inner.failed,
superseded: inner.superseded,
}
}
/// How many units are queued (not running) for `group` — the "#n" a pane
/// shows beside "queued in jera".
pub fn pending_in_group(&self, group: &str) -> usize {
let inner = self.inner.lock().expect("chain lock");
inner
.pending
.iter()
.filter(|p| p.spec.group == group)
.count()
}
/// **The worker.** Drains forever: highest priority first, FIFO within it,
/// one at a time, deferring to the machine's real work. `run_one` is
/// executed on a blocking thread (the folds this exists for are CPU-bound
/// and synchronous), and its `Ok(detail)` / `Err` becomes the job's
/// terminal ledger row.
///
/// Spawn it once per queue: `tokio::spawn(queue.clone().run(runner))`.
pub async fn run<F>(self: Arc<Self>, run_one: F)
where
F: Fn(ChainSpec) -> anyhow::Result<Value> + Send + Sync + 'static,
{
let run_one = Arc::new(run_one);
loop {
let Some(next) = self.take_next() else {
// Nothing to do — sleep until a submit wakes us. A timeout as
// well, so a queue whose notifier was missed still drains.
let _ = tokio::time::timeout(std::time::Duration::from_secs(5), self.wake.notified())
.await;
continue;
};
// Yield to the machine's real work BEFORE claiming the job.
self.await_quiet().await;
self.sink
.upsert(&self.record(&next.job_id, &next.spec, status::RUNNING, None));
let spec = next.spec.clone();
let f = run_one.clone();
let outcome =
tokio::task::spawn_blocking(move || f(spec)).await;
match outcome {
Ok(Ok(detail)) => {
self.write_terminal(&next, status::DONE, detail);
self.inner.lock().expect("chain lock").done += 1;
}
Ok(Err(e)) => {
self.write_terminal(
&next,
status::FAILED,
serde_json::json!({ "error": format!("{e:#}") }),
);
self.inner.lock().expect("chain lock").failed += 1;
}
Err(join) => {
self.write_terminal(
&next,
status::FAILED,
serde_json::json!({ "error": format!("chain worker panicked: {join}") }),
);
self.inner.lock().expect("chain lock").failed += 1;
}
}
self.finish_running(&next);
}
}
/// Run at most `n` units and return how many ran — the deterministic
/// drain a test (or a one-shot backfill) uses instead of the forever loop.
pub async fn drain_n<F>(self: &Arc<Self>, n: usize, run_one: F) -> usize
where
F: Fn(ChainSpec) -> anyhow::Result<Value> + Send + Sync + 'static,
{
let run_one = Arc::new(run_one);
let mut ran = 0;
for _ in 0..n {
let Some(next) = self.take_next() else { break };
self.await_quiet().await;
self.sink
.upsert(&self.record(&next.job_id, &next.spec, status::RUNNING, None));
let spec = next.spec.clone();
let f = run_one.clone();
match tokio::task::spawn_blocking(move || f(spec)).await {
Ok(Ok(detail)) => {
self.write_terminal(&next, status::DONE, detail);
self.inner.lock().expect("chain lock").done += 1;
}
Ok(Err(e)) => {
self.write_terminal(
&next,
status::FAILED,
serde_json::json!({ "error": format!("{e:#}") }),
);
self.inner.lock().expect("chain lock").failed += 1;
}
Err(join) => {
self.write_terminal(
&next,
status::FAILED,
serde_json::json!({ "error": format!("chain worker panicked: {join}") }),
);
self.inner.lock().expect("chain lock").failed += 1;
}
}
self.finish_running(&next);
ran += 1;
}
ran
}
/// **Release the running slot — and re-arm the key if it was asked for
/// again while it ran.** The ONE completion body `run` and `drain_n` share,
/// so the re-arm cannot be implemented in one loop and forgotten in the
/// other (LAW 5).
///
/// The re-armed copy is a NEW job: its own id, its own `queued` row, pushed
/// through the same path a first submission takes, so the ledger shows the
/// second ask exactly as it showed the first.
fn finish_running(self: &Arc<Self>, done: &Pending) {
let rearm = {
let mut inner = self.inner.lock().expect("chain lock");
inner.running = None;
inner.live.remove(&done.spec.key);
inner.rearm.take()
};
if let Some(spec) = rearm {
// Belt and braces: the re-arm is only ever recorded for the running
// key, but a spec that somehow names another key must not be
// pushed under the finished one's name.
if spec.key == done.spec.key {
let _ = self.submit(spec);
}
}
}
/// Pop the next unit: highest [`Priority`] first, FIFO within it.
fn take_next(&self) -> Option<Pending> {
let mut inner = self.inner.lock().expect("chain lock");
if inner.running.is_some() {
return None;
}
let best = inner
.pending
.iter()
.enumerate()
.max_by_key(|(_, p)| (p.spec.priority, std::cmp::Reverse(p.seq)))
.map(|(i, _)| i)?;
let mut p = inner.pending.remove(best);
p.started_micros = Some(chrono::Utc::now().timestamp_micros());
inner.running = Some(p.clone());
Some(p)
}
/// Wait while the machine is busy. Two signals, both non-blocking probes so
/// the chain can never make what it defers to wait for IT:
/// * the ledger's exclusive gate (a bench holding it),
/// * the domain's own predicate from [`with_defer_while`].
async fn await_quiet(&self) {
let ceiling = *self.max_defer.lock().expect("chain lock");
loop {
// The EXCLUSIVE gate first, and alone: while a bench owns the box the
// domain predicate is not even consulted — it reads the whole ledger,
// and asking it every poll while nothing can proceed anyway is pure
// cost. It does NOT expire — a bench keeps the box however long.
// A LOOK, never a take (RA3-8): `try_acquire` held the gate's one permit for an
// instant, so two lanes looking at once saw each other as a bench, and the loser
// slept a whole `defer_poll`. A bench holds the permit; a look only reads that.
if super::exclusive_gate().available_permits() == 0 {
tokio::time::sleep(self.defer_poll).await;
continue;
}
// The DOMAIN signal expires: past the ceiling the chain proceeds
// even though the domain still says busy.
//
// The ceiling is measured from the start of the CURRENT busy
// stretch (`busy_since`), not from this call: once one continuous
// busy signal has cost the chain `max_defer`, every following job
// proceeds, until the predicate has actually been false again.
let pred = self.defer_while.lock().expect("chain lock").clone();
let says_busy = pred.map(|p| p()).unwrap_or(false);
let busy_domain = {
let mut since = self.busy_since.lock().expect("chain lock");
if !says_busy {
*since = None;
false
} else {
let began = *since.get_or_insert_with(std::time::Instant::now);
match ceiling {
Some(c) if began.elapsed() >= c => false,
_ => true,
}
}
};
if !busy_domain {
return;
}
tokio::time::sleep(self.defer_poll).await;
}
}
fn record(
&self,
job_id: &str,
spec: &ChainSpec,
status: &str,
detail: Option<&Value>,
) -> JobRecord {
let now = chrono::Utc::now().timestamp_micros();
JobRecord {
job_id: job_id.to_string(),
kind: spec.kind.clone(),
target: spec.target.clone(),
workspace: spec.workspace.clone(),
status: status.to_string(),
ts_start_micros: now,
ts_end_micros: None,
elapsed_ms: None,
detail_json: serde_json::to_string(detail.unwrap_or(&spec.payload))
.unwrap_or_default(),
result_ref: String::new(),
parent_id: None,
node: None,
phase: None,
// A chain step inherits nothing: the initiator is stamped on the
// handle by the door that created it, not synthesised here.
actor: None,
actor_method: None,
}
}
fn write_terminal(&self, p: &Pending, status: &str, detail: Value) {
let now = chrono::Utc::now().timestamp_micros();
let mut rec = self.record(&p.job_id, &p.spec, status, Some(&detail));
// `elapsed_ms` measures the WORK, not the queue wait — a background
// chain can sit queued for an hour, and reporting that hour as the
// fold's duration would be a lie the Jobs panel renders as a number.
// A superseded job never started, so it reports the wait explicitly.
let start = p.started_micros.unwrap_or(p.queued_micros);
rec.ts_start_micros = start;
rec.ts_end_micros = Some(now);
rec.elapsed_ms = Some((now - start) / 1000);
self.sink.upsert(&rec);
}
}
impl std::fmt::Debug for ChainQueue {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.debug_struct("ChainQueue").field("stats", &self.stats()).finish()
}
}
// The worker is woken by submits; expose that as a method on the Arc so
// `submit` can fire it without holding the inner lock.
impl ChainQueue {
/// Wake the worker (harmless when it is already awake). `submit` calls this
/// for you; a domain that queues through another path can call it directly.
pub fn notify(&self) {
self.wake.notify_one();
}
}
#[cfg(test)]
mod tests {
use super::*;
use std::sync::atomic::{AtomicUsize, Ordering};
/// A sink that remembers every record, so a test can assert on the LEDGER
/// and not just on the queue's own counters.
#[derive(Clone, Default)]
struct Spy(Arc<Mutex<Vec<JobRecord>>>);
impl Spy {
fn sink(&self) -> JobSink {
let log = self.0.clone();
JobSink::remote(move |r| log.lock().unwrap().push(r.clone()))
}
fn last_status_of(&self, target: &str) -> Option<String> {
self.0
.lock()
.unwrap()
.iter()
.filter(|r| r.target == target)
.next_back()
.map(|r| r.status.clone())
}
}
fn spec(key: &str, group: &str, gen: i64) -> ChainSpec {
ChainSpec::new(
super::super::kind::LENS,
key,
group,
gen,
"nordisk",
serde_json::json!({ "key": key }),
)
}
/// LAW 2 — the idempotency anchor. The chain is submitted on EVERY populate
/// tick, so "already in flight" must queue nothing and hand back the SAME
/// id. A test that only checked `pending == 1` would pass on a queue that
/// dropped the first job and queued the second; assert the id too.
#[test]
fn a_resubmitted_key_queues_nothing_and_returns_the_same_id() {
let spy = Spy::default();
let q = ChainQueue::new(spy.sink());
let first = q.submit(spec("lens:change_risk@g1", "lens:nordisk", 1));
let again = q.submit(spec("lens:change_risk@g1", "lens:nordisk", 1));
assert!(first.is_new());
assert!(!again.is_new(), "a duplicate must not queue new work");
assert_eq!(
first.job_id(),
again.job_id(),
"the caller must be handed the ALREADY-QUEUED job, not a fresh one"
);
assert_eq!(q.stats().pending, 1);
}
/// LAW 2 — the supersede anchor, asserted on the LEDGER, not a counter. A
/// cancelled job must be STOPPED (terminal, a pure fact) and never FAILED:
/// yesterday's snapshot not being computed is not a red row.
#[tokio::test]
async fn a_newer_generation_stops_the_unrun_jobs_of_the_older_one() {
let spy = Spy::default();
let q = ChainQueue::new(spy.sink());
q.submit(spec("lens:change_risk@g1", "lens:nordisk", 1));
q.submit(spec("lens:villa@g1", "lens:nordisk", 1));
// A different group is untouched by the supersede.
q.submit(spec("lens:change_risk@korp@g1", "lens:nordisk/korp", 1));
assert_eq!(q.stats().pending, 3);
q.submit(spec("lens:change_risk@g2", "lens:nordisk", 2));
let s = q.stats();
assert_eq!(s.superseded, 2, "both g1 jobs of the group were cancelled");
assert_eq!(
s.pending, 2,
"the g2 job + the untouched other group remain: {s:?}"
);
assert_eq!(
spy.last_status_of("lens:change_risk@g1").as_deref(),
Some(status::STOPPED),
"a superseded job is STOPPED — a pure fact, not a failure"
);
assert_ne!(
spy.last_status_of("lens:villa@g1").as_deref(),
Some(status::FAILED)
);
assert_eq!(
spy.last_status_of("lens:change_risk@korp@g1").as_deref(),
Some(status::QUEUED),
"another group's job must NOT be superseded"
);
// And the worker never sees a superseded job.
let seen = Arc::new(Mutex::new(Vec::<String>::new()));
let s2 = seen.clone();
q.drain_n(5, move |sp| {
s2.lock().unwrap().push(sp.key.clone());
Ok(serde_json::json!({}))
})
.await;
let seen = seen.lock().unwrap().clone();
assert!(
!seen.iter().any(|k| k.ends_with("@g1") && k.contains("change_risk@g1")),
"a superseded job must never run: {seen:?}"
);
assert!(seen.contains(&"lens:change_risk@g2".to_string()));
}
/// A submission that is ALREADY behind what is queued is refused rather
/// than run — otherwise a late-arriving old generation would overwrite a
/// newer answer.
#[test]
fn an_older_generation_submitted_late_is_refused_as_stale() {
let spy = Spy::default();
let q = ChainQueue::new(spy.sink());
q.submit(spec("lens:a@g5", "g", 5));
let late = q.submit(spec("lens:a@g3", "g", 3));
assert_eq!(late, Submitted::Stale(5));
assert_eq!(q.stats().pending, 1);
}
/// LAW 2 — priority actually reorders the drain. A queue that ignored
/// priority would drain the same three jobs and report the same counts;
/// only the ORDER tells the two apart.
#[tokio::test]
async fn high_priority_drains_before_the_low_ones_it_was_queued_behind() {
let spy = Spy::default();
let q = ChainQueue::new(spy.sink());
q.submit(spec("low-1", "g", 1));
q.submit(spec("low-2", "g", 1));
q.submit(spec("urgent", "g", 1).with_priority(Priority::High));
q.submit(spec("low-3", "g", 1));
let order = Arc::new(Mutex::new(Vec::<String>::new()));
let o = order.clone();
let ran = q
.drain_n(4, move |sp| {
o.lock().unwrap().push(sp.key.clone());
Ok(serde_json::json!({}))
})
.await;
assert_eq!(ran, 4);
let order = order.lock().unwrap().clone();
assert_eq!(order[0], "urgent", "High jumps the Low queue: {order:?}");
assert_eq!(
&order[1..],
&["low-1", "low-2", "low-3"],
"within a priority the queue is FIFO: {order:?}"
);
}
/// The runner's outcome reaches the LEDGER: a clean run is `done` carrying
/// the runner's own detail, a failing one is `failed` carrying its error
/// text. A queue that swallowed the error would look identical by count.
#[tokio::test]
async fn the_runners_outcome_becomes_the_ledger_row() {
let spy = Spy::default();
let q = ChainQueue::new(spy.sink());
q.submit(spec("ok", "g", 1));
q.submit(spec("boom", "g", 1));
q.drain_n(2, |sp| {
if sp.key == "boom" {
anyhow::bail!("the fold could not read symbol_facts");
}
Ok(serde_json::json!({ "rows": 51 }))
})
.await;
assert_eq!(spy.last_status_of("ok").as_deref(), Some(status::DONE));
assert_eq!(spy.last_status_of("boom").as_deref(), Some(status::FAILED));
let log = spy.0.lock().unwrap();
let ok = log.iter().filter(|r| r.target == "ok").next_back().unwrap();
assert!(ok.detail_json.contains("\"rows\":51"), "{}", ok.detail_json);
let boom = log.iter().filter(|r| r.target == "boom").next_back().unwrap();
assert!(
boom.detail_json.contains("could not read symbol_facts"),
"the error text must survive to the ledger: {}",
boom.detail_json
);
assert_eq!(q.stats().done, 1);
assert_eq!(q.stats().failed, 1);
}
/// LAW 2 — the deferral anchor. While the domain says the machine is busy,
/// NOTHING runs; the moment it goes quiet the job runs. A test that only
/// asserted the end state would pass on a queue that ignored the predicate
/// entirely, so assert the run count WHILE busy too.
#[tokio::test]
async fn the_worker_waits_while_the_machine_says_it_is_busy() {
let spy = Spy::default();
let q = ChainQueue::new_with_poll(spy.sink(), std::time::Duration::from_millis(10));
let busy = Arc::new(std::sync::atomic::AtomicBool::new(true));
let b = busy.clone();
let q = q.with_defer_while(move || b.load(Ordering::SeqCst));
q.submit(spec("deferred", "g", 1));
let runs = Arc::new(AtomicUsize::new(0));
let r = runs.clone();
let q2 = q.clone();
let worker = tokio::spawn(async move {
q2.drain_n(1, move |_| {
r.fetch_add(1, Ordering::SeqCst);
Ok(serde_json::json!({}))
})
.await
});
tokio::time::sleep(std::time::Duration::from_millis(200)).await;
assert_eq!(
runs.load(Ordering::SeqCst),
0,
"nothing may run while the machine is busy"
);
busy.store(false, Ordering::SeqCst);
tokio::time::sleep(std::time::Duration::from_millis(100)).await;
worker.await.unwrap();
assert_eq!(
runs.load(Ordering::SeqCst),
1,
"and it runs the moment the machine goes quiet"
);
}
/// LAW 2 — **the guard that was missing, written from the field failure it
/// missed.** Measured on oden's live jera ledger 2026-08-22: a non-terminal
/// `workspace_populate` row 5.5 DAYS old (its process had been killed and so
/// never wrote a terminal record), beside 21-hour and 1-hour zombies. A
/// domain predicate reading "is a heavy job non-terminal" was therefore
/// permanently true, and the chain queued, superseded and re-queued for an
/// hour without folding once.
///
/// The deferral test above passes on that build — it only ever flips its
/// predicate to false. This one never does, which is the whole point: the
/// ceiling must let the work through anyway.
#[tokio::test]
async fn a_predicate_that_never_goes_false_does_not_defer_the_chain_for_ever() {
let spy = Spy::default();
let q = ChainQueue::new_with_poll(spy.sink(), std::time::Duration::from_millis(5));
// The zombie: busy, always, exactly as a stuck ledger row reads.
let q = q.with_defer_while(|| true);
let q = q.with_max_defer(Some(std::time::Duration::from_millis(60)));
q.submit(spec("must-still-run", "g", 1));
let ran = Arc::new(AtomicUsize::new(0));
let r = ran.clone();
// A HARD deadline, because the bug this guards is a HANG: without the
// ceiling `drain_n` never returns at all, and a test that hangs reports
// nothing — it has to FAIL.
let n = tokio::time::timeout(
std::time::Duration::from_secs(5),
q.drain_n(1, move |_| {
r.fetch_add(1, Ordering::SeqCst);
Ok(serde_json::json!({}))
}),
)
.await
.expect("the chain deferred FOR EVER — a zombie ledger row must not stop it");
assert_eq!(n, 1);
assert_eq!(
ran.load(Ordering::SeqCst),
1,
"past the ceiling the chain must proceed even though the domain still says busy"
);
assert_eq!(spy.last_status_of("must-still-run").as_deref(), Some(status::DONE));
}
/// The tests that touch the process-global exclusive gate, one at a time.
static GATE_TESTS: tokio::sync::Mutex<()> = tokio::sync::Mutex::const_new(());
/// ★ RA3-8: **lanes started at the same instant all start at once.** The worker's look at the
/// exclusive gate took the gate's one permit for an instant (`try_acquire`), so two lanes
/// looking together saw each other as a bench and the loser slept a whole `defer_poll` (2 s by
/// default; Gunnar worked around it with 20 ms). A look must never take the gate.
#[tokio::test(flavor = "multi_thread", worker_threads = 8)]
async fn lanes_started_at_the_same_instant_all_start_at_once() {
let _one_at_a_time = GATE_TESTS.lock().await;
let lanes: Vec<Arc<ChainQueue>> = (0..16).map(|_| ChainQueue::new(Spy::default().sink())).collect();
for q in &lanes {
q.submit(spec("at-once", "g", 1));
}
let t0 = std::time::Instant::now();
let started = Arc::new(std::sync::Mutex::new(Vec::new()));
let workers: Vec<_> = lanes
.iter()
.map(|q| {
let (q, started) = (q.clone(), started.clone());
tokio::spawn(async move {
q.drain_n(1, move |_| {
started.lock().unwrap().push(t0.elapsed());
Ok(serde_json::json!({}))
})
.await
})
})
.collect();
for w in workers {
w.await.unwrap();
}
let started = started.lock().unwrap().clone();
assert_eq!(started.len(), 16);
let slowest = started.iter().max().unwrap();
assert!(*slowest < std::time::Duration::from_millis(500), "a lane waited {slowest:?} for another lane's look at the gate");
}
/// …and the ceiling must NOT lift the EXCLUSIVE gate. A bench measuring the
/// box keeps it to itself however long the chain has been waiting —
/// otherwise the ceiling would have traded one silent wrong for another.
#[tokio::test]
async fn the_ceiling_does_not_lift_the_exclusive_gate() {
let _one_at_a_time = GATE_TESTS.lock().await;
let spy = Spy::default();
let q = ChainQueue::new_with_poll(spy.sink(), std::time::Duration::from_millis(5));
let q = q.with_max_defer(Some(std::time::Duration::from_millis(20)));
q.submit(spec("waits-for-the-bench", "g", 1));
// Hold the process-global exclusive gate, as a running bench does.
let held = super::super::exclusive_gate()
.clone()
.acquire_owned()
.await
.unwrap();
let ran = Arc::new(AtomicUsize::new(0));
let r = ran.clone();
let q2 = q.clone();
let worker = tokio::spawn(async move {
q2.drain_n(1, move |_| {
r.fetch_add(1, Ordering::SeqCst);
Ok(serde_json::json!({}))
})
.await
});
tokio::time::sleep(std::time::Duration::from_millis(200)).await;
assert_eq!(
ran.load(Ordering::SeqCst),
0,
"the exclusive gate does NOT expire — ten ceilings later a bench still owns the box"
);
drop(held);
tokio::time::sleep(std::time::Duration::from_millis(100)).await;
worker.await.unwrap();
assert_eq!(ran.load(Ordering::SeqCst), 1, "and it runs once the bench lets go");
}
/// The whole chain submits as a unit and every accepted job is visible in
/// the ledger as QUEUED before anything runs — that row is what the pane's
/// "queued in jera (#n)" is counting.
#[test]
fn a_whole_chain_submits_as_a_unit_and_is_visible_queued() {
let spy = Spy::default();
let q = ChainQueue::new(spy.sink());
let specs: Vec<ChainSpec> = ["a", "b", "c"]
.iter()
.map(|n| spec(&format!("lens:{n}"), "lens:nordisk", 7))
.collect();
let out = q.submit_all(specs);
assert_eq!(out.len(), 3);
assert!(out.iter().all(|s| s.is_new()));
assert_eq!(q.pending_in_group("lens:nordisk"), 3);
assert_eq!(q.pending_in_group("lens:other"), 0);
let log = spy.0.lock().unwrap();
assert_eq!(log.iter().filter(|r| r.status == status::QUEUED).count(), 3);
}
/// **A duplicate submit against a RUNNING key re-arms it — exactly once.**
///
/// MEASURED on oden 2026-09-10: a "make lens current" job is keyed without a
/// generation, so a republish that lands mid-fold submits the key, gets
/// `Duplicate` (the key is live), queues nothing — and the fold in progress is
/// stamped with the generation it STARTED at, which is now old. Without this
/// the lens stays stale until the next republish. The queue therefore
/// remembers that the running key was asked for again and pushes ONE fresh
/// pending copy when the run ends. One, not one per duplicate: three
/// republishes during one fold are one refold, because the refold reads the
/// warehouse as it is when it runs.
///
/// RED WHEN REVERTED: without the re-arm, `pending` is 0 after the drain and
/// the second `drain_n` runs nothing.
#[tokio::test]
async fn a_duplicate_submit_while_running_re_arms_the_key_once() {
let spy = Spy::default();
let q = ChainQueue::new_with_poll(spy.sink(), std::time::Duration::from_millis(5));
let first = q.submit(spec("lens:villa", "lens:nordisk", 0));
let first_id = first.job_id().unwrap().to_string();
// The runner blocks until the test says go, so the job is RUNNING while
// the duplicates arrive.
let (started_tx, started_rx) = std::sync::mpsc::channel::<()>();
let (go_tx, go_rx) = std::sync::mpsc::channel::<()>();
let go_rx = Arc::new(Mutex::new(go_rx));
let q2 = q.clone();
let worker = tokio::spawn(async move {
q2.drain_n(1, move |_| {
started_tx.send(()).unwrap();
go_rx.lock().unwrap().recv().unwrap();
Ok(serde_json::json!({}))
})
.await
});
tokio::task::spawn_blocking(move || started_rx.recv().unwrap())
.await
.unwrap();
// Three republishes land while the fold runs.
for _ in 0..3 {
let again = q.submit(spec("lens:villa", "lens:nordisk", 0));
assert!(!again.is_new(), "a live key queues nothing new");
assert_eq!(again.job_id(), Some(first_id.as_str()));
}
assert_eq!(q.stats().pending, 0, "nothing pending while the key is running");
go_tx.send(()).unwrap();
assert_eq!(worker.await.unwrap(), 1);
let s = q.stats();
assert_eq!(s.done, 1);
assert_eq!(
s.pending, 1,
"the run finished with a duplicate on record, so ONE fresh copy must be pending: {s:?}"
);
// The re-armed copy is a NEW job with its own queued row, not the old id
// resurrected — the old id is terminal in the ledger and must stay so.
let log = spy.0.lock().unwrap();
let queued_ids: Vec<&str> = log
.iter()
.filter(|r| r.status == status::QUEUED && r.target == "lens:villa")
.map(|r| r.job_id.as_str())
.collect();
assert_eq!(queued_ids.len(), 2, "one queued row per accepted copy: {queued_ids:?}");
assert_ne!(queued_ids[1], first_id);
drop(log);
assert_eq!(spy.last_status_of("lens:villa").as_deref(), Some(status::QUEUED));
// And the second drain runs it — then nothing more, because no duplicate
// arrived during THAT run.
let ran = Arc::new(AtomicUsize::new(0));
let r = ran.clone();
let n = q
.drain_n(3, move |_| {
r.fetch_add(1, Ordering::SeqCst);
Ok(serde_json::json!({}))
})
.await;
assert_eq!(n, 1, "exactly the one re-armed copy ran");
assert_eq!(q.stats().pending, 0);
}
/// **The domain deferral ceiling is CUMULATIVE across jobs, not restarted per
/// job.**
///
/// MEASURED on oden 2026-09-10: republishes every 2–3 minutes kept
/// `machine_busy` true almost continuously. The ceiling let ONE job through
/// every ten minutes, and then restarted the clock for the next — thirteen
/// lenses became two hours. The ceiling exists so a busy signal cannot hold
/// the chain off for ever; once it has expired the chain must DRAIN, and only
/// go back to yielding after the box has actually been quiet again.
///
/// Measured STRUCTURALLY, not by wall time or by a total: the box this runs
/// on is loaded (load 50+ on oden while this was written) and a 5 ms sleep
/// can take 80 ms, so any count that depends on how many polls fit in the
/// ceiling is a coin toss. What cannot vary: once the ceiling has expired,
/// the cumulative clock consults the predicate EXACTLY ONCE for each
/// following job (asks, sees expired, proceeds), whereas a per-job clock must
/// ask at least twice (the ask that starts its clock, then one more after a
/// sleep). The first job's count is only asserted to be ≥ 2, which is the
/// minimum for having waited at all.
///
/// RED WHEN REVERTED: jobs two and three each consult ≥ 2 times.
#[tokio::test]
async fn the_defer_ceiling_is_cumulative_across_jobs() {
let spy = Spy::default();
let poll = std::time::Duration::from_millis(5);
let ceiling = std::time::Duration::from_millis(80);
let q = ChainQueue::new_with_poll(spy.sink(), poll);
let asked = Arc::new(AtomicUsize::new(0));
let a = asked.clone();
let q = q.with_defer_while(move || {
a.fetch_add(1, Ordering::SeqCst);
true
});
let q = q.with_max_defer(Some(ceiling));
for k in ["a", "b", "c"] {
q.submit(spec(k, "g", 0));
}
// The ask-count as each job STARTS, so the per-job deltas are exact.
let at_start = Arc::new(Mutex::new(Vec::<usize>::new()));
let s = at_start.clone();
let a2 = asked.clone();
let n = tokio::time::timeout(
std::time::Duration::from_secs(10),
q.drain_n(3, move |_| {
s.lock().unwrap().push(a2.load(Ordering::SeqCst));
Ok(serde_json::json!({}))
}),
)
.await
.expect("must not hang");
assert_eq!(n, 3);
let at = at_start.lock().unwrap().clone();
assert_eq!(at.len(), 3);
assert!(at[0] >= 2, "the first job must have waited out the ceiling: asked {} times", at[0]);
assert_eq!(
at[1] - at[0],
1,
"past the ceiling the SECOND job must ask exactly once and proceed \
(a per-job clock would wait the ceiling out again): {at:?}"
);
assert_eq!(at[2] - at[1], 1, "and so must the third: {at:?}");
}
}