crates/jera/src/jobs/report.rs
//! **Job reporting** — a typed, OPT-IN "what is this job doing right now" channel,
//! orthogonal to [`crate::jobs::JobRecord`]'s lifecycle status (`running →
//! done|failed`). Not every job is a progress-bar job:
//!
//! - A long compute/transfer job reports [`JobReport::Progress`] — a determinate
//! `current`/`total` (a progress bar) or an indeterminate spinner (`total: None`).
//! - A long-lived **service** (e.g. a container brought up by
//! [`crate::container`]) reports [`JobReport::Health`] — it's alive / servicing /
//! gone / failed, NOT a percentage. This is the korp "is the FalkorDB container
//! alive or gone" use-case.
//! - Some jobs expose a [`JobReport::Gauge`] a UI reads directly (rows/s, MB,
//! connections, …).
//! - Any job can drop a last human-readable [`JobReport::Message`].
//!
//! A [`Reporter`] is a cheap-to-clone handle a running job uses to emit reports
//! without knowing anything about the ledger; a [`ReportSink`] is where those
//! reports land (mirrors [`crate::jobs::JobSink`]'s write-side seam). The
//! [`JobStore`](crate::jobs::JobStore) keeps the **latest report of each kind**
//! per job id in an in-memory registry — reports are live/ephemeral status, not a
//! durable lifecycle fact, so (unlike [`crate::jobs::JobRecord`]) they are never
//! written to `jobs.redb`: a progress tick every few milliseconds would otherwise
//! amplify into redb writes.
//!
//! Every symbol here is additive: existing [`crate::jobs::JobHandle`] callers that
//! never touch [`Reporter`] keep compiling and behave exactly as before — their
//! report map just stays empty.
use std::collections::BTreeMap;
use std::sync::atomic::{AtomicU64, Ordering};
use std::sync::{Arc, Mutex};
use serde::{Deserialize, Serialize};
/// A job id — a plain alias over the `String` used everywhere in [`crate::jobs`]
/// (kept as its own name here so report-reading signatures self-document).
pub type JobId = String;
// ---------------------------------------------------------------------------
// Health — liveness/health of a long-lived service (not a percentage).
// ---------------------------------------------------------------------------
/// Liveness/health of a long-lived service a job is fronting (e.g. a container).
/// Distinct from [`JobReport::Progress`]: a service doesn't have a "% done", it is
/// either coming up, up, degraded, or gone.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum Health {
/// Bring-up issued, not yet confirmed live (e.g. a container just created).
Starting,
/// Up and reachable, but not yet confirmed to be doing useful work.
Alive,
/// Up AND actively serving (the strongest positive signal).
Servicing,
/// Up but impaired (e.g. failing its own health check) — still alive, worse
/// than [`Health::Alive`].
Degraded,
/// Exited/removed cleanly — no longer present, but not an error (e.g. a
/// container that stopped or was killed on purpose).
Gone,
/// Died / errored out.
Failed,
}
// ---------------------------------------------------------------------------
// JobReport — the typed report model.
// ---------------------------------------------------------------------------
/// One typed report a running job emits about itself. See the module docs for the
/// four kinds and when to use each. `#[serde(tag = "kind", content = "data")]` so
/// the wire form is a self-describing `{"kind": "progress", "data": {...}}` object
/// — stable across additions, and (unlike a purely-internal tag) works for the
/// newtype variants ([`JobReport::Health`]/[`JobReport::Message`]) too.
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
#[serde(tag = "kind", content = "data", rename_all = "snake_case")]
pub enum JobReport {
/// A determinate/indeterminate progress bar: `current` of `total` items/bytes
/// (`total: None` → an indeterminate spinner). `note` is an optional short
/// caption (e.g. "downloading layer 3/5").
Progress { current: u64, total: Option<u64>, note: Option<String> },
/// Liveness/health of a long-lived service — see [`Health`].
Health(Health),
/// A metric/gauge a UI can read directly (e.g. `("rows_per_s", 1234.0,
/// Some("rows/s"))`). `key` distinguishes multiple gauges on the same job —
/// each is tracked as its own latest-value slot (see [`JobReport::report_key`]).
Gauge { key: String, value: f64, unit: Option<String> },
/// The last human-readable status line (a free-form log tail of one).
Message(String),
/// The current **state-machine phase** of a long-lived **daemon** job — a named
/// phase (e.g. `"downloading"`, `"indexing"`, `"cutting"`, `"online"`) plus a
/// `healthy` flag. Distinct from [`JobReport::Progress`] (a `%`) and
/// [`JobReport::Health`] (the service-liveness enum): a daemon carries a *domain*
/// state-machine phase name the Jobs view shows verbatim, while `healthy` is the
/// green/amber signal the infra matrix colours on. Progress% still rides the
/// [`JobReport::Progress`] report alongside this — a daemon typically emits both
/// (phase name + phase %). See [`crate::jobs::kind::DAEMON`].
Phase { name: String, healthy: bool },
}
impl JobReport {
/// The map key this report's *kind* occupies in a [`JobReports`] — the ledger
/// keeps the LATEST report per key, not a full history. `Progress`/`Health`/
/// `Message` each have exactly one slot; `Gauge` gets one slot **per `key`**
/// (`gauge:rows_per_s` and `gauge:connections` coexist independently).
pub fn report_key(&self) -> String {
match self {
JobReport::Progress { .. } => "progress".to_string(),
JobReport::Health(_) => "health".to_string(),
JobReport::Gauge { key, .. } => format!("gauge:{key}"),
JobReport::Message(_) => "message".to_string(),
JobReport::Phase { .. } => "phase".to_string(),
}
}
}
// ---------------------------------------------------------------------------
// JobReports — the latest-per-kind snapshot the ledger keeps for one job.
// ---------------------------------------------------------------------------
/// The latest report of each kind for one job, plus a monotonic `updated_at`
/// sequence (a process-wide Lamport-style counter — **not** wall-clock; see the
/// module docs). A UI diffs `updated_at` across polls to know whether to re-render.
#[derive(Debug, Clone, Default, PartialEq, Serialize, Deserialize)]
pub struct JobReports {
/// `report_key() -> latest report`, e.g. `"progress"`, `"health"`,
/// `"gauge:rows_per_s"`, `"message"`.
pub by_key: BTreeMap<String, JobReport>,
/// Bumped on every accepted report (process-wide, monotonic — not a timestamp).
pub updated_at: u64,
}
impl JobReports {
/// The latest [`JobReport::Progress`], if the job has reported one.
pub fn progress(&self) -> Option<&JobReport> {
self.by_key.get("progress")
}
/// The latest [`Health`], if the job has reported one.
pub fn health(&self) -> Option<Health> {
match self.by_key.get("health") {
Some(JobReport::Health(h)) => Some(*h),
_ => None,
}
}
/// The latest daemon **phase** — `(name, healthy)` — if the job has reported one
/// (see [`JobReport::Phase`]). A daemon-job Jobs-view row shows `name` as the live
/// state-machine phase and colours on `healthy`.
pub fn phase(&self) -> Option<(&str, bool)> {
match self.by_key.get("phase") {
Some(JobReport::Phase { name, healthy }) => Some((name.as_str(), *healthy)),
_ => None,
}
}
/// The latest daemon phase NAME, if any (see [`Self::phase`]).
pub fn phase_name(&self) -> Option<&str> {
self.phase().map(|(n, _)| n)
}
/// The latest daemon phase `healthy` flag, if a phase has been reported (see
/// [`Self::phase`]).
pub fn phase_healthy(&self) -> Option<bool> {
self.phase().map(|(_, h)| h)
}
/// The **determinate** progress fraction in `[0,1]` from the latest
/// [`JobReport::Progress`], or `None` when there is no progress report OR it is
/// indeterminate (`total` absent/zero). A UI host maps this straight onto a job
/// row's progress-bar fill.
pub fn progress_fraction(&self) -> Option<f32> {
match self.by_key.get("progress") {
Some(JobReport::Progress { current, total: Some(t), .. }) if *t > 0 => {
Some((*current as f64 / *t as f64).clamp(0.0, 1.0) as f32)
}
_ => None,
}
}
/// `true` when the job has an active but **indeterminate** progress report — a
/// running spinner: a [`JobReport::Progress`] with no `total`. Distinct from "no
/// progress reported at all" (which returns `false` here).
pub fn progress_indeterminate(&self) -> bool {
matches!(self.by_key.get("progress"), Some(JobReport::Progress { total: None, .. }))
}
/// A compact **meta value** a `facett_jobview` host stamps under its `"progress"`
/// row-meta key: a determinate fraction (e.g. `"0.40"`), the literal
/// `"indeterminate"`, or `None` when the job reported no progress. This is the
/// one-call bridge from jera's report channel to the jobview row-meter
/// convention — jera stays free of any UI-crate dependency (the string contract
/// is all the two sides share). See `facett_jobview::PROGRESS_META_KEY`.
pub fn progress_meta(&self) -> Option<String> {
if let Some(f) = self.progress_fraction() {
Some(format!("{f:.2}"))
} else if self.progress_indeterminate() {
Some("indeterminate".to_string())
} else {
None
}
}
}
/// Process-wide monotonic sequence for [`JobReports::updated_at`] — a Lamport
/// clock, not wall-clock, so ordering is exact even at sub-microsecond report
/// rates and tests never race on `SystemTime` resolution.
pub(crate) fn next_seq() -> u64 {
static SEQ: AtomicU64 = AtomicU64::new(1);
SEQ.fetch_add(1, Ordering::Relaxed)
}
// ---------------------------------------------------------------------------
// ReportRegistry / ReportSink — where a Reporter's emits land.
// ---------------------------------------------------------------------------
/// The in-memory `job_id -> latest reports` table a [`crate::jobs::JobStore`]
/// keeps. `Arc<Mutex<_>>` (not redb): reports are live/ephemeral, cheap to update
/// at a high tick rate, and never need to survive a process restart — unlike
/// [`crate::jobs::JobRecord`], which IS durable.
pub type ReportRegistry = Arc<Mutex<BTreeMap<JobId, JobReports>>>;
/// A fresh, empty [`ReportRegistry`].
pub fn new_registry() -> ReportRegistry {
Arc::new(Mutex::new(BTreeMap::new()))
}
/// The write side of a [`Reporter`]. Cheap to clone (`Arc`), thread/tokio-safe.
/// Mirrors [`crate::jobs::JobSink`]'s shape: a fat client makes one over an
/// in-process [`ReportRegistry`] ([`ReportSink::registry`]); a thin/remote client
/// could later make one from a submit closure ([`ReportSink::remote`]), same as
/// `JobSink::remote` does for `JobRecord`.
#[derive(Clone)]
pub struct ReportSink(Arc<dyn Fn(&str, JobReport) + Send + Sync>);
impl ReportSink {
/// A sink that upserts into `registry` in place — the fat/local path.
pub fn registry(registry: ReportRegistry) -> Self {
ReportSink(Arc::new(move |job_id: &str, report: JobReport| {
let mut g = registry.lock().unwrap_or_else(|e| e.into_inner());
let entry = g.entry(job_id.to_string()).or_default();
entry.by_key.insert(report.report_key(), report);
entry.updated_at = next_seq();
}))
}
/// A sink that hands each `(job_id, report)` to `f` — the seam a thin/remote
/// client would use to forward reports over the wire (mirrors
/// [`crate::jobs::JobSink::remote`]).
pub fn remote(f: impl Fn(&str, JobReport) + Send + Sync + 'static) -> Self {
ReportSink(Arc::new(f))
}
/// A sink that drops every report — the default for a job that never calls
/// [`crate::jobs::JobHandle::with_reports`] (keeps [`Reporter`] unconditionally
/// safe to hold/call even when nothing is listening).
pub fn noop() -> Self {
ReportSink(Arc::new(|_, _| {}))
}
fn emit(&self, job_id: &str, report: JobReport) {
(self.0)(job_id, report)
}
}
// ---------------------------------------------------------------------------
// Reporter — the handle a running job holds and calls into.
// ---------------------------------------------------------------------------
/// A handle a running job uses to emit typed status without knowing anything
/// about the ledger. Cheap to clone (one `String` + an `Arc`-backed sink);
/// `Send + Sync` so it can be handed into a spawned task/thread. Every call is a
/// fire-and-forget upsert — a report is never "rejected", it just replaces the
/// prior latest report of the same kind (see [`JobReport::report_key`]).
#[derive(Clone)]
pub struct Reporter {
job_id: JobId,
sink: ReportSink,
}
impl Reporter {
/// A reporter for `job_id` that emits through `sink`.
pub fn new(job_id: impl Into<JobId>, sink: ReportSink) -> Self {
Reporter { job_id: job_id.into(), sink }
}
/// A reporter that discards everything — a safe default for code paths with no
/// wired [`ReportSink`] (e.g. a [`crate::jobs::JobHandle`] that never called
/// `with_reports`).
pub fn noop() -> Self {
Reporter { job_id: JobId::new(), sink: ReportSink::noop() }
}
/// This reporter's job id.
pub fn job_id(&self) -> &str {
&self.job_id
}
/// Emit a determinate/indeterminate progress update (`total: None` = spinner).
pub fn progress(&self, current: u64, total: Option<u64>) {
self.emit(JobReport::Progress { current, total, note: None });
}
/// Emit a progress update with a short caption (e.g. "downloading layer 3/5").
pub fn progress_note(&self, current: u64, total: Option<u64>, note: impl Into<String>) {
self.emit(JobReport::Progress { current, total, note: Some(note.into()) });
}
/// Emit a [`Health`] transition (e.g. a container going `Starting -> Alive`).
pub fn health(&self, health: Health) {
self.emit(JobReport::Health(health));
}
/// Emit/update a named gauge (e.g. `("rows_per_s", 1234.0, Some("rows/s"))`).
pub fn gauge(&self, key: impl Into<String>, value: f64, unit: Option<&str>) {
self.emit(JobReport::Gauge { key: key.into(), value, unit: unit.map(str::to_string) });
}
/// Emit the last human-readable status line.
pub fn message(&self, msg: impl Into<String>) {
self.emit(JobReport::Message(msg.into()));
}
/// Emit the current **daemon phase** — a named state-machine phase plus a
/// `healthy` flag (see [`JobReport::Phase`]). The one-call bridge a daemon job
/// (e.g. a collector) uses to surface its live phase/health to the Jobs view; it
/// typically pairs this with [`Self::progress`] for the phase %.
pub fn phase(&self, name: impl Into<String>, healthy: bool) {
self.emit(JobReport::Phase { name: name.into(), healthy });
}
fn emit(&self, report: JobReport) {
self.sink.emit(&self.job_id, report);
}
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn report_key_gives_gauges_independent_slots() {
let p = JobReport::Progress { current: 1, total: None, note: None };
let h = JobReport::Health(Health::Alive);
let g1 = JobReport::Gauge { key: "rows_per_s".into(), value: 1.0, unit: None };
let g2 = JobReport::Gauge { key: "connections".into(), value: 2.0, unit: None };
let m = JobReport::Message("hi".into());
assert_eq!(p.report_key(), "progress");
assert_eq!(h.report_key(), "health");
assert_eq!(g1.report_key(), "gauge:rows_per_s");
assert_eq!(g2.report_key(), "gauge:connections");
assert_ne!(g1.report_key(), g2.report_key(), "two gauges get independent slots");
assert_eq!(m.report_key(), "message");
}
/// A reporter's `progress` emit lands in the registry under "progress", latest
/// wins on a second call, and `updated_at` is monotonic (not wall-clock).
#[test]
fn reporter_progress_lands_in_registry_latest_wins() {
let reg = new_registry();
let reporter = Reporter::new("job-1", ReportSink::registry(reg.clone()));
reporter.progress(3, Some(10));
let snap1 = reg.lock().unwrap().get("job-1").cloned().unwrap();
assert_eq!(
snap1.by_key.get("progress"),
Some(&JobReport::Progress { current: 3, total: Some(10), note: None })
);
let seq1 = snap1.updated_at;
reporter.progress(7, Some(10));
let snap2 = reg.lock().unwrap().get("job-1").cloned().unwrap();
assert_eq!(
snap2.by_key.get("progress"),
Some(&JobReport::Progress { current: 7, total: Some(10), note: None }),
"latest progress overwrites the prior one, not appended"
);
assert!(snap2.updated_at > seq1, "updated_at is monotonic across emits");
}
/// Health transitions Starting -> Alive -> Gone are each recorded as the
/// current latest (the ledger's per-job health reflects the most recent call).
#[test]
fn reporter_health_transitions_are_recorded_in_order() {
let reg = new_registry();
let reporter = Reporter::new("svc-1", ReportSink::registry(reg.clone()));
reporter.health(Health::Starting);
assert_eq!(reg.lock().unwrap().get("svc-1").unwrap().health(), Some(Health::Starting));
reporter.health(Health::Alive);
assert_eq!(reg.lock().unwrap().get("svc-1").unwrap().health(), Some(Health::Alive));
reporter.health(Health::Gone);
let final_reports = reg.lock().unwrap().get("svc-1").cloned().unwrap();
assert_eq!(final_reports.health(), Some(Health::Gone), "final health is Gone");
}
/// `reports_snapshot`-shaped data (`Vec<(JobId, JobReports)>`) round-trips
/// through serde — the wire contract a UI polling over JSON relies on.
#[test]
fn reports_snapshot_shape_round_trips_through_serde() {
let mut by_key = BTreeMap::new();
by_key.insert("progress".to_string(), JobReport::Progress { current: 4, total: Some(9), note: None });
by_key.insert("health".to_string(), JobReport::Health(Health::Servicing));
by_key.insert(
"gauge:rows_per_s".to_string(),
JobReport::Gauge { key: "rows_per_s".into(), value: 42.5, unit: Some("rows/s".into()) },
);
by_key.insert("message".to_string(), JobReport::Message("almost done".into()));
let snapshot: Vec<(JobId, JobReports)> =
vec![("job-abc".to_string(), JobReports { by_key, updated_at: 3 })];
let s = serde_json::to_string(&snapshot).unwrap();
let back: Vec<(JobId, JobReports)> = serde_json::from_str(&s).unwrap();
assert_eq!(snapshot, back);
// And the tag shape is the self-describing `{"kind": "...", "data": ...}` form.
assert!(s.contains("\"kind\":\"progress\""), "tagged wire shape: {s}");
assert!(s.contains("\"kind\":\"health\""), "{s}");
assert!(s.contains("\"kind\":\"message\""), "{s}");
}
/// The report→jobview bridge: a determinate progress report yields a fraction
/// and a compact `"0.NN"` meta string; an indeterminate one yields the
/// `"indeterminate"` marker; no report yields `None` (an honest empty meter).
#[test]
fn progress_fraction_and_meta_bridge_for_a_ui_row() {
// Determinate 4/10 → 0.4.
let mut det = JobReports::default();
det.by_key.insert("progress".into(), JobReport::Progress { current: 4, total: Some(10), note: None });
assert_eq!(det.progress_fraction(), Some(0.4));
assert!(!det.progress_indeterminate());
assert_eq!(det.progress_meta().as_deref(), Some("0.40"));
// Indeterminate (no total) → no fraction, but a marker meta.
let mut indet = JobReports::default();
indet.by_key.insert("progress".into(), JobReport::Progress { current: 3, total: None, note: None });
assert_eq!(indet.progress_fraction(), None);
assert!(indet.progress_indeterminate());
assert_eq!(indet.progress_meta().as_deref(), Some("indeterminate"));
// total == 0 → treated as indeterminate (no divide-by-zero, no fraction).
let mut zero = JobReports::default();
zero.by_key.insert("progress".into(), JobReport::Progress { current: 1, total: Some(0), note: None });
assert_eq!(zero.progress_fraction(), None);
// No progress report at all → None (honest empty meter, never a full bar).
let none = JobReports::default();
assert_eq!(none.progress_fraction(), None);
assert!(!none.progress_indeterminate());
assert_eq!(none.progress_meta(), None);
// Over-range current clamps to a full (never >1.0) fraction.
let mut over = JobReports::default();
over.by_key.insert("progress".into(), JobReport::Progress { current: 99, total: Some(10), note: None });
assert_eq!(over.progress_fraction(), Some(1.0));
}
/// The daemon **phase** channel: a reporter's `phase` emit lands under the
/// "phase" slot as `(name, healthy)`, latest wins, and coexists with a Progress
/// report (a daemon shows both a phase name AND a phase %). This is the side
/// channel a daemon-job Jobs-view row reads for its live state-machine phase.
#[test]
fn reporter_phase_and_progress_coexist_latest_phase_wins() {
let reg = new_registry();
let reporter = Reporter::new("daemon-1", ReportSink::registry(reg.clone()));
reporter.phase("downloading", false);
reporter.progress(12, Some(100));
let snap = reg.lock().unwrap().get("daemon-1").cloned().unwrap();
assert_eq!(snap.phase(), Some(("downloading", false)));
assert_eq!(snap.phase_name(), Some("downloading"));
assert_eq!(snap.phase_healthy(), Some(false));
assert_eq!(snap.progress_fraction(), Some(0.12), "phase % rides Progress alongside the phase name");
// A later phase overwrites the prior one (latest wins), progress untouched.
reporter.phase("online", true);
let snap = reg.lock().unwrap().get("daemon-1").cloned().unwrap();
assert_eq!(snap.phase(), Some(("online", true)), "latest phase wins");
assert_eq!(snap.phase_healthy(), Some(true));
assert_eq!(snap.progress_fraction(), Some(0.12), "the Progress slot is independent of the Phase slot");
// The tagged wire shape round-trips.
let one: Vec<(JobId, JobReports)> = vec![("daemon-1".into(), snap)];
let s = serde_json::to_string(&one).unwrap();
assert!(s.contains("\"kind\":\"phase\""), "tagged phase wire shape: {s}");
let back: Vec<(JobId, JobReports)> = serde_json::from_str(&s).unwrap();
assert_eq!(back[0].1.phase(), Some(("online", true)));
}
#[test]
fn noop_reporter_never_panics_and_touches_nothing() {
let r = Reporter::noop();
r.progress(1, Some(2));
r.health(Health::Alive);
r.gauge("x", 1.0, Some("u"));
r.message("hi");
r.phase("online", true);
// Nothing to assert beyond "did not panic" — there is no registry backing it.
}
}