crates/step-board/src/lib.rs
//! **step-board** — the producer side of the step bar: the registry a producer
//! publishes its transitions into, and the [`StepMeter`] a pane reads back out.
//!
//! [`StepCell`], [`StepPhase`] and [`StepMeter`] (`facett-jobview`) describe *what a
//! bar shows*. This module describes *who says so*: one [`StepRun`] per job, held in a
//! process-wide [`Board`], written by the producer at each transition and cloned by the
//! pane once a frame. One publish per transition and one per reading that actually
//! CHANGED — nothing is walked, scanned or polled to draw a bar.
//!
//! # Why this consumes the widget crate's types instead of declaring its own
//!
//! It is the same vocabulary. A registry that kept its own `Pending/Active/Done/Failed`
//! and its own per-step struct would be a second spelling of [`StepPhase`] and
//! [`StepCell`], converted at the boundary — the twin LAW 5 forbids, and the exact place
//! a bar starts disagreeing with the words beside it. So the registry is built **on**
//! those types: [`StepRun::steps`] *is* `Vec<StepCell>`, which is also why a step's
//! measured [`count`](StepCell::count) rides along for free rather than needing a
//! parallel array nobody remembers to resize.
//!
//! Nothing in this module touches `egui`. A producer is usually a worker thread or an
//! RPC poll loop, not a paint.
//!
//! # One writer for every sentence
//!
//! [`StepRun::caption`] and [`StepRun::hover_text`] are the ONLY places these sentences
//! are built. The pane draws them and `state_json` publishes them — the identical
//! `String`, never a re-derivation. Publishing a declaration next to a separately-built
//! drawing is how a `state_json` comes to report a transition the paint never made.
//!
//! # Run identity: why [`StepRun::generation`] exists
//!
//! A job restarts. Without a run number, a bar that read *"3 of 6"* and now reads
//! *"1 of 6"* is indistinguishable from a bar that lost ground, and that ambiguity lands
//! exactly when somebody is watching hardest. [`Board::begin`] bumps `generation` and
//! records [`StepRun::reason`], and both are rendered — so a restart reads as **run #4
//! step 1 of 6**, not as a regression.
//!
//! # Wiring a producer that HAS a counter
//!
//! A step with no counter must not pretend to have one: an active step with no
//! [`StepCell::count`] renders `Indeterminate` — a sweep, meaning *in flight, unmeasured*
//! — and that is the honest default.
//!
//! **It is not the common case, though, and this header used to claim it was.** korp's
//! region chain (`korp::collector_steps::ChainStep`) is six legs and **four of them carry
//! a genuine `(current, total)`**: the planet torrent counts bytes off the sparse file
//! librqbit is growing, the convert counts finished pipeline phases off the JSONL
//! osm-katana appends to, the upload counts objects off its own in-flight atomic, and the
//! opening survey counts artefacts found. The one that stays `None` is the tile pyramid,
//! because `osm_katana::tiles::tile_region` really does expose no per-tile callback and
//! the alternative is walking a 1.6 M-file tree while the tiler writes it.
//!
//! That is the shape to copy: reach for a counter first, and leave the slot empty only
//! where the producer genuinely has no number — not by default.
//!
//! Where a real number exists, [`Board::set_count`] is the whole of it, and it is meant to
//! be no harder than leaving it out. dwarves' populate clones N members and the server
//! reports each one's state on every poll, so the clone step is measured:
//!
//! ```
//! use step_board::{Board, StepDef};
//! use facett_jobview::{ProgressMeter, StepPhase};
//!
//! let board = Board::default();
//! let id = "populate/nordisk";
//! let steps = vec![
//! StepDef::new("clone", "clone the members"),
//! StepDef::new("warehouse", "build the warehouse"),
//! ];
//!
//! board.begin(id, "nordisk", steps, "Re-sync clicked");
//! board.enter(id, 0).expect("step 1 of a fresh run is in order");
//!
//! // …each poll, from whatever the producer actually counted:
//! board.set_count(id, 0, 12, 19);
//!
//! let run = board.get(id).unwrap();
//! assert_eq!(run.steps[0].meter(), ProgressMeter::Determinate(12.0 / 19.0));
//! assert_eq!(run.caption(), "run #1 · step 1 of 2 · clone the members");
//!
//! // The step nobody counted stays honest on its own — no opt-out needed.
//! board.complete_through(id, 0);
//! board.enter(id, 1).expect("clone finished, so warehouse is in order");
//! let run = board.get(id).unwrap();
//! assert_eq!(run.steps[1].meter(), ProgressMeter::Indeterminate);
//! assert_eq!(run.steps[1].phase, StepPhase::Active);
//!
//! // …and publishes NO fraction, because a zero there would read as a measurement.
//! let j = run.state_json();
//! assert!(j["steps"][1]["meter"].get("fraction").is_none());
//! assert_eq!(j["steps"][0]["meter"]["current"], 12);
//! ```
//!
//! Never synthesise a number to fill that slot. An invented percentage is
//! indistinguishable from a measured one on screen, which is what would make every
//! measured one worthless.
//!
//! # The order check
//!
//! [`Board::enter`] returns [`Result<(), OrderFault>`] and **refuses** a step that
//! overtakes its predecessor. For a single-threaded chain that can never fire. It exists
//! because a run can have more than one owner — a populate whose clone phase and
//! warehouse-build phase are driven by different threads — and nothing else stops the
//! second from announcing itself while the first is still going. The refusal is not
//! advisory: the offending step is marked [`StepPhase::Failed`] and carries the reason,
//! so a producer that drops the `Result` still gets a red bar rather than a plausible
//! one.
//!
//! A **serial** producer — one owner, one thread — can skip the indices entirely and
//! drive the chain with [`Board::advance`], which finishes the active step and enters
//! the next one and therefore has no order to violate. The check above is for the case
//! `advance` cannot express: two owners on one run.
//!
//! # Resuming a run that was already part-done
//!
//! A host that can cheaply prove what is already finished hands that verdict to
//! [`Board::seed_done`] once, at start, so the first frame is truthful instead of
//! claiming *"0 of N"* about work that is three quarters done. It takes a predicate over
//! step keys and walks nothing itself — the point of this whole seam is that drawing a
//! bar never costs a directory sweep.
use std::collections::BTreeMap;
use std::sync::{Arc, Mutex, OnceLock};
use facett_jobview::{JobEntry, JobStatus, StepCell, StepMeter, StepPhase};
// ── The step vocabulary ────────────────────────────────────────────────────────
/// One step of a run — its stable key and the sentence a human reads. The static half
/// of a [`StepCell`]: what a producer declares up front, before any of it has happened.
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct StepDef {
/// Stable machine key (`"clone"`, `"a"`) — the `state_json` id a robot asserts on.
pub key: String,
/// The human label (`"clone the members"`).
pub label: String,
}
impl StepDef {
/// Build a step definition.
pub fn new(key: impl Into<String>, label: impl Into<String>) -> StepDef {
StepDef { key: key.into(), label: label.into() }
}
}
/// **A producer published a step out of order.** Not a status of the work — a status of
/// the *producers*, and a bug in whoever emitted it.
#[derive(Clone, Debug, PartialEq, Eq)]
pub enum OrderFault {
/// A step was entered while an earlier one had not finished.
PredecessorUnfinished {
/// Index entered.
entered: usize,
/// Index of the first unfinished earlier step.
blocker: usize,
/// The entered step's key.
entered_key: String,
/// The blocking step's key.
blocker_key: String,
/// What the blocker was actually doing.
blocker_phase: StepPhase,
},
/// A step was entered while a LATER one had already started or finished.
Backwards {
/// Index entered.
entered: usize,
/// Index of the later step already under way.
ahead: usize,
/// The entered step's key.
entered_key: String,
/// The later step's key.
ahead_key: String,
},
}
impl OrderFault {
/// The sentence the bar shows and `state_json` publishes — one writer, as everywhere
/// else here. It names **both** steps, because "out of order" without saying which
/// two is not something anyone can act on at 03:00.
pub fn message(&self) -> String {
match self {
OrderFault::PredecessorUnfinished { entered_key, blocker_key, blocker_phase, .. } => {
format!(
"OUT OF ORDER: step {entered_key} started while step {blocker_key} is still \
{} — two producers are racing on one run",
blocker_phase.key().to_uppercase()
)
}
OrderFault::Backwards { entered_key, ahead_key, .. } => format!(
"OUT OF ORDER: step {entered_key} started after step {ahead_key} had already \
begun — the chain went backwards"
),
}
}
/// The index the fault is pinned to (the step that paints red).
pub fn at(&self) -> usize {
match self {
OrderFault::PredecessorUnfinished { entered, .. }
| OrderFault::Backwards { entered, .. } => *entered,
}
}
}
impl std::fmt::Display for OrderFault {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.write_str(&self.message())
}
}
impl std::error::Error for OrderFault {}
// ── One job's run ──────────────────────────────────────────────────────────────
/// **One job's current (or last) run** — the whole thing a bar needs, and nothing a bar
/// has to compute for itself.
#[derive(Clone, Debug, PartialEq)]
pub struct StepRun {
/// Stable job-instance id (`"populate/nordisk"`).
pub id: String,
/// Display title (`"nordisk"`).
pub title: String,
/// **Run identity.** Monotone per job, bumped by [`Board::begin`]. `0` means no run
/// has started in this process.
pub generation: u64,
/// Why THIS run started (`"Re-sync clicked"`). Rendered next to the run number,
/// because a bar that resets without a reason reads as a fault.
pub reason: String,
/// The steps, in dependency order — phase and measured count included.
pub steps: Vec<StepCell>,
/// A short producer note for what is happening when **no** step is active yet. A run
/// can legitimately be underway before step 1 starts, and a bar that says only
/// *"0 of 3"* tells a watcher nothing about whether it is advancing or wedged.
pub note: String,
/// The fail-loud reason when a step is [`StepPhase::Failed`].
pub error: Option<String>,
/// Last transition (unix seconds).
pub updated_unix: u64,
}
/// Seconds since the epoch, saturating to 0 on a clock before it.
fn now_unix() -> u64 {
std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.map(|d| d.as_secs())
.unwrap_or(0)
}
/// The step line a producer gets by default — `"clone clone the members ACTIVE"`.
fn default_line(key: &str, label: &str, phase: StepPhase) -> String {
format!("{key} {label} {}", phase.tag())
}
impl StepRun {
/// A run with every step [`StepPhase::Pending`] and nothing measured.
pub fn new(id: impl Into<String>, title: impl Into<String>, steps: Vec<StepDef>) -> StepRun {
StepRun {
id: id.into(),
title: title.into(),
generation: 0,
reason: String::new(),
note: String::new(),
steps: steps
.into_iter()
.map(|d| {
let line = default_line(&d.key, &d.label, StepPhase::Pending);
StepCell { key: d.key, label: d.label, phase: StepPhase::Pending, line, count: None }
})
.collect(),
error: None,
updated_unix: 0,
}
}
/// How many steps this run has (the *of N* in the caption).
pub fn total(&self) -> usize {
self.steps.len()
}
/// The index of the step being worked on, if any.
pub fn active(&self) -> Option<usize> {
self.steps.iter().position(|s| s.phase == StepPhase::Active)
}
/// The index of the step that failed, if any.
pub fn failed(&self) -> Option<usize> {
self.steps.iter().position(|s| s.phase == StepPhase::Failed)
}
/// How many steps are finished.
pub fn done(&self) -> usize {
self.steps.iter().filter(|s| s.phase == StepPhase::Done).count()
}
/// Every step's phase, in order — the shape a caller that wants to assert the whole
/// chain at once reaches for.
pub fn phases(&self) -> Vec<StepPhase> {
self.steps.iter().map(|s| s.phase).collect()
}
/// The fraction a whole-run bar fills to — **finished steps only**, never a guess at
/// how far into the active step the work has got. A step's own measured fill is
/// [`StepCell::count`]; this is the coarse one, and it is deliberately not blended
/// with the fine one, because an average of a measurement and a step count is
/// neither.
pub fn fraction(&self) -> f32 {
if self.steps.is_empty() {
return 0.0;
}
self.done() as f32 / self.steps.len() as f32
}
/// Is every step done? **An empty run is not complete** — a zero-step bar reading
/// "complete" is the hollow green this whole seam exists to prevent.
pub fn is_complete(&self) -> bool {
!self.steps.is_empty() && self.steps.iter().all(|s| s.phase == StepPhase::Done)
}
/// Is a step being worked on right now?
pub fn in_progress(&self) -> bool {
self.active().is_some()
}
/// The **1-based** human step number: the active step, else the failed one, else how
/// many are done. The numerator of *"step N of M"*.
pub fn human_index(&self) -> usize {
self.active().or_else(|| self.failed()).map_or_else(|| self.done(), |i| i + 1)
}
/// **The one line the human reads and `state_json` publishes** — built here and
/// nowhere else.
///
/// ```text
/// run #3 · step 1 of 3 · clone the members
/// run #3 · step 2 of 3 FAILED · build the warehouse
/// run #3 · 3 of 3 steps complete
/// not started
/// ```
pub fn caption(&self) -> String {
if self.generation == 0 {
return "not started".to_string();
}
let run = format!("run #{}", self.generation);
if let Some(i) = self.failed() {
return format!("{run} · step {} of {} FAILED · {}", i + 1, self.total(), self.steps[i].label);
}
if let Some(i) = self.active() {
return format!("{run} · step {} of {} · {}", i + 1, self.total(), self.steps[i].label);
}
if self.is_complete() {
return format!("{run} · {} of {} steps complete", self.done(), self.total());
}
let tail = if self.note.trim().is_empty() { "idle".to_string() } else { self.note.clone() };
format!("{run} · {} of {} steps complete · {tail}", self.done(), self.total())
}
/// The **why-this-run-exists** line, shown under the caption on hover. Empty reason ⇒
/// empty string, never a fabricated one.
pub fn reason_line(&self) -> String {
if self.generation == 0 || self.reason.trim().is_empty() {
String::new()
} else {
format!("run #{} started because: {}", self.generation, self.reason)
}
}
/// One step's hover line. Out of range yields an empty string rather than panicking
/// a paint.
pub fn step_line(&self, i: usize) -> String {
self.steps.get(i).map(|s| s.line.clone()).unwrap_or_default()
}
/// The whole hover block: the reason line, then one line per step.
pub fn hover_text(&self) -> String {
let mut lines: Vec<String> = Vec::with_capacity(self.steps.len() + 1);
let r = self.reason_line();
if !r.is_empty() {
lines.push(r);
}
lines.extend(self.steps.iter().map(|s| s.line.clone()));
lines.join("\n")
}
/// **The run as the widget's [`StepMeter`]** — the one conversion, so the pane and
/// `state_json` cannot end up with two different renderings of one run.
pub fn to_meter(&self) -> StepMeter {
StepMeter {
generation: self.generation,
caption: self.caption(),
hover: self.hover_text(),
steps: self.steps.clone(),
}
}
/// The run as `state_json` data. The per-step half is [`StepMeter::state_json`]
/// verbatim — **not** a second encoding of the same steps — so a step that is
/// unmeasured publishes `{"kind":"indeterminate"}` with no `fraction` key here for
/// exactly the same reason it does there.
///
/// `step_index` / `step_total` / `fraction` are the run-level spellings of the same
/// numbers the meter publishes as `index` / `total`. They are carried because the
/// robots that already assert this object assert *those* names, and a rename is a
/// silent break: an assertion on a key that no longer exists reads as a passing
/// `null`, not as a failure.
pub fn state_json(&self) -> serde_json::Value {
let mut v = self.to_meter().state_json();
if let Some(o) = v.as_object_mut() {
o.insert("id".into(), serde_json::json!(self.id));
o.insert("title".into(), serde_json::json!(self.title));
o.insert("reason".into(), serde_json::json!(self.reason));
o.insert("note".into(), serde_json::json!(self.note));
o.insert("error".into(), serde_json::json!(self.error));
o.insert("in_progress".into(), serde_json::json!(self.in_progress()));
o.insert("updated_unix".into(), serde_json::json!(self.updated_unix));
o.insert("step_index".into(), serde_json::json!(self.human_index()));
o.insert("step_total".into(), serde_json::json!(self.total()));
o.insert("fraction".into(), serde_json::json!(self.fraction()));
}
v
}
}
/// **One run as the row a `facett_jobview::JobList` draws.** The one conversion, so a
/// pane and `state_json` cannot end up with two different renderings of one run.
///
/// `kind` is the host's own word for what this run is (`"collector"` for korp's region
/// cuts, `"populate"` for dwarves) — it is what sorts and filters these live rows apart
/// from a durable ledger's rows, which record what *happened* rather than what *is
/// happening*.
///
/// Nothing here re-authors a sentence: the caption and every step line travel from
/// [`StepRun::to_meter`] as data.
pub fn to_job_entry(run: &StepRun, kind: &str) -> JobEntry {
let status = if run.failed().is_some() {
JobStatus::Failed
} else if run.in_progress() {
JobStatus::Running
} else if run.is_complete() {
JobStatus::Done
} else {
JobStatus::Queued
};
let mut e = JobEntry::new(run.id.clone(), kind, status);
e.target = run.title.clone();
e.detail = run.error.clone().unwrap_or_default();
e.with_steps(&run.to_meter())
}
/// Every run on a board as job rows, in id order — **one bar per run**.
pub fn job_entries(board: &Board, kind: &str) -> Vec<JobEntry> {
board.snapshot().iter().map(|r| to_job_entry(r, kind)).collect()
}
// ── The board: one row per job ─────────────────────────────────────────────────
#[derive(Default)]
struct Inner {
runs: BTreeMap<String, StepRun>,
}
/// **A step board** — one [`StepRun`] per job, shared by every producer thread and every
/// pane. Cheap to clone (an `Arc` handle).
#[derive(Clone, Default)]
pub struct Board {
inner: Arc<Mutex<Inner>>,
}
impl Board {
/// Every job's run, in id order — **one bar per entry**.
pub fn snapshot(&self) -> Vec<StepRun> {
self.lock().runs.values().cloned().collect()
}
/// One job's run.
pub fn get(&self, id: &str) -> Option<StepRun> {
self.lock().runs.get(id).cloned()
}
/// **Start (or restart) a run.** Bumps [`StepRun::generation`], records why, and puts
/// every step back to [`StepPhase::Pending`] **with no measured count** — a fresh run
/// has measured nothing yet, and carrying the last run's numbers over is how a bar
/// comes to show a fraction from work that is no longer happening.
///
/// Returns the new generation.
pub fn begin(&self, id: &str, title: &str, steps: Vec<StepDef>, reason: impl Into<String>) -> u64 {
let mut g = self.lock();
let fresh = StepRun::new(id, title, steps);
let row = g.runs.entry(id.to_string()).or_insert_with(|| fresh.clone());
row.title = title.to_string();
row.generation += 1;
row.reason = reason.into();
row.note = String::new();
row.steps = fresh.steps;
row.error = None;
row.updated_unix = now_unix();
row.generation
}
/// **Enter step `idx`** (0-based): it becomes [`StepPhase::Active`] and every later
/// step goes back to [`StepPhase::Pending`].
///
/// Earlier steps are **not** filled in. A producer that finished several steps inside
/// one opaque call says so with [`complete_through`](Self::complete_through); a
/// forward-fill here would silently stamp a still-running predecessor `Done` and hide
/// the very race the order check exists to catch.
///
/// A no-op when the job has no run, or `idx` is out of range — a stray publish must
/// never invent a run.
///
/// # Errors
///
/// [`OrderFault`] when this step overtakes an unfinished predecessor, or steps
/// backwards behind one already under way. On a fault the step is left
/// [`StepPhase::Failed`] carrying [`OrderFault::message`], so a caller that drops the
/// `Result` still gets a red bar.
pub fn enter(&self, id: &str, idx: usize) -> Result<(), OrderFault> {
let mut g = self.lock();
let Some(row) = g.runs.get_mut(id) else { return Ok(()) };
if idx >= row.steps.len() {
return Ok(());
}
let fault = row
.steps
.iter()
.take(idx)
.position(|s| s.phase != StepPhase::Done)
.map(|blocker| OrderFault::PredecessorUnfinished {
entered: idx,
blocker,
entered_key: row.steps[idx].key.clone(),
blocker_key: row.steps[blocker].key.clone(),
blocker_phase: row.steps[blocker].phase,
})
.or_else(|| {
row.steps
.iter()
.enumerate()
.skip(idx + 1)
.find(|(_, s)| matches!(s.phase, StepPhase::Done | StepPhase::Active))
.map(|(ahead, _)| OrderFault::Backwards {
entered: idx,
ahead,
entered_key: row.steps[idx].key.clone(),
ahead_key: row.steps[ahead].key.clone(),
})
});
if let Some(f) = fault {
let msg = f.message();
set_phase(&mut row.steps[idx], StepPhase::Failed);
row.steps[idx].line.push_str(" — ");
row.steps[idx].line.push_str(&msg);
row.error = Some(msg);
row.updated_unix = now_unix();
return Err(f);
}
set_phase(&mut row.steps[idx], StepPhase::Active);
for s in row.steps.iter_mut().skip(idx + 1) {
set_phase(s, StepPhase::Pending);
s.count = None;
}
row.error = None;
row.note = String::new();
row.updated_unix = now_unix();
Ok(())
}
/// **Mark steps `0..=idx` finished** — the explicit claim that replaces an inferred
/// forward-fill. A producer whose engine does several steps inside one opaque call
/// says so here, and only then enters the next step.
pub fn complete_through(&self, id: &str, idx: usize) {
let mut g = self.lock();
let Some(row) = g.runs.get_mut(id) else { return };
for s in row.steps.iter_mut().take(idx.saturating_add(1)) {
set_phase(s, StepPhase::Done);
}
row.updated_unix = now_unix();
}
/// **Finish the active step and enter the next** — the serial producer's ergonomic
/// move, which cannot violate the order by construction (it only ever advances by
/// one, from a step it just marked `Done`). Returns the index entered, or `None`
/// when the run is over or nothing was active.
///
/// A producer with more than one owner should NOT use this: it is `enter`'s order
/// check that catches a race, and `advance` has nothing to catch.
pub fn advance(&self, id: &str) -> Option<usize> {
let next = {
let mut g = self.lock();
let row = g.runs.get_mut(id)?;
let cur = row.steps.iter().position(|s| s.phase == StepPhase::Active)?;
set_phase(&mut row.steps[cur], StepPhase::Done);
row.updated_unix = now_unix();
cur + 1
};
if next >= self.get(id)?.steps.len() {
return None;
}
self.enter(id, next).ok().map(|()| next)
}
/// **Seed the steps a run has ALREADY finished — once, at start.**
///
/// A host that can cheaply prove what is already done (korp's resume chain has just
/// read the disk to decide where to restart) hands that verdict here, so the bar is
/// truthful from the first frame instead of claiming *"0 of 4"* about work that is
/// three quarters finished.
///
/// `is_done` is asked once per step, by [`StepCell::key`]. **Nothing here walks
/// anything** — this only records an answer the caller already has, which is the
/// whole point: a bar must never cost a directory sweep. Returns how many steps were
/// seeded, so a caller can tell "the verdict said nothing" from "the seed did
/// nothing".
pub fn seed_done(&self, id: &str, is_done: impl Fn(&str) -> bool) -> usize {
let mut g = self.lock();
let Some(row) = g.runs.get_mut(id) else { return 0 };
let mut n = 0;
for s in row.steps.iter_mut() {
if is_done(&s.key) {
set_phase(s, StepPhase::Done);
n += 1;
}
}
row.updated_unix = now_unix();
n
}
/// **Attach a MEASURED `(current, total)` to step `idx`.**
///
/// Only call this with a number the producer actually counted — see
/// [`StepCell::count`]. A synthesised fraction is indistinguishable from a measured
/// one on screen, which is what makes every measured one worth drawing.
pub fn set_count(&self, id: &str, idx: usize, current: u64, total: u64) {
let mut g = self.lock();
let Some(row) = g.runs.get_mut(id) else { return };
let Some(s) = row.steps.get_mut(idx) else { return };
s.count = Some((current, total));
s.line = format!(
"{} {} {} {current}/{total}",
s.key,
s.label,
s.phase.tag()
);
row.updated_unix = now_unix();
}
/// **Say what is happening before the first step starts.** Published on change like
/// everything else here — never on a timer, because a periodic dump makes an idle
/// producer indistinguishable from a hung one.
pub fn note(&self, id: &str, note: impl Into<String>) {
let mut g = self.lock();
let Some(row) = g.runs.get_mut(id) else { return };
row.note = note.into();
row.updated_unix = now_unix();
}
/// **The run finished** — every step [`StepPhase::Done`].
pub fn finish(&self, id: &str) {
let mut g = self.lock();
let Some(row) = g.runs.get_mut(id) else { return };
for s in row.steps.iter_mut() {
set_phase(s, StepPhase::Done);
}
row.error = None;
row.updated_unix = now_unix();
}
/// **The run died on the step it was on** (or on `at`, when the caller knows better).
/// Finished steps stay finished, because they are.
pub fn fail(&self, id: &str, at: Option<usize>, error: impl Into<String>) {
let mut g = self.lock();
let Some(row) = g.runs.get_mut(id) else { return };
let idx = at
.or_else(|| row.steps.iter().position(|s| s.phase == StepPhase::Active))
.or_else(|| row.steps.iter().position(|s| s.phase == StepPhase::Pending));
let error = error.into();
if let Some(s) = idx.and_then(|i| row.steps.get_mut(i)) {
set_phase(s, StepPhase::Failed);
s.line.push_str(" — ");
s.line.push_str(&error);
}
row.error = Some(error);
row.updated_unix = now_unix();
}
/// Drop a job's row.
pub fn forget(&self, id: &str) {
self.lock().runs.remove(id);
}
/// Clear the board (tests, and a host tearing a view down).
pub fn reset(&self) {
self.lock().runs.clear();
}
fn lock(&self) -> std::sync::MutexGuard<'_, Inner> {
self.inner.lock().unwrap_or_else(|e| e.into_inner())
}
}
/// Set a cell's phase and rewrite its default line — keeping the two in step, so a
/// producer never has to remember that the tag lives inside a string.
fn set_phase(s: &mut StepCell, phase: StepPhase) {
s.phase = phase;
s.line = match s.count {
Some((c, t)) => format!("{} {} {} {c}/{t}", s.key, s.label, phase.tag()),
None => default_line(&s.key, &s.label, phase),
};
}
/// **A process-wide board keyed by namespace** — the `OnceLock` shape a host uses so its
/// producer threads and its panes share one live view across frames. Each namespace gets
/// its own [`Board`]; two hosts in one process never collide.
pub fn shared(namespace: &str) -> Board {
static B: OnceLock<Mutex<BTreeMap<String, Board>>> = OnceLock::new();
let m = B.get_or_init(Default::default);
let mut g = m.lock().unwrap_or_else(|e| e.into_inner());
g.entry(namespace.to_string()).or_default().clone()
}
/// Every run on a board as `state_json` — the list a robot walks to assert that a bar
/// shows what it shows.
///
/// `entity` names what one run IS in the host's own vocabulary, and becomes the key
/// carrying the count (`"collectors"` for korp, `"jobs"` for a generic host). It is a
/// parameter rather than a fixed word because these objects are already under robot
/// assertion: renaming a host's count key is a silent break, since an assertion on a key
/// that no longer exists reads as a passing `null`.
pub fn board_state_json(board: &Board, entity: &str) -> serde_json::Value {
let runs = board.snapshot();
serde_json::json!({
entity: runs.len(),
"in_progress": runs.iter().filter(|r| r.in_progress()).count(),
"complete": runs.iter().filter(|r| r.is_complete()).count(),
"failed": runs.iter().filter(|r| r.failed().is_some()).count(),
"runs": runs.iter().map(StepRun::state_json).collect::<Vec<_>>(),
})
}