gunnar-common 2.0.0

//! **A box's own background duty as named jera jobs** (feature `jobs`; A32-14, the owner's rule:
//! background work in a server runs as jera ChainQueue jobs, never a raw thread).
//!
//! [`nordisk_pairing::duty`] is the same contract on a named thread of its own; it stays for a loop whose
//! state is not `Send` (the pairing loop's tunnel driver, the twin follower). Everything else
//! runs here: a ticker queues one look at a time on the duty's ChainQueue, and the queue's worker
//! runs it on the runtime's blocking pool.
//!
//! * **Visible.** Every job this process started is in [`running`], with its looks and its
//!   failures; its queue's stats count the done and failed looks.
//! * **A failure is said.** A look that PANICS is an `ALERT <kind>` line (through the job's
//!   [`Say`]), counted, and a FAILED row; the next look runs as due.
//! * **It stops.** At [`Job::stop`] and [`stop_all`] (the box's clean stop), and when a look
//!   answers [`Round::Done`]. The wait between looks ends at the stop.
//! * **The rhythm is the duty's.** The first look runs at once; after a look the next waits the
//!   job's period ([`Round::Again`]) or the look's own wait ([`Round::After`], a backoff).

use std::panic::{catch_unwind, AssertUnwindSafe};
use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
use std::sync::{Arc, Mutex, OnceLock};
use std::time::{Duration, Instant};

use jera::jobs::chain::{ChainQueue, ChainSpec, Priority};
use jera::jobs::JobSink;

pub use nordisk_pairing::duty::{Round, Say};

/// The ledger rows' domain.
const DOMAIN: &str = "gunnar";

/// **How soon a lane asks again** when it lost jera's exclusive-gate probe. jera 0.2.0 probes a
/// one-permit semaphore with `try_acquire`, so two lanes starting a job at the same instant make
/// one of them wait its re-ask (2 s by default). A lost probe now costs this. (The probe itself
/// is jera's to fix.)
pub const DEFER_POLL: Duration = Duration::from_millis(20);

/// **One job, started**: its kind, its queue and what it has done.
pub struct Job {
    kind: String,
    queue: Arc<ChainQueue>,
    stopped: AtomicBool,
    in_look: AtomicBool,
    looks: AtomicU64,
    failures: AtomicU64,
    last_failure: Mutex<Option<String>>,
    wake: tokio::sync::Notify,
    tasks: Mutex<Vec<tokio::task::JoinHandle<()>>>,
}

impl Job {
    /// The job's kind (its jera rows' kind).
    pub fn kind(&self) -> &str {
        &self.kind
    }
    /// The looks run so far (the panicked ones too).
    pub fn looks(&self) -> u64 {
        self.looks.load(Ordering::SeqCst)
    }
    /// The looks that panicked.
    pub fn failures(&self) -> u64 {
        self.failures.load(Ordering::SeqCst)
    }
    /// The last failure, in words.
    pub fn last_failure(&self) -> Option<String> {
        self.last_failure.lock().unwrap_or_else(|p| p.into_inner()).clone()
    }
    /// Stopped: asked to, or a look answered [`Round::Done`].
    pub fn is_stopped(&self) -> bool {
        self.stopped.load(Ordering::SeqCst)
    }
    /// The job's jera queue (its stats: done, failed).
    pub fn queue(&self) -> &Arc<ChainQueue> {
        &self.queue
    }
    /// **Stop**, waiting up to `grace` for the look in flight. True when no look runs any more.
    pub fn stop(&self, grace: Duration) -> bool {
        self.tell_stop();
        let until = Instant::now() + grace;
        while self.in_look.load(Ordering::SeqCst) {
            if Instant::now() >= until {
                return false;
            }
            std::thread::sleep(Duration::from_millis(5));
        }
        true
    }

    /// No look is queued from now on; the ticker and the idle worker end now (a blocking look in
    /// flight cannot be cut, and ends on its own).
    fn tell_stop(&self) {
        self.stopped.store(true, Ordering::SeqCst);
        self.wake.notify_waiters();
        for t in self.tasks.lock().unwrap_or_else(|p| p.into_inner()).drain(..) {
            t.abort();
        }
    }

    fn failed(&self, n: u64, why: &str, say: &Say) {
        self.failures.fetch_add(1, Ordering::SeqCst);
        *self.last_failure.lock().unwrap_or_else(|p| p.into_inner()) = Some(why.to_string());
        say(&format!("ALERT {} look {n} {why}; the next look runs as due", self.kind));
    }
}

/// A panic's words.
fn panic_words(p: &(dyn std::any::Any + Send)) -> String {
    p.downcast_ref::<&str>().map(|s| s.to_string()).or_else(|| p.downcast_ref::<String>().cloned()).unwrap_or_else(|| "(no words)".to_string())
}

fn registry() -> &'static Mutex<Vec<Arc<Job>>> {
    static R: OnceLock<Mutex<Vec<Arc<Job>>>> = OnceLock::new();
    R.get_or_init(Mutex::default)
}

/// **Every job of this process that still runs.**
pub fn running() -> Vec<Arc<Job>> {
    registry().lock().unwrap_or_else(|p| p.into_inner()).iter().filter(|j| !j.is_stopped()).cloned().collect()
}

/// The running job of `kind`, if one runs.
pub fn find(kind: &str) -> Option<Arc<Job>> {
    running().into_iter().find(|j| j.kind == kind)
}

/// **Stop every job** (the box's clean stop), all within the one `grace`. Answers how many ran.
pub fn stop_all(grace: Duration) -> usize {
    let all = running();
    // Every one is told first, so one whose look cannot end soon costs the others nothing.
    for j in &all {
        j.tell_stop();
    }
    let until = Instant::now() + grace;
    for j in &all {
        j.stop(until.saturating_duration_since(Instant::now()));
    }
    // The lanes were cut mid-look: their rows say stopped, not running for ever (RA4-5).
    close_orphans("the box stopped before this job ended");
    all.len()
}

/// A duty looked at more often than this writes no ledger rows (a row every two seconds for the
/// box's life would be redb commits for nothing); its looks are still counted and its failures said.
pub const RECORD_MIN: Duration = Duration::from_secs(30);

/// **Keep this process's jera ledger under `dir`** (the box's `<data>/jobs`, RA3-5): from now on
/// the job queue's jobs, and the looks of every duty looked at no more often than [`RECORD_MIN`],
/// are rows there (queued, running, then done or FAILED), read by `nornir jobs` and the Jera tab.
pub fn set_ledger(dir: &std::path::Path) -> Result<(), String> {
    let store = jera::jobs::JobStore::open(dir).map_err(|e| format!("the jobs' ledger under {} does not open: {e:#}", dir.display()))?;
    *ledger().lock().unwrap_or_else(|p| p.into_inner()) = Some(Arc::new(store));
    let closed = close_orphans("the box restarted before this job ended");
    if closed > 0 {
        tracing::warn!("the jobs' ledger: {closed} row(s) a stopped run left running are marked stopped");
    }
    Ok(())
}

/// **Close the rows a stopped or crashed run left running** (RA4-5): every non-terminal row in
/// the ledger is marked stopped with `why`; jera prunes only finished rows, so a row a killed
/// lane never finished would show running for ever. Answers how many it closed.
pub fn close_orphans(why: &str) -> usize {
    let Some(store) = ledger().lock().unwrap_or_else(|p| p.into_inner()).clone() else { return 0 };
    let open: std::collections::BTreeSet<String> = store.list(&jera::jobs::JobSelector::All).unwrap_or_default().into_iter().filter(|r| !r.is_terminal()).map(|r| r.kind).collect();
    let kinds: Vec<&str> = open.iter().map(String::as_str).collect();
    if kinds.is_empty() {
        0
    } else {
        store.stop_orphans(&kinds, why)
    }
}

fn ledger() -> &'static Mutex<Option<Arc<jera::jobs::JobStore>>> {
    static L: OnceLock<Mutex<Option<Arc<jera::jobs::JobStore>>>> = OnceLock::new();
    L.get_or_init(Mutex::default)
}

/// The sink a new lane writes to: the ledger's, or none.
pub fn ledger_sink() -> JobSink {
    ledger().lock().unwrap_or_else(|p| p.into_inner()).as_ref().map(|s| s.sink()).unwrap_or_else(JobSink::noop)
}

/// The ledger's rows of `kind`, newest last (empty with no ledger).
pub fn ledger_rows(kind: &str) -> Vec<jera::jobs::JobRecord> {
    let store = ledger().lock().unwrap_or_else(|p| p.into_inner()).clone();
    store.and_then(|s| s.list(&jera::jobs::JobSelector::Kind(kind.to_string())).ok()).unwrap_or_default()
}

/// **The runtime a job runs on**: the caller's, or (a caller on a plain thread) this module's own
/// small one, started once. `Err` only when no runtime can be built at all.
pub fn handle() -> Result<tokio::runtime::Handle, String> {
    if let Ok(h) = tokio::runtime::Handle::try_current() {
        return Ok(h);
    }
    static OWN: OnceLock<Result<tokio::runtime::Runtime, String>> = OnceLock::new();
    OWN.get_or_init(|| tokio::runtime::Builder::new_multi_thread().worker_threads(2).thread_name("gunnar-jobs").enable_all().build().map_err(|e| format!("the jobs' runtime cannot start: {e}")))
        .as_ref()
        .map(|rt| rt.handle().clone())
        .map_err(Clone::clone)
}

/// **Start a job of `kind`**: a look at once, then each next look after `every` (or the look's
/// own [`Round::After`]), until it is stopped or a look answers [`Round::Done`]. A job that
/// cannot start is an answer, never a panic.
pub fn start(kind: &str, every: Duration, say: Say, look: impl FnMut() -> Round + Send + 'static) -> Result<Arc<Job>, String> {
    let rt = handle().map_err(|e| format!("the job `{kind}` cannot start: {e}"))?;
    let job = Arc::new(Job {
        kind: kind.to_string(),
        // TODO(RA3-8): drop DEFER_POLL for jera's default once jera 0.2.1 (its gate fix) is published.
        queue: ChainQueue::new_with_poll(if every >= RECORD_MIN { ledger_sink() } else { JobSink::noop() }, DEFER_POLL),
        stopped: AtomicBool::new(false),
        in_look: AtomicBool::new(false),
        looks: AtomicU64::new(0),
        failures: AtomicU64::new(0),
        last_failure: Mutex::new(None),
        wake: tokio::sync::Notify::new(),
        tasks: Mutex::new(Vec::new()),
    });
    // The look's answer goes back to the ticker, which waits for it before it times the next.
    let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel::<Round>();
    let look = Mutex::new(look);
    let j = Arc::clone(&job);
    let worker = rt.spawn(Arc::clone(&job.queue).run(move |spec| {
        let n = u64::try_from(spec.generation).unwrap_or(0);
        if j.is_stopped() {
            let _ = tx.send(Round::Done);
            return Ok(serde_json::json!({ "look": n, "ran": false, "why": "the job is stopped" }));
        }
        j.in_look.store(true, Ordering::SeqCst);
        j.looks.fetch_add(1, Ordering::SeqCst);
        let answer = catch_unwind(AssertUnwindSafe(|| (look.lock().unwrap_or_else(|p| p.into_inner()))()));
        j.in_look.store(false, Ordering::SeqCst);
        match answer {
            Ok(round) => {
                let _ = tx.send(round);
                Ok(serde_json::json!({ "look": n, "done": round == Round::Done }))
            }
            Err(p) => {
                let why = format!("panicked: {}", panic_words(p.as_ref()));
                j.failed(n, &why, &say);
                let _ = tx.send(Round::Again);
                Err(anyhow::anyhow!(why))
            }
        }
    }));
    let j = Arc::clone(&job);
    let ticker = rt.spawn(async move {
        let mut n = 0u64;
        loop {
            n += 1;
            let generation = i64::try_from(n).unwrap_or(i64::MAX);
            let spec = ChainSpec::new(j.kind.as_str(), format!("{}-{n}", j.kind), j.kind.as_str(), generation, DOMAIN, serde_json::json!({ "look": n })).with_target(format!("look {n} of {}", j.kind)).with_priority(Priority::Normal);
            let _ = j.queue.submit(spec);
            let next = match rx.recv().await {
                Some(Round::Done) | None => {
                    j.stopped.store(true, Ordering::SeqCst);
                    return;
                }
                Some(Round::Again) => every,
                Some(Round::After(d)) => d,
            };
            tokio::select! {
                _ = tokio::time::sleep(next) => {}
                _ = j.wake.notified() => {}
            }
            if j.is_stopped() {
                return;
            }
        }
    });
    job.tasks.lock().unwrap_or_else(|p| p.into_inner()).extend([worker, ticker]);
    let mut all = registry().lock().unwrap_or_else(|p| p.into_inner());
    all.retain(|j| !j.is_stopped());
    all.push(Arc::clone(&job));
    drop(all);
    Ok(job)
}

#[cfg(test)]
mod tests {
    use super::*;

    /// The jobs share one registry, and [`stop_all`] stops every one: one test at a time.
    static ONE: Mutex<()> = Mutex::new(());

    fn heard() -> (Say, Arc<Mutex<Vec<String>>>) {
        let lines = Arc::new(Mutex::new(Vec::new()));
        let l = Arc::clone(&lines);
        (Arc::new(move |s: &str| l.lock().unwrap().push(s.to_string())), lines)
    }

    fn until(what: impl Fn() -> bool) -> bool {
        let t = Instant::now();
        while t.elapsed() < Duration::from_secs(10) {
            if what() {
                return true;
            }
            std::thread::sleep(Duration::from_millis(10));
        }
        false
    }

    const FAST: Duration = Duration::from_millis(20);

    /// ★ **A look that panics is said, counted, a FAILED jera job, and the job goes on** (A32-14).
    #[test]
    fn a_look_that_panics_is_said_counted_failed_and_the_job_goes_on() {
        let _one = ONE.lock().unwrap_or_else(|p| p.into_inner());
        let (say, lines) = heard();
        let mut n = 0;
        let j = start("test.panics", FAST, say, move || {
            n += 1;
            if n == 1 {
                panic!("the first look breaks")
            }
            Round::Again
        })
        .expect("a job starts");
        assert!(until(|| j.looks() >= 3), "the looks after the panic run (looks {})", j.looks());
        assert_eq!(j.failures(), 1, "the panic is counted");
        assert!(j.last_failure().unwrap_or_default().contains("the first look breaks"), "{:?}", j.last_failure());
        let said = lines.lock().unwrap().join("\n");
        assert!(said.contains("ALERT test.panics") && said.contains("panicked"), "the panic is an ALERT line: {said:?}");
        assert!(until(|| j.queue().stats().failed >= 1 && j.queue().stats().done >= 2), "the panicked look is a FAILED job, the others DONE: {:?}", j.queue().stats());
        j.stop(Duration::from_secs(1));
    }

    /// ★ **A stopped job looks no more, and is not listed** (the box's clean stop).
    #[test]
    fn a_stopped_job_looks_no_more() {
        let _one = ONE.lock().unwrap_or_else(|p| p.into_inner());
        let (say, _) = heard();
        let j = start("test.stops", FAST, say, || Round::Again).expect("a job starts");
        assert!(until(|| j.looks() >= 1));
        assert!(find("test.stops").is_some(), "a running job is listed");
        assert!(stop_all(Duration::from_secs(1)) >= 1);
        assert!(j.is_stopped() && find("test.stops").is_none(), "a stopped job is not listed");
        std::thread::sleep(Duration::from_millis(100));
        let at = j.looks();
        std::thread::sleep(Duration::from_millis(200));
        assert_eq!(j.looks(), at, "no look runs after the stop");
    }

    /// ★ **The stop ends a long wait at once** (a period of an hour does not hold the stop).
    #[test]
    fn the_stop_ends_a_long_wait_at_once() {
        let _one = ONE.lock().unwrap_or_else(|p| p.into_inner());
        let (say, _) = heard();
        let j = start("test.long-wait", Duration::from_secs(3600), say, || Round::Again).expect("a job starts");
        assert!(until(|| j.looks() == 1), "the first look runs at once");
        let t = Instant::now();
        assert!(j.stop(Duration::from_secs(2)), "stopped");
        assert!(t.elapsed() < Duration::from_secs(1), "the stop did not wait out the period");
    }

    /// ★ **Done ends the job; After is the look's own wait** (a retry with a backoff).
    #[test]
    fn done_ends_the_job_and_after_is_the_looks_own_wait() {
        let _one = ONE.lock().unwrap_or_else(|p| p.into_inner());
        let (say, _) = heard();
        let mut n = 0;
        // A period of an hour: only the looks' own short waits can bring the next looks.
        let j = start("test.after", Duration::from_secs(3600), say, move || {
            n += 1;
            if n >= 3 {
                Round::Done
            } else {
                Round::After(Duration::from_millis(10))
            }
        })
        .expect("a job starts");
        assert!(until(|| j.is_stopped()), "Done stops the job (looks {})", j.looks());
        std::thread::sleep(Duration::from_millis(150));
        assert_eq!(j.looks(), 3, "no look after Done");
        assert!(find("test.after").is_none());
    }

    /// ★ **Jobs started at the same instant all look at once** (jera's exclusive gate is a
    /// one-permit probe: two lanes asking together, one lost and waited its queue's re-ask,
    /// 2 s by default).
    #[test]
    fn jobs_started_at_the_same_instant_all_look_at_once() {
        let _one = ONE.lock().unwrap_or_else(|p| p.into_inner());
        let looked = Arc::new(AtomicU64::new(0));
        let began = Instant::now();
        // A server's runtime: many workers, so the lanes' first probes of the gate meet.
        let rt = tokio::runtime::Builder::new_multi_thread().worker_threads(8).enable_all().build().unwrap();
        let together = Arc::new(std::sync::Barrier::new(8));
        let starters: Vec<_> = (0..8)
            .map(|i| {
                let l = Arc::clone(&looked);
                let together = Arc::clone(&together);
                let h = rt.handle().clone();
                std::thread::spawn(move || {
                    let _on = h.enter();
                    let (say, _) = heard();
                    together.wait();
                    start(&format!("test.burst-{i}"), Duration::from_secs(3600), say, move || {
                        l.fetch_add(1, Ordering::SeqCst);
                        Round::Done
                    })
                    .expect("a job starts")
                })
            })
            .collect();
        let jobs: Vec<_> = starters.into_iter().map(|t| t.join().unwrap()).collect();
        assert!(until(|| looked.load(Ordering::SeqCst) == 8));
        assert!(began.elapsed() < Duration::from_millis(1000), "eight first looks took {:?}", began.elapsed());
        drop(jobs);
    }

    /// ★ **With a ledger, a recorded duty's looks are jera rows, a panicked one a FAILED row**
    /// (RA3-5: the lanes wrote to `JobSink::noop()`, so nothing showed in `nornir jobs`).
    #[test]
    fn with_a_ledger_a_recorded_dutys_looks_are_rows_and_a_panic_a_failed_row() {
        let _one = ONE.lock().unwrap_or_else(|p| p.into_inner());
        let dir = std::env::temp_dir().join(format!("gunnar-jobs-ledger-{}", std::process::id()));
        let _ = std::fs::remove_dir_all(&dir);
        set_ledger(&dir).expect("the ledger opens");
        let (say, _) = heard();
        let mut n = 0;
        let j = start("test.recorded", RECORD_MIN, say, move || {
            n += 1;
            match n {
                1 => Round::After(Duration::from_millis(10)),
                _ => panic!("the second look breaks"),
            }
        })
        .expect("a job starts");
        assert!(until(|| ledger_rows("test.recorded").iter().filter(|r| r.status == "done" || r.status == "failed").count() >= 2), "two finished rows: {:?}", ledger_rows("test.recorded").iter().map(|r| r.status.clone()).collect::<Vec<_>>());
        let rows = ledger_rows("test.recorded");
        assert!(rows.iter().any(|r| r.status == "done") && rows.iter().any(|r| r.status == "failed"), "a done row and a FAILED row");
        j.stop(Duration::from_secs(1));
        // A duty looked at every 20 ms writes nothing.
        let (say, _) = heard();
        let quick = start("test.unrecorded", FAST, say, || Round::Again).expect("a job starts");
        assert!(until(|| quick.looks() >= 3));
        assert!(ledger_rows("test.unrecorded").is_empty(), "a frequent duty writes no rows");
        quick.stop(Duration::from_secs(1));
    }

    /// ★ **A row a stopped run left running reads stopped once the ledger opens again** (RA4-5).
    #[test]
    fn a_row_left_running_reads_stopped_after_the_ledger_opens() {
        let _one = ONE.lock().unwrap_or_else(|p| p.into_inner());
        let dir = std::env::temp_dir().join(format!("gunnar-jobs-orphan-{}", std::process::id()));
        let _ = std::fs::remove_dir_all(&dir);
        {
            let store = jera::jobs::JobStore::open(&dir).expect("a ledger");
            let q = ChainQueue::new(store.sink());
            let _ = q.submit(ChainSpec::new("test.orphan", "k", "test.orphan", 0, DOMAIN, serde_json::json!({})));
        }
        set_ledger(&dir).expect("the ledger opens");
        let rows = ledger_rows("test.orphan");
        assert!(!rows.is_empty() && rows.iter().all(|r| r.status == "stopped"), "{:?}", rows.iter().map(|r| r.status.clone()).collect::<Vec<_>>());
    }

    /// **A job started from a plain thread runs** on the module's own runtime.
    #[test]
    fn a_job_started_with_no_runtime_runs() {
        let _one = ONE.lock().unwrap_or_else(|p| p.into_inner());
        let (say, _) = heard();
        let j = std::thread::spawn(move || start("test.no-runtime", FAST, say, || Round::Done)).join().unwrap().expect("a job starts");
        assert!(until(|| j.is_stopped() && j.looks() == 1));
    }

    /// **A job started on the caller's runtime runs there.**
    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
    #[allow(clippy::await_holding_lock)] // the registry lock is the point: one test at a time
    async fn a_job_started_on_a_runtime_runs_there() {
        let _one = ONE.lock().unwrap_or_else(|p| p.into_inner());
        let (say, _) = heard();
        let j = start("test.on-runtime", FAST, say, || Round::Done).expect("a job starts");
        let t = Instant::now();
        while !(j.is_stopped() && j.looks() == 1) && t.elapsed() < Duration::from_secs(10) {
            tokio::time::sleep(Duration::from_millis(10)).await;
        }
        assert!(j.is_stopped() && j.looks() == 1);
    }
}