gunnar-server 3.0.0

//! **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");
    }
}