crates/jera/tests/job_phase_is_durable.rs
//! A reported PHASE must reach the `jobs.redb` row, or nobody outside the
//! producing process ever sees progress.
//!
//! # The defect (2026-08-27)
//!
//! `ReportSink` / `Reporter` had ZERO uses in nornir and ZERO in dwarves — the
//! whole progress API was written and never wired, which reads as "just add a
//! producer". It is not, and this test is the part that would have caught the
//! obvious fix being wrong:
//!
//! * `JobStore::report_sink()` is a PROCESS-LOCAL in-memory registry.
//! * Only `JobReport::Phase` was ever mirrored durably, and only through
//! `daemon_report_sink()`.
//! * `JobStore::start` / `start_child` / `start_scheduled` wired
//! `report_sink()`, so a `CI_BUILD` job that reported phases diligently left
//! `JobRecord.phase == None` on disk.
//!
//! So progress wired inside a server process would land in that server's memory,
//! where the UI — which reads the ledger row over `Viz.Jobs`, or opens a
//! read-only snapshot — could never see it. A green "the reporter emitted"
//! assertion would have proved nothing: `JobStore::job_reports` reads the same
//! in-memory map the emit just wrote, which is a state round-trip, not applied
//! output.
//!
//! Every assertion here therefore goes through a SECOND `JobStore` opened
//! read-only on the same path. `open_read_only` builds a fresh, EMPTY report
//! registry, so the in-memory sink is structurally incapable of answering — the
//! only thing that can is the row.
//!
//! RED PROOF: put `report_sink()` back in `JobStore::start` and
//! `phase_is_visible_to_a_reader_that_never_saw_the_emit` fails with
//! `phase: None`, which is exactly what every out-of-process reader saw before.
use jera::jobs::{JobSelector, JobStore, kind, status};
/// A fresh store root. Never `/tmp` in anger, but a test's ledger is a few
/// hundred bytes and dies with the tempdir.
fn store() -> (tempfile::TempDir, JobStore) {
let dir = tempfile::tempdir().expect("tempdir");
let s = JobStore::open(dir.path()).expect("open store");
(dir, s)
}
/// Read one row back through a reader that has NO access to the writer's
/// in-memory registry.
fn row_from_a_cold_reader(root: &std::path::Path, job_id: &str) -> jera::jobs::JobRecord {
let cold = JobStore::open_read_only(root).expect("open_read_only");
assert!(
cold.job_reports(job_id).is_none(),
"the cold reader must have an EMPTY report registry, otherwise this test could be \
answered by the in-memory sink and would prove nothing about the row"
);
cold.list(&JobSelector::Kind(kind::CI_BUILD.to_string()))
.expect("list")
.into_iter()
.find(|r| r.job_id == job_id)
.unwrap_or_else(|| panic!("no row for {job_id}"))
}
/// ★ THE ONE. A non-daemon job reports a phase; a reader that never saw the emit
/// must see it.
#[test]
fn phase_is_visible_to_a_reader_that_never_saw_the_emit() {
let (dir, s) = store();
let job = s.start(kind::CI_BUILD, "ci", "ws", serde_json::json!({}));
let id = job.job_id().to_string();
job.reporter().phase("build", true);
let rec = row_from_a_cold_reader(dir.path(), &id);
let phase = rec.phase.as_ref().unwrap_or_else(|| {
panic!(
"JobRecord.phase is None on disk. The producer reported; the sink kept it in this \
process's memory. This is the whole defect: a UI reading the ledger sees no \
progress no matter how well the job reports."
)
});
assert_eq!(phase.name, "build");
assert!(phase.healthy);
}
/// A phase ADVANCING must overwrite, not accumulate or stick. A test that only
/// checks the first phase cannot tell a live mirror from a one-shot write.
#[test]
fn the_latest_phase_wins_on_the_row() {
let (dir, s) = store();
let job = s.start(kind::CI_BUILD, "ci", "ws", serde_json::json!({}));
let id = job.job_id().to_string();
let r = job.reporter();
for (name, healthy) in [("checkout", true), ("build", true), ("test", false)] {
r.phase(name, healthy);
let rec = row_from_a_cold_reader(dir.path(), &id);
let p = rec.phase.expect("a phase must be on the row");
assert_eq!(p.name, name, "the row must carry the LATEST phase");
assert_eq!(p.healthy, healthy);
}
}
/// The progress snapshot rides along with the phase — the fraction the meter
/// needs — while the progress ticks themselves stay in memory.
#[test]
fn a_progress_snapshot_rides_the_phase_but_ticks_alone_do_not_write() {
let (dir, s) = store();
let job = s.start(kind::CI_BUILD, "ci", "ws", serde_json::json!({}));
let id = job.job_id().to_string();
let r = job.reporter();
// A tick with NO phase must not create a phase on the row: the durable
// mirror is Phase-only, and that is deliberate (no per-tick write
// amplification).
r.progress(1, Some(4));
assert!(
row_from_a_cold_reader(dir.path(), &id).phase.is_none(),
"a progress tick must not write a phase — only Phase reports persist"
);
// Once a phase lands, it carries the current progress snapshot with it.
r.progress(3, Some(4));
r.phase("build", true);
let p = row_from_a_cold_reader(dir.path(), &id).phase.expect("phase on the row");
assert_eq!(p.progress_current, Some(3));
assert_eq!(p.progress_total, Some(4));
assert_eq!(p.fraction(), Some(0.75));
}
/// A child job (the per-member sub-jobs of an umbrella op) must be durable too,
/// or a populate shows a parent bar and no detail.
#[test]
fn a_child_jobs_phase_is_durable_as_well() {
let (dir, s) = store();
let parent = s.start(kind::CI_BUILD, "ci", "ws", serde_json::json!({}));
let child = s.start_child(parent.job_id(), kind::CI_BUILD, "step", "ws", serde_json::json!({}));
let id = child.job_id().to_string();
child.reporter().phase("member 2/7", true);
let rec = row_from_a_cold_reader(dir.path(), &id);
assert_eq!(rec.parent_id.as_deref(), Some(parent.job_id()));
assert_eq!(rec.phase.expect("child phase on the row").name, "member 2/7");
}
/// ⚠ THE RETENTION TRAP, asserted so the next harness does not lose two runs to
/// it. `submit` prunes terminal rows past retention INSIDE the same call, and the
/// phase mirror is a read-modify-write that no-ops on a vanished row. A row
/// stamped in the past is accepted and gone.
#[test]
fn a_back_dated_terminal_row_is_pruned_by_the_submit_that_accepted_it() {
let (dir, s) = store();
let old_micros = {
let now = std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.expect("epoch")
.as_micros() as i64;
now - 90 * 24 * 3_600 * 1_000_000
};
let rec = jera::jobs::JobRecord {
job_id: "back-dated-sentinel".into(),
kind: kind::CI_BUILD.into(),
target: "ci".into(),
workspace: "ws".into(),
status: status::DONE.into(),
ts_start_micros: old_micros,
ts_end_micros: Some(old_micros + 1_000),
elapsed_ms: Some(1),
detail_json: String::new(),
result_ref: String::new(),
parent_id: None,
node: None,
phase: None,
actor: None,
actor_method: None,
};
s.submit(&rec).expect("submit reports Ok");
let cold = JobStore::open_read_only(dir.path()).expect("open_read_only");
let rows = cold.list(&JobSelector::Kind(kind::CI_BUILD.to_string())).expect("list");
assert!(
!rows.iter().any(|r| r.job_id == "back-dated-sentinel"),
"documented behaviour: the row is accepted and pruned in the same call. A harness \
that stamps a fixed past epoch and then asserts a read-back is chasing this, not a \
wiring bug."
);
}
/// The row is what `Viz.Jobs` serialises — the handler is `store.list(&sel)` then
/// `serde_json::to_string(&rows)`, with no field mapping — so asserting that same
/// serialisation carries the phase is asserting the wire payload, without
/// standing a server up.
#[test]
fn the_phase_survives_the_exact_serialisation_viz_jobs_puts_on_the_wire() {
let (dir, s) = store();
let job = s.start(kind::CI_BUILD, "ci", "ws", serde_json::json!({}));
let id = job.job_id().to_string();
job.reporter().progress(2, Some(5));
job.reporter().phase("linking", true);
let cold = JobStore::open_read_only(dir.path()).expect("open_read_only");
let rows = cold.list(&JobSelector::Kind(kind::CI_BUILD.to_string())).expect("list");
let json = serde_json::to_string(&rows).expect("the Viz.Jobs payload");
// Round-trip through the wire shape and read the field back, rather than
// grepping the string — a substring match would pass on a field that no
// client can deserialise.
let back: Vec<jera::jobs::JobRecord> = serde_json::from_str(&json).expect("decode");
let rec = back.into_iter().find(|r| r.job_id == id).expect("the row is on the wire");
let p = rec.phase.expect("phase crossed the wire shape");
assert_eq!(p.name, "linking");
assert_eq!(p.progress_current, Some(2));
assert_eq!(p.progress_total, Some(5));
}
// ── the shape the real producers have: a bare `JobHandle::start` over a sink ──
/// ★ THE PRODUCER SHAPE. `dwarves/crates/dwarves/src/engine.rs` and nornir's
/// `start_job` both do `JobHandle::start(sink.clone(), …)` with only a `&JobSink`
/// in scope — no `JobStore`, so historically no `ReportSink` to pass and no way
/// to pass one. `job.reporter()` there emitted into a noop, which is a producer
/// that looks wired and reports nowhere.
///
/// Since the sink carries its reports, that exact call gets a durable reporter
/// with no change at the call site.
///
/// RED PROOF, and stated as measured rather than as intended: with
/// `JobHandle::start` handing out `ReportSink::noop()` again, this test falls AND
/// so do the five `JobStore::start`-based ones above — because the redundant
/// `.with_reports(...)` those used to chain is gone, deliberately (L5: one place
/// decides where a store's jobs report). The two that stay green are the
/// retention-trap test and the remote/noop-sink test, neither of which depends on
/// the sink's report side. This test still earns its place: it is the only one
/// that exercises the CALL SHAPE the real producers use, so if `JobStore::start`
/// ever grows its own wiring back the others could go green while the producers
/// stayed dark.
#[test]
fn a_bare_job_handle_over_a_store_sink_reports_durably() {
let dir = tempfile::tempdir().expect("tempdir");
let s = JobStore::open(dir.path()).expect("open store");
// Deliberately NOT `s.start(...)`: this is the call the real producers make.
let sink = s.sink();
let job = jera::jobs::JobHandle::start(
sink.clone(),
kind::CI_BUILD,
"pipeline",
"ws",
serde_json::json!({ "steps": 3 }),
);
let id = job.job_id().to_string();
job.reporter().phase("step 2/3 build", true);
let rec = row_from_a_cold_reader(dir.path(), &id);
let p = rec.phase.unwrap_or_else(|| {
panic!(
"a bare JobHandle::start over a store sink reported into nowhere. This is the \
producer shape dwarves' engine.rs and nornir's start_job actually use."
)
});
assert_eq!(p.name, "step 2/3 build");
assert!(p.healthy);
}
/// The other direction, so the change is not a blanket "everything writes":
/// a handle over a sink with no local ledger must NOT pretend to persist. A
/// `JobSink::remote` (thin client) has no row here to mirror onto; asserting this
/// keeps the next person from "fixing" it by writing to a store that isn't there.
#[test]
fn a_handle_over_a_remote_or_noop_sink_persists_nothing() {
let seen = std::sync::Arc::new(std::sync::Mutex::new(Vec::<String>::new()));
let seen2 = seen.clone();
let sink = jera::jobs::JobSink::remote(move |rec: &jera::jobs::JobRecord| {
seen2.lock().unwrap().push(rec.status.clone());
});
let job = jera::jobs::JobHandle::start(sink, kind::CI_BUILD, "t", "ws", serde_json::Value::Null);
// Must not panic, must not write anywhere: the reporter is a real object that
// drops what it is given.
job.reporter().phase("build", true);
job.reporter().progress(1, Some(2));
job.finish(serde_json::Value::Null, "");
let statuses = seen.lock().unwrap().clone();
assert_eq!(statuses.first().map(String::as_str), Some(status::RUNNING));
assert!(
statuses.iter().any(|s| s == status::DONE),
"the record path still works: {statuses:?}"
);
}
/// ★ THE SECOND BUG, and the one that made the first fix look like it worked.
///
/// A row has two writers that know different things: the [`jera::jobs::JobHandle`]
/// lifecycle writes its own record, which carries `phase: None` and always will,
/// and the report mirror read-modify-writes only the phase. So the terminal
/// `finish()` upsert ERASED every phase the mirror had written. Mid-run the phase
/// was there; the instant the job finished it was gone — which is worse than never
/// working, because the mechanism looks alive until somebody reads it after the
/// fact, which is when a UI actually reads it.
///
/// Found by a dwarves test that ran a real three-step pipeline and asked for the
/// phase afterwards, not by reading the code. RED PROOF: drop the preserve limb in
/// `redb_upsert` and this test fails with `phase: None` while
/// `the_phase_advances_while_the_pipeline_is_still_running` (dwarves) still
/// passes — the exact split that hid it.
#[test]
fn a_terminal_write_does_not_erase_the_phase_the_mirror_wrote() {
let (dir, s) = store();
let job = s.start(kind::CI_BUILD, "ci", "ws", serde_json::json!({}));
let id = job.job_id().to_string();
job.reporter().phase("step 3/3", true);
assert_eq!(
row_from_a_cold_reader(dir.path(), &id).phase.expect("mid-run phase").name,
"step 3/3"
);
// The lifecycle write that used to clobber it.
job.finish(serde_json::json!({ "status": "passed" }), "brokkr_runs:r1");
let rec = row_from_a_cold_reader(dir.path(), &id);
assert_eq!(rec.status, status::DONE, "the terminal write must still land");
let p = rec.phase.unwrap_or_else(|| {
panic!(
"the phase was on the row mid-run and is None after finish(): the lifecycle writer \
overwrote the mirror's field. Every reader that polls AFTER a job completes - which \
is most of them - saw a job that never reported."
)
});
assert_eq!(p.name, "step 3/3");
}
/// The preserve limb must not become "a phase can never change". A record that
/// CARRIES a phase still overwrites — otherwise the mirror itself would be unable
/// to advance the phase, and the fix would have replaced a clobber with a freeze.
#[test]
fn a_record_that_carries_a_phase_still_overwrites() {
let (dir, s) = store();
let job = s.start(kind::CI_BUILD, "ci", "ws", serde_json::json!({}));
let id = job.job_id().to_string();
job.reporter().phase("first", true);
job.reporter().phase("second", false);
let p = row_from_a_cold_reader(dir.path(), &id).phase.expect("phase");
assert_eq!(p.name, "second", "the mirror must still be able to ADVANCE the phase");
assert!(!p.healthy);
}