crates/gunnar-server/src/job_queue.rs
//! **One worker, one job at a time, FIFO, deduplicated by subject — for any
//! kind of job the box has to do slowly.**
//!
//! This module is the mechanism [`crate::gc_queue`] and [`crate::purge_queue`]
//! were each carrying their own copy of. The two copies were 1165 lines with
//! the same seven nouns in both — *dedup, enqueue, job_id, Queued, Running,
//! Done, Failed, worker* — the same `Mutex`/`Condvar`/`VecDeque`, the same
//! prune, the same in-flight instrument. What differed was two things and only
//! two: **what the work is**, and **what has to happen when it ends**. Both are
//! now the kind's business ([`JobKind::run`], [`JobKind::settle`]) and
//! everything else is here, once.
//!
//! # Why a queue at all
//!
//! `Repos.Gc` and `Accounts.Purge` exist so a console button can start
//! something expensive. The moment such a button exists, ten owners can press
//! ten of them inside one second, and neither job is cheap: a gc seals the live
//! logs, compacts the archive and unlinks the whole lazily-exploded read cache;
//! a purge is minutes of unlink I/O across an account. Ten of those at once is
//! not ten times the work, it is a box that has stopped answering.
//!
//! And the rule has to live in the server. A console can serialise **itself**.
//! It cannot serialise the CLI, a second console, a script, or the next client
//! anybody writes — and every one of those holds a credential that reaches the
//! same disk. A rule only one participant obeys is not an invariant; it is a
//! convention that holds until the first person who did not read it.
//!
//! # One worker per KIND, not one worker for the box
//!
//! [`JobQueue`] is a type, not a singleton: each kind gets its own instance and
//! therefore its own jera lane (A32-14) and its own line. That is deliberate and it
//! is the ruling `purge_queue.rs` already argued for, kept rather than quietly
//! reversed by this extraction:
//!
//! **The line a job waits in is part of what the job costs.** A gc waits
//! holding nothing — it is maintenance, it is re-derivable, and it can sit
//! behind anything without costing anybody anything. A purge waits holding a
//! **total lockout on a customer's account**, for the whole time it is queued.
//! Put them in one line and a user is shut out of their account while the box
//! grinds through an hour of somebody else's repository maintenance, because a
//! gc arrived first. A mirror is a third answer again: it is network-bound, it
//! blocks a repository that is *waiting to become usable*, and making it queue
//! behind a compaction that is only holding the disk would be the same mistake
//! in a third colour.
//!
//! So: one runner, three lines. What is shared is the mechanism and the
//! vocabulary; what is not shared is the waiting.
//!
//! The honest limit of that ruling: two kinds CAN therefore run at once — a gc
//! and a purge together is possible today and this change does not alter it —
//! and both are heavy disk I/O. Nobody has measured that hurting, so nothing
//! here pretends to fix it. If it ever does, the fix is a shared admission
//! token taken around the [`JobKind::run`] call in [`work`] and nowhere else:
//! one place, because there is now only one place.
//!
//! # What a finished job is for
//!
//! The worker does not drop a job when it finishes it. A console polls, and a
//! poller that happened to be between requests at the instant the job completed
//! would otherwise get "no such job" — indistinguishable from "you never
//! asked". So a completed job is kept for [`JobKind::FINISHED_RETENTION`], and
//! beyond that a bounded number of the most recent are kept anyway
//! ([`JobKind::FINISHED_KEPT`]), so a busy server forgets by age and a quiet one
//! does not forget at all.
//!
//! # Adding a kind
//!
//! Implement [`JobKind`]: name it, say what an order is, say what a report is,
//! say which field of the order is the key, and write [`JobKind::run`].
//! Everything else has a default. `tests/job_queue_third_kind.rs` is a whole
//! kind declared **in a test crate**, which is the strongest available proof
//! that a new kind needs no edit here — a test crate cannot change this file
//! even by accident.
use std::collections::{HashMap, VecDeque};
use std::fmt;
use std::ops::Deref;
use std::panic::AssertUnwindSafe;
use std::sync::atomic::{AtomicU64, AtomicUsize, Ordering};
use std::sync::{Arc, Mutex, Weak};
use std::time::{Duration, SystemTime, UNIX_EPOCH};
/// How long a finished job stays readable unless its kind says otherwise.
///
/// Ten minutes is far longer than any console poll interval and far shorter
/// than a session, so a button that completed is still reporting its result
/// when whoever pressed it looks back, and a server that has run thousands is
/// not holding all of them.
pub const DEFAULT_FINISHED_RETENTION: Duration = Duration::from_secs(600);
/// Finished jobs kept regardless of age, most recent first.
///
/// The floor under [`DEFAULT_FINISHED_RETENTION`]: a server that has done
/// nothing for an hour should still be able to say what its last job did.
/// Bounded, because "keep everything" is how a long-lived process grows a leak
/// with a good excuse.
pub const DEFAULT_FINISHED_KEPT: usize = 64;
/// Where a job is. One enum for every kind, because all three status RPCs
/// render the same four words and a per-kind copy of them was three chances to
/// disagree about what `Done` means.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum JobState {
/// Waiting its turn.
Queued,
/// The one job of this kind that is running.
Running,
/// Finished; the kind's report is the answer.
Done,
/// Finished badly; `error` says why.
Failed,
}
impl JobState {
/// Whether a *new* request for the same subject should join this job rather
/// than start another.
///
/// A queued or running job is the answer to "do this to this subject" — a
/// second request wants the same outcome and has nothing to add. A finished
/// one is not: the caller is asking for something that has not happened
/// *since*, and refusing them because a job ran an hour ago would be the
/// queue deciding it knows better.
pub fn is_live(self) -> bool {
matches!(self, JobState::Queued | JobState::Running)
}
}
/// **A kind of job.** The only thing a new kind has to write.
///
/// The queue owns waiting, ordering, dedup, the worker, the retention and the
/// instruments. The kind owns what the work *is* and — if it took something
/// while it ran — what has to be given back when it ends.
pub trait JobKind: Send + Sync + Sized + 'static {
/// The kind's short name. It is the job-id prefix (`gc-7`, `purge-12`,
/// `mirror-3`), the jera lane's job kind (`gc`), and the noun in
/// every message this module writes — so an operator reading a stack dump
/// or a FAILED job knows what it was without a lookup.
const NAME: &'static str;
/// How long a finished job of this kind stays readable.
///
/// Overridable rather than a shared constant because the two queues holding
/// the same number today is a coincidence and not a contract — a purge's
/// result is the more valuable of the two to still have when somebody looks
/// back, and a kind that decides so should be able to say it here rather
/// than by editing a number every kind shares.
const FINISHED_RETENTION: Duration = DEFAULT_FINISHED_RETENTION;
/// Finished jobs of this kind kept regardless of age.
const FINISHED_KEPT: usize = DEFAULT_FINISHED_KEPT;
/// **What was asked for**, carried on the job so the worker can hand it
/// back to [`JobKind::run`] minutes later.
///
/// The whole order and not just the subject: a purge's `remove_principal`
/// and `expect_empty` change what the job *is*, and a queue that kept them
/// in a side map keyed by subject would be a second place the two could
/// disagree — with the more destructive reading winning whenever they did.
type Order: Clone + fmt::Debug + Send + Sync + 'static;
/// **What the job produced**, in the kind's own words: a gc's bytes, a
/// purge's stores and gate, a mirror's objects.
///
/// The queue never looks inside it. It starts at [`Default::default`] —
/// which is why a job that never ran, or was refused, reports zeroes rather
/// than nothing.
type Report: Clone + fmt::Debug + Default + Send + Sync + 'static;
/// **The subject**, and therefore the dedup key and the status key.
///
/// A repository path for a gc or a mirror, an account for a purge. Both
/// questions the queue answers are per subject — "is there already one for
/// this" on enqueue, "how is mine going" on status — so this is the only
/// key it keeps.
fn key(order: &Self::Order) -> &str;
/// **Do the work.** Runs on the lane's blocking thread, outside the queue's lock,
/// inside a [`std::panic::catch_unwind`].
///
/// Report through `progress` as you go rather than in the return value: a
/// job of this sort runs for minutes with a console polling the whole time,
/// and a return value can only be read once the work is over — which is
/// precisely when nobody needs it any more.
fn run(&self, order: &Self::Order, progress: &Reporter<Self>) -> Result<(), String>;
/// **How the job settles — and the hook for anything that must come back.**
///
/// Called on **every** path out of [`JobKind::run`]: after `Ok`, after
/// `Err`, and after a caught panic. That is the whole reason it is the
/// queue's job and not the runner's. A purge takes a total lockout on an
/// account before it is even enqueued and something has to lift it; a
/// runner that returned early, failed, or blew up would each be a different
/// missed unlock, and "release it in the error path too" is a rule that
/// holds until the next early return somebody adds.
///
/// The default is the whole story for a kind that took nothing: `Ok` is
/// [`JobState::Done`], `Err` is [`JobState::Failed`] carrying the reason.
fn settle(
&self,
_order: &Self::Order,
outcome: Result<(), String>,
_progress: &Reporter<Self>,
) -> (JobState, Option<String>) {
match outcome {
Ok(()) => (JobState::Done, None),
Err(why) => (JobState::Failed, Some(why)),
}
}
/// What a job refused by a shutting-down queue says.
///
/// A shutting-down queue answers by name rather than accepting a job it
/// will never run: a console button lit for ever is worse than a refusal
/// that says why. Overridden by a kind whose caller has already paid
/// something for the job — a purge's caller is holding a lockout that
/// nothing is now going to lift, and has to be told so.
fn shutdown_error(&self) -> String {
format!("the {} queue is shutting down", Self::NAME)
}
/// What a job says when the kind's lane could not be started (no runtime).
///
/// The queue cannot run anything without its lane, and a job that sits
/// QUEUED for ever is a console button that never goes down. Said on the
/// job itself.
fn spawn_error(&self, err: &dyn fmt::Display) -> String {
format!("the {} worker could not be started: {err}", Self::NAME)
}
}
/// One job of kind `K`, queued or completed.
///
/// The fields above the line are the queue's and are the same for every kind —
/// which is what lets `GcStatus`, `PurgeStatus` and a future `MirrorStatus`
/// render from one shape. `report` is the kind's own, and [`Job`] dereferences
/// to it so a caller writes `job.bytes_reclaimed` for the gc's field and
/// `job.state` for the queue's, without having to know which half it is in.
pub struct Job<K: JobKind> {
pub job_id: String,
/// The subject, exactly as the queue keys on it — a repository path, an
/// account.
pub key: String,
pub state: JobState,
/// How many jobs are ahead of this one; 0 unless [`JobState::Queued`].
///
/// Computed from the waiting list at read time and never stored, because a
/// stored position is wrong the moment the job in front of it starts, and a
/// queue that reported stale positions would be worse than one that
/// reported none.
pub queue_position: u32,
/// **How far along, as `(done, total)`** — whatever the kind counts in.
///
/// `None` until the job reports one, and for kinds that never do: a gc and
/// a purge each finish in one indivisible act as far as a caller is
/// concerned, and a fabricated `0/1` would be a progress bar that lies. A
/// mirror cannot do without it — a fetch of a large repository is minutes
/// of network with somebody watching, and a job that sat at `Running` and
/// then jumped to `Done` tells that watcher nothing.
pub progress: Option<(u64, u64)>,
/// Only set when [`JobState::Failed`].
pub error: Option<String>,
/// What was asked for.
pub order: K::Order,
/// What the kind produced. Also reachable field-by-field through [`Deref`].
pub report: K::Report,
pub enqueued_at_unix_ms: i64,
/// 0 until the job leaves [`JobState::Queued`].
pub started_at_unix_ms: i64,
/// 0 until the job leaves [`JobState::Running`].
pub finished_at_unix_ms: i64,
}
/// **A job reads as one flat record**, the queue's fields and the kind's
/// together.
///
/// Not inheritance-by-`Deref`: a [`Job`] is an envelope the queue wraps around
/// exactly one `K::Report`, the way a newtype wraps its payload, and the target
/// is a plain data struct with no methods of its own — so there is nothing here
/// for a method call to resolve to by surprise. `job.report` is always
/// available for anyone who would rather be explicit.
impl<K: JobKind> Deref for Job<K> {
type Target = K::Report;
fn deref(&self) -> &K::Report {
&self.report
}
}
// Hand-written rather than derived: `derive` would demand `K: Clone` and
// `K: Debug` of the KIND, which holds the runner closures and is neither.
impl<K: JobKind> Clone for Job<K> {
fn clone(&self) -> Self {
Job {
job_id: self.job_id.clone(),
key: self.key.clone(),
state: self.state,
queue_position: self.queue_position,
progress: self.progress,
error: self.error.clone(),
order: self.order.clone(),
report: self.report.clone(),
enqueued_at_unix_ms: self.enqueued_at_unix_ms,
started_at_unix_ms: self.started_at_unix_ms,
finished_at_unix_ms: self.finished_at_unix_ms,
}
}
}
impl<K: JobKind> fmt::Debug for Job<K> {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
f.debug_struct("Job")
.field("kind", &K::NAME)
.field("job_id", &self.job_id)
.field("key", &self.key)
.field("state", &self.state)
.field("queue_position", &self.queue_position)
.field("progress", &self.progress)
.field("error", &self.error)
.field("order", &self.order)
.field("report", &self.report)
.field("enqueued_at_unix_ms", &self.enqueued_at_unix_ms)
.field("started_at_unix_ms", &self.started_at_unix_ms)
.field("finished_at_unix_ms", &self.finished_at_unix_ms)
.finish()
}
}
/// **The handle a running job reports through.**
///
/// Every method takes `&self` and writes into the job under the queue's lock,
/// so a console polling a status RPC sees a count that moves rather than a job
/// that sits at zero for four minutes and then jumps.
pub struct Reporter<K: JobKind> {
inner: Arc<Inner<K>>,
key: String,
}
impl<K: JobKind> Clone for Reporter<K> {
fn clone(&self) -> Self {
Reporter {
inner: Arc::clone(&self.inner),
key: self.key.clone(),
}
}
}
impl<K: JobKind> Reporter<K> {
/// The subject this job is working on, for a runner that wants it back.
pub fn key(&self) -> &str {
&self.key
}
/// **How far along**, in whatever the kind counts. Overwrites the previous
/// reading; a progress that accumulated across reports would be a second
/// counter to keep right.
pub fn progress(&self, done: u64, total: u64) {
self.edit(|job| job.progress = Some((done, total)));
}
/// Write into the kind's own report.
pub fn report(&self, apply: impl FnOnce(&mut K::Report)) {
self.edit(|job| apply(&mut job.report));
}
/// **The job as it stands right now**, for a runner that has to read back
/// what it reported.
///
/// The one caller is the mirror runner, which publishes an event-stream event
/// carrying the reference list it wrote through [`Reporter::report`]. That
/// list is on the JOB — it has to be, because a status RPC renders it —
/// and a runner that also kept its own copy would be two places the same
/// facts could disagree (one implementation).
///
/// `None` when the job is gone, which nothing does while a runner is
/// running: a live job is never pruned.
pub fn snapshot(&self) -> Option<Job<K>> {
let state = self.inner.state.lock().unwrap_or_else(|p| p.into_inner());
state.jobs.get(&self.key).cloned()
}
/// Edit the job record under the lock. Silently does nothing if the job is
/// gone, which nothing does today — a live job is never pruned.
fn edit(&self, apply: impl FnOnce(&mut Job<K>)) {
let mut state = self.inner.state.lock().unwrap_or_else(|p| p.into_inner());
if let Some(job) = state.jobs.get_mut(&self.key) {
apply(job);
}
}
}
/// The queue: one worker, one job at a time, FIFO, deduplicated by subject.
pub struct JobQueue<K: JobKind> {
inner: Arc<Inner<K>>,
}
impl<K: JobKind> Clone for JobQueue<K> {
fn clone(&self) -> Self {
JobQueue {
inner: Arc::clone(&self.inner),
}
}
}
struct Inner<K: JobKind> {
state: Mutex<State<K>>,
/// **The kind's jera lane** (A32-14): one ChainQueue, one job at a time, FIFO, started on the
/// first enqueue on the caller's runtime (or gunnar-common's own); its worker task is ended
/// when the last handle of the queue goes.
lane: Mutex<
Option<(
Arc<jera::jobs::chain::ChainQueue>,
tokio::task::JoinHandle<()>,
)>,
>,
kind: K,
/// **Runs in flight right now**, and the highest that has ever been.
///
/// The instrument the concurrency gate reads. It is bumped around the call
/// to [`JobKind::run`] rather than around the worker's bookkeeping, so what
/// a test observes is *jobs overlapping*, which is the thing that must
/// never happen — not the queue's opinion of how many are running.
in_flight: AtomicUsize,
peak_in_flight: AtomicUsize,
/// How many times the kind's `run` was CALLED. A dedup test that only
/// compared job ids could pass over a queue that ran the work twice under
/// one id; this is the applied output that cannot.
runs: AtomicU64,
seq: AtomicU64,
}
struct State<K: JobKind> {
/// Subject keys awaiting a turn, in the order they were asked for.
waiting: VecDeque<String>,
/// Every job the queue is holding, live and finished, keyed by SUBJECT and
/// not by job id — both questions anybody asks are per subject, and a job-id
/// map would be a second index over the same rows for no question at all.
jobs: HashMap<String, Job<K>>,
stopping: bool,
}
impl<K: JobKind> Default for State<K> {
fn default() -> Self {
State {
waiting: VecDeque::new(),
jobs: HashMap::new(),
stopping: false,
}
}
}
pub(crate) fn now_ms() -> i64 {
SystemTime::now()
.duration_since(UNIX_EPOCH)
.map(|d| d.as_millis() as i64)
.unwrap_or_default()
}
impl<K: JobKind> JobQueue<K> {
/// A queue over `kind`. No lane is started until something is enqueued.
pub fn new(kind: K) -> Self {
JobQueue {
inner: Arc::new(Inner {
state: Mutex::new(State::default()),
lane: Mutex::new(None),
kind,
in_flight: AtomicUsize::new(0),
peak_in_flight: AtomicUsize::new(0),
runs: AtomicU64::new(0),
seq: AtomicU64::new(0),
}),
}
}
/// **The job id the next enqueue would use**, without enqueuing.
///
/// For a kind that has to name the job before it exists. A purge takes the
/// account's lockout *before* enqueuing — that is what makes the lockout a
/// fact by the time `Accounts.Purge` returns — and the lockout's holder has
/// to be the job id, or the release could not tell this job's lockout from
/// anybody else's.
///
/// Minted from the same counter [`JobQueue::enqueue`] uses, so an id handed
/// out here and never used simply leaves a gap. A gap in a job-id sequence
/// costs nothing; two jobs sharing an id would make the release ambiguous.
pub fn next_job_id(&self) -> String {
self.mint()
}
/// **Ask for the order's subject to be worked on, and return without
/// working on it.**
///
/// Returns the job now responsible for the subject: a new one, or the one
/// already queued or running. Never enqueues a second job for a subject
/// that has one live.
pub fn enqueue(&self, order: K::Order) -> Job<K> {
self.enqueue_inner(order, None)
}
/// [`JobQueue::enqueue`] under an id taken from [`JobQueue::next_job_id`]
/// earlier.
///
/// The caller must not assume this id is the one that ends up responsible:
/// if a job for the same subject is already live, that one is returned and
/// this order's id is never used. `Accounts.Purge` handles exactly that by
/// reading the gate back out of the table rather than trusting the one it
/// wrote.
pub fn enqueue_as(&self, order: K::Order, job_id: String) -> Job<K> {
self.enqueue_inner(order, Some(job_id))
}
/// The job the queue is holding for `key`, if it still has one.
pub fn status(&self, key: &str) -> Option<Job<K>> {
let state = self.inner.state.lock().unwrap_or_else(|p| p.into_inner());
let job = state.jobs.get(key)?.clone();
Some(with_position(&state, job))
}
/// The highest number of jobs of this kind that have ever been running
/// together.
///
/// **Must be 1 after any amount of traffic** — that is the invariant this
/// module exists for, and this is how a test sees it rather than infers it.
/// 0 before anything has run.
pub fn peak_in_flight(&self) -> usize {
self.inner.peak_in_flight.load(Ordering::SeqCst)
}
/// How many times the kind's `run` has been called — i.e. how much work
/// actually happened. The applied output the dedup gate asserts on.
pub fn runs(&self) -> u64 {
self.inner.runs.load(Ordering::SeqCst)
}
/// Run nothing more after the job in flight, and refuse new jobs by name. For tests and the
/// box's stop; the lane's worker ends when the last handle of the queue goes.
pub fn stop(&self) {
let mut state = self.inner.state.lock().unwrap_or_else(|p| p.into_inner());
state.stopping = true;
}
/// **The queue's jera lane**, counted: its done and failed jobs (A32-14). `None` before the
/// first job, when no lane has started.
pub fn jera_stats(&self) -> Option<jera::jobs::chain::ChainStats> {
self.inner
.lane
.lock()
.unwrap_or_else(|p| p.into_inner())
.as_ref()
.map(|(q, _)| q.stats())
}
/// **The kind's lane**, started on the first ask: a jera ChainQueue whose worker runs one job
/// at a time on the runtime's blocking pool. Held weakly by its worker, so a queue nobody
/// holds any more ends it. `Err` only when no runtime can be had.
fn lane(&self) -> Result<Arc<jera::jobs::chain::ChainQueue>, String> {
let mut lane = self.inner.lane.lock().unwrap_or_else(|p| p.into_inner());
if let Some((q, _)) = lane.as_ref() {
return Ok(Arc::clone(q));
}
let rt = gunnar_common::jobs::handle()?;
// Every gc, purge and mirror is a row in the box's ledger once it is open (RA3-5).
// jera's own re-ask (RA3-8: jera 0.2.1 looks at the exclusive gate and never takes it).
let queue = jera::jobs::chain::ChainQueue::new(gunnar_common::jobs::ledger_sink());
let weak: Weak<Inner<K>> = Arc::downgrade(&self.inner);
let worker = rt.spawn(Arc::clone(&queue).run(move |spec| {
let key = spec
.payload
.get("key")
.and_then(|v| v.as_str())
.unwrap_or_default()
.to_owned();
let Some(inner) = weak.upgrade() else {
anyhow::bail!("the {} queue is gone", K::NAME)
};
run_one(&inner, &key).map_err(|why| anyhow::anyhow!(why))
}));
*lane = Some((Arc::clone(&queue), worker));
Ok(queue)
}
fn mint(&self) -> String {
let id = self.inner.seq.fetch_add(1, Ordering::SeqCst) + 1;
format!("{}-{id}", K::NAME)
}
fn enqueue_inner(&self, order: K::Order, preminted: Option<String>) -> Job<K> {
let key = K::key(&order).to_owned();
let mut state = self.inner.state.lock().unwrap_or_else(|p| p.into_inner());
if state.stopping {
let now = now_ms();
let refused: Job<K> = Job {
// Whatever the caller already named the job, so a purge's
// caller can still say which lockout is stranded. Empty for a
// kind that had not named one yet.
job_id: preminted.unwrap_or_default(),
key,
state: JobState::Failed,
queue_position: 0,
progress: None,
error: Some(self.inner.kind.shutdown_error()),
order,
report: K::Report::default(),
enqueued_at_unix_ms: now,
started_at_unix_ms: 0,
finished_at_unix_ms: now,
};
return refused;
}
if let Some(existing) = state.jobs.get(&key) {
if existing.state.is_live() {
// **The double press.** Not an error and not a new job — the
// caller wants this subject dealt with and something is already
// going to do that.
let job = existing.clone();
return with_position(&state, job);
}
}
// Minted here, AFTER the dedup check, so an id is spent only on a job
// that exists. A kind that had to name the job earlier passes its id in.
let job_id = match preminted {
Some(id) => id,
None => self.mint(),
};
let job: Job<K> = Job {
job_id,
key: key.clone(),
state: JobState::Queued,
queue_position: 0,
progress: None,
error: None,
order,
report: K::Report::default(),
enqueued_at_unix_ms: now_ms(),
started_at_unix_ms: 0,
finished_at_unix_ms: 0,
};
state.jobs.insert(key.clone(), job.clone());
state.waiting.push_back(key.clone());
// **A jera job on the kind's lane** (A32-14), one per job, keyed by its id: the lane runs
// them one at a time in the order asked, each a row that ends DONE or FAILED.
let lane = match self.lane() {
Ok(lane) => lane,
Err(err) => {
state.waiting.pop_back();
let message = self.inner.kind.spawn_error(&err);
let failed = state.jobs.get_mut(&key).expect("just inserted");
failed.state = JobState::Failed;
failed.error = Some(message);
failed.finished_at_unix_ms = now_ms();
return failed.clone();
}
};
let spec = jera::jobs::chain::ChainSpec::new(
K::NAME,
job.job_id.clone(),
K::NAME,
0,
"gunnar",
serde_json::json!({ "key": key }),
)
.with_target(format!("{} {key}", K::NAME))
.with_priority(jera::jobs::chain::Priority::Normal);
let _ = lane.submit(spec);
with_position(&state, job)
}
}
/// Fill in `queue_position` from the waiting list.
fn with_position<K: JobKind>(state: &State<K>, mut job: Job<K>) -> Job<K> {
job.queue_position = if job.state == JobState::Queued {
state
.waiting
.iter()
.position(|k| *k == job.key)
.map(|n| n as u32)
.unwrap_or(0)
} else {
0
};
job
}
/// **One job of the kind's lane** — the only place a job of this kind is started, and its lane
/// runs one at a time. `Err` (a FAILED jera row) when the job ended FAILED or panicked; a
/// stopping queue runs nothing (the job stays queued, as the old worker left it).
fn run_one<K: JobKind>(inner: &Arc<Inner<K>>, key: &str) -> Result<serde_json::Value, String> {
let (order, job_id) = {
let mut state = inner.state.lock().unwrap_or_else(|p| p.into_inner());
if state.stopping {
return Ok(
serde_json::json!({ "key": key, "ran": false, "why": "the queue is stopping" }),
);
}
if let Some(at) = state.waiting.iter().position(|k| k == key) {
state.waiting.remove(at);
}
match state.jobs.get_mut(key) {
Some(job) if job.state == JobState::Queued => {
job.state = JobState::Running;
job.started_at_unix_ms = now_ms();
(job.order.clone(), job.job_id.clone())
}
// The job was pruned or already run, which nothing does today. Skip rather than
// invent an order: a purge run from a default would be a purge nobody asked for.
_ => return Ok(serde_json::json!({ "key": key, "ran": false })),
}
};
let progress = Reporter {
inner: Arc::clone(inner),
key: key.to_owned(),
};
// ── outside the lock, deliberately ─────────────────────────────────
//
// These jobs take minutes. Holding the state lock across one would
// block every status poll and every enqueue for the whole run, so the
// console button this queue exists to serve would go grey exactly when
// somebody is watching it.
let live = inner.in_flight.fetch_add(1, Ordering::SeqCst) + 1;
inner.peak_in_flight.fetch_max(live, Ordering::SeqCst);
inner.runs.fetch_add(1, Ordering::SeqCst);
// **The panic is caught here and not left to unwind**: an unwinding job never reaches
// `settle` below, so whatever it took — a purge's lockout on a customer's account — would
// never be given back. Said as an ALERT line, and the lane's next job runs.
//
// `AssertUnwindSafe` because the kind is a `&dyn`-shaped thing we do
// not own and cannot prove unwind-safe. What it touches on this side of
// the boundary is the job record, and the worst a half-applied progress
// report leaves is a count that is short — which the FAILED state and
// the panic message already say not to trust.
let outcome =
match std::panic::catch_unwind(AssertUnwindSafe(|| inner.kind.run(&order, &progress))) {
Ok(result) => result,
Err(panic) => {
let why = format!("the {} panicked: {}", K::NAME, panic_text(&panic));
tracing::error!("ALERT gunnar-{} job {job_id} ({key}): {why}", K::NAME);
Err(why)
}
};
inner.in_flight.fetch_sub(1, Ordering::SeqCst);
// ── the settle, on every path ──────────────────────────────────────
//
// Unconditional, and it runs after a success, after a failure and after
// a panic. This is where a kind gives back whatever it took.
let (state_out, error) = inner.kind.settle(&order, outcome, &progress);
let mut state = inner.state.lock().unwrap_or_else(|p| p.into_inner());
if let Some(job) = state.jobs.get_mut(key) {
job.state = state_out;
job.error = error.clone();
job.finished_at_unix_ms = now_ms();
}
prune::<K>(&mut state);
match state_out {
JobState::Failed => Err(error.unwrap_or_else(|| format!("the {} job failed", K::NAME))),
_ => Ok(serde_json::json!({ "key": key, "job": job_id })),
}
}
impl<K: JobKind> Drop for Inner<K> {
/// The last handle of the queue is gone: its lane's idle worker ends with it.
fn drop(&mut self) {
if let Some((_, worker)) = self.lane.lock().unwrap_or_else(|p| p.into_inner()).take() {
worker.abort();
}
}
}
/// Forget finished jobs that are old, and the oldest of them once there are too
/// many. Live jobs are never pruned — a queue that forgot what it was about to
/// do would be a leak of a different kind, and a console polling a job that is
/// still running would get "no such job".
fn prune<K: JobKind>(state: &mut State<K>) {
let now = now_ms();
let cutoff = now - K::FINISHED_RETENTION.as_millis() as i64;
let mut finished: Vec<(String, i64)> = state
.jobs
.iter()
.filter(|(_, job)| !job.state.is_live())
.map(|(key, job)| (key.clone(), job.finished_at_unix_ms))
.collect();
// Newest first, so the keep-the-last-N floor keeps the useful ones.
finished.sort_by_key(|job| std::cmp::Reverse(job.1));
for (index, (key, finished_at)) in finished.into_iter().enumerate() {
if index >= K::FINISHED_KEPT || finished_at < cutoff {
state.jobs.remove(&key);
}
}
}
/// The message out of a caught panic, or a stand-in when it is neither of the
/// two shapes `panic!` produces.
pub(crate) fn panic_text(panic: &Box<dyn std::any::Any + Send>) -> String {
if let Some(text) = panic.downcast_ref::<&'static str>() {
return (*text).to_owned();
}
if let Some(text) = panic.downcast_ref::<String>() {
return text.clone();
}
"a panic with no message".to_owned()
}
#[cfg(test)]
mod tests {
use super::*;
/// A kind that does nothing, so the bookkeeping below is about the queue
/// and not about any real work.
struct Nothing;
impl JobKind for Nothing {
const NAME: &'static str = "nothing";
type Order = String;
type Report = ();
fn key(order: &String) -> &str {
order
}
fn run(&self, _order: &String, _progress: &Reporter<Self>) -> Result<(), String> {
Ok(())
}
}
/// A kind whose `boom` panics and whose `bad` fails.
struct Flaky;
impl JobKind for Flaky {
const NAME: &'static str = "flaky";
type Order = String;
type Report = ();
fn key(order: &String) -> &str {
order
}
fn run(&self, order: &String, _progress: &Reporter<Self>) -> Result<(), String> {
match order.as_str() {
"boom" => panic!("the boom job breaks"),
"bad" => Err("the bad job fails".to_owned()),
_ => Ok(()),
}
}
}
/// ★ **The queue's jobs are jera jobs** (A32-14): each one is a row on the queue's jera lane,
/// a panicked or failed one a FAILED row, and the next job runs; still one at a time, FIFO.
#[test]
fn the_jobs_run_as_jera_jobs_and_a_panic_is_a_failed_one() {
let q = JobQueue::new(Flaky);
for k in ["a", "boom", "b", "bad", "c"] {
q.enqueue(k.to_owned());
}
let t = std::time::Instant::now();
while ["a", "boom", "b", "bad", "c"]
.iter()
.any(|k| q.status(k).is_none_or(|j| j.state.is_live()))
&& t.elapsed() < Duration::from_secs(10)
{
std::thread::sleep(Duration::from_millis(10));
}
let state = |k: &str| q.status(k).map(|j| j.state);
assert_eq!(
[
state("a"),
state("boom"),
state("b"),
state("bad"),
state("c")
],
[
Some(JobState::Done),
Some(JobState::Failed),
Some(JobState::Done),
Some(JobState::Failed),
Some(JobState::Done)
]
);
assert!(
q.status("boom")
.and_then(|j| j.error)
.unwrap_or_default()
.contains("the boom job breaks"),
"the panic's words are on the job"
);
let s = q.jera_stats().expect("the queue runs on a jera lane");
assert_eq!(
(s.done, s.failed),
(3, 2),
"three done and two failed jera jobs: {s:?}"
);
assert_eq!(q.peak_in_flight(), 1, "one at a time");
q.stop();
}
/// ★ **A stopped queue refuses by name and runs nothing more**; its lane says so.
#[test]
fn a_stopped_queue_refuses_and_runs_nothing_more() {
let q = JobQueue::new(Flaky);
q.enqueue("a".to_owned());
let t = std::time::Instant::now();
while q.status("a").is_none_or(|j| j.state.is_live())
&& t.elapsed() < Duration::from_secs(10)
{
std::thread::sleep(Duration::from_millis(10));
}
assert!(
q.jera_stats().is_some_and(|s| s.done == 1),
"the first job is a done jera job"
);
q.stop();
let refused = q.enqueue("b".to_owned());
assert_eq!(refused.state, JobState::Failed);
assert!(refused.error.unwrap_or_default().contains("shutting down"));
std::thread::sleep(Duration::from_millis(100));
assert_eq!(q.runs(), 1, "nothing ran after the stop");
}
fn finished(key: &str, age: i64) -> Job<Nothing> {
Job {
job_id: format!("nothing-{key}"),
key: key.to_owned(),
state: JobState::Done,
queue_position: 0,
progress: None,
error: None,
order: key.to_owned(),
report: (),
enqueued_at_unix_ms: 0,
started_at_unix_ms: 0,
// Newer as `age` grows.
finished_at_unix_ms: now_ms() + age,
}
}
/// `prune` keeps the newest and drops the rest — asserted on which jobs
/// survive, not on the count, so a prune that kept the wrong ones is red.
#[test]
fn prune_keeps_the_newest_finished_jobs() {
let mut state = State::<Nothing>::default();
for n in 0..(Nothing::FINISHED_KEPT + 5) {
let key = format!("ns/repo{n}");
state.jobs.insert(key.clone(), finished(&key, n as i64));
}
prune::<Nothing>(&mut state);
assert_eq!(state.jobs.len(), Nothing::FINISHED_KEPT);
assert!(
state
.jobs
.contains_key(&format!("ns/repo{}", Nothing::FINISHED_KEPT + 4)),
"the newest finished job must survive"
);
assert!(
!state.jobs.contains_key("ns/repo0"),
"the oldest finished job must be dropped"
);
}
/// A live job is never pruned, however many finished ones crowd it out.
#[test]
fn prune_never_drops_a_live_job() {
let mut state = State::<Nothing>::default();
let mut running = finished("ns/running", 0);
running.state = JobState::Running;
running.finished_at_unix_ms = 0;
state.jobs.insert("ns/running".into(), running);
for n in 0..(Nothing::FINISHED_KEPT + 5) {
let key = format!("ns/repo{n}");
state.jobs.insert(key.clone(), finished(&key, n as i64));
}
prune::<Nothing>(&mut state);
assert!(
state.jobs.contains_key("ns/running"),
"a running job must survive a prune; forgetting it would leave a \
console button lit for ever"
);
}
/// A panic carrying a `String` and one carrying a `&'static str` both come
/// back as their message. Asserted because the fallback text is what an
/// operator would otherwise read instead of the cause.
#[test]
fn a_panics_message_survives_being_caught() {
let owned: Box<dyn std::any::Any + Send> = Box::new("a static message".to_string());
assert_eq!(panic_text(&owned), "a static message");
let borrowed: Box<dyn std::any::Any + Send> = Box::new("a literal message");
assert_eq!(panic_text(&borrowed), "a literal message");
let odd: Box<dyn std::any::Any + Send> = Box::new(7u32);
assert_eq!(panic_text(&odd), "a panic with no message");
}
}