crates/jera/src/jobs.rs
//! Unified **job ledger** — the one durable list of every long-running nornir
//! operation (~1s–minutes): release runs, workspace fetch/republish, docs
//! render/export/book, knowledge scan, deepscan, bench, test matrix, arch
//! generate, index build, vector embed, bakeoff, airgap stages, …
//!
//! Today each op writes its *result* to a domain Iceberg table at the end
//! (`bench_runs`, `test_results`, `architecture_wiring`, `doc_exports`,
//! `clone_events`, …). What was missing is one place that answers **"what's
//! running / did it finish / how long did it take"** — the op *lifecycle*,
//! orthogonal to the result. That is this module.
//!
//! **Why redb, not Iceberg.** A job record is *mutable* (`running → done|failed`)
//! and short-lived; redb overwrites the key in place — no append-only
//! "two-rows-then-coalesce-on-read" dance, no per-event Iceberg snapshot
//! (a snapshot+manifest per `start`/`finish` is the small-files antipattern).
//! Iceberg stays the home for what an op *produced*; redb is the ledger that an
//! op *ran*. `result_ref` (e.g. `bench_runs:<uuid>`, `release_events:<run_id>`)
//! links a job back to its Iceberg result row.
//!
//! **Storage.** Its own redb instance at `<warehouse root>/jobs.redb` — distinct
//! from skade-katalog's `catalog.redb` and the server's `registry.redb`. That rule
//! is now a FUNCTION, [`store_path`], and not a sentence: it was stated here from
//! the first commit and jera still made every consumer re-derive the join, so four
//! spellings went live (see [`store_path`] for the measured list). A consumer with
//! a warehouse root calls `store_path(root)` or hands that root straight to
//! [`JobStore::open`]; it does not invent a sub-directory. A consumer whose ledger
//! genuinely is not at a warehouse root (holger's `<data>/holger-jobs`) passes its
//! own directory, which is legitimate and unchanged.
//!
//! Opening the WRONG directory used to be silent — you got a valid, empty ledger
//! beside the real one, and empty is what a quiet forge looks like. It is now
//! detected and named: see [`stray_ledger_near`] and [`JobStore::stray_ledger`].
//!
//! Access mirrors the warehouse exactly (see `nornir::warehouse`):
//!
//! - **fat client** (local FS via library code, no gRPC): the [`JobHandle`]
//! writes the local `jobs.redb` directly through a redb [`JobSink`].
//! - **thin client** (no local FS): the [`JobHandle`]'s sink submits each record
//! over gRPC (`Jobs.Submit`); the server applies it to *its* `jobs.redb`.
//!
//! Single-writer discipline matches `registry.redb`: the owning process is the
//! sole writer; viz reads via [`JobStore::open_read_only`], which copies the
//! file aside when the server holds the lock (same trick as
//! `nornir::warehouse::IcebergWarehouse::open_read_only`).
use std::path::{Path, PathBuf};
use std::sync::Arc;
use anyhow::{Context, Result};
use redb::{Database, ReadableTable, TableDefinition};
use serde::{Deserialize, Serialize};
/// Functional-status emit bridge (a no-op unless a consumer wires one in).
pub mod selftest;
/// The background **CHAIN** — the queue this ledger never had: submit derived
/// work when a fact lands, run it when the machine gets to it, idempotent on a
/// key, superseded by a newer generation, and deferential to the machine's real
/// work. See [`chain`].
pub mod chain;
pub use chain::{ChainQueue, ChainSpec, ChainStats, Priority, Submitted};
/// Typed **job reporting** — [`JobReport`]/[`Health`]/[`Reporter`], the OPT-IN
/// "what is this job doing right now" channel orthogonal to this module's
/// lifecycle status. See the module docs for the full picture.
pub mod report;
pub use report::{Health, JobId, JobReport, JobReports, ReportRegistry, ReportSink, Reporter};
/// `job_id → serde_json(JobRecord)`. Bytes in, bytes out (like the registry).
const TABLE: TableDefinition<&str, &[u8]> = TableDefinition::new("jobs");
/// Retention applied opportunistically on terminal writes: drop terminal jobs
/// older than this many days, and cap the total row count (oldest terminal
/// pruned first). Keeps `jobs.redb` bounded on a long-running server.
pub const KEEP_DAYS: i64 = 30;
/// Hard cap on retained rows — see [`KEEP_DAYS`].
pub const CAP: usize = 2000;
/// Lifecycle status of a job. Mutable: the same key is overwritten in place as a
/// job advances. `queued` and `running` are the two non-terminal states.
///
/// A `queued` job has been admitted to the ledger but is **waiting on the serial
/// scheduler** (see [`is_exclusive`]) — another exclusive job (e.g. a `bench_run`)
/// is running and this one waits behind it. It transitions `queued → running`
/// when it acquires the exclusive gate, then `running → done|failed` as usual.
/// Non-exclusive jobs never pass through `queued`; they start `running` directly.
pub mod status {
/// Admitted but waiting on the serial scheduler (an exclusive job is running
/// ahead of this one). Non-terminal; visible in `nornir jobs` / the viz list.
pub const QUEUED: &str = "queued";
pub const RUNNING: &str = "running";
pub const DONE: &str = "done";
pub const FAILED: &str = "failed";
/// A job that was intentionally **stopped**. Terminal, but — the headline
/// guarantee — it is **NOT a failure**. Two producers:
///
/// * a **daemon** whose config toggle went off, or whose handle was dropped as
/// the process winds down: a never-ending daemon that stops for a reason
/// other than crashing is a pure fact, not a red row
/// ([`crate::jobs::kind::DAEMON`]);
/// * a **superseded chain job** ([`super::chain`]): a newer generation landed
/// before this one ran, so computing it would have produced a stale answer.
/// Not computing yesterday's snapshot is not a red row either.
///
/// An ordinary fire-and-complete job still never reaches this status. See
/// [`is_failure`].
pub const STOPPED: &str = "stopped";
/// Whether `s` is a terminal status (job will not change further).
pub fn is_terminal(s: &str) -> bool {
s == DONE || s == FAILED || s == STOPPED
}
/// Whether `s` is a terminal status that represents an actual **failure**
/// ([`FAILED`]). [`DONE`] and [`STOPPED`] are terminal but NOT failures — so a
/// daemon that reaches [`STOPPED`] "never counts as failed for not completing".
pub fn is_failure(s: &str) -> bool {
s == FAILED
}
}
/// Canonical job kinds — one per op family. Strings (not an enum) so a new op
/// needs no schema change; these are the conventional tags readers group/filter on.
pub mod kind {
pub const RELEASE_RUN: &str = "release_run";
/// `nornir release undo` (task #32): a reversal run, tracked like a release.
pub const RELEASE_UNDO: &str = "release_undo";
pub const WORKSPACE_FETCH: &str = "workspace_fetch";
pub const WORKSPACE_REPUBLISH: &str = "workspace_republish";
/// The PARENT job of a workspace **populate** (register-eager): the umbrella op
/// that clones every member then builds the warehouse. Its children stream in as
/// [`WORKSPACE_CLONE`] (one per member, as each lands) plus a final
/// [`WORKSPACE_REPUBLISH`] for the build. Readers fold the children under it.
pub const WORKSPACE_POPULATE: &str = "workspace_populate";
/// The clone/fetch of ONE member during a populate — a CHILD of a
/// [`WORKSPACE_POPULATE`] parent (`parent_id` set). Streamed: emitted+finished as
/// each member's clone completes, so the Jobs panel shows members appear
/// running→done one at a time. Distinct from [`WORKSPACE_FETCH`] (the on-demand
/// `Workspaces.Fetch` RPC over the whole workspace).
pub const WORKSPACE_CLONE: &str = "workspace_clone";
pub const DOCS_RENDER: &str = "docs_render";
pub const DOCS_EXPORT: &str = "docs_export";
pub const DOCS_BOOK: &str = "docs_book";
pub const DOCS_BOOK_SVG: &str = "docs_book_svg";
pub const KNOWLEDGE_SCAN: &str = "knowledge_scan";
pub const SCIP_INGEST: &str = "scip_ingest";
pub const DEEPSCAN: &str = "deepscan";
/// An **index snapshot** of one repo's RA/SCIP index INTO the warehouse —
/// `Index.Snapshot` / `Index.UploadSnapshot` (`snapshot_to_iceberg`). Distinct
/// from [`INDEX_BUILD`] (building the on-disk index): this is the
/// build-the-index → land-it-in-Iceberg op the snapshot RPCs run.
pub const SNAPSHOT: &str = "snapshot";
/// A **DWARF/symbol scan** of one built binary — `Introspect.Symbols` /
/// `Introspect.SymbolLookup` / `Introspect.DefinedIn` (`extract_symbols`).
/// Real work (reads + parses the artifact's debug info), tracked so the
/// "symbol-scan" op family shows in the Jobs panel like every other scan.
pub const SYMBOL_SCAN: &str = "symbol_scan";
/// A **warehouse table SCAN** — `Warehouse.Scan`, the read every Mímir/coverage/
/// dataview pane performs when it binds a table.
///
/// Added 2026-08-25 with the SCAN-is-not-a-user-action pass. The panes lost their
/// `Scan` buttons and start their own reads; without a ledger row those reads were
/// INVISIBLE — the Jobs panel showed nothing while the panes were plainly working,
/// which is the same "the pane built to show it could not see it" defect dwarves
/// `e3348b1` fixed for gRPC-triggered runs (both drain sites passed `None` as the
/// jera sink, so every run was historized and never jera-tracked).
///
/// A scan is cheap next to a populate, but "cheap" is not the test — VISIBLE is.
/// An operator who cannot see the read cannot tell a slow warehouse from a dead
/// one, and `jobsfleet.rs:74` already reads an absent ledger as an empty one.
pub const WAREHOUSE_SCAN: &str = "warehouse_scan";
/// A **security / SBOM + vulnerability scan** of one workspace repo —
/// `Mimir.SecurityScan` (`cargo metadata` SBOM + OSV query, persisting
/// `vuln_findings`). Heavy (subprocess + network), so it earns a job.
pub const SECURITY_SCAN: &str = "security_scan";
/// **Per-member CHILD** of a member-iterating populate phase — the leaf of the
/// deep (3-level) populate tree `workspace_populate ← <phase> ← <member>`. The
/// phase parents ([`KNOWLEDGE_SCAN`], [`SECURITY_SCAN`], [`SCIP_INGEST`],
/// [`SNAPSHOT`]) iterate per member internally; these kinds are the one job
/// each member's work emits under its phase parent (`target` = member name,
/// `parent_id` = the phase-parent's `job_id`). The viz Jobs tree nests them by
/// `parent_id`, so each phase becomes expandable into its members.
pub const KNOWLEDGE_MEMBER: &str = "knowledge_member";
/// Per-member child of the [`SECURITY_SCAN`] phase parent — see [`KNOWLEDGE_MEMBER`].
pub const SECURITY_MEMBER: &str = "security_member";
/// Per-member child of the [`SCIP_INGEST`] phase parent — see [`KNOWLEDGE_MEMBER`].
pub const SCIP_MEMBER: &str = "scip_member";
/// Per-member child of the [`SNAPSHOT`] phase parent — see [`KNOWLEDGE_MEMBER`].
pub const SNAPSHOT_MEMBER: &str = "snapshot_member";
pub const BENCH_RUN: &str = "bench_run";
pub const TEST_MATRIX: &str = "test_matrix";
pub const ARCH_GENERATE: &str = "arch_generate";
pub const INDEX_BUILD: &str = "index_build";
pub const VECTOR_EMBED: &str = "vector_embed";
pub const BAKEOFF: &str = "bakeoff";
pub const AIRGAP: &str = "airgap";
/// `nornir coverage` — source line/region coverage measurement.
pub const COVERAGE: &str = "coverage";
/// A container run tracked like any other job: bring-up → running, exit →
/// done|failed, logs captured into `detail_json`. See
/// [`crate::container::run_container`] for the no-shell bollard bring-up.
pub const CONTAINER: &str = "container";
/// The forge cloning a Build Thing's ABSENT declared sibling checkout beside it
/// (`[build.toolchain] siblings`) before the cell runs — one job per sibling.
pub const SIBLING_CLONE: &str = "sibling_clone";
/// A **Build-tab TEST-BOOT** of a built cargo-variant binary (`.exe`) — spawn
/// the built executable as a child process with an interactive console/PROMPT
/// (stdin→child, stdout/stderr→UI). Cheapest boot mode (no container/VM). The
/// job tracks the child's lifecycle bidirectionally: kill the job → SIGKILL the
/// process-group; the child dies on its own → the job flips Killed/Failed.
pub const BOOT_EXE: &str = "boot_exe";
/// A **Build-tab TEST-BOOT** of a `ContainerBuild` image — bring the image up
/// (podman/bollard) and stream its logs to the UI. Bidirectional kill like
/// [`BOOT_EXE`].
pub const BOOT_CONTAINER: &str = "boot_container";
/// A **Build-tab TEST-BOOT** of a `VmBuild` — boot it as a KVM guest via
/// tunnr's boot primitive (`tunnr_vm::boot_test`). Bidirectional kill like
/// [`BOOT_EXE`].
pub const BOOT_VM: &str = "boot_vm";
/// `nornir warehouse compact` — small-file COMPACTION: rewrite a table's
/// (partition's) many small data files into a few large SORTED Parquet files
/// and commit a replace snapshot. Real IO + a catalog commit, so it earns a
/// job row in the Jobs panel.
pub const WAREHOUSE_COMPACT: &str = "warehouse_compact";
/// `nornir warehouse expire` — snapshot EXPIRY + orphan-file GC: drop old
/// snapshots from the catalog and delete their now-unreferenced data/manifest
/// files. Reclaims disk; tracked like every other server op.
pub const WAREHOUSE_EXPIRE: &str = "warehouse_expire";
/// 📥 The 🧠 Models viz tab downloading an LLM / embedding model (via
/// `ollama pull` or HuggingFace hf-hub) — one job per download, with a
/// progress row + cancel. See `src/viz/models_tab.rs`.
pub const MODEL_DOWNLOAD: &str = "model_download";
/// 🦀 A **host toolchain install**: nornir's 🚦 pre-flight popup installing a rust
/// toolchain on a host that has none, by fetching `rustup-init` over HTTPS and
/// exec'ing it with argv. Multi-minute, user-clicked, exactly once per host.
///
/// It is a job because it used to be a bare `std::thread::spawn` — invisible to
/// `nornir jobs`, no lifecycle row, and if the process died mid-install nothing
/// recorded that it had ever started (owner, 2026-08-23: "should be reg[istered]
/// jera long work … but use jera!"). Long work belongs in the ledger.
pub const TOOLCHAIN_INSTALL: &str = "toolchain_install";
/// 🎬 A **Saga capture drain**: the viz `⏺ Record`/`⏹ Stop` toggle draining the
/// process-global `nornir_saga` event buffer and writing the batch to the
/// `saga_events` warehouse table (via `IcebergSagaSink`). One job per Stop.
pub const SAGA_CAPTURE: &str = "saga_capture";
/// 📡 A **monitor fetch sweep**: the `nornir monitor` poll loop fetching a
/// workspace's members (git clone/fetch) on a tick. One job per swept workspace.
pub const MONITOR_FETCH: &str = "monitor_fetch";
/// 🔨 A **brokkr CI pipeline run**: the brokkr engine running one pipeline
/// (trigger → plan → run → historize → notify). The parent job; each stage's
/// step runs as a `start_child` under it. One job per pipeline run.
/// See `src/brokkr/engine.rs`.
pub const CI_BUILD: &str = "ci_build";
/// ♾️ A **never-ending DAEMON** job — a long-running service that *strives* to
/// reach and hold a steady state (e.g. a korp **collector**: planet → regions →
/// tiles → ONLINE), self-healing across restarts. Unlike every other kind here it
/// is **not** fire-and-complete: it stays [`status::RUNNING`] indefinitely,
/// carrying a live state-machine **phase** + progress% + a `healthy` flag through
/// the [`report`] side channel ([`JobReport::Phase`](report::JobReport::Phase)),
/// and it **never counts as failed for not completing** — a daemon that stops
/// records [`status::STOPPED`] (a pure fact), not [`status::FAILED`]. Started via
/// [`JobHandle::start_daemon`] / [`JobStore::start_daemon`], resumed across a
/// server restart via [`JobStore::start_or_resume_daemon`], and stopped via
/// [`JobHandle::stop`]. One daemon job per enabled collector (target = collector
/// name). Its [`JobClass`] is [`JobClass::Daemon`].
pub const DAEMON: &str = "daemon";
/// One **Mimir lens FOLD** submitted as a background [`chain`](super::chain)
/// job: a knowledge snapshot landed, so a lens's answer is recomputed ONCE and
/// written to its `lens_<name>` warehouse table, instead of being folded on
/// every request that asks for it (measured 2026-08-22 on oden:
/// `Mimir.ChangeRisk` folded 13.4 s per call). `target` = `<lens>@<repo
/// scope>`, `detail_json` = the payload `(workspace, repo, snapshot_id,
/// lens)`. [`JobClass::Parallel`] by class but queued at
/// [`chain::Priority::Low`], so it never delays a build, a populate or a
/// bench.
pub const LENS: &str = "lens";
/// One **Mimir FEED PRODUCER** submitted as a background [`chain`](super::chain)
/// job on the SAME queue as the [`LENS`] jobs but at
/// [`chain::Priority::Normal`] — so every pending feed drains BEFORE any Low
/// lens (a lens cannot fold a base table a feed has not written yet). A feed
/// runs the EXISTING producer for a base table with no populate producer —
/// `binary_facts`/`symbol_sizes` (`nornir bin analyze`), `coverage_fn`
/// (`nornir coverage`), `semver_diffs` (the world-coderoom compat edge) — and
/// writes both the base table and a `feed_runs` honesty row (rows written, or
/// WHY it could not run). `target` = `<feed>@<repo scope>`. [`JobClass::Parallel`]
/// by class, like [`LENS`]. See `nornir::feed_chain`.
pub const FEED: &str = "feed";
}
// ---------------------------------------------------------------------------
// Concurrency policy — how aggressively a kind may run alongside its peers.
// ---------------------------------------------------------------------------
/// Concurrency class of a job [`kind`] — the scheduler's policy for it.
///
/// The server is one long-lived process, so "how many of these may run at once"
/// is a process-wide question answered here. Today there are two classes; more
/// (e.g. a per-resource pool) can be added without touching call sites.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum JobClass {
/// **Heavy / serial.** CPU- or IO-saturating work whose results are only
/// meaningful in isolation — a benchmark sharing the box with another
/// benchmark measures noise. At most **one** exclusive job runs process-wide
/// at a time; the rest wait in `queued` (see [`status::QUEUED`]).
Exclusive,
/// **Parallel.** The default — runs immediately, never gated, unaffected by
/// (and not affecting) the exclusive gate. Most ops (docs render, fetch,
/// arch generate, container runs, …) are here.
Parallel,
/// **Daemon.** A never-ending service ([`kind::DAEMON`]): runs immediately (like
/// [`JobClass::Parallel`] — never gated), but is *expected to never terminate on
/// its own*, so it "never counts as failed for not completing" — dropping its
/// handle records [`status::STOPPED`], not [`status::FAILED`]. The class the UI
/// keys off to render a daemon row (live phase, no "did it finish?" expectation)
/// instead of a fire-and-complete row.
Daemon,
}
/// The [`JobClass`] for a job [`kind`]. The default is [`JobClass::Parallel`];
/// only genuinely box-saturating kinds whose measurement demands isolation are
/// [`JobClass::Exclusive`]. Today that is the benchmark runner ([`kind::BENCH_RUN`]):
/// a queued bench waits for the running one to finish so neither perturbs the
/// other's numbers. Add a kind here when it must not run concurrently with its
/// peers.
pub fn class_of(kind: &str) -> JobClass {
match kind {
kind::BENCH_RUN => JobClass::Exclusive,
kind::DAEMON => JobClass::Daemon,
_ => JobClass::Parallel,
}
}
/// `true` when `kind` is [`JobClass::Exclusive`] — i.e. the serial scheduler must
/// admit at most one such job at a time process-wide. Convenience over
/// [`class_of`].
pub fn is_exclusive(kind: &str) -> bool {
matches!(class_of(kind), JobClass::Exclusive)
}
/// `true` when `kind` is a never-ending [`JobClass::Daemon`] (today: [`kind::DAEMON`]).
/// Drives the daemon lifecycle: such a job records [`status::STOPPED`] (not
/// [`status::FAILED`]) when its handle is dropped without an explicit terminal call —
/// "never counts as failed for not completing". Convenience over [`class_of`].
pub fn is_daemon(kind: &str) -> bool {
matches!(class_of(kind), JobClass::Daemon)
}
/// The process-global serial gate for [`JobClass::Exclusive`] jobs — a
/// `Semaphore` with a single permit. Held from the moment an exclusive job leaves
/// `queued` (acquires the permit, flips to `running`) until it reaches a terminal
/// state, so only one exclusive job runs at a time across the whole process.
/// `OnceLock` because the server is the long-lived owner and the gate must be the
/// same instance for every call site.
fn exclusive_gate() -> &'static Arc<tokio::sync::Semaphore> {
static GATE: std::sync::OnceLock<Arc<tokio::sync::Semaphore>> = std::sync::OnceLock::new();
GATE.get_or_init(|| Arc::new(tokio::sync::Semaphore::new(1)))
}
/// The **durable, last-known state-machine phase** of a [`kind::DAEMON`] job,
/// persisted ONTO its [`JobRecord`] in `jobs.redb` so an out-of-process /
/// [`JobStore::open_read_only`] reader can see the daemon's live phase even though
/// the in-memory [`ReportRegistry`] (the fast path) is process-local and starts
/// EMPTY for a fresh/read-only store.
///
/// This is the *durable mirror* of the ephemeral [`JobReport::Phase`] the daemon
/// emits through its [`Reporter`]: written only when [`Reporter::phase`] fires (a
/// rare state-machine transition — planet → regions → tiles → online — NOT the
/// high-frequency progress tick), so it stays durable without reintroducing the
/// per-tick redb write-amplification the [`report`] channel exists to avoid. The
/// progress figures are a **snapshot** captured from the in-memory registry at the
/// instant the phase was persisted (so they can lag the live progress bar between
/// phase transitions — the honest caveat of persisting only on `phase()`).
///
/// `current`/`total` are integers (not an `f32` fraction) so [`JobRecord`] keeps
/// its `Eq` derive; read the fraction via [`Self::fraction`].
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct DaemonPhase {
/// The domain state-machine phase name the Jobs view shows verbatim (e.g.
/// `"downloading"`, `"cutting"`, `"online"`).
pub name: String,
/// The green/amber health signal for this phase.
pub healthy: bool,
/// Latest known progress numerator captured from the report registry when this
/// phase was persisted (`None` when the daemon had reported no progress yet).
pub progress_current: Option<u64>,
/// Latest known progress denominator (`None` = indeterminate / no total).
pub progress_total: Option<u64>,
}
impl DaemonPhase {
/// The determinate progress fraction in `[0,1]`, or `None` when progress is
/// absent or indeterminate — the durable mirror of
/// [`JobReports::progress_fraction`].
pub fn fraction(&self) -> Option<f32> {
match (self.progress_current, self.progress_total) {
(Some(c), Some(t)) if t > 0 => Some((c as f64 / t as f64).clamp(0.0, 1.0) as f32),
_ => None,
}
}
}
/// One job's lifecycle record. The redb value (serde_json). `detail_json` and
/// `result_ref` are free-form (`""` when absent) so the record never needs a
/// schema migration as new ops adopt the ledger.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct JobRecord {
pub job_id: String,
/// One of [`kind`].
pub kind: String,
/// Repo / workspace / artifact the job acts on (human-readable).
pub target: String,
pub workspace: String,
/// One of [`status`]; overwritten in place as the job advances.
pub status: String,
pub ts_start_micros: i64,
/// Set on the terminal write only.
pub ts_end_micros: Option<i64>,
/// `(ts_end - ts_start) / 1000`, set on the terminal write.
pub elapsed_ms: Option<i64>,
/// Free-form JSON detail (args / counts / error chain), or `""`.
pub detail_json: String,
/// Pointer to the domain result row this job produced, e.g.
/// `"bench_runs:<uuid>"`, `"release_events:<run_id>"`, or `""`.
pub result_ref: String,
/// **Hierarchy** — the `job_id` of this job's PARENT, when it is a *sub-job*
/// of a larger op (`None` = a top-level job). A multi-phase / multi-member op
/// emits one parent job plus a child per phase/member: e.g. a workspace
/// *populate* is a [`kind::WORKSPACE_POPULATE`] parent with a
/// [`kind::WORKSPACE_CLONE`] child per member (streamed as each clone lands)
/// and a [`kind::WORKSPACE_REPUBLISH`] child for the build. Readers (the Jobs
/// panel / `Viz.Jobs` / `nornir jobs`) render children folded under their
/// parent. `#[serde(default)]` so old rows (written before this field) decode
/// as top-level — no migration.
#[serde(default)]
pub parent_id: Option<String>,
/// **Which node ran this job** — the fleet machine (roster node id / host)
/// that executed it, stamped when the multi-machine scheduler
/// ([`crate::agent`]) dispatches the job across the constellation
/// (Loki/Odin/NUC + Njord). `None` = ran on the local control-plane / not a
/// distributed job. `#[serde(default)]` so rows written before this field
/// decode as `None` — non-breaking, exactly like [`parent_id`](Self::parent_id).
#[serde(default)]
pub node: Option<String>,
/// **Durable last-known daemon phase** — the [`kind::DAEMON`] state-machine
/// phase persisted onto the row so an out-of-process / [`JobStore::open_read_only`]
/// reader can see it (the in-memory [`ReportRegistry`] fast path is process-local
/// and starts empty for a fresh store). Written when [`Reporter::phase`] fires
/// through a daemon-wired [`JobStore::daemon_report_sink`]; `None` for every
/// non-daemon job and for a daemon that has not reported a phase yet.
/// `#[serde(default)]` so rows written before this field decode as `None` — the
/// same non-breaking pattern as [`parent_id`](Self::parent_id)/[`node`](Self::node).
#[serde(default)]
pub phase: Option<DaemonPhase>,
/// **Who initiated this job** — the authenticated principal's `subject` for a
/// job that came in over gRPC (`nornir_service::provider::Principal::subject`),
/// or the OS user for a job started from a local CLI. `None` = the initiator
/// was never stamped.
///
/// The ledger recorded WHAT ran, WHERE ([`node`](Self::node)) and UNDER WHOM in
/// the process tree ([`parent_id`](Self::parent_id)) — but never WHO ASKED, so
/// a Jobs panel could not tell a release the operator started from one an
/// external client kicked off. The identity already exists at the door; it was
/// simply dropped on the floor. `#[serde(default)]` so rows written before this
/// field decode as `None`, exactly like `parent_id`/`node`.
#[serde(default)]
pub actor: Option<String>,
/// **How that initiator authenticated** — `Principal::method` for a gRPC job
/// (`"bearer"`, `"mtls"`, `"apikey"`, …) or [`actor_method::LOCAL`] for one
/// started from a local CLI on this box.
///
/// This is the field that answers "was it us or an external user": the service
/// account and a stranger can both be a `subject`, but only a local run is
/// [`actor_method::LOCAL`]. Read it with [`JobRecord::is_local_actor`].
#[serde(default)]
pub actor_method: Option<String>,
}
impl JobRecord {
/// `true` when the job is in a terminal state.
pub fn is_terminal(&self) -> bool {
status::is_terminal(&self.status)
}
/// The durable last-known daemon [`DaemonPhase`] on this row, if any — the
/// out-of-process-readable mirror of the live [`JobReport::Phase`]. This is what
/// a [`JobStore::open_read_only`] / snapshot / RPC viewer reads to show a daemon's
/// current phase without the process-local report registry.
pub fn daemon_phase(&self) -> Option<&DaemonPhase> {
self.phase.as_ref()
}
/// `true` when this job was started from a local CLI on the serving box
/// rather than by a remote client — i.e. [`actor_method`](Self::actor_method)
/// is [`actor_method::LOCAL`].
///
/// This is deliberately NOT "the subject looks like ours": a bearer token
/// minted for the service account is still a REMOTE call, and answering the
/// operator's question ("was it us, or an external user?") with a name match
/// would say yes to anyone holding that token.
pub fn is_local_actor(&self) -> bool {
self.actor_method.as_deref() == Some(actor_method::LOCAL)
}
/// One-line initiator for a Jobs panel: `"alice (local)"`,
/// `"ci-runner (bearer)"`, or `"unknown"` when the row predates the field or
/// nothing stamped it. Never invents an initiator.
pub fn actor_label(&self) -> String {
match (self.actor.as_deref(), self.actor_method.as_deref()) {
(Some(a), Some(m)) => format!("{a} ({m})"),
(Some(a), None) => a.to_string(),
(None, _) => "unknown".to_string(),
}
}
}
/// Values for [`JobRecord::actor_method`]. A gRPC job carries the authenticating
/// scheme straight from `nornir_service::provider::Principal::method`; only a job
/// started from a local CLI uses [`LOCAL`].
pub mod actor_method {
/// Started from a CLI running on this machine, writing this machine's ledger —
/// no remote credential involved. The only value [`JobRecord::is_local_actor`]
/// accepts.
pub const LOCAL: &str = "local";
/// Started from a CLI, but a THIN one: it submitted the row to a remote server
/// with a bearer token, so the initiator is a real person but the call crossed
/// the network and carried a credential. Deliberately NOT [`LOCAL`] — the whole
/// point of the distinction is that a credential was presented.
pub const CLI_REMOTE: &str = "cli-remote";
}
/// Which jobs a read returns.
#[derive(Debug, Clone)]
pub enum JobSelector {
/// Every job in the ledger.
All,
/// Only jobs for this workspace.
Workspace(String),
/// Only jobs of this [`kind`].
Kind(String),
}
impl JobSelector {
/// Does this record belong in a read scoped by `self`?
///
/// **Public because a reader outside this crate has to be able to SELECT with the
/// same predicate `list` uses.** `dwarves_embed::read_ledger_jobs` needs two facts
/// from one table scan — *is the ledger empty* (a question about the LEDGER, which
/// a class window must not answer) and *which rows are of this kind* — and a second
/// [`JobStore::list`] call to get the second fact would be a second scan. Filtering
/// in the caller with a hand-rolled `r.kind == k` would be worse: a twin of this
/// match that drifts the first time a selector variant is added (LAW 5). One
/// predicate, one owner, every road.
pub fn matches(&self, r: &JobRecord) -> bool {
match self {
JobSelector::All => true,
JobSelector::Workspace(w) => &r.workspace == w,
JobSelector::Kind(k) => &r.kind == k,
}
}
}
fn now_micros() -> i64 {
chrono::Utc::now().timestamp_micros()
}
// ---------------------------------------------------------------------------
// JobSink — where a JobHandle's records go. Fat = redb; thin = a submit closure
// (the gRPC client, supplied by the binary so the library stays tonic-free).
// ---------------------------------------------------------------------------
/// The write side of a [`JobHandle`], plus the [`ReportSink`] a handle built from
/// it reports through by default. Cheap to clone (`Arc`). A fat client makes one
/// over the local `jobs.redb` ([`JobStore::sink`]); a thin client makes one from a
/// closure that submits the record over gRPC ([`JobSink::remote`]).
///
/// # Why the sink carries reports (2026-08-27)
///
/// [`JobHandle::start`] used to hand every new handle [`ReportSink::noop`], and
/// reports were opt-in via [`JobHandle::with_reports`]. That made a working
/// reporter something each call site had to remember, and the constellation
/// measured the result: `Reporter` had ZERO uses in nornir and ZERO in dwarves,
/// and the two live `JobHandle::start` sites that own real work —
/// `dwarves/crates/dwarves/src/engine.rs` (the `CI_BUILD` pipeline job) and
/// `nornir/src/bin/nornir.rs`'s `start_job` (every CLI-initiated job) — both take
/// a `&JobSink` and never had a `ReportSink` in scope to pass. So `job.reporter()`
/// at either site would have emitted into a noop and looked like a working
/// producer.
///
/// A sink already knows where a job's records go. It is the right place to know
/// where a job's reports go, and then no call site has a decision to forget: the
/// existing bare `JobHandle::start(sink.clone(), …)` gains a live reporter without
/// being touched. [`JobHandle::with_reports`] still overrides, for the cases that
/// genuinely want a different destination.
#[derive(Clone)]
pub struct JobSink {
write: Arc<dyn Fn(&JobRecord) + Send + Sync>,
/// Where a [`JobHandle`] made from this sink reports unless overridden.
reports: ReportSink,
}
impl JobSink {
/// A redb-backed sink: upserts each record into `db`, opportunistically
/// pruning on terminal writes. Best-effort — a write failure is logged and
/// swallowed (a ledger hiccup must never abort the op it tracks).
///
/// Its default reports carry the DURABLE half of
/// [`JobStore::daemon_report_sink`]: a [`JobReport::Phase`] lands on the job's
/// row, so a reader in another process sees it. There is no in-memory registry
/// here to take a progress snapshot from — a bare sink has no store — so
/// `progress_current`/`progress_total` stay `None`. [`JobStore::sink`]
/// overrides with the full sink, which has both.
pub fn redb(db: Arc<Database>) -> Self {
let phase_db = db.clone();
JobSink {
write: Arc::new(move |rec: &JobRecord| {
if let Err(e) = redb_upsert(&db, rec) {
eprintln!(" ⚠ jobs: dropped {} row for {}/{} (non-fatal): {e:#}", rec.status, rec.kind, rec.target);
return;
}
if rec.is_terminal() {
if let Err(e) = redb_prune(&db, KEEP_DAYS, CAP) {
eprintln!(" ⚠ jobs: prune failed (non-fatal): {e:#}");
}
}
}),
reports: ReportSink::remote(move |job_id: &str, report: JobReport| {
if let JobReport::Phase { name, healthy } = &report {
let phase = DaemonPhase {
name: name.clone(),
healthy: *healthy,
progress_current: None,
progress_total: None,
};
if let Err(e) = redb_persist_phase(&phase_db, job_id, &phase) {
eprintln!(" ⚠ jobs: dropped durable phase for {job_id} (non-fatal): {e:#}");
}
}
}),
}
}
/// A sink that hands each record to `f` — the thin-client path, where `f`
/// performs the `Jobs.Submit` RPC. The library never links tonic; the binary
/// supplies the closure.
///
/// ⚠ NAMED GAP, not overlooked: reports default to [`ReportSink::noop`] here.
/// A phase is mirrored onto the ledger ROW, and a thin client does not hold the
/// ledger — carrying it would mean re-submitting the record on every phase
/// change, which is a wire-cost decision, not a bug fix. A thin caller that
/// wants phases can pass its own sink through [`Self::with_reports`].
pub fn remote(f: impl Fn(&JobRecord) + Send + Sync + 'static) -> Self {
JobSink { write: Arc::new(f), reports: ReportSink::noop() }
}
/// A sink that drops every record — for ops that have no ledger configured
/// (keeps the `start`/`finish` call sites unconditional). Reports are dropped
/// too: there is no row to mirror a phase onto.
pub fn noop() -> Self {
JobSink { write: Arc::new(|_| {}), reports: ReportSink::noop() }
}
/// Replace this sink's default [`ReportSink`]. Used by [`JobStore::sink`] to
/// upgrade the row-only phase mirror above to the store's full sink (in-memory
/// registry + durable phase WITH the progress snapshot).
#[must_use]
pub fn with_reports(mut self, reports: ReportSink) -> Self {
self.reports = reports;
self
}
/// This sink's default [`ReportSink`] — what a [`JobHandle`] built from it
/// reports through unless [`JobHandle::with_reports`] overrides.
pub fn reports(&self) -> ReportSink {
self.reports.clone()
}
fn upsert(&self, rec: &JobRecord) {
(self.write)(rec)
}
/// **Write one already-formed record into this sink** — the public half of
/// [`upsert`](Self::upsert).
///
/// Every other path into a sink builds the record as it goes: `JobHandle::start`
/// mints it, `finish`/`fail` mutate and re-write it. That covers a process tracking
/// its OWN work, and it is the only shape jera exposed — so a process that RECEIVES
/// a finished record from somewhere else had no way in, and `JobSink::remote` (which
/// exists precisely so a sink can ship rows elsewhere) had no counterpart on the
/// receiving end.
///
/// korp found it: a THIN client's job rows must land in the SERVER's ledger, and the
/// server holds a `JobRecord` it deserialised from the wire, not a handle it started.
/// Without this the row could be sent and could not be stored (korp-v2 row 1).
///
/// The record is written verbatim — no timestamps are re-stamped and no status is
/// re-derived — because the row already happened somewhere else and the receiver's
/// clock is not where it happened. A caller that must not trust a field (korp
/// overwrites `node` with the identity it resolved for the sender, rather than the
/// one the sender claimed) fixes it up BEFORE calling this.
pub fn record(&self, rec: &JobRecord) {
self.upsert(rec);
}
}
// ---------------------------------------------------------------------------
// JobHandle — wraps one op's lifecycle. start → (finish | fail | Drop=failed).
// ---------------------------------------------------------------------------
/// Tracks a single op: writes a `running` record on [`start`](JobHandle::start),
/// a terminal record on [`finish`](JobHandle::finish)/[`fail`](JobHandle::fail),
/// and — via [`Drop`] — a `failed` record if neither was called (so a panic or
/// early `?` return never leaves a zombie `running` row).
pub struct JobHandle {
sink: JobSink,
rec: JobRecord,
finished: bool,
/// The exclusive-gate permit, held for an [`JobClass::Exclusive`] job from the
/// moment it leaves `queued` until it terminates (then dropped, admitting the
/// next queued exclusive job). `None` for parallel jobs and for jobs not run
/// through the scheduler.
_gate_permit: Option<tokio::sync::OwnedSemaphorePermit>,
/// Where this job's [`Reporter`] emits. Defaults to the SINK's report sink
/// ([`JobSink::reports`]), so a handle built from [`JobStore::sink`] reports
/// durably with no extra call; [`with_reports`](Self::with_reports) overrides.
/// A handle over [`JobSink::noop`] or [`JobSink::remote`] gets
/// [`ReportSink::noop`], because there is no local row to mirror a phase onto.
report_sink: ReportSink,
}
impl JobHandle {
/// Begin a job: build the `running` record and write it through `sink`.
/// `detail` is any JSON (args / counts); pass [`serde_json::Value::Null`]
/// for none.
pub fn start(
sink: JobSink,
kind: &str,
target: &str,
workspace: &str,
detail: serde_json::Value,
) -> Self {
let rec = Self::new_record(kind, target, workspace, detail, status::RUNNING, None);
sink.upsert(&rec);
let report_sink = sink.reports();
JobHandle { sink, rec, finished: false, _gate_permit: None, report_sink }
}
/// Begin a **child** job under `parent_id` — a sub-job of a larger op (e.g. one
/// member's clone under a workspace-populate parent). Identical to
/// [`start`](Self::start) but stamps `parent_id` so readers fold it under its
/// parent. Children are emitted+finished one at a time as the op streams, so the
/// Jobs panel shows them appear progressively.
pub fn start_child(
sink: JobSink,
parent_id: &str,
kind: &str,
target: &str,
workspace: &str,
detail: serde_json::Value,
) -> Self {
let rec = Self::new_record(
kind,
target,
workspace,
detail,
status::RUNNING,
Some(parent_id.to_string()),
);
sink.upsert(&rec);
let report_sink = sink.reports();
JobHandle { sink, rec, finished: false, _gate_permit: None, report_sink }
}
/// Begin a never-ending **daemon** job ([`kind::DAEMON`]) for `target` (the
/// collector / service name). Starts [`status::RUNNING`] like [`start`](Self::start)
/// — a daemon runs immediately, ungated — but its [`Drop`] records
/// [`status::STOPPED`] (a pure fact) instead of [`status::FAILED`]: it "never
/// counts as failed for not completing". Drive its live state-machine phase +
/// progress% + healthy flag through [`reporter`](Self::reporter) (see
/// [`report::Reporter::phase`]); stop it with [`stop`](Self::stop). To survive a
/// server restart without duplicating the ledger row, prefer
/// [`JobStore::start_or_resume_daemon`].
pub fn start_daemon(
sink: JobSink,
target: &str,
workspace: &str,
detail: serde_json::Value,
) -> Self {
Self::start(sink, kind::DAEMON, target, workspace, detail)
}
/// **Restart-safe resume.** Re-attach to an existing (leftover, non-terminal)
/// daemon [`JobRecord`] — reusing its `job_id` and original `ts_start_micros` so
/// the ledger keeps ONE row for this daemon across a process restart (its history
/// and uptime resume rather than a fresh row appearing each boot). Re-stamps the
/// row [`status::RUNNING`] through `sink`. The phase itself resumes from the
/// collector re-emitting it (reports are process-local; see [`report`]).
pub fn resume_daemon(sink: JobSink, mut prev: JobRecord, detail: serde_json::Value) -> Self {
prev.status = status::RUNNING.to_string();
prev.ts_end_micros = None;
prev.elapsed_ms = None;
if !detail.is_null() {
prev.detail_json = detail_to_string(detail);
}
sink.upsert(&prev);
let report_sink = sink.reports();
JobHandle { sink, rec: prev, finished: false, _gate_permit: None, report_sink }
}
/// **Serial scheduler entry.** Begin a job that respects its [`class_of`]
/// concurrency policy:
///
/// - [`JobClass::Parallel`] kinds behave exactly like [`start`](Self::start):
/// the record is written `running` and the handle returns immediately.
/// - [`JobClass::Exclusive`] kinds (e.g. [`kind::BENCH_RUN`]) are admitted to
/// the ledger as [`status::QUEUED`] first, then this future **waits** on the
/// process-global [`exclusive_gate`]. If another exclusive job is running,
/// the row stays `queued` (visible in `nornir jobs` / the viz list) until
/// that job finishes; only then does it flip `queued → running` and the
/// work proceeds. The returned handle holds the gate permit until it reaches
/// a terminal state, so the next queued exclusive job waits behind it.
///
/// Always resolves to a `running` [`JobHandle`]; the caller does its work and
/// calls [`finish`](Self::finish)/[`fail`](Self::fail) as usual.
pub async fn start_scheduled(
sink: JobSink,
kind: &str,
target: &str,
workspace: &str,
detail: serde_json::Value,
) -> Self {
if !is_exclusive(kind) {
return Self::start(sink, kind, target, workspace, detail);
}
// Exclusive: publish a `queued` row so the wait is visible, then block on
// the single-permit gate. The row sits `queued` for the whole wait.
let mut rec = Self::new_record(kind, target, workspace, detail, status::QUEUED, None);
sink.upsert(&rec);
let permit = exclusive_gate()
.clone()
.acquire_owned()
.await
.expect("exclusive job gate is never closed");
// Acquired — promote queued → running in place (same job_id key), and
// re-stamp the start time so `elapsed_ms` measures actual work, not the
// queue wait.
rec.status = status::RUNNING.to_string();
rec.ts_start_micros = now_micros();
sink.upsert(&rec);
let report_sink = sink.reports();
JobHandle { sink, rec, finished: false, _gate_permit: Some(permit), report_sink }
}
/// Build a fresh `JobRecord` in `initial_status` (no sink write).
fn new_record(
kind: &str,
target: &str,
workspace: &str,
detail: serde_json::Value,
initial_status: &str,
parent_id: Option<String>,
) -> JobRecord {
JobRecord {
job_id: uuid::Uuid::new_v4().to_string(),
kind: kind.to_string(),
target: target.to_string(),
workspace: workspace.to_string(),
status: initial_status.to_string(),
ts_start_micros: now_micros(),
ts_end_micros: None,
elapsed_ms: None,
detail_json: detail_to_string(detail),
result_ref: String::new(),
parent_id,
node: None,
phase: None,
actor: None,
actor_method: None,
}
}
/// The job's id — stamp it into a domain `result_ref` for a bidirectional
/// link if desired.
pub fn job_id(&self) -> &str {
&self.rec.job_id
}
/// Stamp **which node** ran this job — the fleet machine id (from the
/// multi-machine [`crate::agent::Roster`]) that executed it. Builder-style and
/// re-upserts, so the ledger row carries the node from the start. A handle that
/// never calls this keeps `node: None` (local / not distributed) — fully
/// additive, no existing call site changes. No-op once the job is terminal.
pub fn on_node(mut self, node: impl Into<String>) -> Self {
if !self.finished {
self.rec.node = Some(node.into());
self.sink.upsert(&self.rec);
}
self
}
/// Stamp **who initiated** this job — the authenticated principal's subject
/// plus the scheme it authenticated with. Builder-style and re-upserts, so the
/// ledger row carries the initiator from the start rather than only after the
/// job ends. A handle that never calls this keeps `actor: None` and reads back
/// as `"unknown"` — fully additive, no existing call site changes. No-op once
/// the job is terminal.
///
/// At a gRPC door pass `principal.subject` and `principal.method`; from a local
/// CLI pass the OS user and [`actor_method::LOCAL`].
pub fn by_actor(mut self, subject: impl Into<String>, method: impl Into<String>) -> Self {
if !self.finished {
self.rec.actor = Some(subject.into());
self.rec.actor_method = Some(method.into());
self.sink.upsert(&self.rec);
}
self
}
/// [`by_actor`](Self::by_actor) for a job started from a CLI on this machine:
/// stamps the OS user (`$USER`, else `uid:<n>`) with [`actor_method::LOCAL`].
pub fn by_local_user(self) -> Self {
// No libc: a service unit can run with neither var set, and inventing a
// name there would be worse than admitting we do not know. The METHOD is
// still `local`, which is the half the operator's question turns on.
let who = std::env::var("USER")
.or_else(|_| std::env::var("LOGNAME"))
.unwrap_or_else(|_| "unknown-local-user".to_string());
self.by_actor(who, actor_method::LOCAL)
}
/// The node this job is stamped to run on, if any (see [`on_node`](Self::on_node)).
pub fn node(&self) -> Option<&str> {
self.rec.node.as_deref()
}
/// OVERRIDE this handle's report destination — builder style.
///
/// ⚠ No longer how a handle gets a working reporter, and that is the fix: the
/// default now comes from the [`JobSink`] ([`JobSink::reports`]). A handle built
/// from [`JobStore::sink`] — including the bare
/// `JobHandle::start(sink.clone(), …)` calls in dwarves' `engine.rs` and
/// nornir's `start_job`, which only ever had a `&JobSink` in scope — reports
/// durably without calling this. Reach for it when a caller genuinely wants a
/// DIFFERENT destination than its sink implies.
pub fn with_reports(mut self, report_sink: ReportSink) -> Self {
self.report_sink = report_sink;
self
}
/// A [`Reporter`] handle for THIS job — hand it to the work being tracked so it
/// can emit [`JobReport`]s without knowing about the ledger. Cheap to clone. It
/// emits wherever this handle's sink says (see [`JobSink::reports`]), so for a
/// store-backed job a `phase` call reaches the `jobs.redb` row; for a
/// [`JobSink::noop`]/[`JobSink::remote`] job it goes nowhere, by design.
pub fn reporter(&self) -> Reporter {
Reporter::new(self.job_id().to_string(), self.report_sink.clone())
}
/// Terminal success. `result_ref` points at the domain row produced (or `""`).
pub fn finish(mut self, detail: serde_json::Value, result_ref: &str) {
self.terminate(status::DONE, detail_to_string(detail), result_ref.to_string());
}
/// Terminal **stop** of a daemon — a pure fact ([`status::STOPPED`]), NOT a
/// failure. Call this when the config toggle turns a collector off (a clean,
/// intentional stop) so the row reads `stopped`, not `failed`. `detail` is any
/// JSON note (reason / final counts) or [`serde_json::Value::Null`]. Dropping a
/// daemon handle without calling this records the same `stopped` status (see the
/// [`Drop`] impl); `stop` is the explicit, reason-carrying form.
pub fn stop(mut self, detail: serde_json::Value) {
self.terminate(status::STOPPED, detail_to_string(detail), String::new());
}
/// Terminal failure, carrying the error chain into `detail_json`.
pub fn fail(mut self, err: &anyhow::Error) {
let detail = serde_json::json!({ "error": format!("{err:#}") }).to_string();
self.terminate(status::FAILED, detail, String::new());
}
/// Terminal failure carrying a caller-supplied `detail` JSON instead of an
/// error chain — for ops that "ran but the result is a failure" (e.g. a
/// container that exited non-zero: the spawn/wait succeeded, so there is no
/// `anyhow::Error`, but the captured logs/exit_code belong on the row).
pub fn fail_with_detail(mut self, detail: serde_json::Value) {
self.terminate(status::FAILED, detail_to_string(detail), String::new());
}
/// Terminal failure that **cross-references its result** — the failure peer of
/// [`finish`](Self::finish), with the same `(detail, result_ref)` shape.
///
/// # Why this exists (found by a test, 2026-08-27)
///
/// Before it, no failing terminal could set [`JobRecord::result_ref`] at all:
/// [`fail`](Self::fail) and [`fail_with_detail`](Self::fail_with_detail) both pass
/// `String::new()`. So a run that PASSED got a row pointing at its historized report
/// and a run that BURNED got `result_ref: ""` — and the burn is the one an operator
/// actually needs to open. Every reader that follows the column (the Jobs pane pushes
/// a `result` meta entry only `if !r.result_ref.is_empty()`) therefore showed a link
/// on exactly the rows nobody needed it on.
///
/// dwarves had worked around it in THREE places identically — `engine.rs` twice and
/// `runa_exec.rs` once — by stuffing `"result_ref": …` into the free-form `detail_json`
/// blob, where no reader looks for it. Three copies of a workaround for a missing API
/// is the API being missing (LAW 5); this is the one writer they now share.
pub fn fail_with_ref(mut self, detail: serde_json::Value, result_ref: &str) {
self.terminate(status::FAILED, detail_to_string(detail), result_ref.to_string());
}
fn terminate(&mut self, st: &str, detail_json: String, result_ref: String) {
let end = now_micros();
self.rec.status = st.to_string();
self.rec.ts_end_micros = Some(end);
self.rec.elapsed_ms = Some((end - self.rec.ts_start_micros) / 1000);
self.rec.detail_json = detail_json;
self.rec.result_ref = result_ref;
self.sink.upsert(&self.rec);
self.finished = true;
}
}
impl Drop for JobHandle {
fn drop(&mut self) {
if !self.finished {
// A never-ending DAEMON that is dropped without an explicit terminal call
// has STOPPED — a pure fact, NOT a failure (the daemon guarantee: "never
// counts as failed for not completing"). Every other kind keeps the
// zombie-guard: a dropped fire-and-complete job failed (panic / early `?`).
if is_daemon(&self.rec.kind) {
let detail =
serde_json::json!({ "note": "daemon stopped (handle dropped)" }).to_string();
self.terminate(status::STOPPED, detail, String::new());
} else {
let detail = serde_json::json!({
"error": "job dropped without finish/fail (panic or early return)"
})
.to_string();
self.terminate(status::FAILED, detail, String::new());
}
}
}
}
fn detail_to_string(detail: serde_json::Value) -> String {
if detail.is_null() {
String::new()
} else {
detail.to_string()
}
}
// ---------------------------------------------------------------------------
// JobStore — the redb file. Fat writer / lock-tolerant reader.
// ---------------------------------------------------------------------------
/// **The one file name jera keeps under a ledger root.** Every join of it goes
/// through [`store_path`]; nothing else in this crate spells it.
pub const STORE_FILE: &str = "jobs.redb";
/// **THE ledger path for a warehouse root — `<warehouse root>/jobs.redb`, and
/// this function is the only writer of that rule.**
///
/// # Why this exists
///
/// The module docs have said `<warehouse root>/jobs.redb` since the ledger was
/// written, but jera never handed anyone a function for it, so every consumer
/// re-derived the join and they did not agree. Measured on this box 2026-08-25,
/// five live ledgers under four different spellings:
///
/// ```text
/// ~/.nornir/warehouse/jobs.redb 548K nornir <root>
/// ~/.nornir/workspaces/default/builds/jobs.redb 1.6M nornir <root>
/// ~/.nornir/workspaces/nordisk/builds/jobs.redb 592K nornir <root>
/// ~/.korp/jobs.redb 580K korp <root>
/// ~/.holger/holger-jobs/jobs.redb 536K holger <root>/holger-jobs
/// ~/.dwarves/warehouse/jobs/jobs.redb — dwarves <root>/jobs
/// ```
///
/// `JobStore::open` takes a `&Path` that means "the directory holding the redb
/// file", and a warehouse root is only *one* of the things a caller can pass. A
/// caller who passes the other one does not get an error — it gets a brand new,
/// empty, perfectly valid ledger a directory away from the real one, and an empty
/// ledger is indistinguishable from a quiet forge. That is the failure this
/// constellation already paid for once, from the writer side: a Live Jobs pane
/// whose emptiness meant "the writer was never wired", not "nothing ran".
///
/// So: a consumer that has a warehouse root calls this and does not spell the
/// join. A consumer whose ledger genuinely does not live at a warehouse root
/// (holger's `<data>/holger-jobs`) keeps passing its own directory to
/// [`JobStore::open`] — that is legitimate and unchanged; what it must not do is
/// invent a *second* rule for a warehouse root it already has.
pub fn store_path(warehouse_root: &Path) -> PathBuf {
warehouse_root.join(STORE_FILE)
}
/// **A ledger the caller is one directory away from** — `Some(path)` when `root`
/// holds no `jobs.redb` but a neighbour a consumer demonstrably confuses it with
/// does.
///
/// Exactly two neighbours are checked, because exactly two spellings exist in the
/// constellation and a third guess is how a partial rule survives:
///
/// * `<root>/jobs/jobs.redb` — the caller passed a warehouse root, the ledger was
/// written one level down (dwarves' spelling, `server.rs` / `main.rs`);
/// * `<root>/../jobs.redb` **and only when `root` is literally named `jobs`** —
/// the caller passed `<warehouse>/jobs`, the ledger is at the warehouse root
/// (nornir's spelling, and the 1.5 MB live one). The name guard is what keeps
/// holger's `<data>/holger-jobs` from being told its data root is a stray.
///
/// Returns `None` the moment `<root>/jobs.redb` exists: a ledger that is already
/// there is the answer, and a second file elsewhere is then somebody's deliberate
/// second warehouse (`~/.dwarves` beside `~/.nornir` is correct and settled —
/// `dwarves/crates/dwarves/src/store.rs:96-99`), not a mistake.
pub fn stray_ledger_near(root: &Path) -> Option<PathBuf> {
if store_path(root).is_file() {
return None;
}
let down = store_path(&root.join("jobs"));
if down.is_file() {
return Some(down);
}
if root.file_name().is_some_and(|n| n == "jobs") {
let up = store_path(root.parent()?);
if up.is_file() {
return Some(up);
}
}
None
}
/// The redb-backed job ledger at `<root>/jobs.redb`. Open once; the owning
/// process is the sole writer (single-writer discipline, like `registry.redb`).
pub struct JobStore {
db: Arc<Database>,
/// A ledger that already existed one directory away when this store was
/// opened — see [`stray_ledger_near`] and [`Self::stray_ledger`]. `None` on
/// every healthy open, which is the overwhelming majority.
stray: Option<PathBuf>,
/// Holds a copied-aside snapshot's temp dir alive (set by
/// [`open_read_only`](Self::open_read_only) when the live file was locked).
_snapshot: Option<tempfile::TempDir>,
/// The in-memory `job_id -> latest reports` table backing [`Self::job_reports`]
/// / [`Self::reports_snapshot`]. Never persisted to `jobs.redb` — see the
/// [`report`] module docs for why. Each `JobStore` instance (including a
/// read-only snapshot) gets its OWN registry: reports are this process's live
/// view, not part of the durable file.
reports: ReportRegistry,
}
impl JobStore {
/// Open (creating if absent) the ledger at `<root>/jobs.redb` for read+write.
///
/// `root` is a **directory**, and for a consumer that has a warehouse root it
/// is that root — see [`store_path`] for the rule and why jera now owns it.
/// If the open has to CREATE a new file while a ledger already sits one
/// directory away, that is recorded on the store ([`Self::stray_ledger`]) and
/// said once on stderr rather than passed off as an empty ledger.
pub fn open(root: &Path) -> Result<Self> {
// Before anything is created: creating `<root>/jobs.redb` is exactly what
// makes the neighbour undetectable afterwards.
let stray = stray_ledger_near(root);
if let Some(other) = &stray {
eprintln!(
"jera: opening an EMPTY job ledger at {} while one already exists at {} — \
the caller is one directory off; an empty ledger reads as a quiet forge",
store_path(root).display(),
other.display()
);
}
std::fs::create_dir_all(root)
.with_context(|| format!("create jobs root {}", root.display()))?;
let path = store_path(root);
let db = Database::create(&path).with_context(|| format!("open {}", path.display()))?;
// Materialize the table so first-time reads don't fail.
let w = db.begin_write()?;
{
let _ = w.open_table(TABLE)?;
}
w.commit()?;
Ok(Self { db: Arc::new(db), _snapshot: None, stray, reports: report::new_registry() })
}
/// Lock-tolerant read open. Like [`open`](Self::open), but if `jobs.redb` is
/// already locked exclusively by another process (the live `nornir-server` —
/// redb is single-writer), it copies the file aside into a temp dir and opens
/// *that*, yielding a point-in-time read-only snapshot. Mirrors
/// `nornir::warehouse::IcebergWarehouse::open_read_only`; the viz local-mode
/// feed uses this so it coexists with a running server instead of deadlocking.
pub fn open_read_only(root: &Path) -> Result<Self> {
match Self::open(root) {
Ok(s) => Ok(s),
Err(e) if is_lock_error(&e) => {
let live = store_path(root);
if !live.exists() {
return Err(e);
}
let tmp = tempfile::Builder::new()
.prefix("nornir-jobs-snapshot-")
.tempdir()
.context("create temp dir for jobs snapshot")?;
let snap = store_path(tmp.path());
copy_redb_consistent(&live, &snap).with_context(|| {
format!("copy locked jobs {} -> snapshot {}", live.display(), snap.display())
})?;
let db = Database::create(&snap)
.with_context(|| format!("open jobs snapshot {}", snap.display()))?;
Ok(Self {
db: Arc::new(db),
_snapshot: Some(tmp),
stray: None,
reports: report::new_registry(),
})
}
Err(e) => Err(e),
}
}
/// **A ledger that already existed one directory away when this store was
/// opened** — `None` on a healthy open.
///
/// `Some(path)` means this store is almost certainly the WRONG one: it was
/// created empty next to a ledger somebody is already writing. The open still
/// succeeds, because a hard error here would take down a best-effort ledger
/// wiring (`JobStore::open(...).ok()` degrading to [`JobSink::noop`] is the
/// pattern in dwarves, korp and holger) and turn "you opened the wrong path"
/// into "you have no ledger at all" — a strictly worse silence. Instead the
/// fact is recorded here and shouted once on stderr, so a cockpit pane can
/// render "reading X, but a ledger exists at Y" instead of an empty list that
/// looks like a quiet forge.
pub fn stray_ledger(&self) -> Option<&Path> {
self.stray.as_deref()
}
/// A redb sink over this store — the fat-client write path.
/// The write sink for this store, carrying this store's FULL report sink
/// ([`Self::daemon_report_sink`]: in-memory registry + durable phase mirror with
/// the progress snapshot). Because [`JobSink`] carries reports, every
/// `JobHandle::start*` built from this — including the bare ones in dwarves'
/// `engine.rs` and nornir's `start_job`, which only ever had a `&JobSink` in
/// scope — gets a live, durable reporter without a caller change.
pub fn sink(&self) -> JobSink {
JobSink::redb(self.db.clone()).with_reports(self.daemon_report_sink())
}
/// A [`ReportSink`] over this store's in-memory report registry — hand it to
/// anything that should be able to emit reports for jobs started elsewhere
/// (e.g. [`crate::boot::BootJobs::with_reports`]).
///
/// ⚠ EPHEMERAL ONLY, and no longer what a store's own handles get. Nothing
/// emitted through this sink survives the process — including
/// [`JobReport::Phase`], which means an out-of-process reader sees
/// `JobRecord.phase == None` no matter how diligently the producer reports. Since
/// 2026-08-27 [`Self::start`] / [`Self::start_child`] / [`Self::start_scheduled`]
/// wire [`Self::daemon_report_sink`] instead; reach for this one only when you
/// deliberately want the in-memory registry and nothing on disk.
pub fn report_sink(&self) -> ReportSink {
report::ReportSink::registry(self.reports.clone())
}
/// A **daemon** [`ReportSink`] over this store: the fast in-memory path of
/// [`Self::report_sink`] PLUS a durable mirror — when a [`JobReport::Phase`] lands
/// (a [`Reporter::phase`] call), the latest phase (name + healthy + a progress
/// snapshot) is persisted ONTO the job's `jobs.redb` row as a [`DaemonPhase`], so
/// an out-of-process / [`Self::open_read_only`] reader — which builds a FRESH,
/// empty in-memory registry — can still see the daemon's current phase.
///
/// Only [`JobReport::Phase`] triggers a redb write; every other report (progress
/// ticks, gauges, health, messages) stays purely in-memory, so the durable mirror
/// never reintroduces the per-tick write-amplification the [`report`] channel
/// exists to avoid.
///
/// ⚠ NOT DAEMON-ONLY ANY MORE (2026-08-27), despite the name. This sink now backs
/// EVERY [`JobHandle`] a store creates — [`Self::start`], [`Self::start_child`],
/// [`Self::start_scheduled`], [`Self::start_daemon`] and
/// [`Self::start_or_resume_daemon`]. The name is kept because it is published API.
///
/// The old split gave non-daemon handles the plain in-memory
/// [`Self::report_sink`], and the reason recorded for it was write amplification —
/// which the `Phase`-only limb above already prevents, for a daemon and a
/// `CI_BUILD` alike. What the split actually bought was a TRAP: a caller who wired
/// progress into a `JobStore::start` job got a phase that lived in that process's
/// memory and nowhere else, so an out-of-process reader (`Viz.Jobs`, a thin
/// client, [`Self::open_read_only`], which builds a FRESH EMPTY registry) saw
/// `JobRecord.phase == None` and a UI showed no progress. The bug looks like a
/// missing producer and is actually a sink chosen one line away from the emit.
///
/// A job's phase is exactly as invisible across a process boundary whether the
/// job is a daemon or not, so there is now ONE sink and no decision to get wrong.
/// [`Self::report_sink`] stays for callers who explicitly want the ephemeral-only
/// registry (e.g. `boot::BootJobs::with_reports`).
pub fn daemon_report_sink(&self) -> ReportSink {
let registry = self.reports.clone();
let db = self.db.clone();
ReportSink::remote(move |job_id: &str, report: JobReport| {
// Fast path: upsert into the in-memory latest-per-kind registry (identical
// to `ReportSink::registry`), capturing the current progress snapshot while
// the lock is held so a Phase persist below doesn't re-lock.
let progress = {
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.clone());
entry.updated_at = report::next_seq();
match entry.progress() {
Some(JobReport::Progress { current, total, .. }) => (Some(*current), *total),
_ => (None, None),
}
};
// Durable path: ONLY a phase transition persists to the row (rare event —
// not the high-frequency progress tick), best-effort like every ledger write.
if let JobReport::Phase { name, healthy } = &report {
let phase = DaemonPhase {
name: name.clone(),
healthy: *healthy,
progress_current: progress.0,
progress_total: progress.1,
};
if let Err(e) = redb_persist_phase(&db, job_id, &phase) {
eprintln!(" ⚠ jobs: dropped durable daemon phase for {job_id} (non-fatal): {e:#}");
}
}
})
}
/// A clone of this store's [`ReportRegistry`] `Arc` — the read side, for a
/// caller that wants to look reports up itself (e.g. to fold them into its own
/// `state_json`) rather than going through [`Self::job_reports`].
pub fn report_registry(&self) -> ReportRegistry {
self.reports.clone()
}
/// The latest report of each kind for job `id`, or `None` if it has never
/// reported (including: it doesn't exist, or it exists but never called a
/// [`Reporter`] method). Local, in-memory — see [`report`] module docs.
pub fn job_reports(&self, id: &str) -> Option<JobReports> {
self.reports.lock().unwrap_or_else(|e| e.into_inner()).get(id).cloned()
}
/// A snapshot of every job's latest reports — what a UI polls to render
/// progress bars / health dots across the whole job list. Newest-`updated_at`
/// ordering is left to the caller; this returns registry iteration order
/// (sorted by job id, since the registry is a `BTreeMap`).
pub fn reports_snapshot(&self) -> Vec<(JobId, JobReports)> {
self.reports
.lock()
.unwrap_or_else(|e| e.into_inner())
.iter()
.map(|(k, v)| (k.clone(), v.clone()))
.collect()
}
/// Begin a job whose lifecycle writes land in this store (fat path). The
/// returned handle's [`JobHandle::reporter`] is already wired to this store's
/// reports — call it to emit [`JobReport`]s the moment the job starts.
///
/// Wired to [`Self::daemon_report_sink`] since 2026-08-27, not the in-memory-only
/// [`Self::report_sink`]: a `Reporter::phase` call on a job started here now
/// reaches the `jobs.redb` row, so `Viz.Jobs` and every read-only reader can see
/// it. See [`Self::daemon_report_sink`] for why the daemon/non-daemon split was a
/// trap rather than an optimisation.
pub fn start(
&self,
kind: &str,
target: &str,
workspace: &str,
detail: serde_json::Value,
) -> JobHandle {
// No `.with_reports` needed: `self.sink()` already carries it (L5 — one
// place decides where a store's jobs report).
JobHandle::start(self.sink(), kind, target, workspace, detail)
}
/// Begin a **child** job under `parent_id` in this store — see
/// [`JobHandle::start_child`]. Used to stream the per-member / per-phase
/// sub-jobs of an umbrella op (a workspace populate) under its parent. Reports
/// are wired the same way as [`Self::start`].
pub fn start_child(
&self,
parent_id: &str,
kind: &str,
target: &str,
workspace: &str,
detail: serde_json::Value,
) -> JobHandle {
JobHandle::start_child(self.sink(), parent_id, kind, target, workspace, detail)
}
/// Begin a never-ending **daemon** job in this store (fat path) — see
/// [`JobHandle::start_daemon`]. `target` is the collector / service name. The
/// returned handle's [`JobHandle::reporter`] is already wired to this store's
/// report registry, so the daemon emits its live phase/%/healthy immediately.
pub fn start_daemon(
&self,
target: &str,
workspace: &str,
detail: serde_json::Value,
) -> JobHandle {
JobHandle::start_daemon(self.sink(), target, workspace, detail)
}
/// **Restart-safe daemon start.** If a leftover, non-terminal [`kind::DAEMON`]
/// row for this `(target, workspace)` already exists in the ledger — the row a
/// previous korp-server process wrote before it exited without a clean
/// [`JobHandle::stop`] — RESUME it ([`JobHandle::resume_daemon`]): reuse that
/// `job_id` + original start so the daemon keeps ONE ledger row across restarts
/// (uptime + history intact), rather than a fresh row appearing every boot.
/// Otherwise start a new daemon ([`Self::start_daemon`]). Either way the returned
/// handle is `running`, daemon-classed, and wired to this store's durable daemon
/// reports ([`Self::daemon_report_sink`]).
///
/// **Dedup.** A crash-and-restart storm (or a race) can leave MORE THAN ONE
/// leftover non-terminal daemon row for the same `(target, workspace)`. Rather than
/// resume one and silently orphan the rest as permanent `running` zombies, this
/// resumes the **oldest** (the longest-lived — its `ts_start_micros` carries the
/// truest uptime) and marks every other leftover [`status::STOPPED`] (a pure fact,
/// not a failure — the daemon guarantee), collapsing back to exactly ONE running
/// row for the collector.
pub fn start_or_resume_daemon(
&self,
target: &str,
workspace: &str,
detail: serde_json::Value,
) -> JobHandle {
let mut leftovers: Vec<JobRecord> = self
.list(&JobSelector::All)
.unwrap_or_default()
.into_iter()
.filter(|r| {
r.kind == kind::DAEMON
&& r.target == target
&& r.workspace == workspace
&& !r.is_terminal()
})
.collect();
if !leftovers.is_empty() {
// Resume the OLDEST; STOP the rest (dedup) so no non-terminal zombie is
// left orphaned. `list` is newest-first, so sort ascending and take [0].
leftovers.sort_by(|a, b| a.ts_start_micros.cmp(&b.ts_start_micros));
let prev = leftovers.remove(0);
for mut dup in leftovers {
let end = now_micros();
dup.status = status::STOPPED.to_string();
dup.ts_end_micros = Some(end);
dup.elapsed_ms = Some((end - dup.ts_start_micros) / 1000);
dup.detail_json = serde_json::json!({
"note": "daemon deduped on resume: superseded by an older running row for the same (target, workspace)"
})
.to_string();
if let Err(e) = redb_upsert(&self.db, &dup) {
eprintln!(" ⚠ jobs: failed to stop deduped daemon {} (non-fatal): {e:#}", dup.job_id);
}
}
return JobHandle::resume_daemon(self.sink(), prev, detail)
.with_reports(self.daemon_report_sink());
}
self.start_daemon(target, workspace, detail)
}
/// **Close the rows a dead process left open.** Every non-terminal row whose
/// `kind` is in `kinds` becomes [`status::STOPPED`] — a pure fact, never a
/// failure — with `why` recorded on it. Returns how many were closed.
///
/// A `running` row is only ever closed by the process that opened it, so a
/// process that died (a power cut, a `--refresh` restart, a kill) leaves its
/// rows `running` for ever. MEASURED on oden 2026-08-22: a `workspace_populate`
/// row 5.5 days old; 2026-09-10: three `workspace_republish` rows from the
/// generations two refreshes had replaced, read as live work by the Jera tab.
///
/// The caller names the kinds, and must name ONLY kinds that this process is
/// the sole runner of in this ledger — for nornir's server: the chain jobs, the
/// sweep's republish, and its phase children. A kind a thin client can submit
/// and run elsewhere (a bench on a laptop) is NOT the server's to close.
/// Call it once, at open, before anything of those kinds is started.
pub fn stop_orphans(&self, kinds: &[&str], why: &str) -> usize {
let orphans: Vec<JobRecord> = self
.list(&JobSelector::All)
.unwrap_or_default()
.into_iter()
.filter(|r| !r.is_terminal() && kinds.contains(&r.kind.as_str()))
.collect();
let mut closed = 0;
for mut row in orphans {
let end = now_micros();
let was = std::mem::replace(&mut row.status, status::STOPPED.to_string());
row.ts_end_micros = Some(end);
row.elapsed_ms = Some((end - row.ts_start_micros).max(0) / 1000);
row.detail_json = serde_json::json!({
"orphaned": why,
"was": was,
"detail": serde_json::from_str::<serde_json::Value>(&row.detail_json)
.unwrap_or(serde_json::Value::Null),
})
.to_string();
match redb_upsert(&self.db, &row) {
Ok(()) => closed += 1,
Err(e) => eprintln!(
" ⚠ jobs: failed to stop orphaned {} {} (non-fatal): {e:#}",
row.kind, row.job_id
),
}
}
closed
}
/// Begin a job through the **serial scheduler** (fat path) — see
/// [`JobHandle::start_scheduled`]. Exclusive kinds (e.g. [`kind::BENCH_RUN`])
/// are admitted `queued` and wait on the process-global gate; parallel kinds
/// start `running` immediately. `.await` resolves to a `running` handle.
pub async fn start_scheduled(
&self,
kind: &str,
target: &str,
workspace: &str,
detail: serde_json::Value,
) -> JobHandle {
JobHandle::start_scheduled(self.sink(), kind, target, workspace, detail).await
}
/// Apply a record submitted by a thin client (`Jobs.Submit`): upsert it and
/// prune on terminal writes. The server's write seam.
pub fn submit(&self, rec: &JobRecord) -> Result<()> {
redb_upsert(&self.db, rec)?;
if rec.is_terminal() {
redb_prune(&self.db, KEEP_DAYS, CAP)?;
}
Ok(())
}
/// Read jobs matching `sel`, newest-first by start time.
pub fn list(&self, sel: &JobSelector) -> Result<Vec<JobRecord>> {
let r = self.db.begin_read()?;
let t = r.open_table(TABLE)?;
let mut out = Vec::new();
for row in t.iter()? {
let (_k, v) = row?;
let rec: JobRecord =
serde_json::from_slice(v.value()).context("decode job record")?;
if sel.matches(&rec) {
out.push(rec);
}
}
out.sort_by(|a, b| b.ts_start_micros.cmp(&a.ts_start_micros));
Ok(out)
}
/// Retention: drop terminal jobs older than `keep_days`, then — if more than
/// `cap` rows remain — drop the oldest terminal jobs down to `cap`. Running
/// jobs are never pruned. Returns how many were removed.
pub fn prune(&self, keep_days: i64, cap: usize) -> Result<usize> {
redb_prune(&self.db, keep_days, cap)
}
}
/// Upsert one record (key = `job_id`), overwriting any prior state for that job.
/// Write `rec` to its row, PRESERVING a `phase` already there when the incoming
/// record carries none.
///
/// # Why the preserve limb exists (2026-08-27, found by a test, not by reading)
///
/// A job's row has TWO writers, and they know different things:
///
/// * the [`JobHandle`] lifecycle (`start` → `on_node`/`by_actor` → `finish`/`fail`
/// /`stop`/`Drop`) writes its own `rec`, which has `phase: None` and always
/// will — the handle never learns what the work reported;
/// * [`redb_persist_phase`], the report mirror, read-modify-writes ONLY the
/// `phase` field.
///
/// So the last lifecycle write ERASED every phase the mirror had put there. A
/// pipeline that reported three phases while it ran left `phase: None` on the row
/// the moment it finished, and a UI polling after the run saw a job that had never
/// reported anything. Mid-run it was all there — which is the worst shape of the
/// two, because the mechanism looks alive right up to the point anybody reads it.
/// For a daemon it was invisible: a daemon's terminal write happens at shutdown.
///
/// Fixed here rather than at the three handle write sites, because here is the ONE
/// writer of a row and a rule stated once cannot be forgotten at a fourth site
/// (L5). A record that carries a phase still overwrites — the mirror and
/// `JobStore::submit` of a phase-bearing record both mean it.
fn redb_upsert(db: &Database, rec: &JobRecord) -> Result<()> {
let w = db.begin_write()?;
{
let mut t = w.open_table(TABLE)?;
// Copy the previous phase out and drop the read guard before the insert.
let carried: Option<DaemonPhase> = if rec.phase.is_some() {
None
} else {
let prev: Option<Vec<u8>> = t.get(rec.job_id.as_str())?.map(|v| v.value().to_vec());
prev.and_then(|b| serde_json::from_slice::<JobRecord>(&b).ok()).and_then(|r| r.phase)
};
let bytes = match carried {
Some(phase) => {
let mut merged = rec.clone();
merged.phase = Some(phase);
serde_json::to_vec(&merged).context("encode job record")?
}
None => serde_json::to_vec(rec).context("encode job record")?,
};
t.insert(rec.job_id.as_str(), bytes.as_slice())?;
}
w.commit()?;
Ok(())
}
/// Persist the durable last-known daemon [`DaemonPhase`] onto job `job_id`'s row —
/// a read-modify-write of the existing record (leaving every other field intact),
/// so an out-of-process reader can see the phase. No-op if the row is gone (e.g.
/// pruned) — the phase mirror never resurrects a record. See
/// [`JobStore::daemon_report_sink`].
fn redb_persist_phase(db: &Database, job_id: &str, phase: &DaemonPhase) -> Result<()> {
let w = db.begin_write()?;
{
let mut t = w.open_table(TABLE)?;
// Copy the bytes out and drop the read guard before the mutable insert.
let current: Option<Vec<u8>> = t.get(job_id)?.map(|v| v.value().to_vec());
if let Some(bytes) = current {
let mut rec: JobRecord =
serde_json::from_slice(&bytes).context("decode job record for phase persist")?;
rec.phase = Some(phase.clone());
let out = serde_json::to_vec(&rec).context("encode job record")?;
t.insert(job_id, out.as_slice())?;
}
}
w.commit()?;
Ok(())
}
/// See [`JobStore::prune`].
fn redb_prune(db: &Database, keep_days: i64, cap: usize) -> Result<usize> {
// Snapshot all rows first (read txn), decide what to drop, then remove in one
// write txn — keeps the borrow of the read table from overlapping the write.
let all: Vec<JobRecord> = {
let r = db.begin_read()?;
let t = r.open_table(TABLE)?;
let mut v = Vec::new();
for row in t.iter()? {
let (_k, val) = row?;
v.push(serde_json::from_slice::<JobRecord>(val.value()).context("decode job record")?);
}
v
};
let cutoff = now_micros() - keep_days * 86_400 * 1_000_000;
let mut victims: Vec<String> = Vec::new();
// 1) Aged-out terminal jobs.
for rec in &all {
if rec.is_terminal() && rec.ts_end_micros.unwrap_or(rec.ts_start_micros) < cutoff {
victims.push(rec.job_id.clone());
}
}
// 2) Over-cap: drop the oldest terminal jobs (by start time) not already a
// victim, until the surviving count is within `cap`. Running jobs stay.
let surviving = all.len().saturating_sub(victims.len());
if surviving > cap {
let mut terminal_left: Vec<&JobRecord> = all
.iter()
.filter(|r| r.is_terminal() && !victims.contains(&r.job_id))
.collect();
terminal_left.sort_by(|a, b| a.ts_start_micros.cmp(&b.ts_start_micros)); // oldest first
let need = surviving - cap;
for rec in terminal_left.into_iter().take(need) {
victims.push(rec.job_id.clone());
}
}
if victims.is_empty() {
return Ok(0);
}
let w = db.begin_write()?;
{
let mut t = w.open_table(TABLE)?;
for id in &victims {
t.remove(id.as_str())?;
}
}
w.commit()?;
Ok(victims.len())
}
/// True when `err` (anywhere in its chain) is redb's exclusive-lock rejection —
/// another process holds the single-writer lock on `jobs.redb`. Matches on the
/// message (redb's `DatabaseAlreadyOpen` / lock wording), like
/// `nornir::warehouse::is_catalog_lock_error`.
fn is_lock_error(err: &anyhow::Error) -> bool {
err.chain().any(|e| {
let m = e.to_string();
m.contains("Database already open") || m.contains("Cannot acquire lock")
})
}
/// Copy a locked `live` redb file → `dst` across a **size-stable window**, so the
/// copy is a complete, openable image even while the owner is committing. redb
/// grows the file in regions (header advertises the larger layout before the file
/// is extended); a blind copy can capture a file shorter than its own header
/// claims and trip redb's length invariant on open. Same rationale and fix as
/// `warehouse::copy_catalog_consistent` — copy, re-stat, retry until the source
/// length is unchanged across the copy and the copy is not short.
fn copy_redb_consistent(live: &Path, dst: &Path) -> Result<()> {
const MAX_ATTEMPTS: usize = 64;
let len_of = |p: &Path| -> Result<u64> {
Ok(std::fs::metadata(p).with_context(|| format!("stat {}", p.display()))?.len())
};
for attempt in 0..MAX_ATTEMPTS {
let before = len_of(live)?;
std::fs::copy(live, dst)
.with_context(|| format!("copy {} -> {}", live.display(), dst.display()))?;
let after = len_of(live)?;
if before == after && len_of(dst)? >= after {
return Ok(());
}
std::thread::sleep(std::time::Duration::from_millis(2 + attempt as u64 / 4));
}
anyhow::bail!(
"snapshot of locked jobs db {} never reached a size-stable window after {MAX_ATTEMPTS} attempts",
live.display()
)
}
// ---------------------------------------------------------------------------
// Workspace-scoped sink helper — the shared answer to "give me a durable job
// sink for my app" that korp (`jobs.rs`) and modgunn (`tools.rs`) had each
// hand-rolled byte-identically. See [`workspace_sink`].
// ---------------------------------------------------------------------------
/// Where an app keeps its job ledger, from its `workspace` tag: the
/// `<WORKSPACE_UPPER>_JOBS` env var (e.g. `KORP_JOBS`, `MODGUNN_JOBS`) when set,
/// else `<temp>/<workspace>_jobs.redb`. The env override lets a rendered systemd
/// unit / a test point the ledger at an explicit path; the temp default keeps a
/// bare GUI run self-contained.
///
/// This is the exact path korp's `ledger_path()` and modgunn's `ledger_path()`
/// each derived — now derived once, parameterised on the workspace, so the two
/// (env-var + file) literals that were all that differed become the argument.
pub fn workspace_ledger_path(workspace: &str) -> PathBuf {
let env_var = format!("{}_JOBS", workspace.to_uppercase());
if let Ok(p) = std::env::var(&env_var) {
if !p.trim().is_empty() {
return PathBuf::from(p);
}
}
std::env::temp_dir().join(format!("{workspace}_jobs.redb"))
}
/// A best-effort durable [`JobSink`] for `workspace` — the missing sibling of
/// [`JobStore::open`]/[`JobStore::sink`] that every jera-consuming app wanted.
///
/// Resolves the ledger path via [`workspace_ledger_path`], opens the file's
/// **parent** as the ledger root (`JobStore` manages `jobs.redb` under a root
/// dir), and returns that store's [`JobSink`]. If the ledger can't be opened (no
/// writable path, a held lock, …) it returns [`JobSink::noop`] so tracking never
/// breaks the caller — a GUI/CLI op runs whether or not a ledger is reachable.
///
/// **REUSE (LAW #5).** korp's `jobs.rs` `sink()` and modgunn's `tools.rs` `sink()`
/// were byte-identical: derive the app's ledger path, open its parent, return the
/// redb sink else `noop()`. Both re-derived what jera should hand them; this is
/// that helper. Each app KEEPS its own domain glue (korp's `track`/`spawn_result_job`
/// /`read_entries`, modgunn's `demo_feed_spec`/`run_demo_feed`).
pub fn workspace_sink(workspace: &str) -> JobSink {
let path = workspace_ledger_path(workspace);
let root = path.parent().map(Path::to_path_buf).unwrap_or_else(|| PathBuf::from("."));
match JobStore::open(&root) {
Ok(store) => store.sink(),
Err(_) => JobSink::noop(),
}
}
#[cfg(test)]
mod tests {
use super::*;
fn tmpdir(tag: &str) -> std::path::PathBuf {
std::env::temp_dir().join(format!("nornir-jobs-{tag}-{}", std::process::id()))
}
/// ★ **A ledger one directory away is FOUND and NAMED, never silently
/// replaced with an empty one.**
///
/// The two spellings are both live in this constellation and jera is what
/// they have in common: nornir opens the warehouse ROOT (`<wh>/jobs.redb` —
/// the 1.5 MB ledger on this box), dwarves opens one level DOWN
/// (`<wh>/jobs/jobs.redb`). jera's module docs have always said the root, so
/// the sub-path is a convention a consumer invented; but the reason it could
/// be invented at all is that `open` takes an untyped directory and cannot
/// tell the two apart. Getting it wrong does not fail — it produces a fresh
/// empty ledger a directory from the real one, and empty is exactly what a
/// quiet forge looks like.
///
/// Asserted in BOTH directions, because a rule that only catches the
/// convention you dislike is just a preference:
///
/// 1. legacy ledger at the warehouse root, caller opens `<wh>/jobs`;
/// 2. ledger written one level down, caller opens the warehouse root;
/// 3. holger's `<data>/holger-jobs` shape must NOT be flagged — a directory
/// that merely sits beside a `jobs.redb` is not a mistake, and a warning
/// that fires on a correct layout is a warning people turn off;
/// 4. a store that IS at the path asked for reports nothing, even when a
/// second warehouse holds its own ledger. Two warehouses (`~/.dwarves`
/// beside `~/.nornir`) is settled and correct, not drift.
#[test]
fn a_ledger_one_directory_away_is_named_not_silently_replaced() {
let wh = tmpdir("stray");
std::fs::remove_dir_all(&wh).ok();
// (1) The nornir spelling on disk, the dwarves spelling in the caller.
let legacy = JobStore::open(&wh).unwrap();
let h = legacy.start(kind::BENCH_RUN, "edda", "nordisk", serde_json::Value::Null);
h.finish(serde_json::Value::Null, "");
drop(legacy);
assert!(
store_path(&wh).is_file(),
"fixture did not write the legacy ledger"
);
let down = wh.join("jobs");
assert_eq!(
stray_ledger_near(&down).as_deref(),
Some(store_path(&wh).as_path()),
"opening {} must find the ledger at {}",
down.display(),
store_path(&wh).display()
);
let wrong = JobStore::open(&down).unwrap();
assert!(
wrong.list(&JobSelector::All).unwrap().is_empty(),
"the wrong path really does open an EMPTY store — that is the whole danger"
);
assert_eq!(
wrong.stray_ledger(),
Some(store_path(&wh).as_path()),
"the store opened at the wrong path said NOTHING about the real ledger — \
an empty pane is then indistinguishable from a quiet forge"
);
drop(wrong);
// (2) The mirror image: written down, opened at the root.
let sub = tmpdir("stray-sub");
std::fs::remove_dir_all(&sub).ok();
let inner = JobStore::open(&sub.join("jobs")).unwrap();
let h = inner.start(kind::BENCH_RUN, "edda", "nordisk", serde_json::Value::Null);
h.finish(serde_json::Value::Null, "");
drop(inner);
let at_root = JobStore::open(&sub).unwrap();
assert_eq!(
at_root.stray_ledger(),
Some(store_path(&sub.join("jobs")).as_path()),
"the ledger written one level down was not reported to a caller at the root"
);
drop(at_root);
// (3) holger's shape is CORRECT and must stay quiet.
let data = tmpdir("stray-holger");
std::fs::remove_dir_all(&data).ok();
std::fs::create_dir_all(&data).unwrap();
std::fs::write(store_path(&data), b"not really a db").unwrap();
let holger = data.join("holger-jobs");
assert_eq!(
stray_ledger_near(&holger),
None,
"`<data>/holger-jobs` is holger's settled layout — flagging it would make \
this warning noise on a correct repo"
);
// (4) A store at the path asked for never reports a neighbour.
let ok = JobStore::open(&wh).unwrap();
assert_eq!(
ok.stray_ledger(),
None,
"the ledger IS here; there is nothing to report"
);
drop(ok);
std::fs::remove_dir_all(&wh).ok();
std::fs::remove_dir_all(&sub).ok();
std::fs::remove_dir_all(&data).ok();
}
/// ★ **`store_path` is the one writer of `<warehouse root>/jobs.redb`, and the
/// live 1.5 MB ledger is still at exactly that path.**
///
/// The migration assertion, and it is worth more than the naming: whatever
/// jera says the rule is, a store already on disk at nornir's location must
/// keep being found by it. If this ever reds, a convention change walked away
/// from real data.
#[test]
fn store_path_is_the_rule_and_an_existing_ledger_still_resolves_through_it() {
let wh = tmpdir("storepath");
std::fs::remove_dir_all(&wh).ok();
let store = JobStore::open(&wh).unwrap();
let id = store
.start(kind::BENCH_RUN, "edda", "nordisk", serde_json::Value::Null)
.job_id()
.to_string();
drop(store);
assert_eq!(store_path(&wh), wh.join("jobs.redb"));
assert_eq!(store_path(&wh).file_name().unwrap(), STORE_FILE);
assert!(
store_path(&wh).is_file(),
"open() did not put the file where store_path says"
);
// Re-open through the derived path's parent — the round trip a consumer
// holding only a warehouse root makes — and find the SAME rows.
let root = store_path(&wh).parent().unwrap().to_path_buf();
let again = JobStore::open(&root).unwrap();
assert!(
again
.list(&JobSelector::All)
.unwrap()
.iter()
.any(|j| j.job_id == id),
"the existing ledger was not found through store_path's root"
);
drop(again);
std::fs::remove_dir_all(&wh).ok();
}
/// The ledger has to say WHO asked, and it has to be honest when nobody
/// stamped it. Three separate things are asserted, because two of them can
/// pass while the third is broken:
///
/// 1. an unstamped job reads back `None` / `"unknown"` — it must NOT invent
/// an initiator, which is the failure mode that would make the Jobs panel
/// confidently wrong rather than blank;
/// 2. a stamped job is readable from the store WHILE STILL RUNNING, not only
/// after it ends — a Jobs panel is watched during a release, and a builder
/// that set the field in memory and left the write to `finish` would show
/// `unknown` for the entire life of the job. (Measured: dropping the
/// re-upsert from `by_actor` does NOT fail the after-`finish` assertions,
/// because `finish` upserts the same in-memory record — so the terminal
/// check alone is blind to it. This is the assertion that is not.);
/// 3. `is_local_actor` keys on the METHOD, not on the subject name — a remote
/// caller authenticating as `alice` is still remote.
#[test]
fn actor_is_recorded_survives_finish_and_is_never_invented() {
let dir = tmpdir("actor");
std::fs::remove_dir_all(&dir).ok();
let store = JobStore::open(&dir).unwrap();
// (1) nobody stamped it.
let h = store.start(kind::BENCH_RUN, "facett", "nordisk", serde_json::Value::Null);
let id_anon = h.job_id().to_string();
h.finish(serde_json::Value::Null, "");
// (2) a remote caller over bearer auth.
let h = store
.start(kind::TEST_MATRIX, "holger", "nordisk", serde_json::Value::Null)
.by_actor("ci-runner", "bearer");
let id_remote = h.job_id().to_string();
h.finish(serde_json::Value::Null, "");
// (3) a local CLI run — inspected BEFORE it finishes.
let h = store
.start(kind::DOCS_BOOK_SVG, "nornir", "nordisk", serde_json::Value::Null)
.by_actor("alice", actor_method::LOCAL);
let id_local = h.job_id().to_string();
{
let live = store.list(&JobSelector::All).unwrap();
let running = live
.iter()
.find(|r| r.job_id == id_local)
.expect("a running job is already in the ledger");
assert!(!running.is_terminal(), "still running");
assert_eq!(
running.actor.as_deref(),
Some("alice"),
"the initiator must be visible while the job RUNS — this is what a Jobs \
panel reads during a release, and it is the only assertion here that \
fails if `by_actor` skips its re-upsert"
);
assert!(running.is_local_actor());
}
h.finish(serde_json::Value::Null, "");
let all = store.list(&JobSelector::All).unwrap();
let get = |id: &str| all.iter().find(|r| r.job_id == id).expect("row present");
let anon = get(&id_anon);
assert_eq!(anon.actor, None, "an unstamped job must not acquire an initiator");
assert_eq!(anon.actor_method, None);
assert_eq!(anon.actor_label(), "unknown");
assert!(!anon.is_local_actor(), "unknown is not local");
let remote = get(&id_remote);
assert_eq!(
remote.actor.as_deref(),
Some("ci-runner"),
"the stamp must survive the terminal `finish` upsert, not just the start write"
);
assert_eq!(remote.actor_method.as_deref(), Some("bearer"));
assert_eq!(remote.actor_label(), "ci-runner (bearer)");
assert!(!remote.is_local_actor(), "a bearer caller is REMOTE however it is named");
let local = get(&id_local);
assert_eq!(local.actor.as_deref(), Some("alice"));
assert!(local.is_local_actor());
assert_eq!(local.actor_label(), "alice (local)");
// (4) THE case the method-vs-name distinction exists for: a REMOTE caller
// whose subject is the operator's own name. Without this row the suite is
// blind — measured: an `is_local_actor` implemented as
// `matches!(self.actor, Some("alice") | Some("dwarves"))` passes every
// other assertion here, because no other row pairs a "local-looking" name
// with a remote method. Someone holding a bearer token minted for the
// operator must NOT read back as a local run.
let h = store
.start(kind::BENCH_RUN, "korp", "nordisk", serde_json::Value::Null)
.by_actor("alice", "bearer");
let id_impostor = h.job_id().to_string();
h.finish(serde_json::Value::Null, "");
let all = store.list(&JobSelector::All).unwrap();
let impostor = all.iter().find(|r| r.job_id == id_impostor).expect("row present");
assert_eq!(impostor.actor.as_deref(), Some("alice"));
assert!(
!impostor.is_local_actor(),
"same subject as the local run, but authenticated over bearer — the METHOD \
decides, never the name"
);
assert_eq!(impostor.actor_label(), "alice (bearer)");
// The three rows must not agree with each other — a store that returned one
// shared record would satisfy every assertion above individually.
assert_ne!(anon.actor_label(), remote.actor_label());
assert_ne!(remote.actor_label(), local.actor_label());
std::fs::remove_dir_all(&dir).ok();
}
#[test]
fn start_finish_fail_and_drop_zombie_guard() {
let dir = tmpdir("life");
std::fs::remove_dir_all(&dir).ok();
let store = JobStore::open(&dir).unwrap();
// 1) finish → done, with a result_ref.
let h = store.start(kind::BENCH_RUN, "facett", "nordisk", serde_json::json!({"n": 1}));
let id_done = h.job_id().to_string();
h.finish(serde_json::json!({"rows": 12}), "bench_runs:abc");
// 2) fail → failed, with the error chain.
let h = store.start(kind::TEST_MATRIX, "holger", "nordisk", serde_json::Value::Null);
h.fail(&anyhow::anyhow!("boom"));
// 3) dropped without terminal → failed (zombie guard).
let id_drop = {
let h = store.start(kind::DOCS_BOOK_SVG, "nornir", "nordisk", serde_json::Value::Null);
let id = h.job_id().to_string();
drop(h);
id
};
let all = store.list(&JobSelector::All).unwrap();
assert_eq!(all.len(), 3, "three jobs recorded");
// Newest-first by start; statuses all terminal.
assert!(all.iter().all(|r| r.is_terminal()), "no zombie running rows");
let done = all.iter().find(|r| r.job_id == id_done).unwrap();
assert_eq!(done.status, status::DONE);
assert_eq!(done.result_ref, "bench_runs:abc");
assert!(done.elapsed_ms.is_some(), "finished job has elapsed");
assert!(done.ts_end_micros.is_some());
let dropped = all.iter().find(|r| r.job_id == id_drop).unwrap();
assert_eq!(dropped.status, status::FAILED, "dropped handle records failed");
assert!(dropped.detail_json.contains("dropped"), "drop reason carried");
// Selector scoping.
assert_eq!(store.list(&JobSelector::Kind(kind::BENCH_RUN.into())).unwrap().len(), 1);
assert_eq!(store.list(&JobSelector::Workspace("nordisk".into())).unwrap().len(), 3);
assert_eq!(store.list(&JobSelector::Workspace("other".into())).unwrap().len(), 0);
std::fs::remove_dir_all(&dir).ok();
}
#[test]
fn running_then_done_overwrites_in_place_no_duplicate() {
let dir = tmpdir("overwrite");
std::fs::remove_dir_all(&dir).ok();
let store = JobStore::open(&dir).unwrap();
let h = store.start(kind::RELEASE_RUN, "ws", "ws", serde_json::Value::Null);
let id = h.job_id().to_string();
// While running, exactly one row, status running.
let running = store.list(&JobSelector::All).unwrap();
assert_eq!(running.len(), 1);
assert_eq!(running[0].status, status::RUNNING);
assert!(running[0].elapsed_ms.is_none());
h.finish(serde_json::Value::Null, "release_events:r1");
// Still one row (same job_id key), now done.
let done = store.list(&JobSelector::All).unwrap();
assert_eq!(done.len(), 1, "finish overwrote the running row, not appended");
assert_eq!(done[0].job_id, id);
assert_eq!(done[0].status, status::DONE);
assert_eq!(done[0].result_ref, "release_events:r1");
std::fs::remove_dir_all(&dir).ok();
}
#[test]
fn prune_drops_aged_terminal_and_caps_count_but_keeps_running() {
let dir = tmpdir("prune");
std::fs::remove_dir_all(&dir).ok();
let store = JobStore::open(&dir).unwrap();
// Insert directly via redb_upsert (NOT submit, which auto-prunes on
// terminal writes) so we exercise `prune` itself.
// An aged-out terminal job (ended well before the cutoff).
let old_end = now_micros() - (KEEP_DAYS + 5) * 86_400 * 1_000_000;
redb_upsert(
&store.db,
&JobRecord {
job_id: "old".into(),
kind: kind::DOCS_BOOK.into(),
target: "t".into(),
workspace: "w".into(),
status: status::DONE.into(),
ts_start_micros: old_end - 1000,
ts_end_micros: Some(old_end),
elapsed_ms: Some(1),
detail_json: String::new(),
result_ref: String::new(),
parent_id: None,
node: None,
phase: None,
actor: None,
actor_method: None,
},
)
.unwrap();
// A fresh running job (must survive pruning).
redb_upsert(
&store.db,
&JobRecord {
job_id: "live".into(),
kind: kind::BENCH_RUN.into(),
target: "t".into(),
workspace: "w".into(),
status: status::RUNNING.into(),
ts_start_micros: now_micros(),
ts_end_micros: None,
elapsed_ms: None,
detail_json: String::new(),
result_ref: String::new(),
parent_id: None,
node: None,
phase: None,
actor: None,
actor_method: None,
},
)
.unwrap();
let removed = store.prune(KEEP_DAYS, CAP).unwrap();
assert_eq!(removed, 1, "only the aged terminal job is pruned");
let left = store.list(&JobSelector::All).unwrap();
assert_eq!(left.len(), 1);
assert_eq!(left[0].job_id, "live", "the running job survives retention");
std::fs::remove_dir_all(&dir).ok();
}
#[test]
fn classify_bench_exclusive_others_parallel() {
assert!(is_exclusive(kind::BENCH_RUN), "bench is the heavy/serial kind");
assert_eq!(class_of(kind::BENCH_RUN), JobClass::Exclusive);
// Everything else stays parallel (the default) for now.
for k in [
kind::RELEASE_RUN,
kind::TEST_MATRIX,
kind::ARCH_GENERATE,
kind::COVERAGE,
kind::CONTAINER,
kind::DOCS_BOOK,
kind::WORKSPACE_FETCH,
] {
assert!(!is_exclusive(k), "{k} should be parallel");
assert_eq!(class_of(k), JobClass::Parallel, "{k}");
}
// The daemon kind is its own class — never-ending, ungated (not exclusive).
assert!(is_daemon(kind::DAEMON), "the daemon kind classifies as a daemon");
assert_eq!(class_of(kind::DAEMON), JobClass::Daemon);
assert!(!is_exclusive(kind::DAEMON), "a daemon runs immediately, ungated");
assert!(!is_daemon(kind::BENCH_RUN), "a bench is not a daemon");
}
/// **The daemon-job lifecycle** (§5 of the collectors-daemon design). A daemon:
/// 1. starts `running` and is daemon-classed (never fire-and-complete);
/// 2. surfaces a live state-machine **phase** + progress% + `healthy` flag
/// through the [`report`] side channel (what the Jobs view renders, not the
/// lifecycle row);
/// 3. **never counts as failed for not completing** — dropping its handle
/// records `stopped` (a pure fact), NOT `failed`;
/// 4. is explicitly stoppable via [`JobHandle::stop`] (also `stopped`).
/// A plain (non-daemon) job dropped the same way still records `failed` (the
/// zombie guard) — the two behaviours are contrasted here.
#[test]
fn daemon_job_reports_phase_and_stops_without_ever_failing() {
let dir = tmpdir("daemon-life");
std::fs::remove_dir_all(&dir).ok();
let store = JobStore::open(&dir).unwrap();
// 1) Start a daemon — running, daemon-classed, not terminal.
let h = store.start_daemon("osm-tiles", "korp", serde_json::json!({ "enabled": true }));
let id = h.job_id().to_string();
let rec = || store.list(&JobSelector::All).unwrap().into_iter().find(|r| r.job_id == id).unwrap();
assert_eq!(rec().status, status::RUNNING, "a daemon starts running");
assert_eq!(rec().kind, kind::DAEMON);
assert!(is_daemon(&rec().kind));
assert!(!rec().is_terminal(), "a daemon is not terminal while it runs");
// 2) It surfaces a live PHASE + progress% + healthy via the report channel —
// the Jobs view shows this, not just the lifecycle row.
let reporter = h.reporter();
reporter.phase("downloading", false);
reporter.progress(12, Some(100));
let reports = store.job_reports(&id).expect("daemon reported");
assert_eq!(reports.phase_name(), Some("downloading"), "live phase surfaced");
assert_eq!(reports.phase_healthy(), Some(false), "not healthy while downloading");
assert_eq!(reports.progress_fraction(), Some(0.12), "phase % surfaced alongside");
reporter.phase("online", true);
assert_eq!(store.job_reports(&id).unwrap().phase_healthy(), Some(true), "healthy at online");
// 3) Dropping the handle WITHOUT finish/fail ⇒ STOPPED, never FAILED.
drop(h);
let r = rec();
assert_eq!(r.status, status::STOPPED, "a dropped daemon STOPS, it does not FAIL");
assert!(!status::is_failure(&r.status), "STOPPED is not a failure");
assert!(status::is_terminal(&r.status), "STOPPED is terminal");
assert!(r.detail_json.contains("daemon stopped"), "the stop reason is recorded: {}", r.detail_json);
// 4) An explicitly stopped daemon is likewise `stopped`, carrying its reason.
let h2 = store.start_daemon("osm-tiles", "korp", serde_json::Value::Null);
let id2 = h2.job_id().to_string();
h2.stop(serde_json::json!({ "reason": "toggled off in Settings" }));
let r2 = store.list(&JobSelector::All).unwrap().into_iter().find(|r| r.job_id == id2).unwrap();
assert_eq!(r2.status, status::STOPPED);
assert!(r2.detail_json.contains("toggled off"), "explicit stop reason carried: {}", r2.detail_json);
// Contrast: a NON-daemon job dropped without a terminal still FAILS (guard).
let non_daemon_id = {
let h = store.start(kind::DOCS_BOOK, "nornir", "korp", serde_json::Value::Null);
let id = h.job_id().to_string();
drop(h);
id
};
let nd = store.list(&JobSelector::All).unwrap().into_iter().find(|r| r.job_id == non_daemon_id).unwrap();
assert_eq!(nd.status, status::FAILED, "a dropped fire-and-complete job still fails (zombie guard)");
std::fs::remove_dir_all(&dir).ok();
}
/// **Restart-safe resume** — the daemon survives a korp-server restart as ONE
/// ledger row. Simulate a process that died mid-run (a leftover `running` daemon
/// row, no clean stop), then [`JobStore::start_or_resume_daemon`] RE-ATTACHES to
/// that same `job_id` (uptime/history intact) instead of appending a fresh row.
#[test]
fn daemon_resumes_the_same_ledger_row_across_a_restart() {
let dir = tmpdir("daemon-resume");
std::fs::remove_dir_all(&dir).ok();
let store = JobStore::open(&dir).unwrap();
// A leftover `running` daemon row from a "previous process" (crash: no stop).
let start_micros = now_micros() - 5_000_000; // 5s ago
redb_upsert(
&store.db,
&JobRecord {
job_id: "prev-daemon".into(),
kind: kind::DAEMON.into(),
target: "osm-tiles".into(),
workspace: "korp".into(),
status: status::RUNNING.into(),
ts_start_micros: start_micros,
ts_end_micros: None,
elapsed_ms: None,
detail_json: String::new(),
result_ref: String::new(),
parent_id: None,
node: None,
phase: None,
actor: None,
actor_method: None,
},
)
.unwrap();
// Restart: resume rather than start-fresh — same row, same start time.
let h = store.start_or_resume_daemon("osm-tiles", "korp", serde_json::json!({ "resumed": true }));
assert_eq!(h.job_id(), "prev-daemon", "resumed the leftover daemon row, not a new one");
let rows = store.list(&JobSelector::Kind(kind::DAEMON.into())).unwrap();
assert_eq!(rows.len(), 1, "still ONE daemon row across the restart");
assert_eq!(rows[0].ts_start_micros, start_micros, "original start (uptime) preserved");
assert_eq!(rows[0].status, status::RUNNING, "resumed running");
// A DIFFERENT target starts its own row (resume is keyed on target+workspace).
let h2 = store.start_or_resume_daemon("dcat", "korp", serde_json::Value::Null);
assert_ne!(h2.job_id(), "prev-daemon", "a different collector gets its own daemon row");
assert_eq!(store.list(&JobSelector::Kind(kind::DAEMON.into())).unwrap().len(), 2);
// Do not let the drop guard stop these before we assert — stop cleanly.
h.stop(serde_json::Value::Null);
h2.stop(serde_json::Value::Null);
std::fs::remove_dir_all(&dir).ok();
}
/// **The adversarial crack, closed.** A daemon's live phase must survive being
/// read from a DIFFERENT store instance than the one that emitted it — the
/// out-of-process / `open_read_only` viewer case, where the in-memory
/// `ReportRegistry` starts EMPTY. We emit a phase through the writer store's
/// [`Reporter`], then open a FRESH read-only store on the same file and prove:
/// (a) its in-memory `job_reports` is empty (the crack is real — reports don't
/// travel), yet (b) the phase IS readable off the durable `JobRecord.phase` row.
#[test]
fn daemon_phase_is_durable_across_a_fresh_read_only_store() {
let dir = tmpdir("daemon-phase-durable");
std::fs::remove_dir_all(&dir).ok();
let writer = JobStore::open(&dir).unwrap();
// A daemon emits a phase (+ a progress tick) through its wired reporter.
let h = writer.start_daemon("osm-tiles", "korp", serde_json::json!({ "enabled": true }));
let id = h.job_id().to_string();
let reporter = h.reporter();
reporter.progress(30, Some(100)); // in-memory only (no redb write)
reporter.phase("cutting", true); // persists the phase (+ progress snapshot) to redb
// In-process, both the fast path and the durable row agree.
assert_eq!(writer.job_reports(&id).unwrap().phase_name(), Some("cutting"));
let row = writer.list(&JobSelector::All).unwrap().into_iter().find(|r| r.job_id == id).unwrap();
let dp = row.daemon_phase().expect("phase persisted onto the durable row");
assert_eq!(dp.name, "cutting");
assert!(dp.healthy);
assert_eq!(dp.fraction(), Some(0.30), "progress snapshot captured at phase() time");
// Now the out-of-process reader. The writer still holds the redb lock, so
// `open_read_only` snapshots the file aside into a FRESH store whose in-memory
// registry is empty — exactly what a viz local-mode feed / RPC viewer gets.
let reader = JobStore::open_read_only(&dir).unwrap();
assert!(
reader.job_reports(&id).is_none(),
"the in-memory report registry does NOT travel to a fresh store (the crack)"
);
let seen = reader
.list(&JobSelector::All)
.unwrap()
.into_iter()
.find(|r| r.job_id == id)
.expect("the daemon row is in the snapshot");
let dp = seen.daemon_phase().expect("phase durable+readable out-of-process");
assert_eq!(dp.name, "cutting", "the read-only viewer sees the live phase off the row");
assert!(dp.healthy);
assert_eq!(dp.fraction(), Some(0.30), "and its progress snapshot");
// A later phase transition overwrites the durable mirror in place, still ONE row.
reporter.phase("online", true);
let reader2 = JobStore::open_read_only(&dir).unwrap();
let seen2 = reader2.list(&JobSelector::All).unwrap().into_iter().find(|r| r.job_id == id).unwrap();
assert_eq!(seen2.daemon_phase().unwrap().name, "online", "latest phase is the durable one");
h.stop(serde_json::Value::Null);
std::fs::remove_dir_all(&dir).ok();
}
/// **Dedup on resume.** MULTIPLE leftover non-terminal daemon rows for one
/// `(target, workspace)` — a crash-restart storm — must collapse to ONE running
/// row: [`JobStore::start_or_resume_daemon`] resumes the OLDEST (truest uptime)
/// and marks the others STOPPED (a fact, not a failure), never orphaning them.
/// **A restart closes the rows the dead process left open — and ONLY the
/// kinds it is told are its own.** Asserted on the ledger read back: the
/// orphaned `running` lens and republish rows are `stopped` (never `failed`)
/// carrying the reason and what they were; the `queued` one too; a `bench_run`
/// (a thin client's, possibly still running on its laptop) is untouched; and a
/// row that was already terminal is not rewritten.
///
/// RED WHEN REVERTED: with no reaping, `list` still shows the three as
/// non-terminal and the count is 0.
#[test]
fn stop_orphans_closes_only_the_named_kinds_and_says_why() {
let dir = tmpdir("stop-orphans");
std::fs::remove_dir_all(&dir).ok();
let store = JobStore::open(&dir).unwrap();
let base = now_micros();
let mk = |job_id: &str, kind: &str, status: &str| JobRecord {
job_id: job_id.into(),
kind: kind.into(),
target: job_id.into(),
workspace: "nordisk".into(),
status: status.into(),
ts_start_micros: base - 60_000_000,
ts_end_micros: None,
elapsed_ms: None,
detail_json: r#"{"why":"changed [dwarves]"}"#.into(),
result_ref: String::new(),
parent_id: None,
node: None,
phase: None,
actor: None,
actor_method: None,
};
redb_upsert(&store.db, &mk("lens-running", kind::LENS, status::RUNNING)).unwrap();
redb_upsert(&store.db, &mk("feed-queued", kind::FEED, status::QUEUED)).unwrap();
redb_upsert(&store.db, &mk("republish-running", kind::WORKSPACE_REPUBLISH, status::RUNNING)).unwrap();
redb_upsert(&store.db, &mk("bench-elsewhere", kind::BENCH_RUN, status::RUNNING)).unwrap();
redb_upsert(&store.db, &mk("lens-done", kind::LENS, status::DONE)).unwrap();
let n = store.stop_orphans(
&[kind::LENS, kind::FEED, kind::WORKSPACE_REPUBLISH],
"the process that opened this row exited before closing it (server restart)",
);
assert_eq!(n, 3, "exactly the three non-terminal rows of the named kinds");
let by_id = |id: &str| {
store
.list(&JobSelector::All)
.unwrap()
.into_iter()
.find(|r| r.job_id == id)
.unwrap()
};
for id in ["lens-running", "feed-queued", "republish-running"] {
let r = by_id(id);
assert_eq!(r.status, status::STOPPED, "{id} must be STOPPED, not failed: {}", r.status);
assert!(r.ts_end_micros.is_some() && r.elapsed_ms.is_some(), "{id} is closed");
let d: serde_json::Value = serde_json::from_str(&r.detail_json).unwrap();
assert!(d["orphaned"].as_str().unwrap().contains("server restart"), "{id}: {d}");
assert!(d["was"] == status::RUNNING || d["was"] == status::QUEUED, "{id}: {d}");
assert_eq!(d["detail"]["why"], "changed [dwarves]", "{id} keeps what it was doing");
}
assert_eq!(by_id("bench-elsewhere").status, status::RUNNING, "not the server's to close");
assert_eq!(by_id("lens-done").status, status::DONE);
assert!(by_id("lens-done").ts_end_micros.is_none(), "a terminal row is not rewritten");
// Idempotent: a second call finds nothing.
assert_eq!(store.stop_orphans(&[kind::LENS], "again"), 0);
}
#[test]
fn start_or_resume_daemon_dedups_multiple_leftover_rows() {
let dir = tmpdir("daemon-dedup");
std::fs::remove_dir_all(&dir).ok();
let store = JobStore::open(&dir).unwrap();
// Three leftover `running` daemon rows for the SAME collector, staggered starts.
let base = now_micros();
let mk = |job_id: &str, start: i64| JobRecord {
job_id: job_id.into(),
kind: kind::DAEMON.into(),
target: "osm-tiles".into(),
workspace: "korp".into(),
status: status::RUNNING.into(),
ts_start_micros: start,
ts_end_micros: None,
elapsed_ms: None,
detail_json: String::new(),
result_ref: String::new(),
parent_id: None,
node: None,
phase: None,
actor: None,
actor_method: None,
};
redb_upsert(&store.db, &mk("oldest", base - 9_000_000)).unwrap();
redb_upsert(&store.db, &mk("middle", base - 5_000_000)).unwrap();
redb_upsert(&store.db, &mk("newest", base - 1_000_000)).unwrap();
// Resume: the OLDEST is kept running, the other two are STOPPED (not failed).
let h = store.start_or_resume_daemon("osm-tiles", "korp", serde_json::Value::Null);
assert_eq!(h.job_id(), "oldest", "the oldest (truest uptime) row is resumed");
let rows = store.list(&JobSelector::Kind(kind::DAEMON.into())).unwrap();
assert_eq!(rows.len(), 3, "no row is deleted — dedup STOPS the extras, keeps them as facts");
let running: Vec<_> = rows.iter().filter(|r| !r.is_terminal()).collect();
assert_eq!(running.len(), 1, "exactly ONE running row survives the dedup");
assert_eq!(running[0].job_id, "oldest");
for id in ["middle", "newest"] {
let r = rows.iter().find(|r| r.job_id == id).unwrap();
assert_eq!(r.status, status::STOPPED, "{id} is STOPPED by dedup");
assert!(!status::is_failure(&r.status), "{id} deduped is NOT a failure");
assert!(r.detail_json.contains("deduped"), "{id} carries the dedup reason: {}", r.detail_json);
}
h.stop(serde_json::Value::Null);
std::fs::remove_dir_all(&dir).ok();
}
/// The scheduler's headline guarantee: two `BENCH_RUN` jobs **serialize**. The
/// second is admitted `queued` while the first is `running`, and only flips to
/// `running` once the first reaches a terminal state. We drive it explicitly:
/// hold the first handle (so it never releases the gate), assert the second is
/// stuck `queued`, then finish the first and observe the second run.
/// The exclusive gate is process-global, so the two scheduler tests below
/// must not overlap (each fully drives the single permit through acquire →
/// release). This std `Mutex` serializes them within the test binary.
static SCHED_TEST_LOCK: std::sync::Mutex<()> = std::sync::Mutex::new(());
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn exclusive_bench_jobs_serialize_second_observes_queued() {
let _guard = SCHED_TEST_LOCK.lock().unwrap_or_else(|e| e.into_inner());
let dir = tmpdir("sched-serialize");
std::fs::remove_dir_all(&dir).ok();
let store = Arc::new(JobStore::open(&dir).unwrap());
// First bench acquires the gate and is `running`.
let first = store
.start_scheduled(kind::BENCH_RUN, "alpha", "ws", serde_json::Value::Null)
.await;
let first_id = first.job_id().to_string();
assert_eq!(
store
.list(&JobSelector::Kind(kind::BENCH_RUN.into()))
.unwrap()
.iter()
.find(|r| r.job_id == first_id)
.unwrap()
.status,
status::RUNNING,
"first bench runs immediately"
);
// Submit a second bench on a task — it must block on the gate, sitting
// `queued` (NOT running, NOT invisible) while `first` holds the permit.
let store2 = store.clone();
let second = tokio::spawn(async move {
let h = store2
.start_scheduled(kind::BENCH_RUN, "beta", "ws", serde_json::Value::Null)
.await;
let id = h.job_id().to_string();
h.finish(serde_json::Value::Null, "");
id
});
// Give the spawned task time to write its `queued` row and block.
let second_id = poll_until(&store, |recs| {
recs.iter().find(|r| r.target == "beta").map(|r| {
assert_eq!(r.status, status::QUEUED, "second bench is queued behind the first");
r.job_id.clone()
})
})
.await;
// The second is definitely NOT running while the first holds the gate.
let recs = store.list(&JobSelector::Kind(kind::BENCH_RUN.into())).unwrap();
let running: Vec<_> = recs.iter().filter(|r| r.status == status::RUNNING).collect();
assert_eq!(running.len(), 1, "exactly one bench runs at a time");
assert_eq!(running[0].job_id, first_id);
// Release the gate: finish the first. The second must now run and finish.
first.finish(serde_json::Value::Null, "");
let got_id = second.await.unwrap();
assert_eq!(got_id, second_id, "the queued bench is the one that ran");
let final_recs = store.list(&JobSelector::Kind(kind::BENCH_RUN.into())).unwrap();
for r in &final_recs {
assert_eq!(r.status, status::DONE, "both benches finished: {r:?}");
}
std::fs::remove_dir_all(&dir).ok();
}
/// A non-exclusive (parallel) job is **not** blocked by a running exclusive
/// job: with a bench holding the gate, a `release_run` still starts `running`
/// immediately (never `queued`).
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn parallel_job_not_blocked_by_running_exclusive() {
let _guard = SCHED_TEST_LOCK.lock().unwrap_or_else(|e| e.into_inner());
let dir = tmpdir("sched-parallel");
std::fs::remove_dir_all(&dir).ok();
let store = JobStore::open(&dir).unwrap();
// Hold an exclusive bench (occupies the single gate permit).
let bench = store
.start_scheduled(kind::BENCH_RUN, "alpha", "ws", serde_json::Value::Null)
.await;
// A parallel job goes straight to `running` despite the bench.
let other = store
.start_scheduled(kind::RELEASE_RUN, "rel", "ws", serde_json::Value::Null)
.await;
let other_rec = store
.list(&JobSelector::Kind(kind::RELEASE_RUN.into()))
.unwrap()
.into_iter()
.next()
.unwrap();
assert_eq!(other_rec.status, status::RUNNING, "parallel job is never gated");
other.finish(serde_json::Value::Null, "");
bench.finish(serde_json::Value::Null, "");
std::fs::remove_dir_all(&dir).ok();
}
/// Poll the ledger until `pick` returns `Some`, briefly yielding between reads.
/// Bounded so a logic bug fails the test instead of hanging it.
async fn poll_until<T>(
store: &JobStore,
pick: impl Fn(&[JobRecord]) -> Option<T>,
) -> T {
for _ in 0..500 {
let recs = store.list(&JobSelector::All).unwrap();
if let Some(v) = pick(&recs) {
return v;
}
tokio::time::sleep(std::time::Duration::from_millis(5)).await;
}
panic!("condition not reached within timeout");
}
#[test]
fn job_record_vec_json_round_trips() {
// The RPC contract (Viz.Jobs / Jobs.Submit) ships JobRecord as JSON.
let recs = vec![JobRecord {
job_id: "j".into(),
kind: kind::ARCH_GENERATE.into(),
target: "skade".into(),
workspace: "nordisk".into(),
status: status::DONE.into(),
ts_start_micros: 1,
ts_end_micros: Some(2),
elapsed_ms: Some(0),
detail_json: "{\"nodes\":310}".into(),
result_ref: "architecture_wiring:xyz".into(),
parent_id: None,
node: None,
phase: None,
actor: None,
actor_method: None,
}];
let s = serde_json::to_string(&recs).unwrap();
let back: Vec<JobRecord> = serde_json::from_str(&s).unwrap();
assert_eq!(recs, back);
}
/// Parent/child hierarchy: a populate-style parent with two streamed children.
/// Each child carries `parent_id`, and the ledger stores+returns all three.
#[test]
fn child_jobs_carry_parent_id_and_persist() {
let dir = tmpdir("hierarchy");
std::fs::remove_dir_all(&dir).ok();
let store = JobStore::open(&dir).unwrap();
// Parent: the umbrella populate job (stays running while children stream).
let parent = store.start(
kind::WORKSPACE_POPULATE,
"nordisk",
"nordisk",
serde_json::json!({ "phase": "populate" }),
);
let pid = parent.job_id().to_string();
// Child 1 streams in and finishes (one member's clone).
let c1 = store.start_child(&pid, kind::WORKSPACE_CLONE, "alpha", "nordisk", serde_json::Value::Null);
let c1_id = c1.job_id().to_string();
c1.finish(serde_json::json!({ "sha": "abc" }), "");
// Child 2 streams in and finishes (second member's clone).
let c2 = store.start_child(&pid, kind::WORKSPACE_CLONE, "beta", "nordisk", serde_json::Value::Null);
c2.finish(serde_json::json!({ "sha": "def" }), "");
// Build child under the same parent.
let build = store.start_child(&pid, kind::WORKSPACE_REPUBLISH, "nordisk", "nordisk", serde_json::Value::Null);
build.finish(serde_json::json!({ "snapshot": "snap-1" }), "");
parent.finish(serde_json::json!({ "members": 2 }), "");
let all = store.list(&JobSelector::Workspace("nordisk".into())).unwrap();
assert_eq!(all.len(), 4, "parent + 2 clone children + 1 build child");
// The parent is top-level.
let p = all.iter().find(|r| r.job_id == pid).unwrap();
assert_eq!(p.parent_id, None, "the populate parent has no parent");
assert_eq!(p.kind, kind::WORKSPACE_POPULATE);
// Every non-parent row is a child of the parent.
let children: Vec<&JobRecord> = all.iter().filter(|r| r.parent_id.as_deref() == Some(pid.as_str())).collect();
assert_eq!(children.len(), 3, "two clones + one build are children of the populate");
assert!(children.iter().any(|r| r.job_id == c1_id && r.kind == kind::WORKSPACE_CLONE));
assert!(children.iter().all(|r| r.is_terminal()), "all children terminal");
std::fs::remove_dir_all(&dir).ok();
}
/// A job dispatched across the fleet records WHICH node ran it, and the node
/// survives the ledger round-trip; a plain (local) job stays `node: None`, and
/// a row written before the field existed decodes as `None` (no migration).
#[test]
fn job_records_its_node_and_persists_it() {
let dir = tmpdir("node");
std::fs::remove_dir_all(&dir).ok();
let store = JobStore::open(&dir).unwrap();
// A distributed job stamped to run on the Odin node.
let h = store
.start(kind::BENCH_RUN, "znippy-zoomies", "nordisk", serde_json::Value::Null)
.on_node("odin");
assert_eq!(h.node(), Some("odin"), "handle exposes its node");
let remote_id = h.job_id().to_string();
h.finish(serde_json::json!({"rows": 3}), "bench_runs:abc");
// A plain local job — no node stamp.
let h2 = store.start(kind::DOCS_RENDER, "edda", "nordisk", serde_json::Value::Null);
let local_id = h2.job_id().to_string();
h2.finish(serde_json::Value::Null, "");
let all = store.list(&JobSelector::All).unwrap();
let remote = all.iter().find(|r| r.job_id == remote_id).unwrap();
// RED-when-broken: the node MUST persist through redb + serde.
assert_eq!(remote.node.as_deref(), Some("odin"), "distributed job records its node");
let local = all.iter().find(|r| r.job_id == local_id).unwrap();
assert_eq!(local.node, None, "a local job carries no node");
// Back-compat: a JSON row written before `node` existed decodes as None.
let legacy = r#"{"job_id":"old","kind":"docs_render","target":"t","workspace":"w",
"status":"done","ts_start_micros":1,"ts_end_micros":2,"elapsed_ms":1,
"detail_json":"","result_ref":""}"#;
let rec: JobRecord = serde_json::from_str(legacy).expect("legacy row decodes");
assert_eq!(rec.node, None, "missing node field ⇒ None (no migration)");
assert_eq!(rec.parent_id, None);
std::fs::remove_dir_all(&dir).ok();
}
/// **NEW (job reporting):** a job started through the store gets a working
/// [`Reporter`] for free (`start` wires it automatically); emitting `Progress`
/// lands in the ledger's report registry and [`JobStore::job_reports`] /
/// [`JobStore::reports_snapshot`] both reflect the current/total exactly as
/// reported — this is what a UI polls for a progress bar.
#[test]
fn job_emits_progress_and_ledger_snapshot_reflects_current_and_total() {
let dir = tmpdir("reports-progress");
std::fs::remove_dir_all(&dir).ok();
let store = JobStore::open(&dir).unwrap();
let handle = store.start(kind::DOCS_EXPORT, "book", "nordisk", serde_json::Value::Null);
let id = handle.job_id().to_string();
let reporter = handle.reporter();
// A job that never reports has no entry yet.
assert!(store.job_reports(&id).is_none(), "no report emitted yet");
reporter.progress(3, Some(10));
let reports = store.job_reports(&id).expect("progress was emitted");
assert_eq!(
reports.by_key.get("progress"),
Some(&JobReport::Progress { current: 3, total: Some(10), note: None })
);
// A second, further-along tick overwrites the latest (not appended).
reporter.progress(8, Some(10));
let reports = store.job_reports(&id).unwrap();
assert_eq!(
reports.by_key.get("progress"),
Some(&JobReport::Progress { current: 8, total: Some(10), note: None }),
"latest progress replaces the prior tick"
);
// The whole-ledger snapshot carries the same job + report.
let snapshot = store.reports_snapshot();
assert_eq!(snapshot.len(), 1);
assert_eq!(snapshot[0].0, id);
assert_eq!(
snapshot[0].1.by_key.get("progress"),
Some(&JobReport::Progress { current: 8, total: Some(10), note: None })
);
// Finishing the job (the durable lifecycle) leaves the report untouched —
// reports and lifecycle status are orthogonal.
handle.finish(serde_json::Value::Null, "");
assert!(store.job_reports(&id).is_some(), "report survives job completion");
std::fs::remove_dir_all(&dir).ok();
}
/// **RED-when-broken: `workspace_sink` yields a durable sink bound to the
/// workspace's env-resolved ledger.** With `<WS>_JOBS` pointing at an isolated
/// dir, a job written through `workspace_sink(ws)` must land in THAT ledger and
/// read back — the seam korp/modgunn's Tools menu stands on. If the env
/// resolution broke (helper ignored `<WS>_JOBS` and used the temp default), the
/// job would not be in this dir and the read would find nothing. Uses a unique
/// workspace so its `<WS>_JOBS` var never collides with a parallel test.
/// (Verified RED by neutralizing — making the helper ignore the env fails this.)
#[test]
fn workspace_sink_writes_a_job_into_the_env_resolved_ledger() {
// Unique per-process workspace ⇒ unique env var, no cross-test collision.
let ws = format!("appshelltest{}", std::process::id());
let env_var = format!("{}_JOBS", ws.to_uppercase());
let dir = tmpdir("workspace-sink");
std::fs::remove_dir_all(&dir).ok();
std::fs::create_dir_all(&dir).unwrap();
let ledger = dir.join(format!("{ws}_jobs.redb"));
// SAFETY: the env var is unique to this test's PID-tagged workspace.
unsafe { std::env::set_var(&env_var, &ledger) };
// The path helper honours the env override (env beats the temp default).
assert_eq!(workspace_ledger_path(&ws), ledger, "env var wins over the temp default");
// A job written through the workspace sink lands in this ledger.
let sink = workspace_sink(&ws);
let h = JobHandle::start(sink, kind::CONTAINER, "demo", &ws, serde_json::json!({"n": 1}));
h.finish(serde_json::Value::Null, "");
// Read it back from the SAME env-resolved root (parent of the ledger path).
let store = JobStore::open_read_only(&dir).unwrap();
let rows = store.list(&JobSelector::Workspace(ws.clone())).unwrap();
let row = rows
.iter()
.find(|r| r.kind == kind::CONTAINER && r.target == "demo")
.expect("the job written through workspace_sink landed in the workspace ledger");
assert_eq!(row.status, status::DONE);
assert_eq!(row.workspace, ws);
unsafe { std::env::remove_var(&env_var) };
std::fs::remove_dir_all(&dir).ok();
}
/// The sink degrades gracefully: an unwritable ledger path yields a `noop` sink
/// (never an error/panic), so an app's Tools menu still runs. A blank env value
/// falls through to the temp default (not "set to empty").
#[test]
fn workspace_sink_noops_on_unwritable_and_blank_env_falls_through() {
let ws = format!("appshellnoop{}", std::process::id());
let env_var = format!("{}_JOBS", ws.to_uppercase());
// A blank env value is treated as unset → temp default path.
unsafe { std::env::set_var(&env_var, " ") };
assert_eq!(
workspace_ledger_path(&ws),
std::env::temp_dir().join(format!("{ws}_jobs.redb")),
"a blank env value falls through to the temp default",
);
// An unwritable path yields a noop sink — the op still runs, nothing panics.
unsafe { std::env::set_var(&env_var, "/proc/nonexistent-appshell/jobs.redb") };
let sink = workspace_sink(&ws);
let h = JobHandle::start(sink, kind::CONTAINER, "demo", &ws, serde_json::Value::Null);
h.finish(serde_json::Value::Null, ""); // does not panic on the noop sink
unsafe { std::env::remove_var(&env_var) };
}
}
#[cfg(test)]
mod receiving_sink_tests {
use super::*;
/// **A sink can receive a record it did not start** — the shape `JobSink::remote`
/// always implied and jera had no counterpart for.
///
/// Asserted on what the sink actually received, not on the call returning: a
/// `record` that dropped the row would pass any "did not panic" check, which is the
/// exact failure this closes on korp's side of the wire.
#[test]
fn a_record_from_elsewhere_reaches_the_sink_verbatim() {
let seen: Arc<std::sync::Mutex<Vec<JobRecord>>> = Arc::new(std::sync::Mutex::new(vec![]));
let w = seen.clone();
let sink = JobSink::remote(move |r: &JobRecord| w.lock().unwrap().push(r.clone()));
// A record that already HAPPENED somewhere else: terminal, with its own clock.
let rec = JobRecord {
job_id: "from-a-thin-client".into(),
kind: "pipeline_run".into(),
target: "knut:sverige@skade".into(),
workspace: "korp".into(),
status: "done".into(),
ts_start_micros: 1_788_000_000_000_000,
ts_end_micros: Some(1_788_000_001_500_000),
elapsed_ms: Some(1_500),
detail_json: r#"{"rows":42}"#.into(),
result_ref: String::new(),
parent_id: None,
node: Some("laptop-14".into()),
phase: None,
actor: None,
actor_method: None,
};
sink.record(&rec);
let got = seen.lock().unwrap();
assert_eq!(got.len(), 1, "the row reached the sink");
// VERBATIM — nothing re-stamped. The row happened elsewhere and the receiver's
// clock is not where it happened, so re-deriving `ts_*` would move the event.
assert_eq!(got[0], rec, "the record is written unchanged");
assert_eq!(got[0].status, "done", "a terminal row stays terminal");
assert_eq!(got[0].ts_start_micros, 1_788_000_000_000_000, "the sender's clock");
assert_eq!(got[0].elapsed_ms, Some(1_500));
}
}